From 4c591ee6553917dde4dbb0778baa9155b9804417 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 12:28:18 -0600 Subject: [PATCH] feat(sp-skill-catalog): add cyclone-store skill (write paths + event contract) --- .superpowers/skills/cyclone-store/SKILL.md | 200 +++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 .superpowers/skills/cyclone-store/SKILL.md diff --git a/.superpowers/skills/cyclone-store/SKILL.md b/.superpowers/skills/cyclone-store/SKILL.md new file mode 100644 index 0000000..8d7d2a2 --- /dev/null +++ b/.superpowers/skills/cyclone-store/SKILL.md @@ -0,0 +1,200 @@ +--- +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. \ No newline at end of file