POST /streams/processed does not exist: the worker 404s on every publish and stream_data never drains

[Xhristin3]

Problem

The processing worker publishes every processed event to POST ${API_URL}/streams/processed (xstreamroll-processing/src/worker.ts), but the API exposes no such route. The only stream routes are GET /streams/pending, POST /streams, GET /streams, GET /streams/:id/analytics, GET /streams/:id/events, GET /streams/:id, PATCH /streams/:id, and DELETE /streams/:id (api/src/streams/streams.controller.ts). A POST to /streams/processed returns 404.

The consequence is total: every event the worker processes hits 404, the publish retry budget (PROCESSING_PUBLISH_MAX_RETRIES, default 3, exponential backoff in xstreamroll-processing/src/session.ts) exhausts, and the event is dead-lettered in memory and lost. stream_events is never written, so GET /streams/:id/events (event replay, issue #396) and GET /streams/:id/analytics are always empty in any real deployment, and stream_data rows are never drained — the worker re-fetches and re-processes the same rows on every poll forever, re-publishing duplicates each cycle.

This is a documented contract, not a hypothetical: ADR-0003 (docs/adr/0003-polling-based-processing.md) states:

Processing status is reported back to the API via standard HTTP POST requests (POST /streams/processed).
POST /streams/processed is idempotent on the server.

The worker integration tests (xstreamroll-processing/__tests__/integration/worker.integration.test.ts) mock the endpoint with nock (.post("/streams/processed")), so CI is green while production is broken.

Root cause

// xstreamroll-processing/src/worker.ts
await axiosInstance.post(`${API_URL}/streams/processed`, event)   // ← 404: route does not exist

api/src/streams/repository/streams-db.repository.ts has getPendingEvents() (reads stream_data) but no counterpart that records a processed event into stream_events or removes the drained stream_data row — the in-memory StreamsRepository.recordEvent() exists, the DB-backed one has no equivalent.

Why this is architecturally hard

1. The endpoint must define the drain semantics. getPendingEvents returns rows from stream_data with no claimed/acked state; a correct POST /streams/processed must make the row disappear (delete or flag) atomically with the stream_events insert, or the worker reprocesses everything forever. That means a transaction — the naive shortcut (insert into stream_events, delete from stream_data, in two statements) can lose data or double-publish under a crash.
2. ADR-0003 promises idempotency, which needs a key to dedupe on (e.g. (stream_id, timestamp, payload) or a client-supplied event id). The worker currently sends no idempotency key — ProcessedStreamEvent (xstreamroll-processing/src/session.ts) has streamId, data, timestamp, processedAt, processingLatencyMs, workerId, sessionId, none of which is a stable event id.
3. Concurrency: with the Postgres lock backend, only one worker owns a stream at a time, but the endpoint itself must still be safe against double-delivery from the in-memory fallback and from retry-after-timeout (worker retries the POST when the API timed out after committing — the classic at-least-once hole).
4. The endpoint also needs a home for the auth story: GET /streams/pending is intentionally unauthenticated, and STREAM_API_KEY is required by api/src/config/env.ts but enforced nowhere — this endpoint is the natural place to start enforcing it, and the decision affects the worker's axios config (xstreamroll-processing/src/worker.ts).

Proposed design

• POST /streams/processed accepts a batch or single ProcessedStreamEvent, runs one transaction: INSERT INTO stream_events (...), DELETE FROM stream_data WHERE id = $eventId (or UPDATE stream_data SET processed = true), and returns 2xx. Resolve the stream_data row identity (add an id to the pending-event payload) so the delete is exact.
• Idempotency: reject or no-op duplicates keyed on the event id (unique index on the chosen key), satisfying ADR-0003.
• Protect the endpoint with STREAM_API_KEY (shared-secret header) and send it from the worker; keep GET /streams/pending consistent with the same policy.
• Replace the nock mocks in worker.integration.test.ts with a provider-style check (or an API integration test that actually exercises the route against the test database) so a regression surfaces in CI.

Downstream impact

• xstreamroll-processing/src/worker.ts and xstreamroll-processing/src/session.ts: the publish target exists once the endpoint lands; no worker change strictly required beyond the auth header.
• xstreamroll-sdk: the SDK's StreamEventRecord/StreamEvent types (packages/types/src/stream-event.ts) already model the replay shape; no type change needed unless the wire payload gains an event id.
• api/src/streams/repository/streams-db.repository.ts: add the processed-event write alongside the existing getPendingEvents.
• Docs: docs/adr/0003-polling-based-processing.md describes the endpoint; the ADR should be updated to match the implemented contract (or the implementation must match the ADR — either way they must agree).

Acceptance criteria

Contract

• POST /streams/processed exists, is documented in Swagger, and returns 2xx for a valid ProcessedStreamEvent.
• After a successful publish, the corresponding stream_data row is no longer returned by GET /streams/pending (drain works).
• The stream_events row inserted by the publish is returned by GET /streams/:id/events and counted by GET /streams/:id/analytics.
• Re-delivering the same event (same idempotency key) does not create a duplicate stream_events row or error (ADR-0003 idempotency).

Security

• POST /streams/processed rejects requests without the shared STREAM_API_KEY secret with 401; the worker sends the key.
• GET /streams/pending is protected by the same policy (or the decision to keep it open is documented in the controller JSDoc and an ADR).

Reliability

• A simulated crash between the stream_events insert and the stream_data delete cannot lose or duplicate an event (single transaction or equivalent recovery).
• Worker publishes no longer fail with 404 in an end-to-end test against the real API (replace the nock mock in xstreamroll-processing/__tests__/integration/worker.integration.test.ts with a route-level assertion, or add an API integration test that covers the same contract).

Documentation

• docs/adr/0003-polling-based-processing.md and the controller JSDoc accurately describe the implemented endpoint (idempotency key, auth, drain semantics).

Out of scope

A claims/lease mechanism to prevent two workers from processing the same event concurrently (the distributed lock in xstreamroll-processing/src/leader-election.ts already covers the common case), and out-of-order replay guarantees.

Getting started

Real files in scope: api/src/streams/streams.controller.ts, api/src/streams/streams.service.ts, api/src/streams/repository/streams-db.repository.ts, api/src/config/env.ts, xstreamroll-processing/src/worker.ts, xstreamroll-processing/__tests__/integration/worker.integration.test.ts, xstreamroll-processing/__tests__/integration/pipeline.integration.test.ts, database/schema.sql (index for the idempotency key).

Verify with:

cd api && npm run typecheck && npm test
cd ../xstreamroll-processing && npm run typecheck && npm test

Good first files to read: xstreamroll-processing/src/worker.ts (the publish handler in start()), api/src/streams/repository/streams-db.repository.ts (getPendingEvents), docs/adr/0003-polling-based-processing.md.

[danfulani0]

Hello maintainer please I will like to work on this issue

[grantfox-oss]

🦊 GrantFox — @danfulani0 has been assigned to this issue as part of the Third Campaign campaign!

Next steps:

1. Open a Pull Request referencing this issue (e.g., Closes #513)
2. Your PR will be reviewed by the XStreamRollz maintainers

Good luck! Track your progress on GrantFox.