diff --git a/.claude/agent-memory/ts-node-backend-engineer/project_processor.md b/.claude/agent-memory/ts-node-backend-engineer/project_processor.md index 902ce3f..d7376c4 100644 --- a/.claude/agent-memory/ts-node-backend-engineer/project_processor.md +++ b/.claude/agent-memory/ts-node-backend-engineer/project_processor.md @@ -1,10 +1,10 @@ --- name: Processor Service -description: processor service: Phase 1 complete (all 11 tasks), key patterns, conventions, and quirks +description: processor service: Phase 1 and Phase 1.5 complete; key patterns, conventions, and quirks type: project --- -Phase 1 complete. All 11 tasks landed. The throughput pipeline is done: consumer + writer + metrics + integration test + Docker + CI. +Phase 1 complete (11 tasks). Phase 1.5 complete (6 tasks, commits b8ebbd0–54f1684). Live broadcast WebSocket endpoint is fully wired and tested. **Architecture divergence from tcp-ingestion:** - ESLint `import/no-restricted-paths` zone: `src/core/` cannot import `src/domain/` (preemptive for Phase 2). @@ -50,3 +50,45 @@ Phase 1 is the throughput pipeline + operational baseline. Phase 2 (domain logic **How to apply:** Phase 2 adds domain logic to `src/domain/` — no changes to `src/core/`. Phase 3 adds graceful shutdown polish, XAUTOCLAIM, and state rehydration. + +--- + +## Phase 1.5 — Live broadcast (Done) + +**ESLint boundary: `src/core/ ↔ src/live/` mutual exclusion.** +Enforced by `import/no-restricted-paths`. Shared code lives in `src/shared/`: +- `src/shared/types.ts`: `Metrics`, `Position`, `AttributeValue` +- `src/shared/codec.ts`: `CodecError`, `decodePosition` +Both `src/core/` modules re-export from shared to preserve existing import paths. + +**Live server (src/live/server.ts):** +- `createLiveServer(config, logger, metrics, onMessage, onClose?, authClient?)` factory +- `LIVE_WS_PORT=0` assigns an OS port but there's no public API to read it back. Integration tests use a two-step approach: probe a free port, then start with that fixed port. +- Auth runs in the `upgrade` handler before completing the WebSocket handshake. + +**Auth/Authz (src/live/auth.ts, src/live/authz.ts):** +- `validate(cookieHeader)` → Directus `/users/me?fields=...`; missing `data` key = error (not unauthorized); `data: null` = unauthorized. +- `canAccessEvent(cookieHeader, eventId)` → Directus `/items/events/:id`; never throws. + +**Subscription registry (src/live/registry.ts):** +- WeakMap> for conn→topics (GC-safe); Map> for topic→conns. +- `SnapshotProvider` injected at construction; default is a stub returning `[]`. +- `fetchSnapshot` in registry wraps in try/catch (fail open = send `subscribed` with empty snapshot on failure). + +**Device-event map (src/live/device-event-map.ts):** +- In-memory `Map>` refreshed every `LIVE_DEVICE_EVENT_REFRESH_MS`. +- Phase 1 deviation: `entry_devices.device_id` is IMEI text in the test fixture (`test/fixtures/test-schema.sql`) but UUID FK to devices.id in the real Directus schema. + +**Broadcast consumer (src/live/broadcast.ts):** +- Per-instance consumer group `live-broadcast-{INSTANCE_ID}`; ACK immediately on consume. +- Test strategy: `makeRedis` stub blocks the second `xreadgroup` call until xack fires (via `stopSignal` Promise), preventing tight-loop OOM in test workers. + +**Snapshot provider (src/live/snapshot.ts):** +- `DISTINCT ON (p.device_id) ... ORDER BY p.device_id, p.ts DESC WHERE p.faulty = false` +- Requires `positions_device_ts_idx ON positions (device_id, ts DESC)` (created in migration 0002). + +**Integration test (test/live.integration.test.ts):** +- Directus stub: `test/helpers/directus-stub.ts` — bare `http.createServer`, no Express. +- Test fixture: `test/fixtures/test-schema.sql` — simplified schema with `entry_devices.device_id TEXT`. +- `waitForMessage(ws, predicate, timeoutMs)` helper for typed WS message assertions. +- Orphan test: `Promise.race([waitForMessage(...), timeout])` → asserts `'timeout'` result. diff --git a/.gitignore b/.gitignore index 382fc0f..cb4929e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ .obsidian/workspace-mobile.json .obsidian/cache .obsidian/plugins/*/data.json +.claude/ \ No newline at end of file