Events: Item Catalog¶

Module: item_catalog Last Updated: 2026-05-10 Generated From: analysis.md

Event Routing¶

Events are published through the transactional outbox and dispatched in-process by the OutboxPoller / EventDispatcher to the @on_event handlers registered for each event type — there are no Kafka topics or partitions. The event_type field inside the envelope is the routing key: a subscribing module registers @on_event(ItemCreated) (etc.) in its application/event_handlers.py. Kafka is deferred until a module is extracted to its own service, at which point these events move onto a grinsystem.item_catalog.events topic with no change to handler signatures. See conventions.md § Event Handlers and the outbox dispatcher runbook.

Produced Events¶

ItemCatalog is a source-of-truth module — it produces events and does not consume any.

Event Envelope¶

All events share the same outer envelope structure:

{
  "event_id": "<UUID>",
  "event_type": "<string>",
  "aggregate_id": "<UUID>",
  "aggregate_type": "Item",
  "tenant_id": "<UUID>",
  "timestamp": "<ISO-8601>",
  "version": 1,
  "payload": { ... },
  "metadata": {
    "correlation_id": "<UUID>",
    "causation_id": "<UUID>"
  }
}

Event Schemas¶

ItemCreated¶

Produced by: Item.create() — triggered by UC-001 (CreateItem) and UC-007 (ImportItems, one event per row).

Consumers: - Manufacturing: creates ManufacturingItemExtension; for ingredient and intermediate types, also auto-creates an IngredientGroup for substitution grouping. - Warehouse: creates WarehouseItemExtension; skipped for service type. - CRM: creates CRMItemExtension.

{
  "event_id": "550e8400-e29b-41d4-a716-446655440001",
  "event_type": "ItemCreated",
  "aggregate_id": "22222222-2222-2222-2222-222222222222",
  "aggregate_type": "Item",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "timestamp": "2026-05-10T09:00:00Z",
  "version": 1,
  "payload": {
    "item_id": "22222222-2222-2222-2222-222222222222",
    "code": "MAK-001",
    "name": "Mąka pszenna typ 500",
    "item_type": "ingredient",
    "storage_unit": "kg",
    "shelf_life_days": null,
    "ean": "5901234123457",
    "is_system": false,
    "tags": ["mąka", "pszenna"],
    "group_id": "11111111-1111-1111-1111-111111111111",
    "created_by": "33333333-3333-3333-3333-333333333333",
    "created_at": "2026-05-10T09:00:00Z"
  },
  "metadata": {
    "correlation_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    "causation_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
  }
}

Payload field reference:

System items: also emitted at provisioning time (one event per active tenant per system item). Consumers treat these identically to tenant-created items.

ItemUpdated¶

Produced by: Item.edit() — triggered by UC-002 (EditItem), only when at least one editable field value changed. No-op edits produce no event.

Consumers: - Warehouse: updates cached shelf_life_days in WarehouseItemExtension; if old_ean ≠ new_ean, also updates the EAN scan lookup index (removes old_ean entry, adds new_ean entry). - CRM: refreshes display name.

{
  "event_id": "550e8400-e29b-41d4-a716-446655440002",
  "event_type": "ItemUpdated",
  "aggregate_id": "22222222-2222-2222-2222-222222222222",
  "aggregate_type": "Item",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "timestamp": "2026-05-10T14:30:00Z",
  "version": 1,
  "payload": {
    "item_id": "22222222-2222-2222-2222-222222222222",
    "name": "Mąka pszenna typ 650",
    "description": "Zaktualizowany opis.",
    "tags": ["mąka", "pszenna", "chlebowa"],
    "shelf_life_days": null,
    "old_ean": "5901234123457",
    "new_ean": "5901234123464",
    "updated_by": "44444444-4444-4444-4444-444444444444",
    "updated_at": "2026-05-10T14:30:00Z"
  },
  "metadata": {
    "correlation_id": "cccccccc-cccc-cccc-cccc-cccccccccccc",
    "causation_id": "dddddddd-dddd-dddd-dddd-dddddddddddd"
  }
}

Payload field reference:

EAN change detection: consumers check old_ean ≠ new_ean to determine whether the EAN scan index needs updating. Both are always present in the payload regardless of whether EAN changed.

ItemDeactivated¶

Produced by: Item.deactivate() — triggered by UC-003 (DeactivateItem).

Consumers: - Manufacturing: warns if item appears in a Draft recipe BOM (informational only — deactivation is already complete at this point). - Warehouse: marks item as unavailable for new incoming stock operations. - CRM: sets orderable = false on CRMItemExtension.

{
  "event_id": "550e8400-e29b-41d4-a716-446655440003",
  "event_type": "ItemDeactivated",
  "aggregate_id": "22222222-2222-2222-2222-222222222222",
  "aggregate_type": "Item",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "timestamp": "2026-05-10T15:00:00Z",
  "version": 1,
  "payload": {
    "item_id": "22222222-2222-2222-2222-222222222222",
    "code": "MAK-001",
    "item_type": "ingredient",
    "deactivated_by": "33333333-3333-3333-3333-333333333333",
    "deactivated_at": "2026-05-10T15:00:00Z"
  },
  "metadata": {
    "correlation_id": "eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee",
    "causation_id": "ffffffff-ffff-ffff-ffff-ffffffffffff"
  }
}

Payload field reference:

ItemReactivated¶

Produced by: Item.reactivate() — triggered by UC-004 (ReactivateItem).

Consumers: - CRM: restores orderable flag on CRMItemExtension based on CRM's own rules.

{
  "event_id": "550e8400-e29b-41d4-a716-446655440004",
  "event_type": "ItemReactivated",
  "aggregate_id": "22222222-2222-2222-2222-222222222222",
  "aggregate_type": "Item",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "timestamp": "2026-05-10T16:00:00Z",
  "version": 1,
  "payload": {
    "item_id": "22222222-2222-2222-2222-222222222222",
    "code": "MAK-001",
    "item_type": "ingredient",
    "reactivated_by": "33333333-3333-3333-3333-333333333333",
    "reactivated_at": "2026-05-10T16:00:00Z"
  },
  "metadata": {
    "correlation_id": "gggggggg-gggg-gggg-gggg-gggggggggggg",
    "causation_id": "hhhhhhhh-hhhh-hhhh-hhhh-hhhhhhhhhhhh"
  }
}

Payload field reference:

Consumed Events¶

ItemCatalog registers no @on_event handlers. It is the source of truth for item identity. The only inbound data flows are HTTP requests from human actors and in-process query calls from sibling modules during deactivation checks (see domain-model.md — Cross-Module Query Ports).

Subscribers (downstream modules)¶

These handlers are owned by the subscribing modules, documented here for cross-reference. Each registers @on_event handlers in its application/event_handlers.py.

Idempotency¶

Dispatch is at-least-once, so each handler must be idempotent using event_id:

1. Before processing, check if event_id exists in a local processed_events table.
2. If found: skip (no duplicate processing).
3. If not found: process and record event_id within the same database transaction as the side effect.

This guards against duplicate delivery (a row re-dispatched after a partial failure). See conventions.md § Handler-side idempotency.

Schema Versioning¶

All events carry "version": 1. When the payload schema must change in a breaking way:

• Increment version to 2.
• Maintain version: 1 support in all consumers during the transition window.
• The event_type remains unchanged — handlers dispatch on event_type then branch on version for deserialization.