Strategy Pattern¶

Every business decision point in Vectis is an abstract strategy interface. The core engine calls strategies; extensions and modules provide implementations.

Strategy ABCs¶

PaymentProcessStrategy¶

The payment strategy is the most feature-rich strategy ABC. It defines the full payment lifecycle:

class PaymentProcessStrategy(ABC):
    @property
    @abstractmethod
    def gateway_name(self) -> str: ...

    @abstractmethod
    async def authorize(self, amount, currency, payment_data) -> PaymentResult: ...

    @abstractmethod
    async def capture(self, transaction_id, amount) -> PaymentResult: ...

    @abstractmethod
    async def refund(self, transaction_id, amount) -> PaymentResult: ...

    async def charge(self, amount, currency, payment_data) -> PaymentResult:
        """Auth + capture in one call. Default: authorize then capture."""
        ...

    async def void(self, transaction_id) -> PaymentResult:
        """Void an authorized but uncaptured transaction."""
        ...

    async def approve_held_transaction(self, transaction_id) -> PaymentResult:
        """Release a transaction held by fraud filters."""
        ...

The PaymentResult dataclass includes a held_for_review flag for fraud filter detection:

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str | None = None
    error_message: str | None = None
    gateway_response: dict | None = None
    held_for_review: bool = False

Creating a Strategy¶

Define an ABC, then register implementations with strategy_resolver.

from abc import ABC, abstractmethod
from decimal import Decimal

class TaxCalculationStrategy(ABC):
    @abstractmethod
    async def calculate(self, line_items: list[dict], address: dict) -> Decimal: ...

Register a default at startup:

from vectis.core.strategy import strategy_resolver

class SimpleSalesTax(TaxCalculationStrategy):
    async def calculate(self, line_items, address):
        subtotal = sum(i["price"] * i["qty"] for i in line_items)
        return subtotal * Decimal("0.08")

strategy_resolver.register(TaxCalculationStrategy, SimpleSalesTax(), name="sales_tax")

Single vs Multi Resolution¶

resolve — returns the last-registered implementation (or a named one). Use for single-active-strategy decision points:

price_strategy = strategy_resolver.resolve(PriceCalculationStrategy)
price = await price_strategy.calculate(variant_id, account_id)

resolve_all — returns every implementation. Use when multiple strategies combine results (e.g., tax jurisdictions):

tax_strategies = strategy_resolver.resolve_all(TaxCalculationStrategy)
total_tax = sum([await s.calculate(items, addr) for s in tax_strategies])

Stage-Sorted Engines (Decided #151, #162, #170, #171)¶

resolve_all returns strategies in registration order — that's fine for independent providers (UPS + USPS shipping). For tax and promotions the order matters: sales tax must complete before excise reads the tax-adjusted subtotal; line-item promos must run before order-subtotal promos so allocation distributes correctly.

The engine (not the resolver) handles ordering. Strategies declare ordering through different mechanisms depending on the engine:

Tax engine — stage class attribute (vectis/modules/tax/engine.py)¶

Tax strategies expose stage: int as an attribute on the strategy class (default 100). group_by_stage() buckets strategies, runs each bucket, then quantizes the emitted LineTax rows to currency decimals (HALF_UP) before the next bucket sees the totals via TaxContext.prior_stage_tax_by_line. So:

class ExciseEngineTaxStrategy(TaxCalculationStrategy):
    stage = 200          # runs after sales-tax tier
    name = "excise_engine"
    ...

Promotion engine — PHASE_* constants (vectis/modules/promotion/engine.py)¶

Promotions don't use stage; the engine has fixed phases the rule's phase field maps into:

Within each phase, stack_group makes siblings mutually exclusive (first non-null wins per group; null groups stack additively). The $0 floor (Decided #170) prevents any line from going negative. PHASE_ORDER allocations get distributed across lines after the phase resolves.

strategy_resolver.register() does not accept a stage or phase kwarg — those are concerns of the engine consuming resolve_all(), not the resolver itself.

Worker Readiness Gate (Decided #233)¶

On boot, the worker iterates every payment method enabled in the DB and calls strategy_resolver.resolve(PaymentProcessStrategy, name=method.gateway_code). In strict mode (default), the worker exits non-zero if any enabled method has no registered strategy. This catches the cluster-state bug where you remove an extension package but forget to disable its payment methods.

Set VECTIS_WORKER_STRICT_STRATEGY_CHECK=false (also accepts 0, no, empty string) to downgrade to warnings — useful for emergency reseeds. The gate lives in vectis/core/extension_readiness.py.

Channel-Scoped Filtering¶

Extensions are activated per-channel via ChannelExtension. Pass channel_enabled to filter strategies at resolution time:

from vectis.core.strategy import _load_channel_extensions

enabled = await _load_channel_extensions(channel_id)
strategies = strategy_resolver.resolve_all(TaxCalculationStrategy, channel_enabled=enabled)

Core defaults (no extension_name) always pass the filter.

Overriding a Default¶

Extensions override by registering the same strategy type. resolve() returns the last-registered implementation, so extensions loaded after core win:

class ExciseEngineTax(TaxCalculationStrategy):
    stage = 200  # tax engine runs sales-tier (100) before excise-tier (200)
    ...

strategy_resolver.register(
    TaxCalculationStrategy,
    ExciseEngineTax(),
    name="excise_engine",
    extension_name="ext-excise-engine",
)