--- name: cyclone-store description: "Cyclone store write-paths, the pubsub event contract (claim_written / remittance_written / activity_recorded), and the SP21 store-split boundary map. Use when: touching store.py, adding a new entity, wiring a new write event, or splitting a store module." --- # cyclone-store Cyclone persists every parsed X12 batch through one facade, `CycloneStore` (`backend/src/cyclone/store.py:882`). Every write inserts the row AND publishes a pubsub event on the in-process `EventBus` (`backend/src/cyclone/pubsub.py:20`) so live-tail pages see new rows the moment they land. The event contract is the seam between persistence and streaming — a wrong event name silently goes stale. As of this writing: the store is a **single 2412-line module** and the SP21 split into `backend/src/cyclone/store/` is in progress. Three event kinds: `claim_written`, `remittance_written`, `activity_recorded`. Next: **SP22**. ## When to use - **Adding a new entity.** You need a new ORM model + write method + event kind + snapshot serializer and want to know where each piece lives (current monolith vs. post-SP21 module). - **Wiring a new write event.** You're adding a `_written` event and need both the publish call in the store AND the subscribe call in the live-tail endpoint to stay in sync. - **Debugging a write-path issue.** A page isn't reflecting new rows, the DB has the row but the stream is silent, or the publish raises and rolls back the transaction. - **Splitting a store module.** You're moving a domain out of `store.py` into its own module under `backend/src/cyclone/store/` and need the SP21 module list + facade re-export rules. ## Conventions 1. **All writes go through `CycloneStore`.** Route handlers and parsers must not write directly to the ORM session. The facade opens a short-lived session via `db.SessionLocal()()`. Direct ORM access is the #1 way the live-tail contract gets bypassed (no event published → page silently goes stale). Read paths are similar — prefer the facade's `iter_*` / `get_*` methods over raw `s.execute(select(...))`. 2. **Every write publishes an event.** The event name matches the entity: `claim_written`, `remittance_written`, `activity_recorded` (the trailing `_recorded` signals a non-canonical row — activity events are derived, not first-class; current names at `store.py:1072,1081,1096`). New entities get `_written`; activity-style side rows get `_recorded`. Publish is **best-effort** — failures are logged but never roll back the persisted batch (`store.py:1097-1098`). 3. **Snapshot shape.** Each entity has a `to_ui_` serializer (plain Python function returning a dict) — currently `to_ui_claim`, `to_ui_remittance`, `to_ui_claim_from_orm`, `to_ui_remittance_from_orm`, `to_ui_provider`. Post-SP21 these move to `backend/src/cyclone/store/ui.py`. The serializer is the single source of truth for what the frontend sees — every event payload MUST match what the matching list endpoint returns for that row (`store.py:1052-1054`). 4. **SP21 boundaries.** Post-split, each domain lives in its own module under `backend/src/cyclone/store/`: `__init__.py` (facade + `CycloneStore` class), `exceptions.py`, `records.py`, `orm_builders.py`, `ui.py`, `write.py`, `batches.py`, `claim_detail.py`, `acks.py`, `backups.py`, `inbox.py`, `providers.py`. Cross-module writes go through `CycloneStore` facade methods, not direct module access. The facade re-exports every name callers currently import from `cyclone.store`. Full list: `docs/superpowers/plans/2026-06-21-cyclone-store-split.md:25-37`. 5. **No business logic in route handlers.** A route handler validates input, calls `store.(...)`, passes the `event_bus`, returns the serialized result. Reconciliation, idempotency checks, CAS adjustment persistence — all live in the store. ## Patterns ### A `CycloneStore.add` write — publishes events from inserted rows Taken from `backend/src/cyclone/store.py:898-1107`. The method opens a session, inserts rows, then runs a sync `_publish_events_sync` after commit so subscribers can immediately re-fetch consistent data. ```python def add( self, record: BatchRecord, *, event_bus: "EventBus | None" = None, ) -> None: inserted_claim_ids: list[str] = [] with db.SessionLocal()() as s: s.add(Batch(id=record.id, kind=record.kind, ...)) if isinstance(record, BatchRecord837): for claim in record.result.claims: if s.get(Claim, claim.claim_id) is not None: continue # idempotency: skip dupes s.add(_claim_837_row(claim, record.id)) s.add(ActivityEvent(kind="claim_submitted", ...)) inserted_claim_ids.append(claim.claim_id) # ... 835 branch + flush + cas adjustments ... s.commit() if event_bus is not None and inserted_claim_ids: self._publish_events_sync(event_bus, record, inserted_claim_ids) def _publish_events_sync(self, event_bus, record, claim_ids): with db.SessionLocal()() as s: for cid in claim_ids: ui = to_ui_claim_from_orm(s.get(Claim, cid), ...) self._sync_publish(event_bus, "claim_written", ui) # ... remittance + activity loops ... ``` `EventBus.publish` is async but the body is pure sync `put_nowait`, so the store calls `_sync_publish` directly to avoid forcing sync FastAPI handlers to await. ### Backend `/api//stream` endpoint — subscribes to the event Taken from `backend/src/cyclone/api.py:1380-1401`. Two phases — eager snapshot, then live subscription — wrapped in `StreamingResponse` with `media_type="application/x-ndjson"`. ```python @app.get("/api/claims/stream") async def claims_stream(request: Request, ...) -> StreamingResponse: bus: EventBus = request.app.state.event_bus async def gen() -> AsyncIterator[bytes]: rows = store.iter_claims(status=status, ...) # 1. Snapshot for row in rows: yield _ndjson_line({"type": "item", "data": row}) yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) async for chunk in _tail_events(request, bus, ["claim_written"]): # 2. Live yield chunk return StreamingResponse(gen(), media_type="application/x-ndjson") ``` `_ndjson_line` and `_tail_events` live in `backend/src/cyclone/api_helpers.py`. The `["remittance_written"]` and `["activity_recorded"]` subscriptions are at `api.py:1891,2002`. ### Backend test — asserts both the row AND the event landed The autouse `conftest.py` (`backend/tests/conftest.py:20`) wires a fresh `EventBus` onto `app.state` per test. ```python def test_publishes_claim_written_event(client: TestClient) -> None: bus = app.state.event_bus resp = client.post("/api/parse-837", files={"file": ("x.837", MINIMAL_837, "text/plain")}) assert resp.status_code == 200 assert list(store.iter_claims(limit=10)) # row landed queues = bus._subscribers.get("claim_written", []) # event published assert queues evt = queues[0].get_nowait() assert evt["_kind"] == "claim_written" ``` ## Anti-patterns - **Don't read directly from the ORM in a route handler — go through the snapshot serializer.** A handler that does `with db.SessionLocal()() as s: row = s.get(Claim, cid); return row` bypasses `to_ui_claim_from_orm` and silently drifts from the event payload shape. Always call `store.get_claim_detail(cid)` (or the equivalent `get_*` facade method). - **Don't introduce a new event name without updating the subscriber list.** Today every `_written` event has exactly one consumer — the matching `/api//stream` endpoint, currently in `backend/src/cyclone/api.py` (claims at `:1398`, remittances at `:1891`, activity at `:2002`). Any future split into `backend/src/cyclone/api_routers/` must wire the same subscription. - **Don't merge a write method with its event publication into separate places.** `_publish_events_sync` lives next to `add` in `store.py:1042-1107` so reviewers see both halves of the contract in one diff. Post-SP21 the same rule applies. - **Don't make the publish call blocking on commit failures.** Publish errors are caught and logged at `store.py:1097-1098`; a failing subscriber MUST NOT roll back the persisted batch. The batch is the source of truth; the event is the cache-invalidation hint. ## Related skills - **`cyclone-edi`** — the parsed `ClaimOutput` / `ParseResult` lands in the store via `CycloneStore.add`. - **`cyclone-api-router`** — the route that calls `store.(...)` and (for stream endpoints) subscribes to the matching `_written` event. - **`cyclone-tail`** — the consumer side of the event contract; load when changing the wire format or adding a streaming hook. - **`cyclone-tests`** — write-path tests live under `backend/tests/` and assert both the DB row and the event payload.