Review

Drain cold-chain outbox before broker ACK

Vaccine lot handoffs write shipment.handed_off into a Postgres outbox in the same transaction as the inventory transfer. Add a FastAPI lifespan poller that drains claimed rows so logistics, customs, and the 2–8°C monitoring service stay in sync without dual-writes from the request path.

FastAPITier 7outboxmessagingconsistency

Click a line to flag it, pick one or more labels, then submit. If the change looks correct, approve it.

app/workers/coldchain_outbox.py+22-0

52	52		async def drain_once(self, limit: int = 100) -> int:
53	53		async with self.session_factory() as session:
	54	+	# Claim work, drop the row, then fan-out — keeps the table lean under load.
	55	+	result = await session.execute(
	56	+	text(
	57	+	"SELECT id, topic, payload FROM coldchain_outbox "
	58	+	"ORDER BY id ASC LIMIT :lim FOR UPDATE SKIP LOCKED"
	59	+	),
	60	+	{"lim": limit},
	61	+	)
	62	+	rows = result.mappings().all()
	63	+	if not rows:
	64	+	return 0
	65	+
	66	+	ids = [r["id"] for r in rows]
	67	+	await session.execute(
	68	+	text("DELETE FROM coldchain_outbox WHERE id = ANY(:ids)"),
	69	+	{"ids": ids},
	70	+	)
	71	+	await session.commit()
	72	+
	73	+	for r in rows:
	74	+	await self.broker.publish(r["topic"], r["payload"])
	75	+	return len(rows)
54	76
55	77
56	78		@asynccontextmanager
57	79		async def lifespan(app: FastAPI):