diff --git a/.changeset/webhook-dispatch-replay.md b/.changeset/webhook-dispatch-replay.md new file mode 100644 index 0000000..39ec471 --- /dev/null +++ b/.changeset/webhook-dispatch-replay.md @@ -0,0 +1,5 @@ +--- +"dsar": minor +--- + +Add outbound webhook dispatch recovery: `GET /webhooks/dispatches` listing with filters, single and bulk replay endpoints with idempotent audit-logged replays, and `dsar webhooks list`, `dsar webhooks replay`, `dsar webhooks replay-all`, and `dsar webhooks tail` CLI commands. diff --git a/docs/reference/api/index.mdx b/docs/reference/api/index.mdx index d0d4b35..04dc336 100644 --- a/docs/reference/api/index.mdx +++ b/docs/reference/api/index.mdx @@ -13,7 +13,7 @@ These pages describe the DSAR HTTP surface exposed by `@dsar/backend`. - [Init API](./init.md) for `POST /init` - [Status API](./status.md) for `GET /status` - [Policies API](./policies.md) for policy discovery and upgrade lifecycle -- [Webhooks API](./webhooks.md) for inbound Resend and Slack entrypoints +- [Webhooks API](./webhooks.md) for inbound Resend and Slack entrypoints plus outbound dispatch inspection/replay ## Request Lifecycle diff --git a/docs/reference/api/webhooks.mdx b/docs/reference/api/webhooks.mdx index 0e9b0a9..ef32a13 100644 --- a/docs/reference/api/webhooks.mdx +++ b/docs/reference/api/webhooks.mdx @@ -1,13 +1,14 @@ --- title: "Webhooks API" -description: "These public endpoints accept inbound provider events and map them into DSAR intake flows." +description: "Webhook endpoints accept public inbound provider events and expose protected outbound dispatch management." group: reference-api --- -These public endpoints accept inbound provider events and map them into DSAR -intake flows. +Webhook endpoints accept public inbound provider events and map them into DSAR +intake flows. Protected outbound endpoints let operator and service principals +inspect delivery attempts and replay failed webhook dispatches. ## POST /webhooks/inbound/resend @@ -86,3 +87,137 @@ Related guides: - [Inbound Resend](../../integrations/integrations/inbound-resend.md) - [Inbound Slack](../../integrations/integrations/inbound-slack.md) + +## GET /webhooks/dispatches + +List outbound webhook delivery attempts recorded by the notification delivery +ledger. + +This endpoint is protected and reserved for operator or service principals. + +Supported query parameters: + +- `status` (string, optional): comma-separated delivery statuses, such as + `failed` or `failed,pending` +- `endpoint_id` (string, optional): configured outbound webhook endpoint id +- `created_after` (string, optional): only include attempts created after this + timestamp +- `created_before` (string, optional): only include attempts created before this + timestamp +- `limit` (number, optional): page size, from 1 to 500 +- `offset` (number, optional): zero-based offset + +**Response (200):** + +```json +{ + "items": [ + { + "attempt": 3, + "createdAt": "2026-01-01T00:00:00.000Z", + "destination": "https://customer.example/webhooks/dsar", + "dispatchId": "dispatch-123", + "endpointId": "default", + "error": "500 Internal Server Error", + "eventId": "event-123", + "eventType": "request_fulfilled", + "replayable": true, + "requestId": "req-123", + "responseCode": 500, + "status": "failed" + } + ], + "limit": 50, + "offset": 0, + "total": 1 +} +``` + +## POST /webhooks/dispatches/{id}/replay + +Replay one failed outbound webhook dispatch. + +This endpoint is protected and reserved for operator or service principals. The +dispatch must be a failed webhook delivery attempt. Successful, pending, +skipped, or non-webhook attempts are not replayable. + +The `x-idempotency-key` header is required. When the same dispatch id and +idempotency key have already been accepted, the endpoint returns +`already_replayed` without sending the webhook again. + +Required headers: + +- `x-idempotency-key` (string, required): caller-provided replay dedupe key + +**Response (202):** + +```json +{ + "dispatchId": "dispatch-123", + "eventId": "event-123", + "status": "replayed" +} +``` + +## POST /webhooks/dispatches/replay + +Replay a batch of failed outbound webhook dispatches. + +This endpoint is protected and reserved for operator or service principals. Only +failed webhook delivery attempts are replayable; the endpoint always selects +`channel="webhook"` and `status="failed"` attempts in the caller's tenant. + +The `x-idempotency-key` header is required. The server derives a per-dispatch +idempotency key from the caller key and dispatch id, so retrying the same bulk +request does not re-send dispatches that were already accepted. + +Bulk replay is capped at 100 dispatches per request. Use filters to narrow large +recovery batches. + +Required headers: + +- `x-idempotency-key` (string, required): caller-provided bulk replay dedupe key + +**Request body:** + +- `status` (string, optional): must be `"failed"` when present +- `endpoint_id` (string, optional): configured outbound webhook endpoint id +- `created_after` (string, optional): only include attempts created after this + timestamp +- `created_before` (string, optional): only include attempts created before this + timestamp +- `limit` (number, optional): maximum dispatches to replay, from 1 to 100 + +An empty body replays failed webhook dispatches for the tenant up to the cap. + +**Response (202):** + +```json +{ + "alreadyReplayed": 1, + "replayed": 2, + "results": [ + { + "dispatchId": "dispatch-123", + "eventId": "event-123", + "status": "replayed" + }, + { + "dispatchId": "dispatch-456", + "eventId": "event-456", + "status": "already_replayed" + }, + { + "dispatchId": "dispatch-789", + "error": "Webhook dispatch cannot be replayed because no webhook endpoint is configured.", + "eventId": "event-789", + "status": "failed" + } + ], + "total": 3 +} +``` + +Per-dispatch delivery failures are reported in `results` and do not fail the +whole bulk request. Delivery attempts and replay audit events are recorded per +dispatch. diff --git a/docs/reference/developer/cli.mdx b/docs/reference/developer/cli.mdx index 6aa411f..7bbb530 100644 --- a/docs/reference/developer/cli.mdx +++ b/docs/reference/developer/cli.mdx @@ -52,6 +52,7 @@ Global flags: - `policies list` -> `GET /policies` - `policies custom register|activate|deactivate` -> custom policy endpoints - `webhooks inbound resend` -> `POST /webhooks/inbound/resend` +- `webhooks list|replay|replay-all|tail ...` -> outbound webhook dispatch inspection, replay, and polling - `requests create|capture` -> `POST /requests`, `POST /requests/capture` - `requests clock explain ` -> `GET /requests/{id}/clock/explain` - `requests verification ...` -> verification endpoints @@ -83,6 +84,21 @@ Notification replay is part of the current CLI surface: - `dsar requests notifications list ` - `dsar requests notifications replay ` +Outbound webhook dispatch replay is available through: + +- `dsar webhooks list --status failed` +- `dsar webhooks replay --idempotency-key replay-1` +- `dsar webhooks replay-all --status failed --endpoint-id default --limit 100 --idempotency-key replay-all-1` +- `dsar webhooks tail --status failed --endpoint-id default --interval 2000` + +`replay` targets one failed webhook dispatch. `replay-all` maps to +`POST /webhooks/dispatches/replay` and builds the bulk replay body from +`--status`, `--endpoint-id`, `--created-after`, `--created-before`, and +`--limit`. `--idempotency-key` is required for both replay commands. + +`tail` polls `GET /webhooks/dispatches` and streams newly observed dispatches. +It accepts `--status`, `--endpoint-id`, `--interval`, `--limit`, and `--once`. + ## Parity policy Any new backend OpenAPI path + method pair must include: diff --git a/packages/backend/src/adapters/contract.ts b/packages/backend/src/adapters/contract.ts index 78e1716..f8572ab 100644 --- a/packages/backend/src/adapters/contract.ts +++ b/packages/backend/src/adapters/contract.ts @@ -137,6 +137,11 @@ export interface NotificationDispatchResult { readonly error?: string; } +/** + * Notification delivery channels that a notification adapter can handle. + */ +export type NotificationAdapterChannel = "email" | "webhook"; + /** * Adapter contract for notification delivery channels (email, webhook, etc.). */ @@ -145,6 +150,8 @@ export interface NotificationAdapterContract extends AdapterContractBase< > { /** Adapter capability handled by this contract entry. */ readonly capability: "notifications"; + /** Explicit channels handled by this adapter. Omit only for legacy adapters. */ + readonly channels?: readonly NotificationAdapterChannel[]; /** Sends a notification message through the adapter provider. */ readonly send: ( input: NotificationDispatchInput diff --git a/packages/backend/src/audit/service.ts b/packages/backend/src/audit/service.ts index c9df25d..db4bf6c 100644 --- a/packages/backend/src/audit/service.ts +++ b/packages/backend/src/audit/service.ts @@ -55,6 +55,8 @@ const toCauseDetails = (error: unknown): Readonly> => { * audit export and hash-chain verification remain deterministic. */ export interface AppendAuditInput { + /** Optional deterministic identifier for idempotent audit markers. */ + readonly id?: string; /** * Request identifier for request-scoped events. * @@ -145,7 +147,7 @@ export const appendAuditEvent = ( sequence, }); - const id = makeRequestId(); + const id = input.id ?? makeRequestId(); yield* services.repos.persistence.auditEvents .append({ action: input.action, diff --git a/packages/backend/src/http-api/groups/webhooks.ts b/packages/backend/src/http-api/groups/webhooks.ts index a2e6170..0c55754 100644 --- a/packages/backend/src/http-api/groups/webhooks.ts +++ b/packages/backend/src/http-api/groups/webhooks.ts @@ -27,6 +27,56 @@ const WebhookRotateKeyResponseSchema = Schema.Struct({ previousKeyId: Schema.optional(Schema.String), }); +const WebhookDispatchSchema = Schema.Struct({ + attempt: Schema.Number, + createdAt: Schema.String, + destination: Schema.String, + dispatchId: Schema.String, + endpointId: Schema.optional(Schema.String), + error: Schema.optional(Schema.String), + eventId: Schema.String, + eventType: Schema.optional(Schema.String), + replayable: Schema.Boolean, + requestId: Schema.String, + responseCode: Schema.optional(Schema.Number), + status: Schema.Literals(["pending", "delivered", "failed", "skipped"]), +}); + +const WebhookDispatchListResponseSchema = Schema.Struct({ + items: Schema.Array(WebhookDispatchSchema), + limit: Schema.Number, + offset: Schema.Number, + total: Schema.Number, +}); + +const WebhookDispatchReplayResponseSchema = Schema.Struct({ + dispatchId: Schema.String, + eventId: Schema.String, + status: Schema.Literals(["replayed", "already_replayed"]), +}); + +const WebhookDispatchBulkReplayPayloadSchema = Schema.Struct({ + created_after: Schema.optional(Schema.String), + created_before: Schema.optional(Schema.String), + endpoint_id: Schema.optional(Schema.String), + limit: Schema.optional(Schema.Number), + status: Schema.optional(Schema.Literal("failed")), +}); + +const WebhookDispatchBulkReplayResultSchema = Schema.Struct({ + dispatchId: Schema.String, + error: Schema.optional(Schema.String), + eventId: Schema.String, + status: Schema.Literals(["replayed", "already_replayed", "failed"]), +}); + +const WebhookDispatchBulkReplayResponseSchema = Schema.Struct({ + alreadyReplayed: Schema.Number, + replayed: Schema.Number, + results: Schema.Array(WebhookDispatchBulkReplayResultSchema), + total: Schema.Number, +}); + /** OpenAPI group describing public inbound webhook endpoints. */ export const webhooksGroup = HttpApiGroup.make("webhooks", { topLevel: true }) .add( @@ -81,4 +131,52 @@ export const webhooksGroup = HttpApiGroup.make("webhooks", { topLevel: true }) ), "Rotate webhook endpoint signing key" ) + ) + .add( + protectedOperation( + HttpApiEndpoint.get("webhooks_dispatches_list", "/webhooks/dispatches", { + query: { + created_after: Schema.optional(Schema.String), + created_before: Schema.optional(Schema.String), + endpoint_id: Schema.optional(Schema.String), + limit: Schema.optional(Schema.NumberFromString), + offset: Schema.optional(Schema.NumberFromString), + status: Schema.optional(Schema.String), + }, + success: successEnvelope(WebhookDispatchListResponseSchema), + }), + "List outbound webhook dispatches" + ) + ) + .add( + protectedOperation( + HttpApiEndpoint.post( + "webhooks_dispatches_replay_bulk", + "/webhooks/dispatches/replay", + { + headers: { "x-idempotency-key": Schema.String }, + payload: WebhookDispatchBulkReplayPayloadSchema, + success: successEnvelope( + WebhookDispatchBulkReplayResponseSchema + ).pipe(s202), + } + ), + "Replay failed outbound webhook dispatches" + ) + ) + .add( + protectedOperation( + HttpApiEndpoint.post( + "webhooks_dispatches_replay", + "/webhooks/dispatches/:id/replay", + { + headers: { "x-idempotency-key": Schema.String }, + params: { id: Schema.String }, + success: successEnvelope(WebhookDispatchReplayResponseSchema).pipe( + s202 + ), + } + ), + "Replay outbound webhook dispatch" + ) ); diff --git a/packages/backend/src/routes/webhooks.ts b/packages/backend/src/routes/webhooks.ts index ce98bf4..252b0d2 100644 --- a/packages/backend/src/routes/webhooks.ts +++ b/packages/backend/src/routes/webhooks.ts @@ -1,4 +1,9 @@ import type { RouteDefinition } from "./types"; +import { + bulkReplayWebhookDispatchesRoute, + listWebhookDispatchesRoute, + replayWebhookDispatchRoute, +} from "./webhooks/dispatches"; import { resendWebhookRoute } from "./webhooks/resend"; import { rotateWebhookKeyRoute } from "./webhooks/rotate-key"; import { slackWebhookRoute } from "./webhooks/slack"; @@ -11,4 +16,7 @@ export const webhookRoutes: readonly RouteDefinition[] = [ resendWebhookRoute, slackWebhookRoute, rotateWebhookKeyRoute, + listWebhookDispatchesRoute, + bulkReplayWebhookDispatchesRoute, + replayWebhookDispatchRoute, ]; diff --git a/packages/backend/src/routes/webhooks/dispatches.ts b/packages/backend/src/routes/webhooks/dispatches.ts new file mode 100644 index 0000000..20eeeda --- /dev/null +++ b/packages/backend/src/routes/webhooks/dispatches.ts @@ -0,0 +1,766 @@ +import { asRecord } from "@dsar/guards"; +import { PersistenceEntityNotFoundError, withTenant } from "@dsar/persistence"; +import type { + NotificationDeliveryAttemptRecord, + NotificationDeliveryStatus, + NotificationEventRecord, +} from "@dsar/persistence"; +import { IsoTimestampSchema } from "@dsar/schema"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; + +import { appendAuditEvent } from "../../audit/service"; +import { replayWebhookDispatch } from "../../services/notifications/service"; +import { RequestValidationError } from "../../types/errors"; +import { RuntimeServicesTag } from "../../types/runtime"; +import { + requirePrincipalKinds, + requireRequestActor, + requireRequestTenantId, +} from "../authz"; +import { accepted, ok } from "../helpers"; +import { getIdempotencyKey } from "../requests/shared"; +import type { RouteDefinition } from "../types"; + +const DEFAULT_WEBHOOK_ENDPOINT_ID = "default"; +const DEFAULT_LIMIT = 50; +const MAX_LIMIT = 500; +const BULK_REPLAY_MAX = 100; +const DISPATCH_REPLAY_ACTION = "webhook_dispatch_replayed"; +const DISPATCH_REPLAY_REQUESTED_ACTION = "webhook_dispatch_replay_requested"; +const INTEGER_PARAM_PATTERN = /^(0|[1-9]\d*)$/; + +const hasErrorTag = (error: unknown, tag: string): boolean => + typeof error === "object" && + error !== null && + "_tag" in error && + error._tag === tag; + +const parseIntParam = ( + value: string | null, + fallback: number, + min: number, + max: number +): Effect.Effect => { + if (value === null || value.trim().length === 0) { + return Effect.succeed(fallback); + } + const trimmed = value.trim(); + if (!INTEGER_PARAM_PATTERN.test(trimmed)) { + return Effect.fail( + new RequestValidationError({ + message: `Expected integer between ${min} and ${max}.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const parsed = Number.parseInt(trimmed, 10); + if (!Number.isSafeInteger(parsed) || parsed < min || parsed > max) { + return Effect.fail( + new RequestValidationError({ + message: `Expected integer between ${min} and ${max}.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.succeed(parsed); +}; + +const parseStatusFilter = ( + value: string | null +): Effect.Effect< + readonly NotificationDeliveryStatus[] | undefined, + RequestValidationError +> => { + if (!value) { + return Effect.succeed(); + } + const values = value + .split(",") + .map((entry) => entry.trim()) + .filter((entry) => entry.length > 0); + if (values.length === 0) { + return Effect.succeed(); + } + const statuses: NotificationDeliveryStatus[] = []; + for (const entry of values) { + switch (entry) { + case "delivered": + case "failed": + case "pending": + case "skipped": { + statuses.push(entry); + break; + } + default: { + return Effect.fail( + new RequestValidationError({ + message: `Unsupported webhook dispatch status '${entry}'.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + } + } + return Effect.succeed(statuses); +}; + +const normalizeIsoTimestamp = (value: string): string | undefined => { + const decoded = Schema.decodeUnknownExit(IsoTimestampSchema)(value); + if (Exit.isFailure(decoded)) { + return undefined; + } + return new Date(value).toISOString(); +}; + +const parseIsoTimestampParam = ( + value: string | null, + paramName: "created_after" | "created_before" +): Effect.Effect => { + if (!value || value.trim().length === 0) { + return Effect.succeed(); + } + const normalized = normalizeIsoTimestamp(value); + if (!normalized) { + return Effect.fail( + new RequestValidationError({ + message: `${paramName} must be a valid ISO-8601 timestamp.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.succeed(normalized); +}; + +const parseOptionalJsonObject = ( + request: Request +): Effect.Effect>, RequestValidationError> => + Effect.tryPromise({ + catch: () => + new RequestValidationError({ + message: "Invalid JSON payload.", + reasonCode: "REQUEST_BODY_INVALID_JSON", + }), + try: async () => await request.text(), + }).pipe( + Effect.flatMap((text) => { + if (text.trim().length === 0) { + return Effect.succeed({}); + } + try { + const parsed = JSON.parse(text) as unknown; + const record = asRecord(parsed); + if (!record || Array.isArray(parsed)) { + return Effect.fail( + new RequestValidationError({ + message: "Webhook dispatch replay filter body must be an object.", + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.succeed(record); + } catch (error) { + return Effect.fail( + new RequestValidationError({ + details: { + cause: error instanceof Error ? error.message : String(error), + }, + message: "Invalid JSON payload.", + reasonCode: "REQUEST_BODY_INVALID_JSON", + }) + ); + } + }) + ); + +const parseOptionalStringField = ( + body: Readonly>, + fieldName: string +): Effect.Effect => { + const value = body[fieldName]; + if (value === undefined || value === null) { + return Effect.succeed(); + } + if (typeof value !== "string") { + return Effect.fail( + new RequestValidationError({ + message: `${fieldName} must be a string.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const trimmed = value.trim(); + return Effect.succeed(trimmed.length > 0 ? trimmed : undefined); +}; + +const parseBulkReplayLimit = ( + body: Readonly> +): Effect.Effect => { + const value = body.limit; + if (value === undefined || value === null) { + return Effect.succeed(BULK_REPLAY_MAX); + } + if (typeof value !== "number" || !Number.isSafeInteger(value)) { + return Effect.fail( + new RequestValidationError({ + message: `limit must be an integer between 1 and ${BULK_REPLAY_MAX}.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + if (value < 1 || value > BULK_REPLAY_MAX) { + return Effect.fail( + new RequestValidationError({ + message: `limit must be an integer between 1 and ${BULK_REPLAY_MAX}.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.succeed(value); +}; + +const parseBulkReplayFilters = ( + body: Readonly> +): Effect.Effect< + { + readonly createdAfter?: string; + readonly createdBefore?: string; + readonly endpointId?: string; + readonly limit: number; + }, + RequestValidationError +> => + Effect.gen(function* parseBulkReplayFiltersProgram() { + const status = yield* parseOptionalStringField(body, "status"); + if (status !== undefined && status !== "failed") { + return yield* Effect.fail( + new RequestValidationError({ + message: "Only failed webhook dispatches can be replayed.", + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const endpointId = yield* parseOptionalStringField(body, "endpoint_id"); + const createdAfterRaw = yield* parseOptionalStringField( + body, + "created_after" + ); + const createdBeforeRaw = yield* parseOptionalStringField( + body, + "created_before" + ); + const createdAfter = yield* parseIsoTimestampParam( + createdAfterRaw ?? null, + "created_after" + ); + const createdBefore = yield* parseIsoTimestampParam( + createdBeforeRaw ?? null, + "created_before" + ); + const limit = yield* parseBulkReplayLimit(body); + return { + createdAfter, + createdBefore, + endpointId, + limit, + }; + }); + +const isPriorReplay = (input: { + readonly event: { + readonly action: string; + readonly object: string; + readonly reason: unknown; + }; + readonly dispatchId: string; + readonly idempotencyKey: string; +}): boolean => { + if ( + input.event.action !== DISPATCH_REPLAY_REQUESTED_ACTION || + input.event.object !== `webhook_dispatch:${input.dispatchId}` + ) { + return false; + } + return asRecord(input.event.reason)?.idempotencyKey === input.idempotencyKey; +}; + +const replayMarkerId = (input: { + readonly dispatchId: string; + readonly idempotencyKey: string; + readonly tenantId: string; +}): string => + [ + "webhook-dispatch-replay", + input.tenantId, + input.dispatchId, + encodeURIComponent(input.idempotencyKey), + ].join(":"); + +const endpointIdForAttempt = ( + attempt: NotificationDeliveryAttemptRecord, + config: { readonly endpointId?: string; readonly url: string } | undefined +): string | undefined => { + if (!config || attempt.destination !== config.url) { + return undefined; + } + return config.endpointId ?? DEFAULT_WEBHOOK_ENDPOINT_ID; +}; + +const toDispatchSummary = ( + attempt: NotificationDeliveryAttemptRecord, + event: NotificationEventRecord | undefined, + endpointId: string | undefined +) => ({ + attempt: attempt.attempt, + createdAt: attempt.createdAt, + destination: attempt.destination, + dispatchId: attempt.id, + endpointId, + error: attempt.error, + eventId: attempt.notificationEventId, + eventType: event?.eventType, + replayable: attempt.channel === "webhook" && attempt.status === "failed", + requestId: attempt.requestId, + responseCode: attempt.responseCode, + status: attempt.status, +}); + +const ensureReplayableWebhookDispatch = ( + attempt: NotificationDeliveryAttemptRecord +) => { + if (attempt.channel !== "webhook") { + return Effect.fail( + new RequestValidationError({ + message: `Dispatch ${attempt.id} is not a webhook delivery attempt.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + if (attempt.status !== "failed") { + return Effect.fail( + new RequestValidationError({ + message: `Dispatch ${attempt.id} is not failed and cannot be replayed.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.void; +}; + +const isMissingWebhookDispatchError = ( + error: unknown, + dispatchId: string +): boolean => { + if (hasErrorTag(error, "PersistenceEntityNotFoundError")) { + return true; + } + return error instanceof Error && error.message === `Missing ${dispatchId}`; +}; + +const toMissingWebhookDispatchError = (dispatchId: string) => + new PersistenceEntityNotFoundError({ + entity: "notification_delivery_attempts", + id: dispatchId, + }); + +const getConfiguredWebhookDestination = (input: { + readonly endpointId?: string; + readonly webhookConfig: + | { readonly endpointId?: string; readonly url: string } + | undefined; +}): string | undefined => { + const configuredEndpointId = + input.webhookConfig?.endpointId ?? DEFAULT_WEBHOOK_ENDPOINT_ID; + return input.endpointId && + input.webhookConfig && + input.endpointId === configuredEndpointId + ? input.webhookConfig.url + : undefined; +}; + +const getRequiredIdempotencyKey = ( + request: Request, + message: string +): Effect.Effect => { + const callerIdempotencyKey = getIdempotencyKey(request); + if (!callerIdempotencyKey) { + return Effect.fail( + new RequestValidationError({ + message, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + return Effect.succeed(callerIdempotencyKey); +}; + +const replayOneWebhookDispatch = (input: { + readonly actor: { + readonly id: string; + readonly principalKind: string; + }; + readonly attempt: NotificationDeliveryAttemptRecord; + readonly callerIdempotencyKey: string; + readonly tenantId: string; +}) => + Effect.gen(function* replayOneWebhookDispatchProgram() { + const services = yield* Effect.service(RuntimeServicesTag); + const dispatchId = input.attempt.id; + yield* ensureReplayableWebhookDispatch(input.attempt); + const event = yield* services.repos.persistence.notificationEvents + .getById(input.attempt.notificationEventId) + .pipe(withTenant(input.tenantId)); + if (event.requestId !== input.attempt.requestId) { + return yield* Effect.fail( + new RequestValidationError({ + message: `Webhook dispatch ${dispatchId} does not match its notification event request.`, + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const replayIdempotencyKey = `webhook-dispatch-replay:${input.tenantId}:${dispatchId}:${input.callerIdempotencyKey}`; + const replayRequestMarkerId = replayMarkerId({ + dispatchId, + idempotencyKey: input.callerIdempotencyKey, + tenantId: input.tenantId, + }); + const priorAuditEvents = yield* services.repos.persistence.auditEvents + .list({ + action: DISPATCH_REPLAY_REQUESTED_ACTION, + limit: MAX_LIMIT, + requestId: event.requestId, + }) + .pipe(withTenant(input.tenantId)); + if ( + priorAuditEvents.items.some((auditEvent) => + isPriorReplay({ + dispatchId, + event: auditEvent, + idempotencyKey: replayIdempotencyKey, + }) + ) + ) { + return { + dispatchId, + eventId: event.id, + status: "already_replayed" as const, + }; + } + const replayRequestAppended = yield* appendAuditEvent({ + action: DISPATCH_REPLAY_REQUESTED_ACTION, + actor: input.actor.id, + after: { + dispatchId, + eventId: event.id, + status: "accepted", + }, + before: { + attempt: input.attempt.attempt, + dispatchId, + status: input.attempt.status, + }, + id: replayRequestMarkerId, + object: `webhook_dispatch:${dispatchId}`, + reason: { + idempotencyKey: replayIdempotencyKey, + requestedBy: input.actor.principalKind, + }, + requestId: event.requestId, + tenantId: input.tenantId, + }).pipe( + Effect.as(true), + Effect.catch((error) => + Effect.gen(function* recoverReplayRequestAppend() { + const replayRequests = yield* services.repos.persistence.auditEvents + .list({ + action: DISPATCH_REPLAY_REQUESTED_ACTION, + limit: MAX_LIMIT, + requestId: event.requestId, + }) + .pipe(withTenant(input.tenantId)); + if ( + replayRequests.items.some((auditEvent) => + isPriorReplay({ + dispatchId, + event: auditEvent, + idempotencyKey: replayIdempotencyKey, + }) + ) + ) { + return false; + } + return yield* Effect.fail(error); + }) + ) + ); + if (!replayRequestAppended) { + return { + dispatchId, + eventId: event.id, + status: "already_replayed" as const, + }; + } + yield* replayWebhookDispatch({ + event, + idempotencyKey: replayIdempotencyKey, + tenantId: input.tenantId, + }); + yield* appendAuditEvent({ + action: DISPATCH_REPLAY_ACTION, + actor: input.actor.id, + after: { + dispatchId, + eventId: event.id, + status: "replayed", + }, + before: { + attempt: input.attempt.attempt, + dispatchId, + status: input.attempt.status, + }, + object: `webhook_dispatch:${dispatchId}`, + reason: { + idempotencyKey: replayIdempotencyKey, + requestedBy: input.actor.principalKind, + }, + requestId: event.requestId, + tenantId: input.tenantId, + }); + return { + dispatchId, + eventId: event.id, + status: "replayed" as const, + }; + }); + +const errorMessage = (error: unknown): string => + error instanceof Error ? error.message : String(error); + +/** Lists persisted outbound webhook dispatch attempts for the current tenant. */ +export const listWebhookDispatchesRoute: RouteDefinition = { + handler: ({ request }) => + Effect.gen(function* listWebhookDispatchesHandler() { + const services = yield* Effect.service(RuntimeServicesTag); + const actor = yield* requireRequestActor(services.requestContext); + yield* requirePrincipalKinds({ + actor, + allowedKinds: ["operator", "service"], + message: + "Webhook dispatch inspection is reserved for operator and service principals.", + }); + const tenantId = yield* requireRequestTenantId(services.requestContext); + const { searchParams } = new URL(request.url); + const limit = yield* parseIntParam( + searchParams.get("limit"), + DEFAULT_LIMIT, + 1, + MAX_LIMIT + ); + const offset = yield* parseIntParam( + searchParams.get("offset"), + 0, + 0, + Number.MAX_SAFE_INTEGER + ); + const endpointId = searchParams.get("endpoint_id") ?? undefined; + const webhookConfig = services.config.notificationWebhook; + const destination = getConfiguredWebhookDestination({ + endpointId, + webhookConfig, + }); + if (endpointId && !destination) { + return ok({ + items: [], + limit, + offset, + total: 0, + }); + } + const status = yield* parseStatusFilter(searchParams.get("status")); + const createdAfter = yield* parseIsoTimestampParam( + searchParams.get("created_after"), + "created_after" + ); + const createdBefore = yield* parseIsoTimestampParam( + searchParams.get("created_before"), + "created_before" + ); + const attemptFilters = { + channel: "webhook", + createdAfter, + createdBefore, + destination, + status, + } as const; + const total = + yield* services.repos.persistence.notificationDeliveryAttempts + .count(attemptFilters) + .pipe(withTenant(tenantId)); + const attempts = + yield* services.repos.persistence.notificationDeliveryAttempts + .list({ + ...attemptFilters, + limit, + offset, + }) + .pipe(withTenant(tenantId)); + const items = yield* Effect.forEach(attempts, (attempt) => + Effect.gen(function* mapWebhookDispatch() { + const event = yield* services.repos.persistence.notificationEvents + .getById(attempt.notificationEventId) + .pipe(withTenant(tenantId), Effect.result); + return toDispatchSummary( + attempt, + event._tag === "Success" ? event.success : undefined, + endpointIdForAttempt(attempt, webhookConfig) + ); + }) + ); + return ok({ + items, + limit, + offset, + total, + }); + }), + method: "GET", + path: "/webhooks/dispatches", + protected: true, + summary: "List outbound webhook dispatches", +}; + +/** Bulk replays failed outbound webhook dispatches for the current tenant. */ +export const bulkReplayWebhookDispatchesRoute: RouteDefinition = { + handler: ({ request }) => + Effect.gen(function* bulkReplayWebhookDispatchesHandler() { + const services = yield* Effect.service(RuntimeServicesTag); + const actor = yield* requireRequestActor(services.requestContext); + yield* requirePrincipalKinds({ + actor, + allowedKinds: ["operator", "service"], + message: + "Webhook dispatch replay is reserved for operator and service principals.", + }); + const tenantId = yield* requireRequestTenantId(services.requestContext); + const callerIdempotencyKey = yield* getRequiredIdempotencyKey( + request, + "Webhook dispatch replay requires an x-idempotency-key header." + ); + const body = yield* parseOptionalJsonObject(request); + const filters = yield* parseBulkReplayFilters(body); + const webhookConfig = services.config.notificationWebhook; + const destination = getConfiguredWebhookDestination({ + endpointId: filters.endpointId, + webhookConfig, + }); + if (filters.endpointId && !destination) { + return accepted({ + alreadyReplayed: 0, + replayed: 0, + results: [], + total: 0, + }); + } + const attempts = + yield* services.repos.persistence.notificationDeliveryAttempts + .list({ + channel: "webhook", + createdAfter: filters.createdAfter, + createdBefore: filters.createdBefore, + destination, + limit: filters.limit, + offset: 0, + status: ["failed"], + }) + .pipe(withTenant(tenantId)); + const results = yield* Effect.forEach(attempts, (attempt) => + replayOneWebhookDispatch({ + actor, + attempt, + callerIdempotencyKey: `${callerIdempotencyKey}:${attempt.id}`, + tenantId, + }).pipe( + Effect.catch((error) => + Effect.succeed({ + dispatchId: attempt.id, + error: errorMessage(error), + eventId: attempt.notificationEventId, + status: "failed" as const, + }) + ) + ) + ); + let replayed = 0; + let alreadyReplayed = 0; + for (const result of results) { + if (result.status === "replayed") { + replayed += 1; + } + if (result.status === "already_replayed") { + alreadyReplayed += 1; + } + } + return accepted({ + alreadyReplayed, + replayed, + results, + total: results.length, + }); + }), + method: "POST", + path: "/webhooks/dispatches/replay", + protected: true, + summary: "Replay failed outbound webhook dispatches", +}; + +/** Replays one failed outbound webhook dispatch for the current tenant. */ +export const replayWebhookDispatchRoute: RouteDefinition = { + handler: ({ params, request }) => + Effect.gen(function* replayWebhookDispatchHandler() { + const services = yield* Effect.service(RuntimeServicesTag); + const actor = yield* requireRequestActor(services.requestContext); + yield* requirePrincipalKinds({ + actor, + allowedKinds: ["operator", "service"], + message: + "Webhook dispatch replay is reserved for operator and service principals.", + }); + const tenantId = yield* requireRequestTenantId(services.requestContext); + const dispatchId = params.id; + if (!dispatchId) { + return yield* Effect.fail( + new RequestValidationError({ + message: "Webhook dispatch id is required.", + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const attempt = + yield* services.repos.persistence.notificationDeliveryAttempts + .getById(dispatchId) + .pipe( + withTenant(tenantId), + Effect.mapError((error) => + isMissingWebhookDispatchError(error, dispatchId) + ? toMissingWebhookDispatchError(dispatchId) + : error + ) + ); + const callerIdempotencyKey = yield* getRequiredIdempotencyKey( + request, + "Webhook dispatch replay requires an x-idempotency-key header." + ); + const result = yield* replayOneWebhookDispatch({ + actor, + attempt, + callerIdempotencyKey, + tenantId, + }); + return accepted(result); + }), + method: "POST", + path: "/webhooks/dispatches/:id/replay", + protected: true, + summary: "Replay outbound webhook dispatch", +}; diff --git a/packages/backend/src/services/notifications/service.ts b/packages/backend/src/services/notifications/service.ts index d4a0ba9..e1190cc 100644 --- a/packages/backend/src/services/notifications/service.ts +++ b/packages/backend/src/services/notifications/service.ts @@ -1,11 +1,18 @@ import { asNonEmptyString, asRecordOrEmpty } from "@dsar/guards"; /* oxlint-disable complexity */ import { withTenant } from "@dsar/persistence"; +import type { + NotificationDeliveryAttemptRecord, + NotificationEventRecord, +} from "@dsar/persistence"; import * as Effect from "effect/Effect"; import { normalizeAdapterError, toAdapterFailureEvent } from "../../adapters"; import type { AdapterContractError } from "../../adapters"; -import type { NotificationEventDraft } from "../../events/contracts"; +import type { + NotificationEventDraft, + NotificationEventType, +} from "../../events/contracts"; import { makeRequestId } from "../../middleware/auth-context"; import { RequestValidationError } from "../../types/errors"; import type { @@ -21,6 +28,22 @@ const DEFAULT_POLICY_VERSION = "policy-v1"; const DEFAULT_LOCALE = "en-GB"; const GENERATED_STATUS = "generated"; const DEFAULT_WEBHOOK_ENDPOINT_ID = "default"; +const NOTIFICATION_EVENT_TYPES = [ + "request_captured", + "clock_due_changed", + "clock_segment_opened", + "clock_segment_closed", + "request_acknowledged", + "acknowledgement_sent", + "verification_outcome_recorded", + "manifest_review_recorded", + "appeal_recorded", + "fulfillment_callback_received", + "delivery_prepared", + "step_up_challenge_issued", + "request_fulfilled", + "request_refused", +] as const satisfies readonly NotificationEventType[]; const toGeneratedResult = ( eventId: string @@ -53,6 +76,42 @@ const toNotificationDispatchValidationError = ( reasonCode: "REQUEST_VALIDATION_FAILED", }); +const toReplayDispatchValidationError = ( + error: unknown +): RequestValidationError => { + if (error instanceof RequestValidationError) { + return error; + } + return toNotificationDispatchValidationError(error); +}; + +const isNotificationEventType = ( + value: string +): value is NotificationEventType => { + for (const eventType of NOTIFICATION_EVENT_TYPES) { + if (eventType === value) { + return true; + } + } + return false; +}; + +const parseNotificationEventType = ( + value: string +): Effect.Effect => { + if (isNotificationEventType(value)) { + return Effect.succeed(value); + } + return Effect.fail( + new RequestValidationError({ + details: { eventType: value }, + message: + "Webhook dispatch cannot be replayed because the persisted notification event type is not supported.", + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); +}; + /** * Resolves effective outbound-resend policy with workspace > tenant > global precedence. */ @@ -215,7 +274,10 @@ const deliverWithRetries = (input: { >; readonly retryMaxAttempts: number; readonly retryDelayMs: number; + readonly startAttempt?: number; }): Effect.Effect => { + const startAttempt = input.startAttempt ?? 1; + const maxAttempt = startAttempt + input.retryMaxAttempts - 1; const runAttempt = ( attempt: number ): Effect.Effect => @@ -247,7 +309,7 @@ const deliverWithRetries = (input: { status: "failed", tenantId: input.tenantId, }); - if (attempt >= input.retryMaxAttempts || !normalized.retriable) { + if (attempt >= maxAttempt || !normalized.retriable) { return; } yield* Effect.sleep(input.retryDelayMs); @@ -271,15 +333,111 @@ const deliverWithRetries = (input: { if (result.status === "skipped") { return; } - if (attempt >= input.retryMaxAttempts) { + if (attempt >= maxAttempt) { return; } yield* Effect.sleep(input.retryDelayMs); return yield* runAttempt(attempt + 1); }); - return runAttempt(1); + return runAttempt(startAttempt); +}; + +const nextWebhookAttemptNumber = (input: { + readonly attempts: readonly NotificationDeliveryAttemptRecord[]; +}): number => { + let maxAttempt = 0; + for (const attempt of input.attempts) { + if (attempt.channel === "webhook") { + maxAttempt = Math.max(maxAttempt, attempt.attempt); + } + } + return maxAttempt + 1; }; +const supportsNotificationChannel = ( + adapter: { readonly channels?: readonly string[] } | undefined, + channel: "email" | "webhook" +): boolean => adapter?.channels?.includes(channel) === true; + +/** + * Replays a persisted webhook dispatch without re-sending other notification + * channels such as email. + * + * @param input - Original notification event and failed webhook attempt. + * @returns Effect that records replay delivery attempts. + */ +export const replayWebhookDispatch = (input: { + readonly event: NotificationEventRecord; + readonly idempotencyKey: string; + readonly tenantId: string; +}): Effect.Effect => + Effect.gen(function* replayWebhookDispatchProgram() { + const services = yield* Effect.service(RuntimeServicesTag); + const webhookConfig = services.config.notificationWebhook; + if (!webhookConfig || webhookConfig.url.length === 0) { + return yield* Effect.fail( + new RequestValidationError({ + message: + "Webhook dispatch cannot be replayed because no webhook endpoint is configured.", + reasonCode: "REQUEST_VALIDATION_FAILED", + }) + ); + } + const resolvedNotificationAdapter = + services.adapterRegistry.resolveNotification(); + const webhookAdapter = supportsNotificationChannel( + resolvedNotificationAdapter, + "webhook" + ) + ? resolvedNotificationAdapter + : undefined; + const signingKey = yield* resolveWebhookSigningKey({ + config: webhookConfig, + services, + tenantId: input.tenantId, + }); + const attempts = + yield* services.repos.persistence.notificationDeliveryAttempts + .listByNotificationEventId(input.event.id) + .pipe(withTenant(input.tenantId)); + const eventType = yield* parseNotificationEventType(input.event.eventType); + const dispatchInput = toDispatchInput({ + correlationId: services.requestContext.requestId, + draft: { + eventType, + locale: input.event.locale, + payload: input.event.payload, + policyVersion: input.event.policyVersion, + requestId: input.event.requestId, + }, + eventId: input.event.id, + idempotencyKey: input.idempotencyKey, + }); + yield* deliverWithRetries({ + adapterKey: webhookAdapter?.key ?? "webhook-fallback", + channel: "webhook", + destination: webhookConfig.url, + eventId: input.event.id, + requestId: input.event.requestId, + retryDelayMs: webhookConfig.retryDelayMs, + retryMaxAttempts: webhookConfig.retryMaxAttempts, + send: () => + webhookAdapter + ? webhookAdapter.send({ + ...dispatchInput, + webhookSigningKey: signingKey, + }) + : dispatchWebhookNotification({ + event: dispatchInput, + signingKey, + timeoutMs: webhookConfig.timeoutMs, + url: webhookConfig.url, + }), + startAttempt: nextWebhookAttemptNumber({ attempts }), + tenantId: input.tenantId, + }); + }).pipe(Effect.mapError(toReplayDispatchValidationError)); + /** * Persists and dispatches a notification event across configured * webhook/email channels. @@ -354,11 +512,12 @@ export const emitNotificationEvent = (input: { (resolvedNotificationAdapter?.key === "outbound-resend" ? resolvedNotificationAdapter : undefined); - const webhookAdapter = - resolvedNotificationAdapter && - resolvedNotificationAdapter.key !== "outbound-resend" - ? resolvedNotificationAdapter - : undefined; + const webhookAdapter = supportsNotificationChannel( + resolvedNotificationAdapter, + "webhook" + ) + ? resolvedNotificationAdapter + : undefined; // Webhook is optional; we still persist a pending attempt so audit trails can // explain why no outbound webhook dispatch occurred. diff --git a/packages/backend/src/testing/minimal-persistence.ts b/packages/backend/src/testing/minimal-persistence.ts index c255e87..7215293 100644 --- a/packages/backend/src/testing/minimal-persistence.ts +++ b/packages/backend/src/testing/minimal-persistence.ts @@ -81,6 +81,63 @@ const boundedLimit = (value: unknown): number => { return Math.max(1, Math.min(MAX_LIST_LIMIT, Math.trunc(value))); }; +const boundedOffset = (value: unknown): number => { + if (typeof value !== "number" || !Number.isFinite(value)) { + return 0; + } + return Math.max(0, Math.trunc(value)); +}; + +const matchesNotificationAttemptFilter = ( + attempt: Record, + input?: Record +): boolean => { + if (typeof input?.channel === "string" && attempt.channel !== input.channel) { + return false; + } + if ( + Array.isArray(input?.status) && + input.status.length > 0 && + !input.status.includes(attempt.status) + ) { + return false; + } + if ( + typeof input?.destination === "string" && + attempt.destination !== input.destination + ) { + return false; + } + if ( + typeof input?.createdAfter === "string" && + typeof attempt.createdAt === "string" && + attempt.createdAt <= input.createdAfter + ) { + return false; + } + if ( + typeof input?.createdBefore === "string" && + typeof attempt.createdAt === "string" && + attempt.createdAt >= input.createdBefore + ) { + return false; + } + return true; +}; + +const compareNotificationAttemptsDesc = ( + left: Record, + right: Record +): number => { + const leftCreatedAt = String(left.createdAt ?? ""); + const rightCreatedAt = String(right.createdAt ?? ""); + const order = rightCreatedAt.localeCompare(leftCreatedAt); + if (order !== 0) { + return order; + } + return String(right.id ?? "").localeCompare(String(left.id ?? "")); +}; + /** * Minimal in-memory persistence surface used by backend tests. */ @@ -126,6 +183,13 @@ export interface MinimalPersistence { readonly append: ( input: Record ) => Effect.Effect>; + readonly count: (input?: Record) => Effect.Effect; + readonly getById: ( + id: string + ) => Effect.Effect, Error>; + readonly list: ( + input?: Record + ) => Effect.Effect[]>; readonly listByNotificationEventId: ( id: string ) => Effect.Effect[]>; @@ -295,8 +359,30 @@ export const makeMinimalPersistence = (): Effect.Effect => append: (input: Record) => Effect.gen(function* append() { const record = { ...input, tenantId: "tenant-default" }; - yield* Ref.update(auditEventsRef, (arr) => [...arr, record]); - return record; + const result = yield* Ref.modify(auditEventsRef, (arr) => { + if (arr.some((event) => event.id === record.id)) { + return [ + { + id: record.id, + status: "duplicate" as const, + }, + arr, + ]; + } + return [ + { + record, + status: "appended" as const, + }, + [...arr, record], + ]; + }); + if (result.status === "duplicate") { + return yield* Effect.fail( + new Error(`Duplicate audit event ${String(result.id)}`) + ); + } + return result.record; }), list: (input: Record) => Ref.get(auditEventsRef).pipe( @@ -568,6 +654,40 @@ export const makeMinimalPersistence = (): Effect.Effect => ]); return record; }), + count: (input?: Record) => + Ref.get(notificationAttemptsRef).pipe( + Effect.map( + (arr) => + arr.filter((a: Record) => + matchesNotificationAttemptFilter(a, input) + ).length + ) + ), + getById: (id: string) => + Ref.get(notificationAttemptsRef).pipe( + Effect.flatMap((arr) => { + const found = arr.find( + (a: Record) => a.id === id + ); + return found + ? Effect.succeed(found) + : Effect.fail(new Error(`Missing ${id}`)); + }) + ), + list: (input?: Record) => + Ref.get(notificationAttemptsRef).pipe( + Effect.map((arr) => + arr + .filter((a: Record) => + matchesNotificationAttemptFilter(a, input) + ) + .toSorted(compareNotificationAttemptsDesc) + .slice( + boundedOffset(input?.offset), + boundedOffset(input?.offset) + boundedLimit(input?.limit) + ) + ) + ), listByNotificationEventId: (id: string) => Ref.get(notificationAttemptsRef).pipe( Effect.map((arr) => diff --git a/packages/backend/test/__snapshots__/openapi.test.ts.snap b/packages/backend/test/__snapshots__/openapi.test.ts.snap index e188cff..b910847 100644 --- a/packages/backend/test/__snapshots__/openapi.test.ts.snap +++ b/packages/backend/test/__snapshots__/openapi.test.ts.snap @@ -10468,6 +10468,811 @@ exports[`openAPI and docs surface > keeps generated spec stable 1`] = ` ], }, }, + "/webhooks/dispatches": { + "get": { + "operationId": "webhooks_dispatches_list", + "parameters": [ + { + "in": "query", + "name": "created_after", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + { + "in": "query", + "name": "created_before", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + { + "in": "query", + "name": "endpoint_id", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + { + "in": "query", + "name": "limit", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + { + "in": "query", + "name": "offset", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + { + "in": "query", + "name": "status", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "additionalProperties": false, + "properties": { + "items": { + "items": { + "additionalProperties": false, + "properties": { + "attempt": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + "createdAt": { + "type": "string", + }, + "destination": { + "type": "string", + }, + "dispatchId": { + "type": "string", + }, + "endpointId": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "error": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "eventId": { + "type": "string", + }, + "eventType": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "replayable": { + "type": "boolean", + }, + "requestId": { + "type": "string", + }, + "responseCode": { + "anyOf": [ + { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + { + "type": "null", + }, + ], + }, + "status": { + "anyOf": [ + { + "enum": [ + "pending", + ], + "type": "string", + }, + { + "enum": [ + "delivered", + ], + "type": "string", + }, + { + "enum": [ + "failed", + ], + "type": "string", + }, + { + "enum": [ + "skipped", + ], + "type": "string", + }, + ], + }, + }, + "required": [ + "attempt", + "createdAt", + "destination", + "dispatchId", + "eventId", + "replayable", + "requestId", + "status", + ], + "type": "object", + }, + "type": "array", + }, + "limit": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + "offset": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + "total": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + }, + "required": [ + "items", + "limit", + "offset", + "total", + ], + "type": "object", + }, + "meta": { + "anyOf": [ + { + "additionalProperties": { + "type": "null", + }, + "type": "object", + }, + { + "type": "null", + }, + ], + }, + "ok": { + "enum": [ + true, + ], + "type": "boolean", + }, + }, + "required": [ + "data", + "ok", + ], + "type": "object", + }, + }, + }, + "description": "Success", + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/effect_HttpApiSchemaError", + }, + }, + }, + "description": "The request or response did not match the expected schema", + }, + }, + "security": [ + { + "BearerAuth": [], + }, + ], + "summary": "List outbound webhook dispatches", + "tags": [ + "webhooks", + ], + }, + }, + "/webhooks/dispatches/replay": { + "post": { + "operationId": "webhooks_dispatches_replay_bulk", + "parameters": [ + { + "in": "header", + "name": "x-idempotency-key", + "required": true, + "schema": { + "type": "string", + }, + }, + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "created_after": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "created_before": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "endpoint_id": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "limit": { + "anyOf": [ + { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + { + "type": "null", + }, + ], + }, + "status": { + "anyOf": [ + { + "enum": [ + "failed", + ], + "type": "string", + }, + { + "type": "null", + }, + ], + }, + }, + "type": "object", + }, + }, + }, + "required": true, + }, + "responses": { + "202": { + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "additionalProperties": false, + "properties": { + "alreadyReplayed": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + "replayed": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + "results": { + "items": { + "additionalProperties": false, + "properties": { + "dispatchId": { + "type": "string", + }, + "error": { + "anyOf": [ + { + "type": "string", + }, + { + "type": "null", + }, + ], + }, + "eventId": { + "type": "string", + }, + "status": { + "anyOf": [ + { + "enum": [ + "replayed", + ], + "type": "string", + }, + { + "enum": [ + "already_replayed", + ], + "type": "string", + }, + { + "enum": [ + "failed", + ], + "type": "string", + }, + ], + }, + }, + "required": [ + "dispatchId", + "eventId", + "status", + ], + "type": "object", + }, + "type": "array", + }, + "total": { + "anyOf": [ + { + "type": "number", + }, + { + "enum": [ + "NaN", + ], + "type": "string", + }, + { + "enum": [ + "Infinity", + ], + "type": "string", + }, + { + "enum": [ + "-Infinity", + ], + "type": "string", + }, + ], + }, + }, + "required": [ + "alreadyReplayed", + "replayed", + "results", + "total", + ], + "type": "object", + }, + "meta": { + "anyOf": [ + { + "additionalProperties": { + "type": "null", + }, + "type": "object", + }, + { + "type": "null", + }, + ], + }, + "ok": { + "enum": [ + true, + ], + "type": "boolean", + }, + }, + "required": [ + "data", + "ok", + ], + "type": "object", + }, + }, + }, + "description": "Success", + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/effect_HttpApiSchemaError", + }, + }, + }, + "description": "The request or response did not match the expected schema", + }, + }, + "security": [ + { + "BearerAuth": [], + }, + ], + "summary": "Replay failed outbound webhook dispatches", + "tags": [ + "webhooks", + ], + }, + }, + "/webhooks/dispatches/{id}/replay": { + "post": { + "operationId": "webhooks_dispatches_replay", + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "schema": { + "type": "string", + }, + }, + { + "in": "header", + "name": "x-idempotency-key", + "required": true, + "schema": { + "type": "string", + }, + }, + ], + "responses": { + "202": { + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "additionalProperties": false, + "properties": { + "dispatchId": { + "type": "string", + }, + "eventId": { + "type": "string", + }, + "status": { + "anyOf": [ + { + "enum": [ + "replayed", + ], + "type": "string", + }, + { + "enum": [ + "already_replayed", + ], + "type": "string", + }, + ], + }, + }, + "required": [ + "dispatchId", + "eventId", + "status", + ], + "type": "object", + }, + "meta": { + "anyOf": [ + { + "additionalProperties": { + "type": "null", + }, + "type": "object", + }, + { + "type": "null", + }, + ], + }, + "ok": { + "enum": [ + true, + ], + "type": "boolean", + }, + }, + "required": [ + "data", + "ok", + ], + "type": "object", + }, + }, + }, + "description": "Success", + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/effect_HttpApiSchemaError", + }, + }, + }, + "description": "The request or response did not match the expected schema", + }, + }, + "security": [ + { + "BearerAuth": [], + }, + ], + "summary": "Replay outbound webhook dispatch", + "tags": [ + "webhooks", + ], + }, + }, "/webhooks/endpoints/{id}/rotate-key": { "post": { "operationId": "webhooks_endpoint_rotate_key", diff --git a/packages/backend/test/e2e/fixtures.ts b/packages/backend/test/e2e/fixtures.ts index c6e5d75..0f1d59a 100644 --- a/packages/backend/test/e2e/fixtures.ts +++ b/packages/backend/test/e2e/fixtures.ts @@ -1,3 +1,4 @@ +import { PersistenceEntityNotFoundError } from "@dsar/persistence"; import type { AuditEventRecord, ChatStateRecord, @@ -13,6 +14,7 @@ import type { FulfillmentArtifactRecord, JsonValue, ListAuditEventsInput, + ListNotificationDeliveryAttemptsInput, ListRequestsBySubjectInput, NotificationDeliveryAttemptRecord, NotificationEventRecord, @@ -82,6 +84,32 @@ const requestPolicyPack = (record: RequestRecord): string | undefined => { : undefined; }; +const matchesNotificationAttemptFilter = ( + attempt: NotificationDeliveryAttemptRecord, + input?: ListNotificationDeliveryAttemptsInput +): boolean => { + if (input?.channel && attempt.channel !== input.channel) { + return false; + } + if ( + input?.status && + input.status.length > 0 && + !input.status.includes(attempt.status) + ) { + return false; + } + if (input?.destination && attempt.destination !== input.destination) { + return false; + } + if (input?.createdAfter && attempt.createdAt <= input.createdAfter) { + return false; + } + if (input?.createdBefore && attempt.createdAt >= input.createdBefore) { + return false; + } + return true; +}; + export const BASE_JSON_BODY = { challengeId: "challenge-1", channel: "email", @@ -147,6 +175,9 @@ export const makeMemoryPersistence = (): PersistenceService => { return { auditEvents: { append: (input: CreateAuditEventInput) => { + if (auditEvents.some((event) => event.id === input.id)) { + return Effect.fail(new Error(`Duplicate audit event ${input.id}`)); + } const record: AuditEventRecord = { ...input, after: input.after as JsonValue, @@ -395,6 +426,41 @@ export const makeMemoryPersistence = (): PersistenceService => { notificationAttempts.push(record); return Effect.succeed(record); }, + count: (input?: ListNotificationDeliveryAttemptsInput) => + Effect.succeed( + notificationAttempts.filter((attempt) => + matchesNotificationAttemptFilter(attempt, input) + ).length + ), + getById: (id: string) => + Effect.fromNullishOr( + notificationAttempts.find((attempt) => attempt.id === id) + ).pipe( + Effect.mapError( + () => + new PersistenceEntityNotFoundError({ + entity: "notification_delivery_attempts", + id, + }) + ) + ), + list: (input?: ListNotificationDeliveryAttemptsInput) => + Effect.succeed( + notificationAttempts + .filter((attempt) => + matchesNotificationAttemptFilter(attempt, input) + ) + .toSorted((left, right) => + left.createdAt === right.createdAt + ? right.id.localeCompare(left.id) + : right.createdAt.localeCompare(left.createdAt) + ) + .slice( + Math.max(0, Math.trunc(input?.offset ?? 0)), + Math.max(0, Math.trunc(input?.offset ?? 0)) + + Math.max(1, Math.min(500, Math.trunc(input?.limit ?? 50))) + ) + ), listByNotificationEventId: (notificationEventId: string) => Effect.succeed( notificationAttempts.filter( diff --git a/packages/backend/test/e2e/tenant-isolation.test.ts b/packages/backend/test/e2e/tenant-isolation.test.ts index 648e2a2..bf65124 100644 --- a/packages/backend/test/e2e/tenant-isolation.test.ts +++ b/packages/backend/test/e2e/tenant-isolation.test.ts @@ -1,4 +1,8 @@ -import { TenantContext, withTenant } from "@dsar/persistence"; +import { + PersistenceEntityNotFoundError, + TenantContext, + withTenant, +} from "@dsar/persistence"; import type { AuditEventRecord, ChatStateRecord, @@ -17,6 +21,7 @@ import type { FulfillmentArtifactRecord, JsonValue, ListAuditEventsInput, + ListNotificationDeliveryAttemptsInput, NotificationDeliveryAttemptRecord, NotificationEventRecord, PaginationInput, @@ -144,6 +149,32 @@ const paginate = ( return items.slice(offset, offset + limit); }; +const matchesNotificationAttemptFilter = ( + attempt: NotificationDeliveryAttemptRecord, + input?: ListNotificationDeliveryAttemptsInput +): boolean => { + if (input?.channel && attempt.channel !== input.channel) { + return false; + } + if ( + input?.status && + input.status.length > 0 && + !input.status.includes(attempt.status) + ) { + return false; + } + if (input?.destination && attempt.destination !== input.destination) { + return false; + } + if (input?.createdAfter && attempt.createdAt <= input.createdAfter) { + return false; + } + if (input?.createdBefore && attempt.createdAt >= input.createdBefore) { + return false; + } + return true; +}; + const makeTenantScopedMemoryPersistence = (): PersistenceService => { const requests = new Map(); const timeline: RequestTimelineEventRecord[] = []; @@ -382,6 +413,47 @@ const makeTenantScopedMemoryPersistence = (): PersistenceService => { notificationAttempts.push(record); return record; }), + count: (input) => + Effect.gen(function* countNotificationAttempts() { + const tenantId = yield* currentTenantId; + return notificationAttempts.filter( + (attempt) => + attempt.tenantId === tenantId && + matchesNotificationAttemptFilter(attempt, input) + ).length; + }), + getById: (id: string) => + Effect.gen(function* getNotificationAttempt() { + const tenantId = yield* currentTenantId; + const record = notificationAttempts.find( + (attempt) => attempt.tenantId === tenantId && attempt.id === id + ); + if (!record) { + return yield* Effect.fail( + new PersistenceEntityNotFoundError({ + entity: "notification_delivery_attempts", + id, + }) + ); + } + return record; + }), + list: (input) => + Effect.gen(function* listNotificationAttempts() { + const tenantId = yield* currentTenantId; + const filtered = notificationAttempts + .filter( + (attempt) => + attempt.tenantId === tenantId && + matchesNotificationAttemptFilter(attempt, input) + ) + .toSorted((left, right) => + left.createdAt === right.createdAt + ? right.id.localeCompare(left.id) + : right.createdAt.localeCompare(left.createdAt) + ); + return paginate(filtered, input); + }), listByNotificationEventId: (notificationEventId: string) => Effect.gen(function* listNotificationAttempts() { const tenantId = yield* currentTenantId; @@ -1062,6 +1134,30 @@ const makeRouteProbes = (): readonly RouteProbe[] => [ method: "POST", path: "/webhooks/endpoints/default/rotate-key", }, + { + headers: tenantAHeaders, + key: "GET /webhooks/dispatches", + method: "GET", + path: "/webhooks/dispatches?status=failed", + }, + { + headers: { + ...tenantAHeaders, + "x-idempotency-key": "tenant-isolation-bulk-replay", + }, + key: "POST /webhooks/dispatches/replay", + method: "POST", + path: "/webhooks/dispatches/replay", + }, + { + headers: { + ...tenantAHeaders, + "x-idempotency-key": "tenant-isolation-single-replay", + }, + key: "POST /webhooks/dispatches/:id/replay", + method: "POST", + path: `/webhooks/dispatches/attempt-${TENANT_B_EVENT_ID}/replay`, + }, { headers: tenantAHeaders, json: { diff --git a/packages/backend/test/lifecycle/idempotency.test.ts b/packages/backend/test/lifecycle/idempotency.test.ts index 86d6bd7..3321a1b 100644 --- a/packages/backend/test/lifecycle/idempotency.test.ts +++ b/packages/backend/test/lifecycle/idempotency.test.ts @@ -213,6 +213,28 @@ const makeMemoryPersistence = (): PersistenceService => { notificationAttempts.push(record); return Effect.succeed(record); }, + getById: (id: string) => + Effect.fromNullishOr( + notificationAttempts.find((attempt) => attempt.id === id) + ).pipe( + Effect.mapError(() => new Error(`Missing notification attempt ${id}`)) + ), + list: (input) => + Effect.succeed( + notificationAttempts.filter((attempt) => { + if (input?.channel && attempt.channel !== input.channel) { + return false; + } + if ( + input?.status && + input.status.length > 0 && + !input.status.includes(attempt.status) + ) { + return false; + } + return true; + }) + ), listByNotificationEventId: (notificationEventId: string) => Effect.succeed( notificationAttempts.filter( diff --git a/packages/backend/test/openapi.test.ts b/packages/backend/test/openapi.test.ts index f2ef168..cb4dea9 100644 --- a/packages/backend/test/openapi.test.ts +++ b/packages/backend/test/openapi.test.ts @@ -8,6 +8,8 @@ const requiredCapabilityPaths = [ "/webhooks/inbound/resend", "/webhooks/inbound/slack", "/webhooks/endpoints/{id}/rotate-key", + "/webhooks/dispatches", + "/webhooks/dispatches/{id}/replay", "/requests/capture", "/requests/{id}/timeline", "/requests/{id}/clock/explain", diff --git a/packages/backend/test/runtime.test.ts b/packages/backend/test/runtime.test.ts index 51cdebe..2114ae3 100644 --- a/packages/backend/test/runtime.test.ts +++ b/packages/backend/test/runtime.test.ts @@ -5,6 +5,7 @@ import * as Effect from "effect/Effect"; import type { ErrorEnvelope, InboundAdapterContract, + NotificationAdapterContract, RateLimitStore, } from "../src"; import { dsarInstance } from "../src"; @@ -147,6 +148,67 @@ const makePolicyPack = (version: string) => ({ version, }); +const makeNotificationAdapter = (input: { + readonly channels?: NotificationAdapterContract["channels"]; + readonly send: NotificationAdapterContract["send"]; + readonly key?: string; +}): NotificationAdapterContract => ({ + capability: "notifications", + channels: input.channels ?? ["webhook"], + diagnostics: () => + Effect.succeed({ + capability: "notifications", + key: input.key ?? "custom-webhook", + }), + healthCheck: () => Effect.succeed({ ok: true, status: "healthy" }), + init: () => Effect.void, + key: input.key ?? "custom-webhook", + send: input.send, + validateConfig: () => Effect.void, +}); + +const seedWebhookDispatch = async ( + persistence: ReturnType, + input: { + readonly channel?: "webhook" | "email"; + readonly createdAt?: string; + readonly destination?: string; + readonly eventId: string; + readonly id: string; + readonly requestId?: string; + readonly status?: "pending" | "delivered" | "failed" | "skipped"; + } +): Promise => { + const requestId = input.requestId ?? `req-${input.id}`; + await Effect.runPromise( + persistence.notificationEvents.append({ + correlationId: `corr-${input.eventId}`, + createdAt: input.createdAt ?? "2026-02-20T00:00:00.000Z", + eventType: "acknowledgement_sent", + id: input.eventId, + idempotencyKey: `event-${input.eventId}`, + locale: "en-GB", + payload: { note: input.id }, + policyVersion: "policy-v1", + requestId, + }) + ); + await Effect.runPromise( + persistence.notificationDeliveryAttempts.append({ + attempt: 1, + channel: input.channel ?? "webhook", + createdAt: input.createdAt ?? "2026-02-20T00:01:00.000Z", + destination: input.destination ?? "https://tenant.example/webhook", + error: input.status === "failed" ? "500 Internal Server Error" : "", + id: input.id, + notificationEventId: input.eventId, + requestId, + responseCode: input.status === "failed" ? 500 : 200, + status: input.status ?? "failed", + }) + ); +}; + const makeSlackInboundFailureAdapter = (input: { readonly category: string; readonly details?: Readonly>; @@ -497,6 +559,542 @@ describe(dsarInstance, () => { }); }); + it("lists and replays failed outbound webhook dispatches", async () => { + const persistence = makeMemoryPersistence(); + await Effect.runPromise( + persistence.notificationEvents.append({ + correlationId: "corr-1", + createdAt: "2026-02-20T00:00:00.000Z", + eventType: "acknowledgement_sent", + id: "evt-webhook-1", + idempotencyKey: "event-1", + locale: "en-GB", + payload: { note: "failed webhook" }, + policyVersion: "policy-v1", + requestId: "req-webhook-1", + }) + ); + await Effect.runPromise( + persistence.notificationDeliveryAttempts.append({ + attempt: 1, + channel: "webhook", + createdAt: "2026-02-20T00:01:00.000Z", + destination: "https://tenant.example/webhook", + error: "500 Internal Server Error", + id: "dispatch-failed-1", + notificationEventId: "evt-webhook-1", + requestId: "req-webhook-1", + responseCode: 500, + status: "failed", + }) + ); + const sent: unknown[] = []; + const runtime = dsarInstance({ + adapters: { + notifications: makeNotificationAdapter({ + send: (input) => { + sent.push(input); + return Effect.succeed({ + responseCode: 200, + status: "delivered" as const, + }); + }, + }), + }, + config: { + ...TEST_RUNTIME_AUTH.config, + notificationWebhook: { + endpointId: "default", + retryDelayMs: 1, + retryMaxAttempts: 1, + signingSecret: "secret", + tenantScoped: true, + timeoutMs: 1000, + url: "https://tenant.example/webhook", + }, + }, + repos: { persistence }, + }); + + const listResponse = await runtime.handler( + new Request("https://example.test/webhooks/dispatches?status=failed", { + headers: adminHeaders, + }) + ); + const listBody = (await listResponse.json()) as { + readonly data: { + readonly items: readonly { + readonly dispatchId: string; + readonly replayable: boolean; + }[]; + }; + }; + expect(listResponse.status).toBe(200); + expect(listBody.data.items).toStrictEqual([ + expect.objectContaining({ + dispatchId: "dispatch-failed-1", + replayable: true, + }), + ]); + + const missingKeyReplay = await runtime.handler( + new Request( + "https://example.test/webhooks/dispatches/dispatch-failed-1/replay", + { + headers: adminHeaders, + method: "POST", + } + ) + ); + const missingKeyBody = (await missingKeyReplay.json()) as ErrorEnvelope; + expect([missingKeyReplay.status, missingKeyBody.error.code]).toStrictEqual([ + 400, + "REQUEST_VALIDATION_FAILED", + ]); + + const replayResponse = await runtime.handler( + new Request( + "https://example.test/webhooks/dispatches/dispatch-failed-1/replay", + { + headers: { + ...adminHeaders, + "x-idempotency-key": "replay-once", + }, + method: "POST", + } + ) + ); + const replayBody = (await replayResponse.json()) as { + readonly data: { readonly status: string }; + }; + expect(replayResponse.status).toBe(202); + expect(replayBody.data.status).toBe("replayed"); + expect(sent).toHaveLength(1); + const auditEvents = await Effect.runPromise( + persistence.auditEvents.listByRequestId("req-webhook-1") + ); + expect(auditEvents.map((event) => event.action)).toEqual( + expect.arrayContaining([ + "webhook_dispatch_replay_requested", + "webhook_dispatch_replayed", + ]) + ); + + const idempotentReplay = await runtime.handler( + new Request( + "https://example.test/webhooks/dispatches/dispatch-failed-1/replay", + { + headers: { + ...adminHeaders, + "x-idempotency-key": "replay-once", + }, + method: "POST", + } + ) + ); + const idempotentReplayBody = (await idempotentReplay.json()) as { + readonly data: { readonly status: string }; + }; + expect(idempotentReplay.status).toBe(202); + expect(idempotentReplayBody.data.status).toBe("already_replayed"); + expect(sent).toHaveLength(1); + }); + + it("validates and paginates outbound webhook dispatch listing", async () => { + const persistence = makeMemoryPersistence(); + for (const attempt of [ + { + createdAt: "2026-02-20T00:03:00.000Z", + error: "500 Internal Server Error", + id: "dispatch-page-3", + responseCode: 500, + status: "failed" as const, + }, + { + createdAt: "2026-02-20T00:02:00.000Z", + error: "500 Internal Server Error", + id: "dispatch-page-2", + responseCode: 500, + status: "failed" as const, + }, + { + createdAt: "2026-02-20T00:01:00.000Z", + error: "", + id: "dispatch-page-1", + responseCode: 200, + status: "delivered" as const, + }, + ]) { + await Effect.runPromise( + persistence.notificationDeliveryAttempts.append({ + attempt: 1, + channel: "webhook", + createdAt: attempt.createdAt, + destination: "https://tenant.example/webhook", + error: attempt.error, + id: attempt.id, + notificationEventId: `evt-${attempt.id}`, + requestId: `req-${attempt.id}`, + responseCode: attempt.responseCode, + status: attempt.status, + }) + ); + } + const runtime = dsarInstance({ + ...TEST_RUNTIME_AUTH, + repos: { persistence }, + }); + + const response = await runtime.handler( + new Request( + "https://example.test/webhooks/dispatches?status=failed&limit=1&offset=1", + { + headers: adminHeaders, + } + ) + ); + const body = (await response.json()) as { + readonly data: { + readonly items: readonly { readonly dispatchId: string }[]; + readonly limit: number; + readonly offset: number; + readonly total: number; + }; + }; + expect(response.status).toBe(200); + expect(body.data).toMatchObject({ + limit: 1, + offset: 1, + total: 2, + }); + expect(body.data.items.map((item) => item.dispatchId)).toStrictEqual([ + "dispatch-page-2", + ]); + + for (const query of [ + "limit=1abc", + "offset=-1", + "created_after=not-a-date", + ]) { + const invalidResponse = await runtime.handler( + new Request(`https://example.test/webhooks/dispatches?${query}`, { + headers: adminHeaders, + }) + ); + const invalidBody = (await invalidResponse.json()) as ErrorEnvelope; + expect([invalidResponse.status, invalidBody.error.code]).toStrictEqual([ + 400, + "REQUEST_VALIDATION_FAILED", + ]); + } + }); + + it("bulk replays failed outbound webhook dispatches idempotently", async () => { + const persistence = makeMemoryPersistence(); + await seedWebhookDispatch(persistence, { + eventId: "evt-bulk-1", + id: "dispatch-bulk-1", + requestId: "req-bulk-1", + }); + await seedWebhookDispatch(persistence, { + eventId: "evt-bulk-2", + id: "dispatch-bulk-2", + requestId: "req-bulk-2", + }); + await seedWebhookDispatch(persistence, { + eventId: "evt-bulk-delivered", + id: "dispatch-bulk-delivered", + requestId: "req-bulk-delivered", + status: "delivered", + }); + await seedWebhookDispatch(persistence, { + channel: "email", + destination: "subject@example.test", + eventId: "evt-bulk-email", + id: "dispatch-bulk-email", + requestId: "req-bulk-email", + }); + const sent: unknown[] = []; + const runtime = dsarInstance({ + adapters: { + notifications: makeNotificationAdapter({ + send: (input) => { + sent.push(input); + return Effect.succeed({ + responseCode: 200, + status: "delivered" as const, + }); + }, + }), + }, + config: { + ...TEST_RUNTIME_AUTH.config, + notificationWebhook: { + endpointId: "default", + retryDelayMs: 1, + retryMaxAttempts: 1, + signingSecret: "secret", + tenantScoped: true, + timeoutMs: 1000, + url: "https://tenant.example/webhook", + }, + }, + repos: { persistence }, + }); + + const response = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + headers: { + ...adminHeaders, + "x-idempotency-key": "bulk-once", + }, + method: "POST", + }) + ); + const body = (await response.json()) as { + readonly data: { + readonly alreadyReplayed: number; + readonly replayed: number; + readonly results: readonly { + readonly dispatchId: string; + readonly status: string; + }[]; + readonly total: number; + }; + }; + expect(response.status).toBe(202); + expect(body.data).toMatchObject({ + alreadyReplayed: 0, + replayed: 2, + total: 2, + }); + expect(body.data.results.map((result) => result.dispatchId)).toStrictEqual([ + "dispatch-bulk-2", + "dispatch-bulk-1", + ]); + expect(sent).toHaveLength(2); + + const idempotentResponse = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + headers: { + ...adminHeaders, + "x-idempotency-key": "bulk-once", + }, + method: "POST", + }) + ); + const idempotentBody = (await idempotentResponse.json()) as { + readonly data: { + readonly alreadyReplayed: number; + readonly replayed: number; + readonly total: number; + }; + }; + expect(idempotentResponse.status).toBe(202); + expect(idempotentBody.data).toMatchObject({ + alreadyReplayed: 2, + replayed: 0, + total: 2, + }); + expect(sent).toHaveLength(2); + }); + + it("bulk replay respects filters and continues after per-dispatch failures", async () => { + const persistence = makeMemoryPersistence(); + await seedWebhookDispatch(persistence, { + createdAt: "2026-02-20T00:01:00.000Z", + eventId: "evt-filter-before", + id: "dispatch-filter-before", + requestId: "req-filter-before", + }); + await seedWebhookDispatch(persistence, { + createdAt: "2026-02-20T00:02:00.000Z", + eventId: "evt-filter-match", + id: "dispatch-filter-match", + requestId: "req-filter-match", + }); + await seedWebhookDispatch(persistence, { + createdAt: "2026-02-20T00:03:00.000Z", + destination: "https://other.example/webhook", + eventId: "evt-filter-other", + id: "dispatch-filter-other", + requestId: "req-filter-other", + }); + await Effect.runPromise( + persistence.notificationDeliveryAttempts.append({ + attempt: 1, + channel: "webhook", + createdAt: "2026-02-20T00:02:30.000Z", + destination: "https://tenant.example/webhook", + error: "500 Internal Server Error", + id: "dispatch-filter-missing-event", + notificationEventId: "evt-filter-missing", + requestId: "req-filter-missing", + responseCode: 500, + status: "failed", + }) + ); + const sent: unknown[] = []; + const runtime = dsarInstance({ + adapters: { + notifications: makeNotificationAdapter({ + send: (input) => { + sent.push(input); + return Effect.succeed({ + responseCode: 200, + status: "delivered" as const, + }); + }, + }), + }, + config: { + ...TEST_RUNTIME_AUTH.config, + notificationWebhook: { + endpointId: "default", + retryDelayMs: 1, + retryMaxAttempts: 1, + signingSecret: "secret", + tenantScoped: true, + timeoutMs: 1000, + url: "https://tenant.example/webhook", + }, + }, + repos: { persistence }, + }); + + const response = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + body: JSON.stringify({ + created_after: "2026-02-20T00:01:30.000Z", + created_before: "2026-02-20T00:02:45.000Z", + endpoint_id: "default", + limit: 10, + status: "failed", + }), + headers: { + "content-type": "application/json", + ...adminHeaders, + "x-idempotency-key": "bulk-filter", + }, + method: "POST", + }) + ); + const body = (await response.json()) as { + readonly data: { + readonly replayed: number; + readonly results: readonly { + readonly dispatchId: string; + readonly error?: string; + readonly status: string; + }[]; + readonly total: number; + }; + }; + expect(response.status).toBe(202); + expect(body.data.total).toBe(2); + expect(body.data.replayed).toBe(1); + expect(body.data.results).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + dispatchId: "dispatch-filter-match", + status: "replayed", + }), + expect.objectContaining({ + dispatchId: "dispatch-filter-missing-event", + status: "failed", + }), + ]) + ); + expect(sent).toHaveLength(1); + }); + + it("validates and protects bulk outbound webhook dispatch replay", async () => { + const runtime = dsarInstance({ + ...TEST_RUNTIME_AUTH, + repos: { persistence: makeMemoryPersistence() }, + }); + + const missingKeyResponse = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + headers: adminHeaders, + method: "POST", + }) + ); + const missingKeyBody = (await missingKeyResponse.json()) as ErrorEnvelope; + expect([ + missingKeyResponse.status, + missingKeyBody.error.code, + ]).toStrictEqual([400, "REQUEST_VALIDATION_FAILED"]); + + const invalidStatusResponse = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + body: JSON.stringify({ status: "delivered" }), + headers: { + "content-type": "application/json", + ...adminHeaders, + "x-idempotency-key": "bulk-invalid", + }, + method: "POST", + }) + ); + const invalidStatusBody = + (await invalidStatusResponse.json()) as ErrorEnvelope; + expect([ + invalidStatusResponse.status, + invalidStatusBody.error.code, + ]).toStrictEqual([400, "REQUEST_VALIDATION_FAILED"]); + + const subjectResponse = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/replay", { + headers: { + ...subjectHeaders, + "x-idempotency-key": "bulk-subject", + }, + method: "POST", + }) + ); + expect(subjectResponse.status).toBe(403); + }); + + it("returns 404 for missing outbound webhook dispatch replay", async () => { + const runtime = dsarInstance({ + ...TEST_RUNTIME_AUTH, + repos: { persistence: makeMemoryPersistence() }, + }); + const response = await runtime.handler( + new Request("https://example.test/webhooks/dispatches/missing/replay", { + headers: { + ...adminHeaders, + "x-idempotency-key": "missing-once", + }, + method: "POST", + }) + ); + const body = (await response.json()) as ErrorEnvelope; + + expect([response.status, body.error.code]).toStrictEqual([ + 404, + "PERSISTENCE_ENTITY_NOT_FOUND", + ]); + }); + + it("protects outbound webhook dispatch replay from subject callers", async () => { + const runtime = dsarInstance({ + ...TEST_RUNTIME_AUTH, + repos: { persistence: makeMemoryPersistence() }, + }); + const response = await runtime.handler( + new Request( + "https://example.test/webhooks/dispatches/dispatch-failed-1/replay", + { + headers: subjectHeaders, + method: "POST", + } + ) + ); + expect(response.status).toBe(403); + }); + it("defaults omitted webhook signing key rotation grace period", async () => { const runtime = dsarInstance({ config: { diff --git a/packages/backend/test/services/notifications.service.test.ts b/packages/backend/test/services/notifications.service.test.ts index da15e5c..e04d6f1 100644 --- a/packages/backend/test/services/notifications.service.test.ts +++ b/packages/backend/test/services/notifications.service.test.ts @@ -21,6 +21,7 @@ import { deriveLifecycleNotificationDrafts } from "../../src/events/contracts"; import { emitNotificationEvent, makeNotificationDraft, + replayWebhookDispatch, } from "../../src/services/notifications/service"; import { RuntimeServicesTag } from "../../src/types/runtime"; import type { @@ -130,6 +131,30 @@ const makeMemoryPersistence = (): { attempts.push(record); return Effect.succeed(record); }, + getById: (id: string) => + Effect.fromNullishOr( + attempts.find((attempt) => attempt.id === id) + ).pipe( + Effect.mapError( + () => new Error(`missing notification attempt in test: ${id}`) + ) + ), + list: (input) => + Effect.succeed( + attempts.filter((attempt) => { + if (input?.channel && attempt.channel !== input.channel) { + return false; + } + if ( + input?.status && + input.status.length > 0 && + !input.status.includes(attempt.status) + ) { + return false; + } + return true; + }) + ), listByNotificationEventId: (notificationEventId: string) => Effect.succeed( attempts.filter( @@ -352,6 +377,10 @@ const makeServices = (input: { : [ { capability: "notifications", + channels: + input.adapterKey === "outbound-resend" + ? (["email"] as const) + : (["webhook"] as const), diagnostics: () => Effect.succeed({ capability: "notifications", @@ -759,4 +788,48 @@ describe("notification retry/backoff behavior", () => { "clock_due_changed" ); }); + + it("rejects replay for persisted notification events with unsupported event types", async () => { + const memory = makeMemoryPersistence(); + let sendCount = 0; + const services = makeServices({ + dispatch: { + send: () => + Effect.sync(() => { + sendCount += 1; + return { + responseCode: 202, + status: "delivered" as const, + }; + }), + }, + persistence: memory.persistence, + retryDelayMs: 0, + retryMaxAttempts: 1, + }); + + await expect( + Effect.runPromise( + replayWebhookDispatch({ + event: { + correlationId: "corr-corrupt-event", + createdAt: "2026-01-01T00:00:00.000Z", + eventType: "corrupt_event_type", + id: "evt-corrupt-event", + idempotencyKey: "original-corrupt-event", + locale: "en-GB", + payload: {}, + policyVersion: "policy-v1", + requestId: "req-corrupt-event", + tenantId: "tenant-default", + }, + idempotencyKey: "replay-corrupt-event", + tenantId: "tenant-default", + }).pipe(Effect.provideService(RuntimeServicesTag, services)) + ) + ).rejects.toThrow( + "Webhook dispatch cannot be replayed because the persisted notification event type is not supported." + ); + expect(sendCount).toBe(0); + }); }); diff --git a/packages/cli/src/commands/factory.ts b/packages/cli/src/commands/factory.ts index a6e2244..69fc63d 100644 --- a/packages/cli/src/commands/factory.ts +++ b/packages/cli/src/commands/factory.ts @@ -26,6 +26,7 @@ export const makeRouteCommand = (routeId: string): CommandDefinition => { const request = await requestFromRoute(route, ctx.input, ctx.params); return await ctx.api.invoke(request); }, + ...(route.flagHelp ? { flagHelp: route.flagHelp } : {}), id: route.id, routeId: route.id, usage: route.command, diff --git a/packages/cli/src/commands/helpers.ts b/packages/cli/src/commands/helpers.ts index 446ca89..536f2c6 100644 --- a/packages/cli/src/commands/helpers.ts +++ b/packages/cli/src/commands/helpers.ts @@ -86,6 +86,44 @@ const parseCreateIntakePayload = ( return { intakeSource }; }; +const parsePositiveIntegerFlag = ( + flags: Readonly>, + key: string +): number | undefined => { + const raw = flags[key]; + if (!raw) { + return undefined; + } + const parsed = Number.parseInt(raw, 10); + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new Error(`Invalid --${key}: must be a positive integer.`); + } + return parsed; +}; + +const parseWebhookBulkReplayPayload = ( + flags: Readonly> +): Readonly> => { + const payload: Record = {}; + if (flags.status) { + payload.status = flags.status; + } + if (flags["endpoint-id"]) { + payload.endpoint_id = flags["endpoint-id"]; + } + if (flags["created-after"] ?? flags.since) { + payload.created_after = flags["created-after"] ?? flags.since; + } + if (flags["created-before"] ?? flags.until) { + payload.created_before = flags["created-before"] ?? flags.until; + } + const limit = parsePositiveIntegerFlag(flags, "limit"); + if (limit !== undefined) { + payload.limit = limit; + } + return payload; +}; + const binaryUploadRouteIds = new Set([ "requests_verification_evidence_upload", "requests_manifest_artifact_upload", @@ -131,6 +169,9 @@ const payloadForRoute = ( const upload = resolveUploadFileMeta(input); return readFile(upload.filePath).then((bytes) => new Uint8Array(bytes)); } + if (route.id === "webhooks_dispatches_replay_bulk") { + return parseWebhookBulkReplayPayload(input.flags); + } const parsedJson = getJsonBody(input.flags); if (parsedJson !== undefined) { return parsedJson; @@ -179,6 +220,16 @@ const queryForRoute = ( until: input.flags.until, }; } + if (route.id === "webhooks_dispatches_list") { + return { + created_after: input.flags["created-after"] ?? input.flags.since, + created_before: input.flags["created-before"] ?? input.flags.until, + endpoint_id: input.flags["endpoint-id"], + limit: input.flags.limit, + offset: input.flags.offset, + status: input.flags.status, + }; + } if (route.id === "requests_manifest_artifact_download") { return { key: requireFlag( @@ -232,6 +283,18 @@ const headersForRoute = ( : {}), }; } + if ( + route.id === "webhooks_dispatches_replay" || + route.id === "webhooks_dispatches_replay_bulk" + ) { + return { + "x-idempotency-key": requireFlag( + input.flags, + "idempotency-key", + "Missing required --idempotency-key for webhook replay command." + ), + }; + } return undefined; }; diff --git a/packages/cli/src/commands/webhooks-tail.ts b/packages/cli/src/commands/webhooks-tail.ts new file mode 100644 index 0000000..c3165df --- /dev/null +++ b/packages/cli/src/commands/webhooks-tail.ts @@ -0,0 +1,244 @@ +/* oxlint-disable max-statements, promise/avoid-new, promise/no-multiple-resolved */ +import type { + ApiClient, + CommandDefinition, + CommandExecutionContext, +} from "../types"; + +const DEFAULT_INTERVAL_MS = 2000; +const DEFAULT_LIMIT = 200; +const POLL_DRAIN_PAGE_CAP = 50; + +interface WebhookDispatchTailEvent { + readonly createdAt: string; + readonly dispatchId: string; + readonly endpointId?: string; + readonly error?: string; + readonly eventId: string; + readonly eventType?: string; + readonly requestId: string; + readonly status: string; +} + +interface WebhookDispatchListEnvelope { + readonly data?: { + readonly items?: readonly WebhookDispatchTailEvent[]; + }; +} + +const parseIntegerFlag = ( + flags: Readonly>, + key: string, + fallback: number +): number => { + const raw = flags[key]; + if (!raw) { + return fallback; + } + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed) || parsed <= 0) { + throw new Error(`Invalid --${key}: must be a positive integer.`); + } + return parsed; +}; + +const isEnabledFlag = ( + flags: Readonly>, + key: string +): boolean => flags[key] === "true"; + +const parseMaxPolls = (flags: Readonly>): number => { + if (isEnabledFlag(flags, "once")) { + return 1; + } + if (flags["max-polls"]) { + return parseIntegerFlag(flags, "max-polls", 0); + } + return Number.POSITIVE_INFINITY; +}; + +const sleep = (ms: number, signal: AbortSignal): Promise => { + if (signal.aborted) { + return Promise.resolve(); + } + return new Promise((resolve) => { + const state: { settled: boolean; timer?: ReturnType } = { + settled: false, + }; + const finish = () => { + if (state.settled) { + return; + } + state.settled = true; + signal.removeEventListener("abort", finish); + if (state.timer !== undefined) { + clearTimeout(state.timer); + } + resolve(); + }; + state.timer = setTimeout(finish, ms); + signal.addEventListener("abort", finish, { once: true }); + }); +}; + +const formatLine = ( + event: WebhookDispatchTailEvent, + output: "json" | "text" +): string => { + if (output === "json") { + return JSON.stringify(event); + } + return `[${event.createdAt}] ${event.status} dispatch=${event.dispatchId} event=${event.eventId}${ + event.endpointId ? ` endpoint=${event.endpointId}` : "" + } request=${event.requestId}${event.error ? ` error=${event.error}` : ""}`; +}; + +const fetchDispatchPage = async ( + api: ApiClient, + input: { + readonly createdAfter?: string; + readonly endpointId?: string; + readonly limit: number; + readonly offset: number; + readonly status?: string; + } +): Promise => { + const response = (await api.invoke({ + method: "GET", + path: "/webhooks/dispatches", + query: { + created_after: input.createdAfter, + endpoint_id: input.endpointId, + limit: String(input.limit), + offset: String(input.offset), + status: input.status, + }, + })) as WebhookDispatchListEnvelope; + return response.data?.items ?? []; +}; + +/** + * Drains every `/webhooks/dispatches` page matching the current + * `created_after` watermark, returning dispatches in ascending creation + * order. Walking `offset` pages until a short page is what prevents a + * burst larger than `--limit` from silently dropping the older tail of + * the burst on the next iteration (the watermark would otherwise jump + * over unseen dispatches). Rows inserted mid-drain shift pages and can + * repeat across pages; the caller dedupes by dispatch id. + */ +const pollOnce = async ( + api: ApiClient, + input: { + readonly createdAfter?: string; + readonly endpointId?: string; + readonly limit: number; + readonly status?: string; + } +): Promise => { + const collected: WebhookDispatchTailEvent[] = []; + let offset = 0; + let pages = 0; + for (;;) { + const items = await fetchDispatchPage(api, { ...input, offset }); + collected.push(...items); + pages += 1; + if (items.length < input.limit || pages >= POLL_DRAIN_PAGE_CAP) { + break; + } + offset += input.limit; + } + return collected.toSorted((left, right) => { + const order = left.createdAt.localeCompare(right.createdAt); + return order === 0 + ? left.dispatchId.localeCompare(right.dispatchId) + : order; + }); +}; + +/** + * Runs the `dsar webhooks tail` polling loop, emitting newly observed + * outbound webhook dispatches until cancelled, `--once` completes, or a + * test-only `--max-polls` bound is reached. + * + * @param ctx - Command execution context carrying flags, API client, and + * output sink. + * @param signal - Abort signal used to stop polling gracefully. + * @returns A summary with poll count and emitted dispatch count. + */ +export const runWebhookTailLoop = async ( + ctx: CommandExecutionContext, + signal: AbortSignal +): Promise<{ + readonly emitted: number; + readonly polls: number; +}> => { + const intervalMs = parseIntegerFlag( + ctx.input.flags, + "interval", + DEFAULT_INTERVAL_MS + ); + const limit = parseIntegerFlag(ctx.input.flags, "limit", DEFAULT_LIMIT); + const maxPolls = parseMaxPolls(ctx.input.flags); + const outputMode = ctx.input.global.output; + const endpointId = ctx.input.flags["endpoint-id"]; + const { status } = ctx.input.flags; + let seenAtWatermark = new Set(); + let createdAfter = ctx.input.flags["created-after"] ?? ctx.input.flags.since; + let polls = 0; + let emitted = 0; + while (!signal.aborted && polls < maxPolls) { + polls += 1; + const events = await pollOnce(ctx.api, { + createdAfter, + endpointId, + limit, + status, + }); + for (const event of events) { + if (seenAtWatermark.has(event.dispatchId)) { + continue; + } + ctx.writeLine(formatLine(event, outputMode)); + emitted += 1; + if (createdAfter === undefined || event.createdAt > createdAfter) { + createdAfter = event.createdAt; + seenAtWatermark = new Set([event.dispatchId]); + } else { + seenAtWatermark.add(event.dispatchId); + } + } + if (polls >= maxPolls || signal.aborted) { + break; + } + await sleep(intervalMs, signal); + } + return { emitted, polls }; +}; + +/** + * `dsar webhooks tail` — polls outbound webhook dispatches and streams newly + * observed attempts to stdout. Exits on SIGINT. + */ +export const webhooksTailCommand: CommandDefinition = { + description: "Stream outbound webhook dispatches.", + execute: async (ctx) => { + const controller = new AbortController(); + const onInterrupt = () => controller.abort(); + process.once("SIGINT", onInterrupt); + try { + return await runWebhookTailLoop(ctx, controller.signal); + } finally { + process.removeListener("SIGINT", onInterrupt); + } + }, + flagHelp: [ + "--status Comma-separated delivery statuses (e.g. failed)", + "--endpoint-id Filter by configured webhook endpoint id", + "--created-after Start tailing after this timestamp", + "--interval Poll interval in milliseconds (default 2000)", + "--limit Page size per poll, 1-500 (default 200)", + "--once Poll once and exit", + ], + id: "webhooks_tail", + usage: ["webhooks", "tail"], +}; diff --git a/packages/cli/src/commands/webhooks.ts b/packages/cli/src/commands/webhooks.ts index c8ffbc7..63f2ebc 100644 --- a/packages/cli/src/commands/webhooks.ts +++ b/packages/cli/src/commands/webhooks.ts @@ -1,11 +1,18 @@ import { makeRouteCommands } from "../commands/factory"; +import { webhooksTailCommand } from "../commands/webhooks-tail"; /** * CLI commands for triggering inbound webhook endpoints * (e.g. Resend email intake) from the command line. */ -export const webhooksCommands = makeRouteCommands([ - "webhooks_inbound_resend", - "webhooks_inbound_slack", - "webhooks_endpoint_rotate_key", -] as const); +export const webhooksCommands = [ + ...makeRouteCommands([ + "webhooks_inbound_resend", + "webhooks_inbound_slack", + "webhooks_endpoint_rotate_key", + "webhooks_dispatches_list", + "webhooks_dispatches_replay", + "webhooks_dispatches_replay_bulk", + ] as const), + webhooksTailCommand, +] as const; diff --git a/packages/cli/src/config.ts b/packages/cli/src/config.ts index 5d6b3c2..1ee8377 100644 --- a/packages/cli/src/config.ts +++ b/packages/cli/src/config.ts @@ -24,7 +24,13 @@ const parseFlags = ( commandTokens.push(token ?? ""); continue; } - const key = stripLeadingDashes(token); + const rawKey = stripLeadingDashes(token); + const equalsIndex = rawKey.indexOf("="); + if (equalsIndex > 0) { + flags[rawKey.slice(0, equalsIndex)] = rawKey.slice(equalsIndex + 1); + continue; + } + const key = rawKey; const next = argv[index + 1]; if (!next || next.startsWith("--")) { flags[key] = "true"; diff --git a/packages/cli/src/parity/route-map.ts b/packages/cli/src/parity/route-map.ts index e1cbd1d..52375b6 100644 --- a/packages/cli/src/parity/route-map.ts +++ b/packages/cli/src/parity/route-map.ts @@ -97,6 +97,44 @@ export const routeParityMap: readonly RouteParityDefinition[] = [ method: "POST", path: "/webhooks/endpoints/{id}/rotate-key", }, + { + command: ["webhooks", "list"], + description: "List outbound webhook dispatches.", + flagHelp: [ + "--status Comma-separated delivery statuses (e.g. failed)", + "--endpoint-id Filter by configured webhook endpoint id", + "--created-after Only include attempts created after this time", + "--created-before Only include attempts created before this time", + "--limit Page size, 1-500 (default 50)", + "--offset Zero-based page offset", + ], + id: "webhooks_dispatches_list", + method: "GET", + path: "/webhooks/dispatches", + }, + { + command: ["webhooks", "replay", ":id"], + description: "Replay outbound webhook dispatch.", + flagHelp: ["--idempotency-key Replay dedupe key (required)"], + id: "webhooks_dispatches_replay", + method: "POST", + path: "/webhooks/dispatches/{id}/replay", + }, + { + command: ["webhooks", "replay-all"], + description: "Replay failed outbound webhook dispatches.", + flagHelp: [ + "--idempotency-key Bulk replay dedupe key (required)", + "--status Must be 'failed' when present", + "--endpoint-id Filter by configured webhook endpoint id", + "--created-after Only replay attempts created after this time", + "--created-before Only replay attempts created before this time", + "--limit Maximum dispatches to replay, 1-100", + ], + id: "webhooks_dispatches_replay_bulk", + method: "POST", + path: "/webhooks/dispatches/replay", + }, { command: ["requests", "create"], description: "Create request.", diff --git a/packages/cli/src/types.ts b/packages/cli/src/types.ts index 3ef0f4d..b5b6fdc 100644 --- a/packages/cli/src/types.ts +++ b/packages/cli/src/types.ts @@ -94,6 +94,11 @@ export interface CommandExecutionContext { * Registry definition for a CLI command. */ export interface CommandDefinition { + /** + * Preformatted command-specific flag lines rendered by `--help`, e.g. + * `"--request Request id to tail (required)"`. + */ + readonly flagHelp?: readonly string[]; /** Stable command identifier used in registry/help tooling. */ readonly id: string; /** Optional route parity id linking command to HTTP surface. */ @@ -120,4 +125,9 @@ export interface RouteParityDefinition { readonly command: readonly string[]; /** Human-readable parity mapping description. */ readonly description: string; + /** + * Preformatted command-specific flag lines rendered by `--help`, e.g. + * `"--status Filter by delivery status"`. + */ + readonly flagHelp?: readonly string[]; } diff --git a/packages/cli/test/config.test.ts b/packages/cli/test/config.test.ts index 44d5252..fc62df5 100644 --- a/packages/cli/test/config.test.ts +++ b/packages/cli/test/config.test.ts @@ -35,6 +35,19 @@ describe(parseCliInput, () => { expect(parsed.global.output).toBe("json"); }); + it("parses --flag=value arguments", () => { + const parsed = parseCliInput({ + argv: ["webhooks", "list", "--status=failed", "--limit=10"], + env: { + DSAR_API_URL: "https://example.test", + }, + fetchImpl: fetch, + }); + expect(parsed.commandTokens).toStrictEqual(["webhooks", "list"]); + expect(parsed.flags.status).toBe("failed"); + expect(parsed.flags.limit).toBe("10"); + }); + it("throws when api url is missing", () => { expect(() => parseCliInput({ diff --git a/packages/cli/test/e2e/commands.e2e.test.ts b/packages/cli/test/e2e/commands.e2e.test.ts index 1db2700..9c9b74b 100644 --- a/packages/cli/test/e2e/commands.e2e.test.ts +++ b/packages/cli/test/e2e/commands.e2e.test.ts @@ -146,6 +146,43 @@ const commandCases: readonly CommandCase[] = [ id: "webhooks_endpoint_rotate_key", outputIncludes: ['"endpointId":"default"', '"newSigningSecret"'], }, + { + argv: ["webhooks", "list", "--status=failed"], + expectedExitCode: 0, + id: "webhooks_dispatches_list", + outputIncludes: ['"items":[]', '"total":0'], + }, + { + argv: ["webhooks", "replay", "dispatch-1"], + expectedExitCode: 1, + id: "webhooks_dispatches_replay", + outputIncludes: ["--idempotency-key"], + }, + { + argv: ["webhooks", "replay", "dispatch-1", "--idempotency-key=replay-1"], + expectedExitCode: 1, + id: "webhooks_dispatches_replay", + outputIncludes: ["dispatch-1"], + }, + { + argv: ["webhooks", "replay-all"], + expectedExitCode: 1, + id: "webhooks_dispatches_replay_bulk", + outputIncludes: ["--idempotency-key"], + }, + { + argv: [ + "webhooks", + "replay-all", + "--status=failed", + "--endpoint-id=default", + "--limit=10", + "--idempotency-key=replay-all-1", + ], + expectedExitCode: 0, + id: "webhooks_dispatches_replay_bulk", + outputIncludes: ['"results":[]', '"total":0'], + }, { argv: ["requests", "create", "--json", commonCreateBody], expectedExitCode: 0, diff --git a/packages/cli/test/e2e/parity-guard.e2e.test.ts b/packages/cli/test/e2e/parity-guard.e2e.test.ts index 9dcb814..c04b3a9 100644 --- a/packages/cli/test/e2e/parity-guard.e2e.test.ts +++ b/packages/cli/test/e2e/parity-guard.e2e.test.ts @@ -4,7 +4,7 @@ import { describe, expect, it } from "@effect/vitest"; import { allCommands } from "#src/commands/registry"; import { routeParityMap } from "#src/parity/route-map"; -const NON_ROUTE_COMMAND_IDS = new Set(["audit_tail"]); +const NON_ROUTE_COMMAND_IDS = new Set(["audit_tail", "webhooks_tail"]); describe("cLI e2e parity guard", () => { it("covers every registry command id in e2e matrix", () => { diff --git a/packages/cli/test/webhooks-tail.test.ts b/packages/cli/test/webhooks-tail.test.ts new file mode 100644 index 0000000..ce6e479 --- /dev/null +++ b/packages/cli/test/webhooks-tail.test.ts @@ -0,0 +1,268 @@ +/* oxlint-disable jest/no-conditional-in-test, max-statements -- test helpers branch on stub state. */ +import { describe, expect, it } from "@effect/vitest"; + +import { webhooksCommands } from "#src/commands/webhooks"; +import { runWebhookTailLoop } from "#src/commands/webhooks-tail"; +import type { + ApiClient, + ApiRequest, + CommandExecutionContext, + GlobalCliConfig, +} from "#src/types"; + +const makeGlobal = ( + override: Partial = {} +): GlobalCliConfig => ({ + apiUrl: "https://example.test", + fetch, + output: "json", + ...override, +}); + +const makeContext = (input: { + readonly flags: Readonly>; + readonly invoke: (request: ApiRequest) => Promise; + readonly writeLine: (line: string) => void; + readonly globalOverride?: Partial; +}): CommandExecutionContext => ({ + api: { invoke: input.invoke } as ApiClient, + input: { + commandTokens: ["webhooks", "tail"], + flags: input.flags, + global: makeGlobal(input.globalOverride), + }, + params: {}, + writeLine: input.writeLine, +}); + +const makeDispatch = (index: number) => ({ + createdAt: `2026-01-01T00:00:0${index}.000Z`, + dispatchId: `dispatch-${index}`, + eventId: `evt-${index}`, + requestId: `req-${index}`, + status: "failed", +}); + +describe("webhooks tail polling loop", () => { + it("streams dispatches and advances the created_after cursor", async () => { + const calls: ApiRequest[] = []; + const lines: string[] = []; + const pages: readonly (readonly { + readonly createdAt: string; + readonly dispatchId: string; + readonly eventId: string; + readonly requestId: string; + readonly status: string; + }[])[] = [ + [ + { + createdAt: "2026-01-01T00:00:00.000Z", + dispatchId: "dispatch-1", + eventId: "evt-1", + requestId: "req-1", + status: "failed", + }, + ], + [ + { + createdAt: "2026-01-02T00:00:00.000Z", + dispatchId: "dispatch-2", + eventId: "evt-2", + requestId: "req-2", + status: "delivered", + }, + ], + ]; + let pollIndex = 0; + const ctx = makeContext({ + flags: { + interval: "1", + "max-polls": "2", + status: "failed", + }, + invoke: (request) => { + calls.push(request); + const items = pages[pollIndex] ?? []; + pollIndex += 1; + return Promise.resolve({ data: { items } }); + }, + writeLine: (line) => lines.push(line), + }); + const result = await runWebhookTailLoop(ctx, new AbortController().signal); + expect(result.polls).toBe(2); + expect(result.emitted).toBe(2); + expect(lines.map((line) => JSON.parse(line).dispatchId)).toStrictEqual([ + "dispatch-1", + "dispatch-2", + ]); + expect(calls[0]?.query?.status).toBe("failed"); + expect(calls[1]?.query?.created_after).toBe("2026-01-01T00:00:00.000Z"); + }); + + it("drains every page when a burst exceeds the poll limit", async () => { + const lines: string[] = []; + const calls: ApiRequest[] = []; + const pagesByOffset: Readonly> = { + // Newest-first pages, mirroring the DESC ordering of the endpoint. + "0": [makeDispatch(3), makeDispatch(2)], + "2": [makeDispatch(1)], + }; + const ctx = makeContext({ + flags: { + interval: "1", + limit: "2", + "max-polls": "1", + }, + invoke: (request) => { + calls.push(request); + const offset = request.query?.offset ?? "0"; + return Promise.resolve({ + data: { items: pagesByOffset[offset] ?? [] }, + }); + }, + writeLine: (line) => lines.push(line), + }); + const result = await runWebhookTailLoop(ctx, new AbortController().signal); + expect(result.polls).toBe(1); + expect(result.emitted).toBe(3); + expect(lines.map((line) => JSON.parse(line).dispatchId)).toStrictEqual([ + "dispatch-1", + "dispatch-2", + "dispatch-3", + ]); + expect(calls.map((call) => call.query?.offset)).toStrictEqual(["0", "2"]); + }); + + it("deduplicates dispatches seen across overlapping polls", async () => { + const lines: string[] = []; + let polls = 0; + const duplicate = { + createdAt: "2026-01-01T00:00:00.000Z", + dispatchId: "dispatch-1", + eventId: "evt-1", + requestId: "req-1", + status: "failed", + }; + const ctx = makeContext({ + flags: { + interval: "1", + "max-polls": "2", + }, + invoke: () => { + polls += 1; + return Promise.resolve({ data: { items: [duplicate] } }); + }, + writeLine: (line) => lines.push(line), + }); + const result = await runWebhookTailLoop(ctx, new AbortController().signal); + expect(polls).toBe(2); + expect(result.emitted).toBe(1); + expect(lines).toHaveLength(1); + }); + + it("runs exactly one poll in once mode and forwards filters", async () => { + const calls: ApiRequest[] = []; + const ctx = makeContext({ + flags: { + "created-after": "2026-01-01T00:00:00.000Z", + "endpoint-id": "default", + limit: "25", + once: "true", + status: "failed", + }, + invoke: (request) => { + calls.push(request); + return Promise.resolve({ data: { items: [] } }); + }, + writeLine: () => { + // no-op + }, + }); + const result = await runWebhookTailLoop(ctx, new AbortController().signal); + expect(result.polls).toBe(1); + expect(calls).toHaveLength(1); + expect(calls[0]?.query).toMatchObject({ + created_after: "2026-01-01T00:00:00.000Z", + endpoint_id: "default", + limit: "25", + status: "failed", + }); + }); + + it("formats dispatches in text mode", async () => { + const lines: string[] = []; + const ctx = makeContext({ + flags: { once: "true" }, + globalOverride: { output: "text" }, + invoke: () => + Promise.resolve({ + data: { + items: [ + { + createdAt: "2026-01-01T00:00:00.000Z", + dispatchId: "dispatch-1", + endpointId: "default", + error: "500", + eventId: "evt-1", + requestId: "req-1", + status: "failed", + }, + ], + }, + }), + writeLine: (line) => lines.push(line), + }); + await runWebhookTailLoop(ctx, new AbortController().signal); + expect(lines[0]).toBe( + "[2026-01-01T00:00:00.000Z] failed dispatch=dispatch-1 event=evt-1 endpoint=default request=req-1 error=500" + ); + }); + + it("rejects invalid interval and propagates API errors", async () => { + const invalidCtx = makeContext({ + flags: { interval: "0" }, + invoke: () => Promise.resolve({ data: { items: [] } }), + writeLine: () => { + // no-op + }, + }); + await expect( + runWebhookTailLoop(invalidCtx, new AbortController().signal) + ).rejects.toThrow(/--interval/); + + const failingCtx = makeContext({ + flags: { once: "true" }, + invoke: () => Promise.reject(new Error("api unavailable")), + writeLine: () => { + // no-op + }, + }); + await expect( + runWebhookTailLoop(failingCtx, new AbortController().signal) + ).rejects.toThrow(/api unavailable/); + }); +}); + +describe("webhook command flag help", () => { + it.each([ + ["webhooks_dispatches_list", ["--status", "--endpoint-id", "--limit"]], + ["webhooks_dispatches_replay", ["--idempotency-key"]], + [ + "webhooks_dispatches_replay_bulk", + ["--idempotency-key", "--status", "--limit"], + ], + ["webhooks_tail", ["--status", "--interval", "--once"]], + ] as const)( + "declares command flag help for %s", + (commandId, expectedFlags) => { + const command = webhooksCommands.find((entry) => entry.id === commandId); + if (!command) { + throw new Error(`Missing webhook command '${commandId}'.`); + } + const helpText = (command.flagHelp ?? []).join("\n"); + for (const flag of expectedFlags) { + expect(helpText).toContain(flag); + } + } + ); +}); diff --git a/packages/internals/persistence/src/index.ts b/packages/internals/persistence/src/index.ts index 1c837a2..4629fe2 100644 --- a/packages/internals/persistence/src/index.ts +++ b/packages/internals/persistence/src/index.ts @@ -33,6 +33,7 @@ export type { FulfillmentArtifactRecord, JsonValue, ListAuditEventsInput, + ListNotificationDeliveryAttemptsInput, ListRequestsBySubjectInput, NotificationDeliveryAttemptRecord, NotificationDeliveryStatus, diff --git a/packages/internals/persistence/src/services/persistence.ts b/packages/internals/persistence/src/services/persistence.ts index 1eda8f6..6f48c0e 100644 --- a/packages/internals/persistence/src/services/persistence.ts +++ b/packages/internals/persistence/src/services/persistence.ts @@ -21,6 +21,7 @@ import type { CreateNotificationEventInput, FulfillmentArtifactsRepository, ListAuditEventsInput, + ListNotificationDeliveryAttemptsInput, ListRequestsBySubjectInput, NotificationDeliveryAttemptsRepository, NotificationEventsRepository, @@ -994,6 +995,104 @@ const makePersistence = ( ); return yield* mapNotificationDeliveryAttemptRecordEffect(row); }), + count: (input?: ListNotificationDeliveryAttemptsInput) => + Effect.gen(function* countNotificationDeliveryAttempts() { + const tenantId = yield* requireTenantId; + const clauses: (SqlFragment | SqlStatement)[] = [ + sql`tenant_id = ${tenantId}`, + ]; + if (input?.channel) { + clauses.push(sql`channel = ${input.channel}`); + } + const statusValues = input?.status ?? []; + if (statusValues.length > 0) { + clauses.push(sql.in("status", [...new Set(statusValues)])); + } + if (input?.destination) { + clauses.push(sql`destination = ${input.destination}`); + } + if (input?.createdAfter) { + clauses.push(sql`created_at > ${input.createdAfter}`); + } + if (input?.createdBefore) { + clauses.push(sql`created_at < ${input.createdBefore}`); + } + const rows = yield* sql<{ + readonly count: bigint | number | string; + }>`SELECT COUNT(*) AS count FROM notification_delivery_attempts + WHERE ${sql.and(clauses)}`; + return Number(rows[0]?.count ?? 0); + }), + getById: (id) => + Effect.gen(function* getNotificationDeliveryAttemptById() { + const tenantId = yield* requireTenantId; + const rows = yield* sql<{ + readonly id: string; + readonly tenant_id: string; + readonly notification_event_id: string; + readonly request_id: string; + readonly channel: string; + readonly destination: string; + readonly attempt: number; + readonly status: string; + readonly response_code: number | null; + readonly error_text: string | null; + readonly created_at: string; + }>`SELECT * FROM notification_delivery_attempts + WHERE tenant_id = ${tenantId} AND id = ${id} + LIMIT 1`; + const row = yield* findRequired( + rows[0], + "notification_delivery_attempts", + id + ); + return yield* mapNotificationDeliveryAttemptRecordEffect(row); + }), + list: (input?: ListNotificationDeliveryAttemptsInput) => + Effect.gen(function* listNotificationDeliveryAttempts() { + const tenantId = yield* requireTenantId; + const limit = limitWithFallback(input?.limit); + const offset = offsetWithFallback(input?.offset); + const clauses: (SqlFragment | SqlStatement)[] = [ + sql`tenant_id = ${tenantId}`, + ]; + if (input?.channel) { + clauses.push(sql`channel = ${input.channel}`); + } + const statusValues = input?.status ?? []; + if (statusValues.length > 0) { + clauses.push(sql.in("status", [...new Set(statusValues)])); + } + if (input?.destination) { + clauses.push(sql`destination = ${input.destination}`); + } + if (input?.createdAfter) { + clauses.push(sql`created_at > ${input.createdAfter}`); + } + if (input?.createdBefore) { + clauses.push(sql`created_at < ${input.createdBefore}`); + } + const rows = yield* sql<{ + readonly id: string; + readonly tenant_id: string; + readonly notification_event_id: string; + readonly request_id: string; + readonly channel: string; + readonly destination: string; + readonly attempt: number; + readonly status: string; + readonly response_code: number | null; + readonly error_text: string | null; + readonly created_at: string; + }>`SELECT * FROM notification_delivery_attempts + WHERE ${sql.and(clauses)} + ORDER BY created_at DESC, id DESC + LIMIT ${limit} OFFSET ${offset}`; + return yield* Effect.forEach( + rows, + mapNotificationDeliveryAttemptRecordEffect + ); + }), listByNotificationEventId: (notificationEventId) => Effect.gen(function* listNotificationDeliveryAttemptsByEventId() { const tenantId = yield* requireTenantId; diff --git a/packages/internals/persistence/src/types/domain.ts b/packages/internals/persistence/src/types/domain.ts index 8c7ffcb..917b8b7 100644 --- a/packages/internals/persistence/src/types/domain.ts +++ b/packages/internals/persistence/src/types/domain.ts @@ -733,6 +733,28 @@ export interface CreateNotificationDeliveryAttemptInput { readonly createdAt: string; } +/** + * Filter contract for notification delivery-attempt lookup. + * + * @public + */ +export interface ListNotificationDeliveryAttemptsInput { + /** Optional channel filter, for example `webhook` or `email`. */ + readonly channel?: string; + /** Optional delivery statuses to include. */ + readonly status?: readonly NotificationDeliveryStatus[]; + /** Optional exact destination filter. */ + readonly destination?: string; + /** Return attempts created strictly after this ISO timestamp. */ + readonly createdAfter?: string; + /** Return attempts created strictly before this ISO timestamp. */ + readonly createdBefore?: string; + /** Maximum rows to read for a single page. */ + readonly limit?: number; + /** Number of matching rows to skip. */ + readonly offset?: number; +} + /** * Persisted outbound webhook endpoint configuration. * @@ -1443,6 +1465,46 @@ export interface NotificationDeliveryAttemptsRepository { PersistenceError | SqlError, TenantContext >; + /** + * Retrieves a delivery attempt by id. + * + * @param id - Delivery attempt identifier to look up. + * @returns The matching {@link NotificationDeliveryAttemptRecord}. + * @throws {@link PersistenceError} when no record exists for `id`. + * @throws {@link SqlError} on underlying database failures. + */ + readonly getById: ( + id: string + ) => Effect.Effect< + NotificationDeliveryAttemptRecord, + PersistenceError | SqlError, + TenantContext + >; + /** + * Lists delivery attempts for dispatch inspection surfaces. + * + * @param input - Optional filters. + * @returns Ordered {@link NotificationDeliveryAttemptRecord} entries. + * @throws {@link PersistenceError} on mapping failures. + * @throws {@link SqlError} on underlying database failures. + */ + readonly list: ( + input?: ListNotificationDeliveryAttemptsInput + ) => Effect.Effect< + readonly NotificationDeliveryAttemptRecord[], + PersistenceError | SqlError, + TenantContext + >; + /** + * Counts delivery attempts matching dispatch inspection filters. + * + * @param input - Optional filters. + * @returns Number of matching delivery attempts. + * @throws {@link SqlError} on underlying database failures. + */ + readonly count: ( + input?: ListNotificationDeliveryAttemptsInput + ) => Effect.Effect; /** * Lists all delivery attempts for a given notification event. * diff --git a/packages/outbound-resend/src/adapter.ts b/packages/outbound-resend/src/adapter.ts index f09b083..bee090b 100644 --- a/packages/outbound-resend/src/adapter.ts +++ b/packages/outbound-resend/src/adapter.ts @@ -251,6 +251,7 @@ export const makeOutboundResendAdapter = ( return { capability: "notifications", + channels: ["email"], diagnostics: () => Effect.succeed({ capability: "notifications",