From 327a3ade53b502854364595f547129f5143114da Mon Sep 17 00:00:00 2001 From: Moskovkin Ilya Date: Sun, 19 Jul 2026 02:42:53 +0700 Subject: [PATCH] feat: add protected event ingest foundation --- README.md | 12 +- package.json | 8 +- scripts/test-signalops-ingest.mjs | 266 ++++++++++++++++++++++++++++++ src/app/api/events/route.ts | 17 ++ src/lib/signalops/ingest-auth.ts | 28 ++++ src/lib/signalops/ingest-sink.ts | 168 +++++++++++++++++++ src/lib/signalops/ingest.ts | 204 +++++++++++++++++++++++ tsconfig.json | 1 + 8 files changed, 702 insertions(+), 2 deletions(-) create mode 100644 scripts/test-signalops-ingest.mjs create mode 100644 src/app/api/events/route.ts create mode 100644 src/lib/signalops/ingest-auth.ts create mode 100644 src/lib/signalops/ingest-sink.ts create mode 100644 src/lib/signalops/ingest.ts diff --git a/README.md b/README.md index 9909abc..f27b00f 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,9 @@ SignalOps is a React dashboard for operating AI image-generation products. It is - Recharts - Lucide React -## Run Locally +## Run locally + +Requires Node.js 24 or newer and pnpm 10.24.0, matching CI and the native TypeScript contract checks. ```bash pnpm install @@ -39,6 +41,14 @@ pnpm dev The dev server is pinned to [http://localhost:3020](http://localhost:3020) to avoid colliding with other local portfolio/product apps. +## Event API boundaries + +- `POST /api/events/validate` is public and verification-only. It normalizes and redacts events but stores nothing. +- `POST /api/events` is a protected development/test ingest seam. Set `SIGNALOPS_INGEST_TOKEN` and send that exact value using the Bearer authorization scheme. +- Accepted local events always use the validator's bounded JSON, batch, normalization, and redaction contract. Exact retries return the same receipt reference and do not duplicate retained events. +- The local memory sink retains at most 1,000 events and can evict the oldest retained event. Receipts expose stored, duplicate, evicted, and retained counts so this behavior is not presented as durability. +- Production intentionally returns `503 storage_not_ready` after authentication because this slice does not configure a durable store. The endpoint is not production-ready until a separate durable sink is implemented and explicitly wired. + ## Demo script For a fast portfolio review, the dashboard opens with a **Guided incident replay** rail directly under the header. It turns the surface into a self-explaining demo — a first-run reviewer can follow one incident end to end in well under 90 seconds. diff --git a/package.json b/package.json index fa8cf0d..ff0e447 100644 --- a/package.json +++ b/package.json @@ -2,6 +2,10 @@ "name": "signalops", "version": "0.1.0", "private": true, + "engines": { + "node": ">=24" + }, + "packageManager": "pnpm@10.24.0", "scripts": { "dev": "next dev -p 3020", "dev:any": "next dev", @@ -10,7 +14,9 @@ "lint": "eslint", "typecheck": "tsc --noEmit --pretty false", "test:home-navigation": "node scripts/test-home-navigation.mjs", - "check": "pnpm lint && pnpm typecheck && pnpm test:home-navigation && pnpm build" + "test:events": "node scripts/test-signalops-events.mjs", + "test:ingest": "node scripts/test-signalops-ingest.mjs", + "check": "pnpm lint && pnpm typecheck && pnpm test:home-navigation && pnpm test:events && pnpm test:ingest && pnpm build" }, "dependencies": { "@base-ui/react": "^1.5.0", diff --git a/scripts/test-signalops-ingest.mjs b/scripts/test-signalops-ingest.mjs new file mode 100644 index 0000000..5e4cb2f --- /dev/null +++ b/scripts/test-signalops-ingest.mjs @@ -0,0 +1,266 @@ +import assert from "node:assert/strict"; + +import { SIGNALOPS_EVENT_LIMITS } from "../src/lib/signalops/events.ts"; +import { + authenticateSignalEventIngest, + isSignalEventIngestConfigured, +} from "../src/lib/signalops/ingest-auth.ts"; +import { + createMemorySignalEventSink, + getSignalEventSinkForRuntime, +} from "../src/lib/signalops/ingest-sink.ts"; +import { handleSignalEventIngest } from "../src/lib/signalops/ingest.ts"; + +const configuredEnv = { SIGNALOPS_INGEST_TOKEN: "test-token-with-enough-entropy" }; + +assert.equal(isSignalEventIngestConfigured(configuredEnv), true); +assert.equal(isSignalEventIngestConfigured({}), false); + +const unauthorizedCases = [ + new Headers(), + new Headers({ authorization: "Basic dGVzdDp0ZXN0" }), + new Headers({ authorization: "Bearer" }), + new Headers({ authorization: "Bearer wrong-token" }), + new Headers({ authorization: "Bearer test-token-with-enough-entropy" }), +]; + +for (const headers of unauthorizedCases) { + assert.equal(authenticateSignalEventIngest(headers, configuredEnv), false); +} + +assert.equal( + authenticateSignalEventIngest( + new Headers({ authorization: "Bearer test-token-with-enough-entropy" }), + configuredEnv, + ), + true, +); + +const sink = createMemorySignalEventSink({ capacity: 2 }); +const eventOne = { eventId: "evt_one", occurredAt: "2026-07-19T00:00:00.000Z" }; +const eventTwo = { eventId: "evt_two", occurredAt: "2026-07-19T00:01:00.000Z" }; +const eventThree = { eventId: "evt_three", occurredAt: "2026-07-19T00:02:00.000Z" }; + +const firstWrite = await sink.store([eventOne, eventTwo]); +assert.deepEqual(firstWrite.storedEventIds, ["evt_one", "evt_two"]); +assert.equal(firstWrite.duplicateEvents, 0); +assert.equal(firstWrite.evictedEvents, 0); +assert.equal(firstWrite.retainedEvents, 2); + +const boundedWrite = await sink.store([eventTwo, eventThree]); +assert.deepEqual(boundedWrite.duplicateEventIds, ["evt_two"]); +assert.deepEqual(boundedWrite.storedEventIds, ["evt_three"]); +assert.equal(boundedWrite.evictedEvents, 1); +assert.deepEqual( + sink.snapshot().map((event) => event.eventId), + ["evt_two", "evt_three"], +); + +sink.reset(); +assert.deepEqual(sink.snapshot(), []); + +assert.equal(getSignalEventSinkForRuntime({ NODE_ENV: "production" }).adapter, "unconfigured"); +assert.equal(getSignalEventSinkForRuntime({ VERCEL: "1" }).adapter, "unconfigured"); + +const repeatedBatchSink = createMemorySignalEventSink(); +const repeatedBatchWrite = await repeatedBatchSink.store([eventOne, eventOne]); +assert.equal(repeatedBatchWrite.storedEvents, 1); +assert.equal(repeatedBatchWrite.duplicateEvents, 1); +assert.equal(repeatedBatchWrite.retainedEvents, 1); + +function ingestRequest(body, authorization) { + return new Request("http://localhost/api/events", { + method: "POST", + headers: { + "content-type": "application/json", + ...(authorization ? { authorization } : {}), + }, + body, + }); +} + +const fixedRequestId = () => "req_test_ingest"; +const validBody = JSON.stringify({ + type: "generation.completed", + occurredAt: "2026-07-19T00:00:00.000Z", + generationId: "gen_ingest_001", + providerId: "fal", + modelId: "flux-kontext-pro", + user: "private@example.com", + prompt: "private prompt", +}); + +const missingConfigResponse = await handleSignalEventIngest(ingestRequest(validBody), { + env: {}, + sink: createMemorySignalEventSink(), + requestIdFactory: fixedRequestId, +}); +assert.equal(missingConfigResponse.status, 503); +assert.equal(missingConfigResponse.body.code, "ingest_not_configured"); + +const unauthorizedResponses = await Promise.all( + [undefined, "Basic bad", "Bearer", "Bearer wrong-token"].map((authorization) => + handleSignalEventIngest(ingestRequest(validBody, authorization), { + env: configuredEnv, + sink: createMemorySignalEventSink(), + requestIdFactory: fixedRequestId, + }), + ), +); +for (const response of unauthorizedResponses) { + assert.equal(response.status, 401); + assert.deepEqual(response.body, { + ok: false, + code: "unauthorized", + error: "Unauthorized.", + requestId: "req_test_ingest", + }); +} + +const productionSink = createMemorySignalEventSink(); +const productionResponse = await handleSignalEventIngest( + ingestRequest(validBody, "Bearer test-token-with-enough-entropy"), + { + env: { ...configuredEnv, NODE_ENV: "production" }, + sink: productionSink, + requestIdFactory: fixedRequestId, + }, +); +assert.equal(productionResponse.status, 503); +assert.equal(productionResponse.body.code, "storage_not_ready"); +assert.deepEqual(productionSink.snapshot(), []); + +const ingestSink = createMemorySignalEventSink({ capacity: 4 }); +const ingestOptions = { + env: configuredEnv, + sink: ingestSink, + requestIdFactory: fixedRequestId, + currentTimeMs: Date.parse("2026-07-19T00:05:00.000Z"), +}; +const authorizedResponse = await handleSignalEventIngest( + ingestRequest(validBody, "Bearer test-token-with-enough-entropy"), + ingestOptions, +); +assert.equal(authorizedResponse.status, 200); +assert.equal(authorizedResponse.body.ok, true); +assert.equal(authorizedResponse.body.partial, false); +assert.equal(authorizedResponse.body.receipt.acceptedEvents, 1); +assert.equal(authorizedResponse.body.receipt.rejectedEvents, 0); +assert.equal(authorizedResponse.body.receipt.storedEvents, 1); +assert.equal(authorizedResponse.body.receipt.duplicateEvents, 0); +assert.match(authorizedResponse.body.receipt.receiptId, /^rcpt_[0-9a-f]{16}$/); +assert.match(authorizedResponse.body.receipt.eventRefs[0], /^evtref_[0-9a-f]{16}$/); +assert.equal(ingestSink.snapshot()[0].user, "[redacted]"); +assert.equal(ingestSink.snapshot()[0].prompt, "[redacted]"); + +const duplicateResponse = await handleSignalEventIngest( + ingestRequest(validBody, "Bearer test-token-with-enough-entropy"), + ingestOptions, +); +assert.equal(duplicateResponse.status, 200); +assert.equal(duplicateResponse.body.receipt.storedEvents, 0); +assert.equal(duplicateResponse.body.receipt.duplicateEvents, 1); +assert.equal( + duplicateResponse.body.receipt.receiptId, + authorizedResponse.body.receipt.receiptId, +); +assert.equal(ingestSink.snapshot().length, 1); + +const conflictingResponse = await handleSignalEventIngest( + ingestRequest( + JSON.stringify({ ...JSON.parse(validBody), cost: 99 }), + "Bearer test-token-with-enough-entropy", + ), + ingestOptions, +); +assert.equal(conflictingResponse.status, 409); +assert.equal(conflictingResponse.body.code, "idempotency_conflict"); +assert.equal(ingestSink.snapshot().length, 1); +assert.equal(ingestSink.snapshot()[0].cost, undefined); + +const atomicConflictSink = createMemorySignalEventSink(); +await atomicConflictSink.store([{ eventId: "evt_existing", cost: 1 }]); +const atomicConflict = await atomicConflictSink.store([ + { eventId: "evt_new", cost: 2 }, + { eventId: "evt_existing", cost: 3 }, +]); +assert.equal(atomicConflict.conflictEvents, 1); +assert.equal(atomicConflict.storedEvents, 0); +assert.deepEqual( + atomicConflictSink.snapshot().map((event) => event.eventId), + ["evt_existing"], +); + +const partialSink = createMemorySignalEventSink(); +const partialResponse = await handleSignalEventIngest( + ingestRequest( + JSON.stringify({ + events: [ + JSON.parse(validBody), + { ...JSON.parse(validBody), eventId: "invalid-row", durationMs: -1 }, + ], + }), + "Bearer test-token-with-enough-entropy", + ), + { ...ingestOptions, sink: partialSink }, +); +assert.equal(partialResponse.status, 200); +assert.equal(partialResponse.body.partial, true); +assert.equal(partialResponse.body.receipt.acceptedEvents, 1); +assert.equal(partialResponse.body.receipt.rejectedEvents, 1); +assert.equal(partialResponse.body.receipt.storedEvents, 1); +assert.equal(partialSink.snapshot().length, 1); + +const invalidJsonResponse = await handleSignalEventIngest( + ingestRequest("{not-json", "Bearer test-token-with-enough-entropy"), + { ...ingestOptions, sink: createMemorySignalEventSink() }, +); +assert.equal(invalidJsonResponse.status, 400); +assert.equal(invalidJsonResponse.body.code, "invalid_json"); + +const oversizedSink = createMemorySignalEventSink(); +const oversizedResponse = await handleSignalEventIngest( + new Request("http://localhost/api/events", { + method: "POST", + headers: { + authorization: "Bearer test-token-with-enough-entropy", + "content-type": "application/json", + "content-length": String(SIGNALOPS_EVENT_LIMITS.maxBodyBytes + 1), + }, + body: "{}", + }), + { ...ingestOptions, sink: oversizedSink }, +); +assert.equal(oversizedResponse.status, 413); +assert.equal(oversizedResponse.body.code, "payload_too_large"); +assert.deepEqual(oversizedSink.snapshot(), []); + +const rejectedSink = createMemorySignalEventSink(); +const rejectedResponse = await handleSignalEventIngest( + ingestRequest( + JSON.stringify({ ...JSON.parse(validBody), durationMs: -1 }), + "Bearer test-token-with-enough-entropy", + ), + { ...ingestOptions, sink: rejectedSink }, +); +assert.equal(rejectedResponse.status, 422); +assert.equal(rejectedResponse.body.code, "invalid_event"); +assert.deepEqual(rejectedSink.snapshot(), []); + +const throwingSink = { + adapter: "memory", + durable: false, + capacity: 1, + async store() { + throw new Error("secret backend detail"); + }, +}; +const unexpectedResponse = await handleSignalEventIngest( + ingestRequest(validBody, "Bearer test-token-with-enough-entropy"), + { ...ingestOptions, sink: throwingSink }, +); +assert.equal(unexpectedResponse.status, 500); +assert.equal(unexpectedResponse.body.code, "internal_error"); +assert.doesNotMatch(JSON.stringify(unexpectedResponse.body), /secret backend detail/); + +console.log("signalops protected ingest auth, storage, and receipt checks passed"); diff --git a/src/app/api/events/route.ts b/src/app/api/events/route.ts new file mode 100644 index 0000000..dd1fe0c --- /dev/null +++ b/src/app/api/events/route.ts @@ -0,0 +1,17 @@ +import { NextResponse } from "next/server"; + +import { getSignalEventSinkForRuntime } from "@/lib/signalops/ingest-sink"; +import { handleSignalEventIngest } from "@/lib/signalops/ingest"; + +export const runtime = "nodejs"; + +export async function POST(request: Request) { + const response = await handleSignalEventIngest(request, { + sink: getSignalEventSinkForRuntime(), + }); + + return NextResponse.json(response.body, { + status: response.status, + headers: { "cache-control": "no-store" }, + }); +} diff --git a/src/lib/signalops/ingest-auth.ts b/src/lib/signalops/ingest-auth.ts new file mode 100644 index 0000000..15ad2cf --- /dev/null +++ b/src/lib/signalops/ingest-auth.ts @@ -0,0 +1,28 @@ +import { createHash, timingSafeEqual } from "node:crypto"; + +type RuntimeEnv = Record; + +function digestSecret(value: string) { + return createHash("sha256").update(value, "utf8").digest(); +} + +function extractBearerToken(headers: Headers) { + const authorization = headers.get("authorization") ?? ""; + const match = /^Bearer ([^\s]+)$/.exec(authorization); + return match?.[1] ?? ""; +} + +export function isSignalEventIngestConfigured(env: RuntimeEnv = process.env) { + return Boolean(env.SIGNALOPS_INGEST_TOKEN?.trim()); +} + +export function authenticateSignalEventIngest( + headers: Headers, + env: RuntimeEnv = process.env, +) { + const expectedToken = env.SIGNALOPS_INGEST_TOKEN ?? ""; + const suppliedToken = extractBearerToken(headers); + const secretsMatch = timingSafeEqual(digestSecret(suppliedToken), digestSecret(expectedToken)); + + return Boolean(suppliedToken) && isSignalEventIngestConfigured(env) && secretsMatch; +} diff --git a/src/lib/signalops/ingest-sink.ts b/src/lib/signalops/ingest-sink.ts new file mode 100644 index 0000000..e0c04d1 --- /dev/null +++ b/src/lib/signalops/ingest-sink.ts @@ -0,0 +1,168 @@ +import { createHash } from "node:crypto"; + +export const DEFAULT_MEMORY_INGEST_CAPACITY = 1_000; + +export type IngestibleSignalEvent = { + eventId: string; + receivedAt?: string; +}; + +export type SignalEventSinkWriteResult = { + acceptedEvents: number; + storedEvents: number; + duplicateEvents: number; + conflictEvents: number; + evictedEvents: number; + retainedEvents: number; + storedEventIds: string[]; + duplicateEventIds: string[]; + conflictEventIds: string[]; +}; + +export type SignalEventSink = { + adapter: string; + durable: boolean; + capacity: number | null; + store(events: readonly IngestibleSignalEvent[]): Promise; +}; + +export type MemorySignalEventSink = SignalEventSink & { + adapter: "memory"; + durable: false; + capacity: number; + reset(): void; + snapshot(): IngestibleSignalEvent[]; +}; + +const globalIngestState = globalThis as typeof globalThis & { + __signalopsDevelopmentIngestSink?: MemorySignalEventSink; +}; + +function payloadFingerprint(event: IngestibleSignalEvent) { + const stablePayload = Object.entries(event) + .filter(([key, value]) => key !== "receivedAt" && value !== undefined) + .sort(([left], [right]) => left.localeCompare(right)); + + return createHash("sha256").update(JSON.stringify(stablePayload), "utf8").digest("hex"); +} + +export function createMemorySignalEventSink( + options: { capacity?: number } = {}, +): MemorySignalEventSink { + const capacity = options.capacity ?? DEFAULT_MEMORY_INGEST_CAPACITY; + if (!Number.isInteger(capacity) || capacity < 1 || capacity > 10_000) { + throw new RangeError("Memory ingest capacity must be an integer between 1 and 10000."); + } + + const events = new Map< + string, + { event: IngestibleSignalEvent; payloadFingerprint: string } + >(); + + return { + adapter: "memory", + durable: false, + capacity, + async store(incomingEvents) { + const duplicateEventIds: string[] = []; + const conflictEventIds: string[] = []; + let evictedEvents = 0; + + const incoming = incomingEvents.map((event) => ({ + event, + payloadFingerprint: payloadFingerprint(event), + })); + const knownFingerprints = new Map( + [...events].map(([eventId, stored]) => [eventId, stored.payloadFingerprint]), + ); + const pendingEvents = new Map< + string, + { event: IngestibleSignalEvent; payloadFingerprint: string } + >(); + + for (const candidate of incoming) { + const knownFingerprint = knownFingerprints.get(candidate.event.eventId); + if (knownFingerprint === undefined) { + knownFingerprints.set(candidate.event.eventId, candidate.payloadFingerprint); + pendingEvents.set(candidate.event.eventId, candidate); + } else if (knownFingerprint !== candidate.payloadFingerprint) { + conflictEventIds.push(candidate.event.eventId); + } else { + duplicateEventIds.push(candidate.event.eventId); + } + } + + if (conflictEventIds.length > 0) { + return { + acceptedEvents: incomingEvents.length, + storedEvents: 0, + duplicateEvents: 0, + conflictEvents: new Set(conflictEventIds).size, + evictedEvents: 0, + retainedEvents: events.size, + storedEventIds: [], + duplicateEventIds: [], + conflictEventIds: [...new Set(conflictEventIds)], + }; + } + + for (const candidate of pendingEvents.values()) { + events.set(candidate.event.eventId, { + event: { ...candidate.event }, + payloadFingerprint: candidate.payloadFingerprint, + }); + + if (events.size > capacity) { + const oldestEventId = events.keys().next().value; + if (oldestEventId) { + events.delete(oldestEventId); + evictedEvents += 1; + } + } + } + + return { + acceptedEvents: incomingEvents.length, + storedEvents: pendingEvents.size, + duplicateEvents: duplicateEventIds.length, + conflictEvents: 0, + evictedEvents, + retainedEvents: events.size, + storedEventIds: [...pendingEvents.keys()], + duplicateEventIds, + conflictEventIds, + }; + }, + reset() { + events.clear(); + }, + snapshot() { + return [...events.values()].map(({ event }) => ({ ...event })); + }, + }; +} + +export function getDevelopmentSignalEventSink() { + if (!globalIngestState.__signalopsDevelopmentIngestSink) { + globalIngestState.__signalopsDevelopmentIngestSink = createMemorySignalEventSink(); + } + + return globalIngestState.__signalopsDevelopmentIngestSink; +} + +const unavailableProductionSink: SignalEventSink = { + adapter: "unconfigured", + durable: false, + capacity: null, + async store() { + throw new Error("Production ingest storage is unavailable."); + }, +}; + +export function getSignalEventSinkForRuntime( + env: Record = process.env, +) { + return env.NODE_ENV === "production" || Boolean(env.VERCEL) + ? unavailableProductionSink + : getDevelopmentSignalEventSink(); +} diff --git a/src/lib/signalops/ingest.ts b/src/lib/signalops/ingest.ts new file mode 100644 index 0000000..ba2f77d --- /dev/null +++ b/src/lib/signalops/ingest.ts @@ -0,0 +1,204 @@ +import { createHash } from "node:crypto"; + +import { + SIGNALOPS_EVENT_LIMITS, + acceptsApplicationJson, + normalizeSignalEventBatch, + readBoundedRequestBody, + requestContentLengthExceeds, +} from "./events.ts"; +import { + authenticateSignalEventIngest, + isSignalEventIngestConfigured, +} from "./ingest-auth.ts"; +import type { SignalEventSink } from "./ingest-sink.ts"; + +type RuntimeEnv = Record; + +type IngestResponse = { + status: number; + body: Record & { + ok: boolean; + requestId: string; + code?: string; + error?: string; + }; +}; + +function createRequestId() { + return `req_${crypto.randomUUID()}`; +} + +function isProductionRuntime(env: RuntimeEnv) { + return env.NODE_ENV === "production" || Boolean(env.VERCEL); +} + +function safeReference(prefix: "evtref" | "rcpt", value: string) { + const digest = createHash("sha256").update(value, "utf8").digest("hex").slice(0, 16); + return `${prefix}_${digest}`; +} + +function errorResponse( + status: number, + requestId: string, + code: string, + error: string, +): IngestResponse { + return { + status, + body: { ok: false, code, error, requestId }, + }; +} + +export async function handleSignalEventIngest( + request: Request, + options: { + env?: RuntimeEnv; + sink: SignalEventSink; + requestIdFactory?: () => string; + currentTimeMs?: number; + }, +): Promise { + const env = options.env ?? process.env; + const requestId = (options.requestIdFactory ?? createRequestId)(); + + if (!isSignalEventIngestConfigured(env)) { + return errorResponse( + 503, + requestId, + "ingest_not_configured", + "Protected ingest is not configured.", + ); + } + + if (!authenticateSignalEventIngest(request.headers, env)) { + return errorResponse(401, requestId, "unauthorized", "Unauthorized."); + } + + if (isProductionRuntime(env) && !options.sink.durable) { + return errorResponse( + 503, + requestId, + "storage_not_ready", + "Durable ingest storage is not configured.", + ); + } + + try { + if (!acceptsApplicationJson(request)) { + return errorResponse(415, requestId, "unsupported_media_type", "Use application/json."); + } + + if (requestContentLengthExceeds(request, SIGNALOPS_EVENT_LIMITS.maxBodyBytes)) { + return errorResponse( + 413, + requestId, + "payload_too_large", + `Request body must be ${SIGNALOPS_EVENT_LIMITS.maxBodyBytes} bytes or smaller.`, + ); + } + + const bodyResult = await readBoundedRequestBody(request, SIGNALOPS_EVENT_LIMITS.maxBodyBytes); + if (!bodyResult.ok) { + return errorResponse( + 413, + requestId, + "payload_too_large", + `Request body must be ${SIGNALOPS_EVENT_LIMITS.maxBodyBytes} bytes or smaller.`, + ); + } + + let payload: unknown; + try { + payload = JSON.parse(bodyResult.text); + } catch { + return errorResponse(400, requestId, "invalid_json", "Body must contain valid JSON."); + } + + let batch; + try { + batch = normalizeSignalEventBatch(payload, { + privacyMode: "redact", + currentTimeMs: options.currentTimeMs, + maxBatchEvents: SIGNALOPS_EVENT_LIMITS.maxBatchEvents, + }); + } catch (error) { + return errorResponse( + 422, + requestId, + "invalid_batch", + error instanceof Error ? error.message : "Invalid event batch.", + ); + } + + if (batch.events.length === 0) { + return { + status: 422, + body: { + ok: false, + code: "invalid_event", + error: "No valid events were accepted.", + requestId, + rejected: batch.rejected, + }, + }; + } + + const writeResult = await options.sink.store(batch.events); + const eventIds = batch.events.map((event) => event.eventId); + + if (writeResult.conflictEvents > 0) { + return { + status: 409, + body: { + ok: false, + code: "idempotency_conflict", + error: "An event identifier was already used for a different payload.", + requestId, + conflictEventRefs: writeResult.conflictEventIds.map((eventId) => + safeReference("evtref", eventId), + ), + }, + }; + } + + const receiptFingerprint = JSON.stringify({ + eventIds, + rejected: batch.rejected, + }); + + return { + status: 200, + body: { + ok: true, + partial: batch.rejected.length > 0, + requestId, + receipt: { + type: "signalops.ingest_receipt", + receiptId: safeReference("rcpt", receiptFingerprint), + acceptedEvents: batch.events.length, + rejectedEvents: batch.rejected.length, + storedEvents: writeResult.storedEvents, + duplicateEvents: writeResult.duplicateEvents, + evictedEvents: writeResult.evictedEvents, + retainedEvents: writeResult.retainedEvents, + eventRefs: eventIds.map((eventId) => safeReference("evtref", eventId)), + storedEventRefs: writeResult.storedEventIds.map((eventId) => + safeReference("evtref", eventId), + ), + duplicateEventRefs: writeResult.duplicateEventIds.map((eventId) => + safeReference("evtref", eventId), + ), + storage: { + adapter: options.sink.adapter, + durable: options.sink.durable, + capacity: options.sink.capacity, + }, + }, + rejected: batch.rejected, + }, + }; + } catch { + return errorResponse(500, requestId, "internal_error", "Unexpected ingest error."); + } +} diff --git a/tsconfig.json b/tsconfig.json index cf9c65d..2e1433a 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -9,6 +9,7 @@ "esModuleInterop": true, "module": "esnext", "moduleResolution": "bundler", + "allowImportingTsExtensions": true, "resolveJsonModule": true, "isolatedModules": true, "jsx": "react-jsx",