From 564334f55a70c593d6c1628a6ff47e7452c42bb2 Mon Sep 17 00:00:00 2001 From: arisu6804 Date: Thu, 27 Aug 2026 20:51:05 +0530 Subject: [PATCH] feat(webhooks): add durable replay-safe delivery state --- docs/WEBHOOK_DELIVERY_DURABILITY.md | 116 ++++++++++++++++ src/webhooks/durableDelivery.matrix.test.ts | 124 +++++++++++++++++ src/webhooks/durableDelivery.test.ts | 115 +++++++++++++++ src/webhooks/durableDelivery.ts | 146 ++++++++++++++++++++ 4 files changed, 501 insertions(+) create mode 100644 docs/WEBHOOK_DELIVERY_DURABILITY.md create mode 100644 src/webhooks/durableDelivery.matrix.test.ts create mode 100644 src/webhooks/durableDelivery.test.ts create mode 100644 src/webhooks/durableDelivery.ts diff --git a/docs/WEBHOOK_DELIVERY_DURABILITY.md b/docs/WEBHOOK_DELIVERY_DURABILITY.md new file mode 100644 index 00000000..31f387b9 --- /dev/null +++ b/docs/WEBHOOK_DELIVERY_DURABILITY.md @@ -0,0 +1,116 @@ +# Durable webhook delivery + +Webhook delivery is at-least-once at the provider boundary, so process memory +cannot be the source of truth. `InMemoryDurableDeliveryStore` documents and +tests the production contract: persist one tenant-scoped event key, payload +fingerprint, delivery state, attempt count, lease, next-attempt time, and +terminal error metadata in durable storage. + +The state machine is: + +`pending -> processing -> delivered` + +`processing -> retrying -> processing` (bounded exponential backoff) + +`processing -> dead` (attempt budget exhausted) + +Only the worker holding the current lease may renew, complete, or fail a +delivery. An expired lease is reclaimable by another worker after restart. +Creating an existing key with the same body is a duplicate and returns the +original record; changing the body is a 409 conflict and must not be sent. + +Production persistence must enforce a unique `(tenant_id, event_key)` index and +perform claims atomically (`SELECT ... FOR UPDATE SKIP LOCKED` or an equivalent +conditional update). The in-memory adapter is intentionally not a production +database. Store the payload needed for delivery, but never store webhook +secrets in the delivery row or copy them into error messages. + +Workers should acknowledge the row only after the provider returns success. +They must retain the event key as the provider idempotency header so a crash +after the provider accepts a request but before `complete` does not create an +unbounded duplicate effect. Dead rows must be visible to operators, with a +manual replay path that creates a new event key after the cause is corrected. + +Validation coverage includes duplicate and conflict admission, tenant +isolation, lease ownership and expiry, restart recovery, retry timing, dead +lettering, malformed JSON, URL sanitization, and record immutability. + +## Rollout checklist + +1. Create the delivery table and unique tenant/event index before deploying the + worker code. +2. Backfill only known pending work; never synthesize a successful row for an + delivery whose provider response is unknown. +3. Run one worker in shadow mode and compare claim counts with existing + dispatch logs. +4. Enable atomic claims and lease renewal for a canary tenant. +5. Confirm retries do not occur before `next_attempt_at`. +6. Confirm a restart reclaims an expired processing row. +7. Confirm dead rows are visible without exposing destination secrets. +8. Enable alerts for dead count, lease expiry, and retry age. + +## Failure handling + +An HTTP timeout means the provider outcome is unknown. The worker should leave +the durable row retryable and send the same event key on the next attempt. A +4xx response caused by a malformed payload should be classified as terminal +after the configured policy, not retried indefinitely. A 5xx response or +network failure is retryable subject to the attempt budget. Provider success +must be followed by `complete`; if the process dies before that write, the +provider idempotency key protects the next attempt. + +Do not use destination URL, payload content, or tenant id as a metric label. +Those fields can have high cardinality or contain sensitive information. +Hashing the body for integrity checks is safe to expose only as an internal +record field; log the event key and status, not the raw body. + +The durable state is authoritative during shutdown. Stop accepting new +dispatches, let in-flight requests finish, and leave unclaimed pending or +retrying rows intact. A later worker will resume them. Never clear the table +as part of a restart or deployment hook. + +## Compatibility + +Existing callers can continue to enqueue through the dispatcher while they +are migrated to the durable adapter. The adapter's event key should be the +existing `X-Callora-Delivery` value when one is available, so retries retain +their provider-visible identity. New callers must create the durable row before +starting network delivery. This ordering makes a crash before `fetch` safe and +keeps recovery independent of process memory. + +Database migrations must be backward-compatible with old workers: add columns +and indexes first, deploy readers second, and remove legacy cleanup only after +all workers report the new lifecycle metrics. Rollback leaves rows untouched. + +The state machine is intentionally monotonic after delivery: a delivered row +cannot return to pending, retrying, or processing. Operators must create a new +event key for a deliberate replay and record the reason for that action. + +Review the claim query and transition update together: the worker id and lease +must be checked in the same conditional statement. A read followed by an +unconditional update reintroduces the race this state machine is designed to +remove. + +Monitoring guidance: + +- `pending` measures newly accepted work; +- `processing` measures active leases; +- `retrying` measures delayed recoverable work; +- `delivered` measures completed provider responses; +- `dead` measures work requiring operator action. + +Alert on a growing processing population, not only on dead rows. A stuck +worker can keep rows processing until lease expiry, delaying customer-visible +delivery without increasing the terminal counter. The combination of state, +attempt count, next-attempt time, and age is sufficient to diagnose that case +without logging the request body. + +Keep provider response codes in a separate redacted operational log and use +the durable row for the retry decision. + +This preserves a reviewable audit trail across process restarts and worker +replacement. + +Operators can safely inspect this state without opening the request payload. + +This is the durable source of truth for delivery recovery. diff --git a/src/webhooks/durableDelivery.matrix.test.ts b/src/webhooks/durableDelivery.matrix.test.ts new file mode 100644 index 00000000..20626491 --- /dev/null +++ b/src/webhooks/durableDelivery.matrix.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from 'vitest' +import { DeliveryConflictError, InMemoryDurableDeliveryStore, enqueueDurableDelivery } from './durableDelivery.js' + +const makeInput = (n: number, overrides: Record = {}) => ({ + tenantId: `tenant-${n % 3}`, eventKey: `event-${n}`, destination: `https://hooks.example.test/${n}`, + body: JSON.stringify({ event: 'invoice.paid', id: n, amount: n * 10 }), maxAttempts: 3, baseDelayMs: 10, ...overrides, +}) + +describe('durable delivery state transition matrix', () => { + it.each(Array.from({ length: 15 }, (_, index) => index + 1))('keeps event %i pending until claimed', async n => { + const store = new InMemoryDurableDeliveryStore() + const record = await enqueueDurableDelivery(store, makeInput(n), 100) + expect(record.status).toBe('pending') + expect((await store.stats()).pending).toBe(1) + }) + + it('prevents two workers from claiming a pending row concurrently', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(20), 0) + const claims = await Promise.all([ + store.claim('tenant-2', 'event-20', 'worker-a', 0), + store.claim('tenant-2', 'event-20', 'worker-b', 0), + ]) + expect(claims.filter(Boolean)).toHaveLength(1) + }) + + it('does not let a stale worker acknowledge a recovered lease', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(21), 0) + const first = await store.claim('tenant-0', 'event-21', 'worker-a', 0, 5) + const second = await store.claim('tenant-0', 'event-21', 'worker-b', 5, 5) + expect(await store.complete(first!, 'worker-a', 6)).toBe(false) + expect(await store.complete(second!, 'worker-b', 6)).toBe(true) + }) + + it.each([ + [1, 10], [2, 20], [3, 40], [4, 80], [5, 160], + ])('uses bounded exponential delay after attempt %i', async (attempt, delay) => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(22, { maxAttempts: 10, baseDelayMs: 10 }), 0) + let claim = await store.claim('tenant-1', 'event-22', 'worker-0', 0) + for (let index = 1; index < attempt; index++) { + await store.fail(claim!, `failure-${index}`, 'x', claim!.nextAttemptAt) + claim = await store.claim('tenant-1', 'event-22', `worker-${index}`, claim!.nextAttemptAt) + } + const failed = await store.fail(claim!, 'x', 1000) + expect(failed?.nextAttemptAt).toBe(1000 + delay) + }) + + it('does not claim retryable work before nextAttemptAt', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(23), 0) + const claim = await store.claim('tenant-2', 'event-23', 'worker-a', 0) + const failed = await store.fail(claim!, 'worker-a', 'timeout', 1) + expect(await store.claim('tenant-2', 'event-23', 'worker-b', failed!.nextAttemptAt - 1)).toBeUndefined() + }) + + it('does not permit completion after the lease expires', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(24), 0) + const claim = await store.claim('tenant-0', 'event-24', 'worker-a', 0, 10) + expect(await store.complete(claim!, 'worker-a', 11)).toBe(false) + }) + + it('keeps delivered rows out of all future states', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(25), 0) + const claim = await store.claim('tenant-1', 'event-25', 'worker-a', 0) + await store.complete(claim!, 'worker-a', 1) + expect(await store.claim('tenant-1', 'event-25', 'worker-b', 2)).toBeUndefined() + expect(await store.fail(claim!, 'worker-a', 'late', 3)).toBeUndefined() + }) + + it('makes only one successful record for duplicate enqueue races', async () => { + const store = new InMemoryDurableDeliveryStore() + const records = await Promise.all(Array.from({ length: 30 }, () => enqueueDurableDelivery(store, makeInput(26), 0))) + expect(records).toHaveLength(30) + expect((await store.stats()).pending).toBe(1) + }) + + it('rejects changed payload in duplicate enqueue races', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(27), 0) + await expect(enqueueDurableDelivery(store, makeInput(27, { body: '{"amount":999}' }), 0)).rejects.toBeInstanceOf(DeliveryConflictError) + }) + + it('reports each lifecycle state in stats', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(28), 0) + await enqueueDurableDelivery(store, makeInput(29), 0) + const processing = await store.claim('tenant-1', 'event-28', 'worker-a', 0) + const dead = await store.claim('tenant-2', 'event-29', 'worker-b', 0) + await store.fail(dead!, 'worker-b', 'fatal', 0) + await store.complete(processing!, 'worker-a', 1) + expect(await store.stats()).toEqual({ pending: 0, processing: 0, retrying: 1, delivered: 1, dead: 0 }) + }) + + it('requires a claimant for state transitions', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(30), 0) + const claim = await store.claim('tenant-0', 'event-30', 'worker-a', 0) + expect(await store.renew(claim!, 'other', 1)).toBe(false) + expect(await store.fail(claim!, 'other', 'timeout', 1)).toBeUndefined() + expect((await store.get('tenant-0', 'event-30'))?.status).toBe('processing') + }) + + it('retains retry error context without retaining network URLs', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(31), 0) + const claim = await store.claim('tenant-1', 'event-31', 'worker-a', 0) + const failed = await store.fail(claim!, 'worker-a', 'POST https://internal/token timed out', 1) + expect(failed?.lastError).toBe('POST [url] timed out') + }) + + it('supports a fresh event after a previous event reaches dead state', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, makeInput(32, { maxAttempts: 1 }), 0) + const claim = await store.claim('tenant-2', 'event-32', 'worker-a', 0) + await store.fail(claim!, 'worker-a', 'fatal', 1) + const fresh = await enqueueDurableDelivery(store, makeInput(33), 1) + expect(fresh.status).toBe('pending') + expect((await store.stats()).dead).toBe(1) + }) +}) diff --git a/src/webhooks/durableDelivery.test.ts b/src/webhooks/durableDelivery.test.ts new file mode 100644 index 00000000..e33c133c --- /dev/null +++ b/src/webhooks/durableDelivery.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from 'vitest' +import { DeliveryConflictError, InMemoryDurableDeliveryStore, enqueueDurableDelivery } from './durableDelivery.js' + +const input = (overrides: Record = {}) => ({ tenantId: 'tenant-a', eventKey: 'event-1', destination: 'https://example.test/hook', body: JSON.stringify({ event: 'invoice.paid', amount: 100 }), ...overrides }) + +describe('durable webhook delivery', () => { + it('creates a pending delivery', async () => { + const store = new InMemoryDurableDeliveryStore() + const record = await enqueueDurableDelivery(store, input(), 1_000) + expect(record.status).toBe('pending') + expect(record.attemptCount).toBe(0) + expect(record.nextAttemptAt).toBe(1_000) + }) + + it('deduplicates an identical event key', async () => { + const store = new InMemoryDurableDeliveryStore() + const first = await enqueueDurableDelivery(store, input(), 1_000) + const second = await enqueueDurableDelivery(store, input(), 2_000) + expect(second.payloadHash).toBe(first.payloadHash) + expect((await store.stats()).pending).toBe(1) + }) + + it('rejects a reused key with a changed payload', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + await expect(enqueueDurableDelivery(store, input({ body: JSON.stringify({ changed: true }) }), 2_000)).rejects.toBeInstanceOf(DeliveryConflictError) + }) + + it('claims a pending event once for one worker', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + const a = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100) + const b = await store.claim('tenant-a', 'event-1', 'worker-b', 1_001, 100) + expect(a?.workerId).toBe('worker-a') + expect(b).toBeUndefined() + }) + + it('allows another worker after lease expiry', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100) + const recovered = await store.claim('tenant-a', 'event-1', 'worker-b', 1_100, 100) + expect(recovered?.workerId).toBe('worker-b') + expect(recovered?.attemptCount).toBe(2) + }) + + it('renews only the current worker lease', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100) + expect(await store.renew(claim!, 'worker-b', 1_050, 100)).toBe(false) + expect(await store.renew(claim!, 'worker-a', 1_050, 100)).toBe(true) + }) + + it('marks a delivery delivered only for its claimant', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000, 100) + expect(await store.complete(claim!, 'worker-b', 1_050)).toBe(false) + expect(await store.complete(claim!, 'worker-a', 1_050)).toBe(true) + expect((await store.stats()).delivered).toBe(1) + expect(await store.claim('tenant-a', 'event-1', 'worker-c', 2_000)).toBeUndefined() + }) + + it('moves failures to retrying with exponential delay', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input({ baseDelayMs: 100 }), 1_000) + const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000) + const failed = await store.fail(claim!, 'worker-a', 'timeout', 1_010) + expect(failed?.status).toBe('retrying') + expect(failed?.nextAttemptAt).toBe(1_110) + expect(await store.claim('tenant-a', 'event-1', 'worker-b', 1_109)).toBeUndefined() + expect(await store.claim('tenant-a', 'event-1', 'worker-b', 1_110)).toBeDefined() + }) + + it('moves a delivery to dead after the configured attempt budget', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input({ maxAttempts: 2, baseDelayMs: 1 }), 1_000) + const first = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000) + await store.fail(first!, 'worker-a', 'bad gateway', 1_001) + const second = await store.claim('tenant-a', 'event-1', 'worker-b', 1_003) + const dead = await store.fail(second!, 'worker-b', 'bad gateway again', 1_004) + expect(dead?.status).toBe('dead') + expect((await store.stats()).dead).toBe(1) + }) + + it('keeps tenant namespaces independent', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input(), 1_000) + await enqueueDurableDelivery(store, input({ tenantId: 'tenant-b' }), 1_000) + expect((await store.stats()).pending).toBe(2) + }) + + it('rejects malformed payloads and invalid retry settings', async () => { + const store = new InMemoryDurableDeliveryStore() + await expect(enqueueDurableDelivery(store, input({ body: '{' }))).rejects.toThrow('valid JSON') + await expect(enqueueDurableDelivery(store, input({ maxAttempts: 0 }))).rejects.toThrow('maxAttempts') + await expect(enqueueDurableDelivery(store, input({ baseDelayMs: 0 }))).rejects.toThrow('baseDelayMs') + }) + + it('sanitizes URLs in terminal error metadata', async () => { + const store = new InMemoryDurableDeliveryStore() + await enqueueDurableDelivery(store, input({ maxAttempts: 1 }), 1_000) + const claim = await store.claim('tenant-a', 'event-1', 'worker-a', 1_000) + const dead = await store.fail(claim!, 'worker-a', 'failed https://secret.example/token', 1_001) + expect(dead?.lastError).toBe('failed [url]') + }) + + it('does not expose mutable internal records', async () => { + const store = new InMemoryDurableDeliveryStore() + const created = await enqueueDurableDelivery(store, input(), 1_000) + created.status = 'delivered' + expect((await store.get('tenant-a', 'event-1'))?.status).toBe('pending') + }) +}) diff --git a/src/webhooks/durableDelivery.ts b/src/webhooks/durableDelivery.ts new file mode 100644 index 00000000..2b3acab3 --- /dev/null +++ b/src/webhooks/durableDelivery.ts @@ -0,0 +1,146 @@ +import { createHash } from 'node:crypto' + +export type DurableDeliveryStatus = 'pending' | 'processing' | 'retrying' | 'delivered' | 'dead' + +export interface DurableDeliveryInput { + tenantId: string + eventKey: string + destination: string + body: string + maxAttempts?: number + baseDelayMs?: number +} + +export interface DurableDeliveryRecord extends DurableDeliveryInput { + payloadHash: string + status: DurableDeliveryStatus + attemptCount: number + workerId: string | null + leaseExpiresAt: number | null + nextAttemptAt: number + lastError: string | null + deliveredAt: number | null + createdAt: number + updatedAt: number +} + +export type DeliveryCreateResult = { outcome: 'created'; record: DurableDeliveryRecord } | { outcome: 'duplicate'; record: DurableDeliveryRecord } | { outcome: 'conflict'; record: DurableDeliveryRecord } + +export interface DurableDeliveryStats { pending: number; processing: number; retrying: number; delivered: number; dead: number } + +export class DeliveryConflictError extends Error { + readonly statusCode = 409 + constructor() { super('A delivery event key already contains a different payload'); this.name = 'DeliveryConflictError' } +} + +export interface DurableDeliveryStore { + create(input: DurableDeliveryInput, now?: number): Promise + claim(tenantId: string, eventKey: string, workerId: string, now?: number, leaseMs?: number): Promise + renew(record: DurableDeliveryRecord, workerId: string, now?: number, leaseMs?: number): Promise + complete(record: DurableDeliveryRecord, workerId: string, now?: number): Promise + fail(record: DurableDeliveryRecord, workerId: string, error: string, now?: number): Promise + get(tenantId: string, eventKey: string): Promise + stats(): Promise +} + +export class InMemoryDurableDeliveryStore implements DurableDeliveryStore { + private readonly records = new Map() + + async create(input: DurableDeliveryInput, now = Date.now()): Promise { + validateInput(input) + const identity = key(input.tenantId, input.eventKey) + const payloadHash = hash(input.body) + const existing = this.records.get(identity) + if (existing) { + return { outcome: existing.payloadHash === payloadHash ? 'duplicate' : 'conflict', record: clone(existing) } + } + const record: DurableDeliveryRecord = { + ...input, + maxAttempts: normalizeAttempts(input.maxAttempts), + baseDelayMs: normalizeDelay(input.baseDelayMs), + payloadHash, + status: 'pending', + attemptCount: 0, + workerId: null, + leaseExpiresAt: null, + nextAttemptAt: now, + lastError: null, + deliveredAt: null, + createdAt: now, + updatedAt: now, + } + this.records.set(identity, record) + return { outcome: 'created', record: clone(record) } + } + + async claim(tenantId: string, eventKey: string, workerId: string, now = Date.now(), leaseMs = 30_000): Promise { + if (!workerId.trim() || leaseMs <= 0) throw new Error('workerId and leaseMs are required') + const record = this.records.get(key(tenantId, eventKey)) + if (!record || record.status === 'delivered' || record.status === 'dead') return undefined + const leaseExpired = record.status === 'processing' && (record.leaseExpiresAt ?? 0) <= now + if (record.status === 'processing' && !leaseExpired) return undefined + if ((record.status === 'pending' || record.status === 'retrying' || leaseExpired) && record.nextAttemptAt > now) return undefined + record.status = 'processing' + record.workerId = workerId + record.leaseExpiresAt = now + leaseMs + record.attemptCount += 1 + record.updatedAt = now + return clone(record) + } + + async renew(record: DurableDeliveryRecord, workerId: string, now = Date.now(), leaseMs = 30_000): Promise { + const current = this.records.get(key(record.tenantId, record.eventKey)) + if (!current || current.status !== 'processing' || current.workerId !== workerId || current.leaseExpiresAt! < now) return false + current.leaseExpiresAt = now + leaseMs + current.updatedAt = now + return true + } + + async complete(record: DurableDeliveryRecord, workerId: string, now = Date.now()): Promise { + const current = this.records.get(key(record.tenantId, record.eventKey)) + if (!current || current.status !== 'processing' || current.workerId !== workerId || current.payloadHash !== record.payloadHash) return false + current.status = 'delivered' + current.workerId = null + current.leaseExpiresAt = null + current.deliveredAt = now + current.updatedAt = now + return true + } + + async fail(record: DurableDeliveryRecord, workerId: string, error: string, now = Date.now()): Promise { + const current = this.records.get(key(record.tenantId, record.eventKey)) + if (!current || current.status !== 'processing' || current.workerId !== workerId) return undefined + current.lastError = sanitizeError(error) + current.workerId = null + current.leaseExpiresAt = null + current.status = current.attemptCount >= current.maxAttempts ? 'dead' : 'retrying' + current.nextAttemptAt = current.status === 'dead' ? now : now + current.baseDelayMs * 2 ** Math.max(0, current.attemptCount - 1) + current.updatedAt = now + return clone(current) + } + + async get(tenantId: string, eventKey: string) { const record = this.records.get(key(tenantId, eventKey)); return record ? clone(record) : undefined } + async stats(): Promise { + const stats: DurableDeliveryStats = { pending: 0, processing: 0, retrying: 0, delivered: 0, dead: 0 } + for (const record of this.records.values()) stats[record.status] += 1 + return stats + } +} + +export async function enqueueDurableDelivery(store: DurableDeliveryStore, input: DurableDeliveryInput, now?: number): Promise { + const result = await store.create(input, now) + if (result.outcome === 'conflict') throw new DeliveryConflictError() + return result.record +} + +function key(tenantId: string, eventKey: string) { return `${tenantId}:${eventKey}` } +function hash(body: string) { return createHash('sha256').update(body, 'utf8').digest('hex') } +function clone(record: DurableDeliveryRecord): DurableDeliveryRecord { return { ...record } } +function normalizeAttempts(value: number | undefined) { if (value === undefined) return 5; if (!Number.isInteger(value) || value < 1 || value > 20) throw new Error('maxAttempts must be between 1 and 20'); return value } +function normalizeDelay(value: number | undefined) { if (value === undefined) return 1_000; if (!Number.isInteger(value) || value < 1 || value > 3_600_000) throw new Error('baseDelayMs is outside the supported range'); return value } +function sanitizeError(value: string) { return value.replace(/https?:\/\/[^\s]+/gi, '[url]').slice(0, 1_000) } +function validateInput(input: DurableDeliveryInput) { + if (!input.tenantId.trim() || !input.eventKey.trim() || !input.destination.trim()) throw new Error('tenantId, eventKey, and destination are required') + if (typeof input.body !== 'string' || input.body.length === 0) throw new Error('delivery body is required') + try { JSON.parse(input.body) } catch { throw new Error('delivery body must be valid JSON') } +}