diff --git a/docs/current/SPECIFICATION.md b/docs/current/SPECIFICATION.md index d0a21a7..9fdddfa 100644 --- a/docs/current/SPECIFICATION.md +++ b/docs/current/SPECIFICATION.md @@ -386,6 +386,8 @@ The subscription sits in the `pending_verification` status until WebSub intent v | `POST` | `/eep/subscriptions/{subscription_id}/pause` | Pause an `active` subscription | `200` + subscription object | `write:subscriptions` | | `POST` | `/eep/subscriptions/{subscription_id}/resume` | Resume a `paused` subscription | `200` + subscription object | `write:subscriptions` | | `POST` | `/eep/subscriptions/{subscription_id}/test` | Trigger a synthetic test delivery | `202` | `write:subscriptions` | +| `POST` | `/eep/subscriptions/{subscription_id}/redeliver` | Re-send specific events (§5.1.2) | `202` | `write:subscriptions` | +| `GET` | `/eep/subscriptions/{subscription_id}/delivery-log` | Per-attempt delivery history (§5.1.2) | `200` | `read:subscriptions` | Publishers MUST NOT return the `delivery_secret` on any of these responses; it is disclosed exactly once, in the `201` body of the creating `POST /eep/subscribe`. @@ -397,6 +399,74 @@ This table makes normative the API that [How to subscribe](../guides/how-to-subs > **Compatibility note (v0.1).** Reference middleware released before this section served the member operations under `/eep/subscribe/{subscription_id}`. Implementations SHOULD continue to accept those paths as deprecated aliases through the `0.1.x` line and SHOULD advertise only the `/eep/subscriptions` forms. +#### 5.1.2 Event history and redelivery (normative) + +SSE subscribers recover missed events with `Last-Event-ID` (§4.3, minimum 24h +retention). Layer 3 clients recover with `system`/`replay` from a `seq` +(§6.3.1). **Webhook subscribers had no equivalent at all.** + +That gap is load-bearing. §10 moves a subscription to `paused` after repeated +delivery failures — precisely when the subscriber's endpoint was down and it +missed the most. Without catch-up, resuming produces a silent hole: the +subscription works again and the events from the outage are simply gone. + +Publishers MUST expose event history: + +```http +GET /eep/events?source={did}&since={event_id}&limit={n} +Authorization: Bearer {API_KEY} +``` + +| Parameter | Required | Meaning | +|---|---|---| +| `source` | no | Restrict to one entity; defaults to everything the caller may read | +| `since` | no | Return events strictly **after** this event `id` | +| `until` | no | Return events at or before this event `id` | +| `limit` | no | Page size; publishers MUST cap it and MUST document the cap | + +- The retention floor is the same as SSE replay: **at least 24 hours** (§4.3). +- Events MUST be returned in emission order for a given `source`. +- The response MUST include a `next_cursor` when more events remain, and MUST + omit it when they do not. A subscriber pages until `next_cursor` is absent. +- `since` naming an event outside the retention window MUST produce `410 Gone` + with the oldest retained id, so the subscriber reconciles from Layer 1 state + rather than silently believing it caught up. Publishers MUST NOT fabricate + history, and MUST NOT return an empty page for an unsatisfiable cursor — + that is indistinguishable from "you are up to date". +- Access is scoped to the caller. History MUST NOT reveal events for entities + or tiers the caller could not have subscribed to, and gate evaluation + (§3.4) applies to each event as it would have at delivery time. + +This is the endpoint metered by the "Event stream history queries" quota in +§13, which previously referred to no defined endpoint. + +**Redelivery.** Publishers SHOULD accept a targeted redelivery request: + +```http +POST /eep/subscriptions/{subscription_id}/redeliver +{ "event_ids": ["01HN3QK7GX-1708123456000"] } +``` + +Redelivered events MUST carry their original `id`, so a subscriber that already +processed one discards it by the ordinary idempotency rule +([delivery_guarantees.md](./delivery_guarantees.md) §6) rather than +double-processing. They MUST be re-signed with a current +`webhook-timestamp` (§5.3), and MUST NOT reset the subscription's failure +counter. Publishers MUST return `202 Accepted` on enqueue and MUST cap the +number of ids per request. + +**Delivery log.** Publishers MUST expose per-attempt delivery history for a +subscription: + +```http +GET /eep/subscriptions/{subscription_id}/delivery-log +``` + +The fields and the 30-day retention floor are specified in +[delivery_guarantees.md](./delivery_guarantees.md) §4. This endpoint is how a +subscriber distinguishes "the publisher never sent it" from "my endpoint +rejected it", which is otherwise unanswerable from the subscriber's side. + ### 5.2 Webhook delivery format The publisher MUST `POST` the following payload to the `delivery_url`: @@ -1217,7 +1287,7 @@ Recommended default limits per subscriber: | Subscription creation | 100/day | | SSE connections | 5 concurrent | | Webhook deliveries received | 10,000/day | -| Event stream history queries | 60/hour | +| Event stream history queries (`GET /eep/events`, §5.1.2) | 60/hour | ### 13.1 Cold-start DID trust progression (normative) diff --git a/packages/@eep-dev/middleware/src/core/eep-server.test.ts b/packages/@eep-dev/middleware/src/core/eep-server.test.ts index 61cb028..11f00c7 100644 --- a/packages/@eep-dev/middleware/src/core/eep-server.test.ts +++ b/packages/@eep-dev/middleware/src/core/eep-server.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import { parseGateConfig, type GateProof, type ProofVerifier } from "@eep-dev/gates"; import { EEPServer } from "./eep-server.js"; +import { InMemoryEventStore } from "./event-store.js"; import { TEST_DELIVERY_EVENT_TYPE, DEFAULT_LEASE_SECONDS, @@ -303,7 +304,7 @@ describe("EEPServer", () => { did: "did:web:example.com" }); const routes = server.getRouteDefinitions(); - expect(routes.length).toBe(18); + expect(routes.length).toBe(21); const operationIds = routes.map((route) => route.operationId); expect(operationIds).toContain("subscribe"); expect(operationIds).toContain("subscriptionStatus"); @@ -312,6 +313,9 @@ describe("EEPServer", () => { expect(operationIds).toContain("resumeSubscription"); expect(operationIds).toContain("pauseSubscription"); expect(operationIds).toContain("testSubscriptionDelivery"); + expect(operationIds).toContain("eventHistory"); + expect(operationIds).toContain("redeliver"); + expect(operationIds).toContain("deliveryLog"); }); // SPECIFICATION.md §5.1.1: creation stays on `POST /eep/subscribe` (it is @@ -600,6 +604,193 @@ describe("EEPServer", () => { }); }); + // SPECIFICATION.md §5.1.2 — webhooks had no catch-up mechanism at all, + // which bites hardest right after §10 pauses a subscription: the endpoint + // was down, it missed the most, and resuming produced a silent hole. + describe("event history and redelivery (§5.1.2)", () => { + const makeServer = () => { + const db = new RecordingDBAdapter(); + const bus = new RecordingEventBusAdapter(); + const store = new InMemoryEventStore(); + const server = new EEPServer({ + baseUrl: "https://api.example.com", + did: "did:web:example.com", + dbAdapter: db, + eventBusAdapter: bus, + eventStore: store + }); + return { db, bus, store, server }; + }; + + const evt = (id: string) => ({ + id, + type: "com.example.entity.updated", + source: "did:web:acme.example", + time: "2026-01-01T00:00:00.000Z", + data: {} + }); + + const createSubscription = async (server: EEPServer): Promise => { + const res = await server.getSubscribeHandler()({ + method: "POST", + path: "/eep/subscribe", + headers: {}, + body: { + source_did: "did:web:agent.example", + delivery_method: "webhook", + delivery_url: "https://hook.example/notify", + event_types: ["com.example.entity.updated"] + } + }); + return (res.body as { subscription_id: string }).subscription_id; + }; + + it("returns history for a subscriber catching up", async () => { + const { server } = makeServer(); + for (const id of ["e1", "e2", "e3"]) await server.recordEvent(evt(id)); + + const res = await server.getEventHistoryHandler()({ + method: "GET", + path: "/eep/events", + headers: {}, + query: { since: "e1" } + }); + expect(res.status).toBe(200); + const body = res.body as { events: Array<{ id: string }>; next_cursor?: string }; + expect(body.events.map((e) => e.id)).toEqual(["e2", "e3"]); + expect(body.next_cursor).toBeUndefined(); + }); + + // 410, not an empty 200: an empty page is indistinguishable from "you are + // up to date", which would let a subscriber believe it caught up. + it("returns 410 with the oldest retained id for an unsatisfiable cursor", async () => { + const { server } = makeServer(); + await server.recordEvent(evt("e1")); + const res = await server.getEventHistoryHandler()({ + method: "GET", + path: "/eep/events", + headers: {}, + query: { since: "evicted" } + }); + expect(res.status).toBe(410); + expect(res.body).toMatchObject({ + error: "retention_window_exceeded", + oldest_retained_event_id: "e1" + }); + }); + + it("pages with next_cursor", async () => { + const { server } = makeServer(); + for (const id of ["e1", "e2", "e3"]) await server.recordEvent(evt(id)); + const res = await server.getEventHistoryHandler()({ + method: "GET", + path: "/eep/events", + headers: {}, + query: { limit: "2" } + }); + const body = res.body as { events: Array<{ id: string }>; next_cursor?: string }; + expect(body.events).toHaveLength(2); + expect(body.next_cursor).toBe("e2"); + }); + + it("re-publishes requested events and names the ones it cannot", async () => { + const { server, bus } = makeServer(); + await server.recordEvent(evt("e1")); + const id = await createSubscription(server); + bus.published.length = 0; + + const res = await server.getRedeliverHandler()({ + method: "POST", + path: `/eep/subscriptions/${id}/redeliver`, + headers: {}, + params: { subscriptionId: id }, + body: { event_ids: ["e1", "gone"] } + }); + + expect(res.status).toBe(202); + expect(res.body).toMatchObject({ redelivered: ["e1"], unavailable: ["gone"] }); + // Redelivered events keep their original id so an already-processed + // event is discarded by the ordinary idempotency rule. + expect(bus.events.at(-1)).toMatchObject({ type: "com.example.entity.updated" }); + }); + + it("rejects a redeliver request with no event ids", async () => { + const { server } = makeServer(); + const id = await createSubscription(server); + const res = await server.getRedeliverHandler()({ + method: "POST", + path: `/eep/subscriptions/${id}/redeliver`, + headers: {}, + params: { subscriptionId: id }, + body: { event_ids: [] } + }); + expect(res.status).toBe(400); + }); + + it("rejects a redeliver request above the per-request cap", async () => { + const { server } = makeServer(); + const id = await createSubscription(server); + const res = await server.getRedeliverHandler()({ + method: "POST", + path: `/eep/subscriptions/${id}/redeliver`, + headers: {}, + params: { subscriptionId: id }, + body: { event_ids: Array.from({ length: 101 }, (_, i) => `e${i}`) } + }); + expect(res.status).toBe(400); + }); + + it("returns 404 when redelivering for an unknown subscription", async () => { + const { server } = makeServer(); + const res = await server.getRedeliverHandler()({ + method: "POST", + path: "/eep/subscriptions/nope/redeliver", + headers: {}, + params: { subscriptionId: "nope" }, + body: { event_ids: ["e1"] } + }); + expect(res.status).toBe(404); + }); + + it("exposes the delivery log for a subscription", async () => { + const { server, store } = makeServer(); + const id = await createSubscription(server); + await store.recordDelivery({ + subscription_id: id, + event_id: "e1", + attempt: 1, + timestamp: new Date().toISOString(), + status_code: 500, + response_time_ms: 42, + final_status: "failed" + }); + + const res = await server.getDeliveryLogHandler()({ + method: "GET", + path: `/eep/subscriptions/${id}/delivery-log`, + headers: {}, + params: { subscriptionId: id } + }); + expect(res.status).toBe(200); + const body = res.body as { attempts: Array<{ final_status: string }> }; + expect(body.attempts).toHaveLength(1); + // This is what lets a subscriber tell "never sent" from "my endpoint + // rejected it". + expect(body.attempts[0]?.final_status).toBe("failed"); + }); + + it("returns 404 for the delivery log of an unknown subscription", async () => { + const { server } = makeServer(); + const res = await server.getDeliveryLogHandler()({ + method: "GET", + path: "/eep/subscriptions/nope/delivery-log", + headers: {}, + params: { subscriptionId: "nope" } + }); + expect(res.status).toBe(404); + }); + }); + // SPECIFICATION.md §3.2.1 — Layer 1 is the polled surface, and nothing // emitted ETag or honoured If-None-Match, so every poll re-downloaded the // whole document. diff --git a/packages/@eep-dev/middleware/src/core/eep-server.ts b/packages/@eep-dev/middleware/src/core/eep-server.ts index dc2b59a..aa2db6a 100644 --- a/packages/@eep-dev/middleware/src/core/eep-server.ts +++ b/packages/@eep-dev/middleware/src/core/eep-server.ts @@ -11,6 +11,12 @@ import { } from "@eep-dev/gates"; import { SSRFError, validateEventTypePattern, validateSSRF } from "@eep-dev/validator"; import { withConditional } from "./conditional.js"; +import { + InMemoryEventStore, + RetentionWindowExceededError, + clampLimit, + type EventStore +} from "./event-store.js"; import { TEST_DELIVERY_EVENT_TYPE, DEFAULT_LEASE_SECONDS, @@ -40,6 +46,12 @@ export type EEPServerOptions = { eventBusAdapter?: EventBusAdapter; dbAdapter?: DBAdapter; proofVerifiers?: ProofVerifier[]; + /** + * Retained event history and delivery log (SPECIFICATION.md §5.1.2). + * Defaults to an in-process store; back it with durable storage to + * survive a restart. + */ + eventStore?: EventStore; }; class InMemoryDBAdapter implements DBAdapter { @@ -115,6 +127,9 @@ function clampLeaseSeconds(requested: unknown): number { return seconds; } +/** Per-request cap on redelivery ids (§5.1.2). */ +const MAX_REDELIVER_IDS = 100; + const DEFAULT_GATE_CONFIG = parseGateConfig({ default_tier: "public", tiers: { @@ -135,6 +150,7 @@ export class EEPServer { private readonly dbAdapter: DBAdapter; private readonly eventBusAdapter: EventBusAdapter; private readonly verifierRegistry: ProofVerifierRegistry; + private readonly eventStore: EventStore; constructor(options: EEPServerOptions) { this.baseUrl = options.baseUrl.replace(/\/+$/, ""); @@ -145,6 +161,7 @@ export class EEPServer { this.authAdapter = options.authAdapter ?? new HeaderProofAuthAdapter(); this.dbAdapter = options.dbAdapter ?? new InMemoryDBAdapter(); this.eventBusAdapter = options.eventBusAdapter ?? new NullEventBusAdapter(); + this.eventStore = options.eventStore ?? new InMemoryEventStore(); this.verifierRegistry = new ProofVerifierRegistry(); for (const verifier of options.proofVerifiers ?? []) { this.verifierRegistry.register(verifier); @@ -637,6 +654,134 @@ export class EEPServer { })); } + + /** + * `GET /eep/events` — event history for catch-up (§5.1.2). + * + * This is what a webhook subscriber uses after an outage: SSE has + * `Last-Event-ID` and Layer 3 has `system`/`replay`, and webhooks had + * nothing, so resuming a `paused` subscription silently lost the events + * from the outage that caused the pause. + */ + getEventHistoryHandler(): RequestHandler { + return async (request) => { + const query = request.query ?? {}; + const limitRaw = query.limit === undefined ? undefined : Number.parseInt(query.limit, 10); + try { + const page = await this.eventStore.history({ + source: query.source, + since: query.since, + until: query.until, + limit: clampLimit(limitRaw) + }); + return { status: 200, body: page }; + } catch (err) { + if (err instanceof RetentionWindowExceededError) { + // 410, not an empty page: an empty page is indistinguishable from + // "you are up to date", which would let the subscriber believe it + // caught up while silently losing events. + return { + status: 410, + body: { + error: "retention_window_exceeded", + message: err.message, + oldest_retained_event_id: err.oldestRetainedId + } + }; + } + throw err; + } + }; + } + + /** + * `POST /eep/subscriptions/:subscriptionId/redeliver` — re-send specific + * events (§5.1.2). + * + * Redelivered events keep their original `id` so a subscriber that already + * processed one discards it by the ordinary idempotency rule rather than + * double-processing. + */ + getRedeliverHandler(): RequestHandler { + return async (request) => { + const subscriptionId = request.params?.subscriptionId; + if (!subscriptionId) { + return { status: 400, body: { error: "invalid_request", message: "subscription_id is required" } }; + } + const subscription = await this.dbAdapter.getSubscription(subscriptionId); + if (!subscription) { + return { status: 404, body: { error: "not_found", message: `subscription ${subscriptionId} does not exist` } }; + } + + const body = (request.body ?? {}) as { event_ids?: unknown }; + const ids = Array.isArray(body.event_ids) + ? body.event_ids.filter((id): id is string => typeof id === "string") + : []; + if (ids.length === 0) { + return { status: 400, body: { error: "invalid_request", message: "event_ids must be a non-empty array of event ids" } }; + } + if (ids.length > MAX_REDELIVER_IDS) { + return { + status: 400, + body: { + error: "invalid_request", + message: `event_ids exceeds the per-request cap of ${MAX_REDELIVER_IDS}` + } + }; + } + + const events = await this.eventStore.getByIds(ids); + const found = new Set(events.map((e) => e.id)); + const missing = ids.filter((id) => !found.has(id)); + + for (const event of events) { + await this.eventBusAdapter.publish(event); + } + + return { + status: 202, + body: { + status: "accepted", + subscription_id: subscriptionId, + redelivered: events.map((e) => e.id), + // Named explicitly rather than silently dropped: an id outside the + // retention window is something the subscriber needs to know about. + unavailable: missing + } + }; + }; + } + + /** `GET /eep/subscriptions/:subscriptionId/delivery-log` (§5.1.2). */ + getDeliveryLogHandler(): RequestHandler { + return async (request) => { + const subscriptionId = request.params?.subscriptionId; + if (!subscriptionId) { + return { status: 400, body: { error: "invalid_request", message: "subscription_id is required" } }; + } + const subscription = await this.dbAdapter.getSubscription(subscriptionId); + if (!subscription) { + return { status: 404, body: { error: "not_found", message: `subscription ${subscriptionId} does not exist` } }; + } + const limitRaw = request.query?.limit; + const entries = await this.eventStore.deliveryLog( + subscriptionId, + limitRaw === undefined ? undefined : Number.parseInt(limitRaw, 10) + ); + return { status: 200, body: { subscription_id: subscriptionId, attempts: entries } }; + }; + } + + /** Record an event in history so it is available for catch-up (§5.1.2). */ + async recordEvent(event: CloudEvent): Promise { + await this.eventStore.append(event); + } + + /** The store backing history and the delivery log. */ + getEventStore(): EventStore { + return this.eventStore; + } + getRouteDefinitions(): RouteDefinition[] { return [ { method: "GET", path: "/.well-known/eep.json", operationId: "manifest", handler: this.getManifestHandler() }, @@ -662,6 +807,9 @@ export class EEPServer { // under `/eep/subscribe/:id` before §5.1.1 existed. Remove at 0.2. { method: "GET", path: "/eep/subscribe/:subscriptionId", operationId: "subscriptionStatusDeprecated", handler: this.getSubscriptionStatusHandler() }, { method: "DELETE", path: "/eep/subscribe/:subscriptionId", operationId: "unsubscribeDeprecated", handler: this.getUnsubscribeHandler() }, + { method: "GET", path: "/eep/events", operationId: "eventHistory", handler: this.getEventHistoryHandler() }, + { method: "POST", path: "/eep/subscriptions/:subscriptionId/redeliver", operationId: "redeliver", handler: this.getRedeliverHandler() }, + { method: "GET", path: "/eep/subscriptions/:subscriptionId/delivery-log", operationId: "deliveryLog", handler: this.getDeliveryLogHandler() }, { method: "GET", path: "/eep/audit-log", operationId: "auditLog", handler: this.getAuditLogHandler() }, { method: "GET", path: "/eep/pulse", operationId: "pulseUpgrade", handler: this.getPulseUpgradeHandler() } ]; diff --git a/packages/@eep-dev/middleware/src/core/event-store.test.ts b/packages/@eep-dev/middleware/src/core/event-store.test.ts new file mode 100644 index 0000000..0969b5b --- /dev/null +++ b/packages/@eep-dev/middleware/src/core/event-store.test.ts @@ -0,0 +1,183 @@ +import { describe, expect, it } from "vitest"; +import { + InMemoryEventStore, + RetentionWindowExceededError, + clampLimit, + DEFAULT_HISTORY_LIMIT, + MAX_HISTORY_LIMIT, +} from "./event-store.js"; +import type { CloudEvent } from "./request-handler.js"; + +function event(id: string, source = "did:web:acme.example"): CloudEvent { + return { + id, + type: "com.example.entity.updated", + source, + time: "2026-01-01T00:00:00.000Z", + data: {}, + }; +} + +async function seeded(ids: string[], source?: string) { + const store = new InMemoryEventStore(); + for (const id of ids) await store.append(event(id, source)); + return store; +} + +describe("clampLimit", () => { + it("defaults for absent or nonsensical values", () => { + expect(clampLimit(undefined)).toBe(DEFAULT_HISTORY_LIMIT); + expect(clampLimit(Number.NaN)).toBe(DEFAULT_HISTORY_LIMIT); + expect(clampLimit(0)).toBe(DEFAULT_HISTORY_LIMIT); + expect(clampLimit(-5)).toBe(DEFAULT_HISTORY_LIMIT); + }); + + it("caps at the maximum", () => { + expect(clampLimit(999_999)).toBe(MAX_HISTORY_LIMIT); + }); + + it("honours a sensible value", () => { + expect(clampLimit(25)).toBe(25); + }); +}); + +describe("InMemoryEventStore history (§5.1.2)", () => { + it("returns events in emission order", async () => { + const store = await seeded(["e1", "e2", "e3"]); + const page = await store.history({}); + expect(page.events.map((e) => e.id)).toEqual(["e1", "e2", "e3"]); + }); + + it("returns events strictly after `since`", async () => { + const store = await seeded(["e1", "e2", "e3"]); + const page = await store.history({ since: "e1" }); + // Strictly after: the cursor event itself is not repeated. + expect(page.events.map((e) => e.id)).toEqual(["e2", "e3"]); + }); + + it("omits next_cursor when the subscriber is caught up", async () => { + const store = await seeded(["e1", "e2"]); + const page = await store.history({}); + // Absence of the cursor is how a subscriber knows it is up to date. + expect(page.next_cursor).toBeUndefined(); + }); + + it("emits next_cursor when more events remain, and pages to the end", async () => { + const store = await seeded(["e1", "e2", "e3", "e4"]); + const first = await store.history({ limit: 2 }); + expect(first.events.map((e) => e.id)).toEqual(["e1", "e2"]); + expect(first.next_cursor).toBe("e2"); + + const second = await store.history({ since: first.next_cursor, limit: 2 }); + expect(second.events.map((e) => e.id)).toEqual(["e3", "e4"]); + expect(second.next_cursor).toBeUndefined(); + }); + + it("filters by source", async () => { + const store = new InMemoryEventStore(); + await store.append(event("a1", "did:web:a.example")); + await store.append(event("b1", "did:web:b.example")); + await store.append(event("a2", "did:web:a.example")); + const page = await store.history({ source: "did:web:a.example" }); + expect(page.events.map((e) => e.id)).toEqual(["a1", "a2"]); + }); + + it("honours `until` as an inclusive upper bound", async () => { + const store = await seeded(["e1", "e2", "e3"]); + const page = await store.history({ until: "e2" }); + expect(page.events.map((e) => e.id)).toEqual(["e1", "e2"]); + }); + + // An empty page is indistinguishable from "you are up to date", which + // would let a subscriber believe it caught up while silently losing + // events. §5.1.2 requires 410 instead. + it("throws rather than returning an empty page for an unsatisfiable cursor", async () => { + const store = await seeded(["e1", "e2"]); + await expect(store.history({ since: "evicted-long-ago" })).rejects.toBeInstanceOf( + RetentionWindowExceededError + ); + }); + + it("names the oldest retained event so the subscriber can reconcile", async () => { + const store = await seeded(["e1", "e2"]); + await store.history({ since: "gone" }).catch((err: RetentionWindowExceededError) => { + expect(err.oldestRetainedId).toBe("e1"); + }); + expect.assertions(1); + }); + + it("reports a null oldest id when nothing is retained", async () => { + const store = new InMemoryEventStore(); + await store.history({ since: "gone" }).catch((err: RetentionWindowExceededError) => { + expect(err.oldestRetainedId).toBeNull(); + }); + expect.assertions(1); + }); + + it("prunes events past the retention window", async () => { + const store = new InMemoryEventStore(0); + await store.append(event("e1")); + // The cutoff is `now - retention`, so with zero retention the event is + // only strictly older than the cutoff once the clock has moved. + await new Promise((resolve) => setTimeout(resolve, 5)); + const page = await store.history({}); + expect(page.events).toEqual([]); + }); +}); + +describe("InMemoryEventStore getByIds", () => { + it("returns only the requested events", async () => { + const store = await seeded(["e1", "e2", "e3"]); + const found = await store.getByIds(["e1", "e3"]); + expect(found.map((e) => e.id)).toEqual(["e1", "e3"]); + }); + + it("silently omits ids it does not hold, so callers can report them", async () => { + const store = await seeded(["e1"]); + const found = await store.getByIds(["e1", "missing"]); + expect(found.map((e) => e.id)).toEqual(["e1"]); + }); +}); + +describe("InMemoryEventStore delivery log (delivery_guarantees.md §4)", () => { + const entry = (overrides: Partial[0]> = {}) => ({ + subscription_id: "sub_1", + event_id: "e1", + attempt: 1, + timestamp: new Date().toISOString(), + status_code: 200, + response_time_ms: 12, + final_status: "delivered" as const, + ...overrides, + }); + + it("records and returns attempts for a subscription", async () => { + const store = new InMemoryEventStore(); + await store.recordDelivery(entry()); + const log = await store.deliveryLog("sub_1"); + expect(log).toHaveLength(1); + expect(log[0]).toMatchObject({ event_id: "e1", final_status: "delivered" }); + }); + + it("scopes the log to the requested subscription", async () => { + const store = new InMemoryEventStore(); + await store.recordDelivery(entry({ subscription_id: "sub_1" })); + await store.recordDelivery(entry({ subscription_id: "sub_2" })); + expect(await store.deliveryLog("sub_1")).toHaveLength(1); + }); + + it("returns the most recent attempts when limited", async () => { + const store = new InMemoryEventStore(); + for (let i = 1; i <= 5; i++) { + await store.recordDelivery(entry({ event_id: `e${i}` })); + } + const log = await store.deliveryLog("sub_1", 2); + expect(log.map((e) => e.event_id)).toEqual(["e4", "e5"]); + }); + + it("prunes attempts past the retention window", async () => { + const store = new InMemoryEventStore(undefined, 0); + await store.recordDelivery(entry({ timestamp: new Date(Date.now() - 1000).toISOString() })); + expect(await store.deliveryLog("sub_1")).toEqual([]); + }); +}); diff --git a/packages/@eep-dev/middleware/src/core/event-store.ts b/packages/@eep-dev/middleware/src/core/event-store.ts new file mode 100644 index 0000000..140e63c --- /dev/null +++ b/packages/@eep-dev/middleware/src/core/event-store.ts @@ -0,0 +1,184 @@ +/** + * Event history and delivery-log storage (SPECIFICATION.md §5.1.2). + * + * SSE subscribers recover missed events with `Last-Event-ID`; Layer 3 clients + * recover with `system`/`replay` from a `seq`. Webhook subscribers had no + * equivalent, which mattered most in exactly the case §10 creates: a + * subscription is `paused` after repeated failures — that is, after the + * subscriber's endpoint was down and it missed the most — and resuming + * produced a silent hole. + */ +import type { CloudEvent } from "./request-handler.js"; + +/** One retained event, plus the ordering key history pages on. */ +export interface StoredEvent { + /** Monotonic, publisher-assigned. Ordering is by this, not by `time`. */ + seq: number; + event: CloudEvent; + /** Epoch millis at which this event was stored, for retention pruning. */ + stored_at: number; +} + +/** One delivery attempt, per delivery_guarantees.md §4. */ +export interface DeliveryLogEntry { + subscription_id: string; + event_id: string; + /** 1-based attempt number within the retry schedule. */ + attempt: number; + /** RFC 3339. */ + timestamp: string; + /** HTTP status of the attempt, absent when the request never completed. */ + status_code?: number; + response_time_ms: number; + final_status: "delivered" | "failed" | "undeliverable"; +} + +export interface EventHistoryPage { + events: CloudEvent[]; + /** Present only when more events remain. Absent means caught up. */ + next_cursor?: string; +} + +/** + * Thrown when `since` names an event older than the retention window. + * + * Surfaced as `410 Gone` rather than an empty page: an empty page is + * indistinguishable from "you are up to date", which would let a subscriber + * believe it caught up when it silently lost events. + */ +export class RetentionWindowExceededError extends Error { + constructor(public readonly oldestRetainedId: string | null) { + super( + oldestRetainedId + ? `cursor is older than the retention window; oldest retained event is ${oldestRetainedId}` + : "cursor is older than the retention window and no events are retained" + ); + this.name = "RetentionWindowExceededError"; + } +} + +export interface EventStore { + append(event: CloudEvent): Promise; + /** Events strictly after `since`, in emission order. */ + history(options: { + source?: string; + since?: string; + until?: string; + limit?: number; + }): Promise; + getByIds(eventIds: string[]): Promise; + recordDelivery(entry: DeliveryLogEntry): Promise; + deliveryLog(subscriptionId: string, limit?: number): Promise; +} + +/** §5.1.2 retention floor, matching SSE replay (§4.3). */ +export const MIN_EVENT_RETENTION_MS = 24 * 60 * 60 * 1000; +/** delivery_guarantees.md §4 retention floor for the delivery log. */ +export const MIN_DELIVERY_LOG_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; +export const DEFAULT_HISTORY_LIMIT = 100; +export const MAX_HISTORY_LIMIT = 1000; + +/** + * In-memory reference implementation. + * + * Suitable for a single process and for tests. A deployment that must survive + * a restart backs this interface with the same store its subscriptions live + * in — the interface, not this class, is what §5.1.2 requires. + */ +export class InMemoryEventStore implements EventStore { + private readonly events: StoredEvent[] = []; + private readonly deliveries: DeliveryLogEntry[] = []; + private nextSeq = 1; + + constructor( + private readonly retentionMs: number = MIN_EVENT_RETENTION_MS, + private readonly deliveryLogRetentionMs: number = MIN_DELIVERY_LOG_RETENTION_MS + ) {} + + async append(event: CloudEvent): Promise { + this.events.push({ seq: this.nextSeq++, event, stored_at: Date.now() }); + this.prune(); + } + + async history(options: { + source?: string; + since?: string; + until?: string; + limit?: number; + }): Promise { + this.prune(); + + let startSeq = 0; + if (options.since !== undefined) { + const cursor = this.events.find((e) => e.event.id === options.since); + if (!cursor) { + // The cursor is either older than retention or was never ours. + // Either way the subscriber cannot trust a page built from it. + throw new RetentionWindowExceededError(this.events[0]?.event.id ?? null); + } + startSeq = cursor.seq; + } + + let endSeq = Number.POSITIVE_INFINITY; + if (options.until !== undefined) { + const cursor = this.events.find((e) => e.event.id === options.until); + if (cursor) endSeq = cursor.seq; + } + + const limit = clampLimit(options.limit); + const matching = this.events.filter( + (e) => + e.seq > startSeq && + e.seq <= endSeq && + (options.source === undefined || e.event.source === options.source) + ); + + const page = matching.slice(0, limit); + const hasMore = matching.length > page.length; + return { + events: page.map((e) => e.event), + // Only present when more remain; its absence is how a subscriber + // knows it is caught up. + ...(hasMore ? { next_cursor: page[page.length - 1]!.event.id } : {}), + }; + } + + async getByIds(eventIds: string[]): Promise { + this.prune(); + const wanted = new Set(eventIds); + return this.events.filter((e) => wanted.has(e.event.id)).map((e) => e.event); + } + + async recordDelivery(entry: DeliveryLogEntry): Promise { + this.deliveries.push(entry); + this.prune(); + } + + async deliveryLog(subscriptionId: string, limit = DEFAULT_HISTORY_LIMIT): Promise { + this.prune(); + return this.deliveries + .filter((d) => d.subscription_id === subscriptionId) + .slice(-clampLimit(limit)); + } + + private prune(): void { + const eventCutoff = Date.now() - this.retentionMs; + while (this.events.length > 0 && this.events[0]!.stored_at < eventCutoff) { + this.events.shift(); + } + const deliveryCutoff = Date.now() - this.deliveryLogRetentionMs; + while ( + this.deliveries.length > 0 && + Date.parse(this.deliveries[0]!.timestamp) < deliveryCutoff + ) { + this.deliveries.shift(); + } + } +} + +export function clampLimit(limit: unknown): number { + if (typeof limit !== "number" || !Number.isFinite(limit) || limit < 1) { + return DEFAULT_HISTORY_LIMIT; + } + return Math.min(Math.floor(limit), MAX_HISTORY_LIMIT); +} diff --git a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts index ba2649f..bb72d8b 100644 --- a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts +++ b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts @@ -1,6 +1,7 @@ import { EEPSigner } from "@eep-dev/signer"; import { matchesAnyPattern } from "@eep-dev/validator"; import { TEST_DELIVERY_EVENT_TYPE } from "../core/request-handler.js"; +import type { EventStore } from "../core/event-store.js"; import type { CloudEvent, DBAdapter, @@ -89,6 +90,13 @@ export type WebhookDispatcherOptions = { deliveryTimeoutMs?: number; /** Optional observer invoked once per subscription after its retries end. */ onDeliveryResult?: (result: DeliveryResult) => void; + /** + * Records every delivery attempt for `GET + * /eep/subscriptions/:id/delivery-log` (SPECIFICATION.md §5.1.2). This is + * how a subscriber tells "the publisher never sent it" apart from "my + * endpoint rejected it" — otherwise unanswerable from its side. + */ + eventStore?: EventStore; }; const defaultHttpClient: WebhookHttpClient = async (url, { headers, body, signal }) => { @@ -150,6 +158,7 @@ export class WebhookDispatcher { private readonly pauseAfterFailures: number; private readonly deliveryTimeoutMs: number; private readonly onDeliveryResult?: (result: DeliveryResult) => void; + private readonly eventStore?: EventStore; private stopped = false; constructor(options: WebhookDispatcherOptions) { @@ -163,6 +172,7 @@ export class WebhookDispatcher { this.pauseAfterFailures = options.pauseAfterFailures ?? DEFAULT_PAUSE_AFTER_FAILURES; this.deliveryTimeoutMs = options.deliveryTimeoutMs ?? DEFAULT_DELIVERY_TIMEOUT_MS; this.onDeliveryResult = options.onDeliveryResult; + this.eventStore = options.eventStore; } /** @@ -246,8 +256,12 @@ export class WebhookDispatcher { }); } + const startedAt = Date.now(); const outcome = await this.attemptDelivery(event, sub); lastStatus = outcome.status; + await this.logAttempt(sub, event, attempt + 1, outcome, Date.now() - startedAt, { + isFinalAttempt: attempt === this.retrySchedule.length - 1 + }); if (outcome.ok) { await this.recordSuccess(sub.subscription_id); return this.report({ @@ -324,6 +338,41 @@ export class WebhookDispatcher { } /** Reset the consecutive-failure counter after a delivery lands. */ + /** + * Append one attempt to the delivery log. + * + * `undeliverable` is reserved for the last attempt of an exhausted schedule: + * an interim failure is `failed` because a later attempt may still succeed, + * and conflating them would make the log read as if every retry were fatal. + */ + private async logAttempt( + sub: SubscriptionRecord, + event: CloudEvent, + attempt: number, + outcome: { ok: boolean; status?: number }, + elapsedMs: number, + context: { isFinalAttempt: boolean } + ): Promise { + if (!this.eventStore) return; + try { + await this.eventStore.recordDelivery({ + subscription_id: sub.subscription_id, + event_id: event.id, + attempt, + timestamp: new Date().toISOString(), + ...(outcome.status === undefined ? {} : { status_code: outcome.status }), + response_time_ms: elapsedMs, + final_status: outcome.ok + ? "delivered" + : context.isFinalAttempt + ? "undeliverable" + : "failed" + }); + } catch { + // Observability must never break delivery. + } + } + private async recordSuccess(subscriptionId: string): Promise { const current = await this.db.getSubscription(subscriptionId); if (current && current.failure_count > 0) { diff --git a/packages/@eep-dev/middleware/src/index.ts b/packages/@eep-dev/middleware/src/index.ts index 12d5e46..5338561 100644 --- a/packages/@eep-dev/middleware/src/index.ts +++ b/packages/@eep-dev/middleware/src/index.ts @@ -47,5 +47,18 @@ export { InMemoryEventBusAdapter } from "./event-bus/in-memory.js"; export { RedisEventBusAdapter, type RedisClientLike } from "./event-bus/redis.js"; export { KafkaEventBusAdapter, type KafkaProducerLike, type KafkaConsumerLike } from "./event-bus/kafka.js"; +export { + InMemoryEventStore, + RetentionWindowExceededError, + MIN_EVENT_RETENTION_MS, + MIN_DELIVERY_LOG_RETENTION_MS, + DEFAULT_HISTORY_LIMIT, + MAX_HISTORY_LIMIT, + type EventStore, + type StoredEvent, + type DeliveryLogEntry, + type EventHistoryPage +} from "./core/event-store.js"; + export { InMemoryDBAdapter } from "./db/in-memory.js"; export { PostgresDBAdapter, type SQLClientLike } from "./db/postgres.js";