feat(sp-skill-catalog): add cyclone-store skill (write paths + event contract)
This commit is contained in:
Tyler
@@ -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 `<entity>_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
|
||||
`<entity>_written`; activity-style side rows get
|
||||
`<entity>_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_<entity>` 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.<method>(...)`, 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/<resource>/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 `<entity>_written` event has exactly one
|
||||
consumer — the matching `/api/<resource>/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<EDI>`
|
||||
lands in the store via `CycloneStore.add`.
|
||||
- **`cyclone-api-router`** — the route that calls `store.<method>(...)`
|
||||
and (for stream endpoints) subscribes to the matching
|
||||
`<entity>_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.
|
||||
Reference in New Issue
Block a user