feat(bi-export): land the inbound delivery timeline in Bronze (#70)

Adds DeliveryAnnouncedV1, DockSlotAssignedV1, DeliveryArrivedV1 integration events for the InboundDelivery transitions (announce/dock-slot/arrival) that previously never crossed a service boundary as an event.
Publishes each through the transactional outbox from the corresponding handler (AnnounceDeliveryHandler, AssignDockSlotHandler, RegisterArrivalHandler), same pattern as the existing ConfirmReceiptHandler → GoodsReceiptConfirmedV1.
Adds a sink-only BiExport consumer per event (own logistics.bi-export queue, no domain writes), landing Bronze columns for BronzeInboundDelivery.announced_at / dock_code / dock_slot_at / arrived_at (docs/BI_Plan.md §5.3).
This is item #1a ("delivery timeline export") from the domain-docs gap analysis — expected/received qty (ASN accuracy) was intentionally left out of this slice.

Author: mkasperczyk90

docs/warehouse_datafoundry.md

@@ -0,0 +1,154 @@
+# Warehouse ↔ DataFoundry integration contract
+
+> This is the producer-side contract for **DataFoundry** (the Databricks lakehouse project that
+> consumes this data) — what actually lands in S3 today, in what shape, at what path, with what
+> guarantees. For the *why* behind each design choice (cloud-agnostic CDC, Auto Loader vs.
+> alternatives, Silver/Gold design, the metrics catalog) see [`BI_Plan.md`](BI_Plan.md) — this file
+> only states what DataFoundry can rely on right now.
+>
+> Nothing below has been deployed to a live AWS environment yet (`BI_LAKE_BUCKET` has never been set
+> on a real `terraform apply`) — every guarantee here has been verified locally against real
+> Postgres/Debezium/RabbitMQ/S3-compatible (LocalStack) containers, not the real thing. Status per
+> row is called out explicitly; nothing is glossed over as "done" that hasn't actually run.
+
+## 1. The landing zone
+
+One S3 bucket, on/off with a single Terraform variable:
+
+- **Bucket**: `warehouse-bi-lake-<account-id>` (override via `bi_lake_bucket_name`,
+  `infra/terraform/bootstrap/variables.tf`)
+- **Switch**: setting the `BI_LAKE_BUCKET` repo variable turns on *both* producer paths below at
+  once (`docs/deploy.md`) — there's no way to enable one without the other today.
+- **IAM**: a single shared ECS task role can `s3:PutObject` under this bucket's root; nothing else
+  writes to it.
+
+Three independent writers land into the same bucket, under three separate prefixes. DataFoundry's
+Auto Loader jobs should point at prefixes, not the bucket root.
+
+| Prefix | Writer | Format | Trigger |
+|---|---|---|---|
+| `bronze/<source_system>/<event_type>/date=YYYY-MM-DD/<event_id>.parquet` | `S3ParquetBronzeExportWriter` (`Warehouse.ServiceDefaults.BiExport`) — one instance per backend service | Parquet, one row per file | Each service's own published integration events, via a dedicated `<service>.bi-export` RabbitMQ queue |
+| `bronze-cdc/<schema>/<table>/date=YYYY-MM-DD/<lsn>.json` | `Warehouse.CdcBridge` (one shared worker) | Raw Debezium change-event JSON (`before`/`after`/`op`/`source`), byte-for-byte, schemas stripped | Postgres logical replication → Debezium Server → RabbitMQ → CdcBridge, for internal state that never becomes an integration event |
+| `bronze/identity-service/UserSnapshot/date=YYYY-MM-DD/<event_id>.parquet` | `Warehouse.IdentityBridge` (reuses the same Parquet writer) | Parquet, one row per user per poll | Periodic poll of Keycloak's Admin REST API (no CDC/events possible — Keycloak dev-mode has no persistent DB) |
+
+**Idempotency**: every key is deterministic — the event-driven path keys on the Wolverine outbox
+message id, CDC keys on the Postgres LSN (the WAL position of the change). A redelivery from the
+at-least-once outbox, or from Debezium, overwrites the same S3 object rather than producing a
+duplicate row. DataFoundry's ingestion does **not** need its own dedup step for either path.
+
+**Schema-on-read**: every Parquet column (`bronze/...`) is written as a string, including numbers and
+timestamps (ISO-8601, `"O"` format) — typing happens in Silver, not Bronze, by design. The CDC path
+is native JSON with no imposed schema at all; Debezium's own envelope carries the typed `before`/
+`after` row images.
+
+## 2. Common envelope (event-driven path only)
+
+Every `bronze/...` Parquet file carries these columns before the event's own fields:
+
+```
+_source_system : string   // "warehouse-service" | "logistics-service" | "masterdata-service" | "identity-service"
+_event_type    : string   // e.g. "GoodsReceivedV1"
+_event_id      : string   // Wolverine's envelope id — the S3 key's dedup key
+_occurred_at   : string   // ISO-8601 — when the event happened, not when it was exported; partitions by business date
+```
+
+The CDC path has no equivalent envelope of its own — `source.schema`/`source.table`/`source.lsn`/
+`ts_ms` inside Debezium's native JSON serve the same role.
+
+## 3. Event-driven catalog (`bronze/`) — every field, as actually implemented
+
+One row per published integration event that has a BiExport consumer today. Field names here are the
+literal Parquet column names (payload keys), not the aspirational Bronze table shapes in
+`BI_Plan.md` §5 — those two mostly agree, but this table is the wire truth.
+
+### `warehouse-service` (queue `warehousing.bi-export`, exchange `inventory`)
+
+| `_event_type` | Payload columns |
+|---|---|
+| `GoodsReceivedV1` | `delivery_id`, `warehouse_code`, `buffer_location`, `line_count` |
+| `PutAwayCompletedV1` | `delivery_id` |
+| `StockReservedV1` | `order_id`, `fully` |
+| `PicksPlannedV1` | `order_id`, `pick_count`, `picks_json` (nested array as a JSON string — explode in Silver) |
+
+### `logistics-service` (queue `logistics.bi-export`, exchange `logistics`)
+
+| `_event_type` | Payload columns |
+|---|---|
+| `GoodsReceiptConfirmedV1` | `delivery_id`, `warehouse_code`, `line_count`, `lines_json` (nested — explode in Silver) |
+| `DeliveryAnnouncedV1` | `delivery_id`, `warehouse_code`, `supplier_party_id`, `planned_at` |
+| `DockSlotAssignedV1` | `delivery_id`, `warehouse_code`, `dock_code`, `dock_slot_at`, `dock_slot_to` |
+| `DeliveryArrivedV1` | `delivery_id`, `warehouse_code`, `arrived_at` |
+| `OutboundOrderPlacedV1` | `order_id`, `customer_role_id`, `warehouse_code`, `line_count`, `lines_json` (nested) |
+| `OutboundOrderCancelledV1` | `order_id` |
+| `PickingReleasedV1` | `order_id`, `warehouse_code` |
+| `PickConfirmedV1` | `order_id`, `sequence`, `location`, `sku`, `batch_number`, `quantity`, `unit` |
+| `ShipmentDispatchedV1` | `order_id`, `shipment_id`, `warehouse_code`, `carrier_role_id`, `tracking_number` (nullable) |
+
+`DeliveryAnnouncedV1`/`DockSlotAssignedV1`/`DeliveryArrivedV1` together give DataFoundry the full
+inbound-delivery timeline (`announced_at`/`dock_code`/`dock_slot_at`/`arrived_at`) that
+`GoodsReceiptConfirmedV1` alone didn't cover — join all four on `delivery_id` in Silver to build one
+`fact_inbound_delivery` row per delivery. Note: `supplier_party_id`/`customer_role_id`/
+`carrier_role_id` are **opaque `PartyRoleRef` strings today** (dev-seed free-text labels, not a real
+Partner id/code) — resolving them to a Partner dimension needs the party-role crosswalk work, not yet
+built.
+
+### `masterdata-service` (queue `masterdata.bi-export`, exchange `catalog`)
+
+| `_event_type` | Payload columns |
+|---|---|
+| `ProductDefinedV2` | `sku`, `name`, `base_unit`, `requires_cold_chain`, `is_hazardous`, `is_batch_tracked`, `unit_weight_kg`, `unit_volume_m3`, `min_celsius` (nullable), `max_celsius` (nullable) |
+
+### Identity (`bronze/identity-service/`)
+
+| `_event_type` | Payload columns |
+|---|---|
+| `UserSnapshot` | `user_id` (Keycloak `username`, the badge number), `display_name` (nullable), `role` (comma-joined if multiple, nullable), `home_warehouse_code` (nullable), `active` |
+
+## 4. CDC catalog (`bronze-cdc/`) — tables configured, per Debezium source instance
+
+CDC exists for internal state that **never crosses a service boundary as an integration event** —
+adding an event purely to feed BI would couple domain code to an analytics concern it has no other
+reason to know about (see `BI_Plan.md` §11 for the full rationale). One Debezium Server instance per
+source database, each with its own `table.include.list`; DataFoundry gets the raw row image, not an
+intent-carrying fact, so dedup/point-in-time logic belongs entirely in Silver.
+
+| Source DB | Tables | Status |
+|---|---|---|
+| `warehouse` | `inventory.stock_items` | **Verified** — a real Postgres `INSERT` flowed through a real Debezium container to S3 in a local smoke test |
+| `warehouse` | `inventory.stock_movements`, `stock_reservations`, `batches`, `stocktakes`, `stocktake_count_lines`, `allocations`, `handling_units`, `handling_unit_lines` | Configured, same instance/config as the verified table — not individually run |
+| `warehouse` | `topology.warehouses`, `rooms`, `locations`, `docks` | Configured, not run |
+| `logistics` | `inbound_deliveries`, `inbound_delivery_lines`, `outbound_orders`, `outbound_order_lines`, `pick_lists`, `pick_tasks`, `shipments`, `packages` | Configured only — this Debezium instance has never been started, even in the smoke test |
+| `masterdata` | `partners.parties`, `party_roles`, `customer_shipping_addresses`, `catalog.product_types` | Configured only — same, never run |
+
+`catalog.product_unit_conversions` is deliberately **not** in scope — nobody has asked for it.
+
+**Important overlap to know about**: `logistics.inbound_deliveries` is in the CDC list above *and*
+`DeliveryAnnouncedV1`/`DockSlotAssignedV1`/`DeliveryArrivedV1` (§3) cover much of the same columns via
+events instead. That's a deliberate, documented exception (`BI_Plan.md` §11) — those three are named
+business milestones, cheaper for Silver to consume as facts than re-derived from a row diff — not an
+oversight. Don't build a Silver transform that expects the CDC path to be the only source for those
+columns.
+
+## 5. What's deliberately not here yet
+
+- **ASN accuracy** (`expected_qty`/`received_qty`/`discrepancy_qty` per delivery line) — the delivery
+  *timeline* (§3) is covered; the line-level expected-vs-actual comparison from UC-02 is a separate,
+  not-yet-built slice.
+- **Cost/valuation** (`stock_value`, `Money` anywhere) — no source system carries cost data yet; a
+  future payments service is the anticipated origin (`BI_Plan.md` §9).
+- **Delivery/shipment confirmation past `Dispatched`** — out of scope for this domain today; proposed
+  as a separate future `fleet-telemetry-service` (`BI_Plan.md` §10).
+- **Party-role → Partner resolution** — `PartyRoleRef` fields travel as opaque strings; a crosswalk
+  resolving them to real Partner attributes is a real backlog item, blocked on real Partner
+  ids/codes reaching the producing services' seed/prod data.
+- **No Databricks-side objects exist yet** — no workspace, no Auto Loader job, no Bronze/Silver/Gold
+  Delta tables. Everything above describes what the Warehouse side delivers into S3; wiring
+  Databricks to actually read it is DataFoundry's next step, not built from this repo.
+
+## 6. Versioning contract
+
+Every event name is suffixed `V<n>` (e.g. `GoodsReceiptConfirmedV1`). Wire shapes are **additive-only
+and immutable**: a new field ships as a new `V<n+1>` event, never as an edit to an existing one, and
+old and new versions can be in flight at the same time. DataFoundry's Silver mappings should key on
+`_event_type` (the exact versioned name), not assume a field list is stable across versions of "the
+same" event.

src/Contracts/Warehouse.Contracts/Logistics/DeliveryAnnouncedV1.cs

@@ -0,0 +1,16 @@
+namespace Warehouse.Contracts.Logistics;
+
+/// <summary>
+/// Integration event: an inbound delivery (ASN) was announced (UC-01). Published by Logistics
+/// through the transactional outbox right after <c>InboundDelivery.Announce</c>; feeds the delivery
+/// timeline in Bronze (docs/BI_Plan.md §5.3 — <c>BronzeInboundDelivery.announced_at</c>).
+///
+/// <b>Additive-only.</b> This is the V1 wire shape and must never change — a new field ships as
+/// <c>DeliveryAnnouncedV2</c>, never as an edit here. Primitives only.
+/// </summary>
+public sealed record DeliveryAnnouncedV1(
+    Guid DeliveryId,
+    string WarehouseCode,
+    string SupplierRoleId,
+    DateTimeOffset PlannedAt,
+    DateTimeOffset OccurredAt);

src/Contracts/Warehouse.Contracts/Logistics/DeliveryArrivedV1.cs

@@ -0,0 +1,15 @@
+namespace Warehouse.Contracts.Logistics;
+
+/// <summary>
+/// Integration event: the truck checked in at the dock for an announced delivery (UC-02). Published
+/// by Logistics through the transactional outbox right after <c>InboundDelivery.RegisterArrival</c>;
+/// feeds the delivery timeline in Bronze (docs/BI_Plan.md §5.3 —
+/// <c>BronzeInboundDelivery.arrived_at</c>).
+///
+/// <b>Additive-only.</b> This is the V1 wire shape and must never change — a new field ships as
+/// <c>DeliveryArrivedV2</c>, never as an edit here. Primitives only.
+/// </summary>
+public sealed record DeliveryArrivedV1(
+    Guid DeliveryId,
+    string WarehouseCode,
+    DateTimeOffset OccurredAt);

src/Contracts/Warehouse.Contracts/Logistics/DockSlotAssignedV1.cs

@@ -0,0 +1,18 @@
+namespace Warehouse.Contracts.Logistics;
+
+/// <summary>
+/// Integration event: a dock + time window was booked for an announced delivery (UC-01). Published
+/// by Logistics through the transactional outbox right after <c>InboundDelivery.AssignDockSlot</c>;
+/// feeds the delivery timeline in Bronze (docs/BI_Plan.md §5.3 —
+/// <c>BronzeInboundDelivery.dock_code</c>/<c>dock_slot_at</c>).
+///
+/// <b>Additive-only.</b> This is the V1 wire shape and must never change — a new field ships as
+/// <c>DockSlotAssignedV2</c>, never as an edit here. Primitives only.
+/// </summary>
+public sealed record DockSlotAssignedV1(
+    Guid DeliveryId,
+    string WarehouseCode,
+    string DockCode,
+    DateTimeOffset SlotFrom,
+    DateTimeOffset SlotTo,
+    DateTimeOffset OccurredAt);

src/Services/Logistics/Modules/Warehouse.Logistics.Core/Application/BiExport/DeliveryAnnouncedBiExportConsumer.cs

@@ -0,0 +1,34 @@
+using Warehouse.Contracts.Logistics;
+using Warehouse.ServiceDefaults.BiExport;
+using Wolverine;
+
+namespace Warehouse.Logistics.Core.Application.BiExport;
+
+/// <summary>
+/// Exports a copy of every <see cref="DeliveryAnnouncedV1"/> to the Bronze landing zone
+/// (docs/BI_Plan.md §5.3/§11 — feeds <c>BronzeInboundDelivery.announced_at</c>). Same shape as
+/// <c>GoodsReceiptConfirmedBiExportConsumer</c>: a pure side observer on its own dedicated queue, no
+/// domain writes, no outbox.
+/// </summary>
+public sealed class DeliveryAnnouncedBiExportConsumer(IBronzeExportWriter writer)
+{
+    private const string SourceSystem = "logistics-service";
+
+    public Task Handle(DeliveryAnnouncedV1 message, Envelope envelope, CancellationToken cancellationToken)
+    {
+        var bronzeEvent = new BronzeEvent(
+            SourceSystem,
+            EventType: nameof(DeliveryAnnouncedV1),
+            EventId: envelope.Id.ToString(),
+            message.OccurredAt,
+            Payload: new Dictionary<string, string?>
+            {
+                ["delivery_id"] = message.DeliveryId.ToString(),
+                ["warehouse_code"] = message.WarehouseCode,
+                ["supplier_party_id"] = message.SupplierRoleId,
+                ["planned_at"] = message.PlannedAt.ToString("O"),
+            });
+
+        return writer.WriteAsync(bronzeEvent, cancellationToken);
+    }
+}

src/Services/Logistics/Modules/Warehouse.Logistics.Core/Application/BiExport/DeliveryArrivedBiExportConsumer.cs

@@ -0,0 +1,33 @@
+using Warehouse.Contracts.Logistics;
+using Warehouse.ServiceDefaults.BiExport;
+using Wolverine;
+
+namespace Warehouse.Logistics.Core.Application.BiExport;
+
+/// <summary>
+/// Exports a copy of every <see cref="DeliveryArrivedV1"/> to the Bronze landing zone
+/// (docs/BI_Plan.md §5.3/§11 — feeds <c>BronzeInboundDelivery.arrived_at</c>). Same shape as
+/// <c>GoodsReceiptConfirmedBiExportConsumer</c>: a pure side observer on its own dedicated queue, no
+/// domain writes, no outbox.
+/// </summary>
+public sealed class DeliveryArrivedBiExportConsumer(IBronzeExportWriter writer)
+{
+    private const string SourceSystem = "logistics-service";
+
+    public Task Handle(DeliveryArrivedV1 message, Envelope envelope, CancellationToken cancellationToken)
+    {
+        var bronzeEvent = new BronzeEvent(
+            SourceSystem,
+            EventType: nameof(DeliveryArrivedV1),
+            EventId: envelope.Id.ToString(),
+            message.OccurredAt,
+            Payload: new Dictionary<string, string?>
+            {
+                ["delivery_id"] = message.DeliveryId.ToString(),
+                ["warehouse_code"] = message.WarehouseCode,
+                ["arrived_at"] = message.OccurredAt.ToString("O"),
+            });
+
+        return writer.WriteAsync(bronzeEvent, cancellationToken);
+    }
+}

src/Services/Logistics/Modules/Warehouse.Logistics.Core/Application/BiExport/DockSlotAssignedBiExportConsumer.cs

@@ -0,0 +1,35 @@
+using Warehouse.Contracts.Logistics;
+using Warehouse.ServiceDefaults.BiExport;
+using Wolverine;
+
+namespace Warehouse.Logistics.Core.Application.BiExport;
+
+/// <summary>
+/// Exports a copy of every <see cref="DockSlotAssignedV1"/> to the Bronze landing zone
+/// (docs/BI_Plan.md §5.3/§11 — feeds <c>BronzeInboundDelivery.dock_code</c>/<c>dock_slot_at</c>).
+/// Same shape as <c>GoodsReceiptConfirmedBiExportConsumer</c>: a pure side observer on its own
+/// dedicated queue, no domain writes, no outbox.
+/// </summary>
+public sealed class DockSlotAssignedBiExportConsumer(IBronzeExportWriter writer)
+{
+    private const string SourceSystem = "logistics-service";
+
+    public Task Handle(DockSlotAssignedV1 message, Envelope envelope, CancellationToken cancellationToken)
+    {
+        var bronzeEvent = new BronzeEvent(
+            SourceSystem,
+            EventType: nameof(DockSlotAssignedV1),
+            EventId: envelope.Id.ToString(),
+            message.OccurredAt,
+            Payload: new Dictionary<string, string?>
+            {
+                ["delivery_id"] = message.DeliveryId.ToString(),
+                ["warehouse_code"] = message.WarehouseCode,
+                ["dock_code"] = message.DockCode,
+                ["dock_slot_at"] = message.SlotFrom.ToString("O"),
+                ["dock_slot_to"] = message.SlotTo.ToString("O"),
+            });
+
+        return writer.WriteAsync(bronzeEvent, cancellationToken);
+    }
+}

src/Services/Logistics/Modules/Warehouse.Logistics.Core/Application/Deliveries/AnnounceDelivery/AnnounceDeliveryHandler.cs

@@ -1,20 +1,23 @@
+using Warehouse.Contracts.Logistics;
 using Warehouse.Logistics.Core.Application.Abstractions;
 using Warehouse.Logistics.Core.Domain;
-using Warehouse.SharedKernel.Application;
+using Warehouse.Logistics.Core.Infrastructure.Persistence;
 using Warehouse.SharedKernel.Domain;
 using Warehouse.SharedKernel.ValueObjects;
+using Wolverine.EntityFrameworkCore;
 
 namespace Warehouse.Logistics.Core.Application.Deliveries.AnnounceDelivery;
 
 /// <summary>
 /// Validates every announced SKU against the local Catalog replica (ADR-0003 — no cross-service
 /// query) and creates the <see cref="InboundDelivery"/>. An ASN that names an unknown SKU is
 /// rejected so it never enters the receiving flow with a code Inventory cannot resolve.
+/// Announces the ASN to Bronze via the transactional outbox (<see cref="DeliveryAnnouncedV1"/>).
 /// </summary>
 public sealed class AnnounceDeliveryHandler(
     IInboundDeliveryRepository deliveries,
     ICatalogProductReplica catalog,
-    IUnitOfWork unitOfWork)
+    IDbContextOutbox<LogisticsDbContext> outbox)
 {
     public async Task<Guid> HandleAsync(AnnounceDeliveryCommand command, CancellationToken cancellationToken = default)
     {
@@ -52,7 +55,10 @@ public async Task<Guid> HandleAsync(AnnounceDeliveryCommand command, Cancellatio
             lines);
 
         deliveries.Add(delivery);
-        await unitOfWork.SaveChangesAsync(cancellationToken);
+        await outbox.PublishAsync(new DeliveryAnnouncedV1(
+            delivery.Id.Value, delivery.Warehouse.Code, delivery.Supplier.Value, delivery.PlannedAt, DateTimeOffset.UtcNow));
+
+        await outbox.SaveChangesAndFlushMessagesAsync(cancellationToken);
         return delivery.Id.Value;
     }
 }