Events and Messaging¶

Vectis has two event systems: the in-process EventBus for synchronous module-to-module communication, and Redpanda (Kafka-compatible) for asynchronous inter-service messaging.

In-Process EventBus¶

Handlers run inline within the current request. Use for audit logging, cache invalidation, or denormalization that must complete before the response.

from vectis.core.events import Event, event_bus

async def on_order_created(event: Event):
    order_id = event.data["order_id"]
    # write audit log, update cache, etc.

event_bus.subscribe("order.created", on_order_created, priority=50)

Handlers are sorted by priority (lower = earlier, default 50). Emit events from services:

await event_bus.emit(Event(
    type="product.created",
    data={"product_id": product.id, "name": product.name},
    source="product_service",
    user_id=current_user.id,
))

A handler can raise EventCancelled to stop later handlers:

from vectis.core.exceptions import EventCancelled

async def validate_order(event: Event):
    if event.data.get("total", 0) > 100_000:
        raise EventCancelled("Order exceeds maximum amount")

event_bus.subscribe("order.pre_create", validate_order, priority=10)

Redpanda (Async Messaging)¶

For inter-service events — ERP sync, analytics, warehouse integration.

Topic Naming + Schema Registry¶

Topics follow vectis.<domain>.<event>.v<N>. Every topic has a matching JSON Schema at backend/vectis/events/schemas/<domain>/<event>_v<N>.json. The schema registry (backend/vectis/events/registry.py) loads schemas at boot and validates every emitted payload:

• Development & test: strict — invalid payloads raise.
• Production: warn-and-publish (logs structured warning) so a bad payload never blocks a transaction; you fix the producer and ship.

The full topic list, schema paths, and consumer ownership lives in vectis/docs/REDPANDA_TOPICS.md.

Topics¶

Registered schemas live under backend/vectis/events/schemas/. Current schema set:

Publishing¶

from vectis.core.events import publish_event

await publish_event(
    topic="vectis.orders.placed.v1",
    event_type="order.placed",
    payload={"order_id": order.id, "grand_total": str(order.grand_total)},
    key=str(order.id),
)

The producer is lazy-initialized. Call shutdown_producer() on app shutdown.

Consuming¶

Register handlers with @on_event and run as a separate process:

from vectis.core.events.consumer import on_event

@on_event("order.created")
async def handle_order_created(event_type: str, payload: dict) -> None:
    # sync to ERP, trigger analytics, etc.
    ...

python -m vectis.core.events.consumer

The standalone consumer's core topic set (CORE_CONSUMED_TOPICS) is vectis.orders, vectis.inventory.stock_changed.v1, and vectis.inventory.restocked.v1. Extension-declared topic subscriptions (event_topics on the manifest) are implemented: derive_consumed_topics() merges the core set with every extension's declared consume/both topics at boot, so the actual subscribed list is core ∪ extension topics. For example, lead.created moved off the legacy vectis.accounts topic to vectis.leads.created.v1, which ext_odoo_leads declares in its manifest event_topics — so the lead topic is consumed only while a lead consumer is installed, not core-hardcoded.

Extension Handlers in the Consumer¶

The standalone consumer runs discover_and_load(run_lifecycle=False) at boot, so extensions' on_activate hooks run there too (without install/upgrade lifecycle or migrations — the consumer fails fast on schema drift instead). Every consumed Redpanda envelope is then bridged onto the in-process event_bus as an Event with source="redpanda" — which means an extension's event_bus.subscribe(...) handlers registered in on_activate fire in the consumer process for consumed topics, not just in the api.

Two event types are excluded from the bridge (DUAL_DISPATCHED_EVENT_TYPES = {"order.created", "order.status_changed"}), for two different reasons:

• order.status_changed is true dual dispatch: the api both emits it on the in-process bus and publishes it to Redpanda in the same flow, so handlers fire in the api and bridging would double-fire them.
• order.created is Redpanda-only: no process emits it on the event_bus at all — the in-process emits on the placement paths are order.placed / order.reissued. It is guarded to avoid double-notifying placement logic already delivered as order.placed. Consequence: an event_bus subscriber to order.created never fires today — subscribe to order.placed for placement logic.

Choosing Between EventBus and Redpanda¶

Fault Topic¶

vectis.workflow.fault.v1 is the dedicated channel for workflow-level faults — anything a Temporal workflow trapped that the operator should triage. The schema requires:

{
  "workflow_id": "string",
  "workflow_type": "string",
  "run_id": "string",
  "occurred_at": "ISO-8601",
  "fault_code": "string (kebab-case)",
  "severity": "warn | error | fatal",
  "context": { "...": "domain payload" },
  "remediation_hint": "string (optional)"
}

Faults are persisted to workflow_faults and surface in the admin Workflow Faults inbox (/workflow-faults). Each row links to the originating workflow run for re-execution. Use this channel for B2B approval timeouts, recurate cascade validation failures, refund execution errors, and recurring-order pauses — anything an operator needs to manually unblock.

Stale-Approval Pricing-Drift Gate¶

When a cart re-enters the workflow after lying idle through a price change, the order-placement gate writes a CartRejectionEvent row tagged pricing_drift (against the cart's approved grand total stored on Cart.cart_approved_grand_total) and emits the corresponding vectis.cart.rejected.v1 event. The buyer is shown the new price and asked to confirm; on confirmation the cart re-enters the cart-approval phase.