diff --git a/.changeset/22756-webhooks-redeliver-member.md b/.changeset/22756-webhooks-redeliver-member.md new file mode 100644 index 00000000000..bf062897686 --- /dev/null +++ b/.changeset/22756-webhooks-redeliver-member.md @@ -0,0 +1,13 @@ +--- +"@objectstack/plugin-webhooks": minor +--- + +feat(plugin-webhooks): the `webhooks` service serves the redeliver door from a web-standard `Request` (`IWebhookService.handleRedeliver`), and the redelivery veto no longer waits for realtime (#22756) + +Clause-②: yes + +- **What is new.** The plugin registers the `webhooks` service and implements its optional `IWebhookService.handleRedeliver(request)` member: the webhook redeliver door, `POST /api/v1/webhooks/redeliver`, from a `Request` to a `Response`, with no raw app, no Hono context and no `http.server`. A hosted tenant kernel owns no socket, so this is how the door reaches it; its one caller is the runtime HTTP dispatcher's `POST /webhooks/redeliver` domain, which lands separately. +- **One door, two faces.** The member and the self-hosted route run the same code: the same session check, the same active-organization tenant, the same replay through `MessagingService.redeliverHttp` and the same answer for every outcome. For the same request they answer the same status, the same `Content-Type: application/json` and the same bytes: `401 UNAUTHENTICATED`, `400 INVALID_REQUEST`, `400 MISSING_REQUIRED_FIELD`, `404 RESOURCE_NOT_FOUND`, `409 DELIVERY_NOT_ELIGIBLE`, `409 DELIVERY_NEVER_SENT`, `500 INTERNAL_ERROR`, or `200` with the row's id and new status. +- **The redelivery veto is installed wherever the door is.** The veto that refuses to replay a webhook delivery whose subscription is gone, or whose signing secret cannot be recovered, now needs only the data engine and the messaging service. It used to be installed only when realtime was present and `autoEnqueue` was not `false`, so on other kernels such a delivery was replayed. It is now refused with `409 DELIVERY_NOT_ELIGIBLE` there too. The `webhooks` service is registered only where the veto is installed. +- **The route is found under `http.server` too.** The plugin reads `http.server` first, then the `http-server` alias. A host that registers its server only as `http.server` now gets the self-hosted route; one that registers it under both names gets it once. +- **The self-hosted route is otherwise unchanged.** Where it was mounted before, it answers the same status, headers and bytes for the same request, and it is never mounted twice. diff --git a/packages/plugins/plugin-webhooks/package.json b/packages/plugins/plugin-webhooks/package.json index 8ef671e4bcf..c120ffc98ee 100644 --- a/packages/plugins/plugin-webhooks/package.json +++ b/packages/plugins/plugin-webhooks/package.json @@ -45,6 +45,7 @@ "@objectstack/metadata-core": "workspace:*", "@objectstack/objectql": "workspace:*", "@types/node": "^26.6.3", + "hono": "^4.13.9", "tsx": "^4.23.15", "typescript": "^6.0.3", "vitest": "^4.1.11" diff --git a/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts b/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts index 60dc0b99b98..d157cbdc6ca 100644 --- a/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts +++ b/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts @@ -6,6 +6,7 @@ import type { II18nService, IMetadataService, IRealtimeService, + IWebhookService, } from '@objectstack/spec/contracts'; import type { EnqueueHttpInput } from '@objectstack/service-messaging'; import { inProcessSessionReadInput } from '@objectstack/types'; @@ -57,6 +58,49 @@ interface MessagingHttpSurface { ): void; } +/** + * [#22756] The service slot this plugin fills: `IWebhookService` + * (`@objectstack/spec/contracts`), named after the plugin's platform + * capability token, as the contract's docblock declares it. + */ +const WEBHOOKS_SLOT = 'webhooks'; + +/** + * [#22756] The slice of a Hono context the redeliver door reads and answers + * through, and nothing more: the request's headers (the caller's credentials), + * its JSON body, and `json(body, status)`. + * + * Both faces of the door hand {@link WebhookOutboxPlugin.serveRedeliver} one of + * these. The self-hosted mount hands it the real Hono context, so its answers + * are the ones it always gave. The `webhooks` service's `handleRedeliver` hands + * it {@link requestDoorContext} over a web-standard `Request`. So both faces run + * ONE copy of the door's rules: who may press it, whose rows it reaches, and + * what each refusal answers. + */ +interface RedeliverDoorContext { + req: { raw: { headers: Headers }; json(): Promise }; + json(body: unknown, status?: number): R; +} + +/** + * [#22756] A web-standard `Request` as the door's context, answering the way + * Hono's `c.json` answers: the body is `JSON.stringify(body)`, the one header is + * `Content-Type: application/json`, and the status defaults to `200`. + * + * The body is read with `request.json()`, which parses exactly what Hono's + * `c.req.json()` parses (the body's text, then `JSON.parse`), so a body that is + * not JSON is refused the same way at both faces. + */ +function requestDoorContext(request: Request): RedeliverDoorContext { + return { + req: { raw: request, json: () => request.json() }, + json: (body, status) => new Response(JSON.stringify(body), { + status, + headers: { 'Content-Type': 'application/json' }, + }), + }; +} + export interface WebhookOutboxPluginOptions { /** * Auto-enqueue config. When enabled (default `true` if the realtime + data @@ -81,7 +125,9 @@ export interface WebhookOutboxPluginOptions { * - the `sys_webhook` configuration object, * - the {@link AutoEnqueuer} that turns `data.record.*` events into outbox * rows (`source: 'webhook'`), and - * - the redeliver admin endpoint. + * - the redeliver door: the self-hosted `POST /api/v1/webhooks/redeliver` + * mount, and the `webhooks` service's transport-neutral `handleRedeliver` + * (`IWebhookService`), which serve one set of rules. * * End-to-end flow: * @@ -174,6 +220,12 @@ export class WebhookOutboxPlugin implements Plugin { // deployment without realtime would silently fail to materialize // declared webhooks — the very no-op this bridge closes (#3461). await this.bootDeclaredWebhooks(ctx); + // [#22756] The redelivery veto, and beside it the `webhooks` + // service — gated on the engine and messaging alone, like the + // step above, and never on the auto-enqueue prerequisites. It + // runs before the enqueuer writes its first row (#8069), and + // before the mount below is mounted. + this.bootRedeliverDoor(ctx, autoEnqueueOpt); await this.bootAutoEnqueue(ctx, autoEnqueueOpt); this.registerAdminRoutes(ctx); }); @@ -303,10 +355,11 @@ export class WebhookOutboxPlugin implements Plugin { } const enqOpts = (typeof opt === 'object' ? opt : {}) as AutoEnqueuerOptions; - // [#8069] Install the redelivery veto BEFORE the enqueuer starts - // writing rows, so no delivery row can ever exist while the refusal - // that protects it does not. - this.installRedeliverGuard(ctx, messaging, engine, enqOpts.subscriptionsObject); + // [#8069] The redelivery veto is already installed: `bootRedeliverDoor` + // ran first, so no delivery row can ever exist while the refusal that + // protects it does not. [#22756] It is installed THERE, and not here, + // because this function returns early without realtime or with + // `autoEnqueue: false`, and the door serves on those kernels too. this.autoEnqueuer = new AutoEnqueuer( engine, realtime, @@ -318,6 +371,46 @@ export class WebhookOutboxPlugin implements Plugin { ctx.logger.info?.('[webhook-auto-enqueuer] started (enqueues source=webhook onto sys_http_delivery)'); } + /** + * [#22756] Install the #8069 redelivery veto, and register the `webhooks` + * service beside it. + * + * The veto needs the data engine (it reads the subscription) and messaging + * (it is installed there). It does NOT need realtime or auto-enqueue: it + * judges a row being REPLAYED, and a row can be replayed on a kernel that + * enqueues nothing. It used to be installed inside {@link bootAutoEnqueue}, + * which returns early without realtime or with `autoEnqueue: false`, so on + * those kernels the door served while the veto was never installed. + * + * The service is registered only once the veto is installed, so no kernel + * serves `handleRedeliver` without it. Where the veto cannot be installed + * the slot stays empty, and the member's caller answers its typed absence + * (the contract's rule) rather than a door with no veto. + * + * `subscriptionsObject` is read from the auto-enqueue options, as before, + * so a host that renamed its subscriptions object keeps the veto reading it. + */ + private bootRedeliverDoor(ctx: PluginContext, opt: boolean | AutoEnqueuerOptions): void { + const engine = this.tryGetService(ctx, ['objectql', 'data']); + const messaging = this.getMessaging(ctx); + if (!engine || !messaging) { + ctx.logger.warn?.( + '[webhook-outbox] redeliver veto not installed and the webhooks service not registered ' + + '— the data engine or the messaging service is not available', + { hasEngine: !!engine, hasMessaging: !!messaging }, + ); + return; + } + const subscriptionsObject = typeof opt === 'object' ? opt.subscriptionsObject : undefined; + if (!this.installRedeliverGuard(ctx, messaging, engine, subscriptionsObject)) return; + + const service: IWebhookService = { + handleRedeliver: (request: Request) => + this.serveRedeliver(ctx, messaging, requestDoorContext(request)), + }; + ctx.registerService(WEBHOOKS_SLOT, service); + } + /** * [#8069] Register {@link createWebhookRedeliverGuard} with messaging, so * `redeliver()` refuses a webhook row whose signing configuration is no @@ -329,13 +422,16 @@ export class WebhookOutboxPlugin implements Plugin { * dispatcher still reports a delivery, while the fail-closed signing * guarantee the system claims (#7799) is not actually being kept. That is a * durability/consistency degradation wearing a functional one's clothes. + * + * [#22756] Returns whether the guard is installed: the `webhooks` service + * is registered only when it is. */ private installRedeliverGuard( ctx: PluginContext, messaging: MessagingHttpSurface, engine: IDataEngine, subscriptionsObject?: string, - ): void { + ): boolean { if (typeof messaging.registerRedeliverGuard !== 'function') { ctx.logger.error?.( '[webhook-outbox] messaging service exposes no registerRedeliverGuard() — redelivery ' @@ -343,16 +439,17 @@ export class WebhookOutboxPlugin implements Plugin { + 'pressing redeliver may send a delivery that can no longer be authenticated ' + '(an unsigned send, which webhook delivery otherwise never makes). The POST ' + '/api/v1/webhooks/redeliver endpoint is reachable by any ' - + 'authenticated user. Fix: upgrade @objectstack/service-messaging to a build that ' - + 'implements registerRedeliverGuard.', + + 'authenticated user; the webhooks service is not registered. Fix: upgrade ' + + '@objectstack/service-messaging to a build that implements registerRedeliverGuard.', ); - return; + return false; } messaging.registerRedeliverGuard( 'webhook', createWebhookRedeliverGuard(engine, subscriptionsObject), ); ctx.logger.debug?.('[webhook-outbox] redeliver guard installed for source=webhook'); + return true; } private tryGetService(ctx: PluginContext, names: string[]): T | undefined { @@ -369,8 +466,45 @@ export class WebhookOutboxPlugin implements Plugin { /** * Mount POST /api/v1/webhooks/redeliver on the host Hono app, if one is - * available. Delegates to `messaging.redeliverHttp(deliveryId, …)`. Auth is - * the better-auth session cookie — every authenticated user counts. + * available. The door itself is {@link serveRedeliver}, handed the Hono + * context as it is, so the mount answers what it always answered. + * + * [#22756] The server is read as `http.server` first, then the deprecated + * `http-server` alias, as `plugin-approvals` reads it. The canonical name + * is the only one present on every provider path: `runtime.ts`'s + * `config.server` path registers no alias, so an alias-only read never + * mounted the door there. `tryGetService` asks each name in its own `try`, + * because `getService` throws on an empty slot. Where both names are + * registered (`plugin-hono-server` registers one server under both), the + * first answers, so the path is mounted once. + */ + private registerAdminRoutes(ctx: PluginContext): void { + const http = this.tryGetService(ctx, ['http.server', 'http-server']); + if (!http || typeof http.getRawApp !== 'function') { + ctx.logger.debug?.('[webhook-outbox] HTTP server not available; redeliver endpoint not mounted'); + return; + } + const rawApp = http.getRawApp(); + const messaging = this.getMessaging(ctx); + if (!rawApp || !messaging) return; + + rawApp.post('/api/v1/webhooks/redeliver', async (c: any) => this.serveRedeliver(ctx, messaging, c)); + + ctx.logger.info?.('[webhook-outbox] redeliver endpoint mounted at POST /api/v1/webhooks/redeliver'); + } + + /** + * The redeliver door: `POST /api/v1/webhooks/redeliver`. Delegates to + * `messaging.redeliverHttp(deliveryId, …)`. Auth is the better-auth session + * cookie — every authenticated user counts. + * + * [#22756] ONE body, two faces. The self-hosted mount + * ({@link registerAdminRoutes}) hands it the Hono context; the `webhooks` + * service's `handleRedeliver` hands it a `Request` through + * {@link requestDoorContext}. Neither face carries a rule of its own, so + * they cannot drift: the same caller, the same tenant, the same veto and + * the same answer for every refusal. The mount is not a caller of the + * member, and nothing here mounts anything. * * [#10740] Which is precisely why the caller's ACTIVE ORGANIZATION is * resolved here and threaded into the call. `sys_http_delivery` is @@ -385,79 +519,70 @@ export class WebhookOutboxPlugin implements Plugin { * the correct outcome. ⛔ It is never repaired with `bypassTenantAudit`, * which would silence the report without closing anything. */ - private registerAdminRoutes(ctx: PluginContext): void { - const http = this.tryGetService(ctx, ['http-server']); - if (!http || typeof http.getRawApp !== 'function') { - ctx.logger.debug?.('[webhook-outbox] HTTP server not available; redeliver endpoint not mounted'); - return; + private async serveRedeliver( + ctx: PluginContext, + messaging: MessagingHttpSurface, + c: RedeliverDoorContext, + ): Promise { + const session = await this.resolveSession(ctx, c); + const userId = session?.user?.id; + if (typeof userId !== 'string' || userId.length === 0) { + return c.json( + { success: false, error: { code: 'UNAUTHENTICATED', message: 'Sign in to redeliver webhook deliveries.' } }, + 401, + ); } - const rawApp = http.getRawApp(); - const messaging = this.getMessaging(ctx); - if (!rawApp || !messaging) return; - - rawApp.post('/api/v1/webhooks/redeliver', async (c: any) => { - const session = await this.resolveSession(ctx, c); - const userId = session?.user?.id; - if (typeof userId !== 'string' || userId.length === 0) { - return c.json( - { success: false, error: { code: 'UNAUTHENTICATED', message: 'Sign in to redeliver webhook deliveries.' } }, - 401, - ); - } - let body: any; - try { - body = await c.req.json(); - } catch { - return c.json({ success: false, error: { code: 'INVALID_REQUEST', message: 'Request body must be JSON.' } }, 400); - } - const deliveryId = typeof body?.deliveryId === 'string' ? body.deliveryId.trim() : ''; - if (!deliveryId) { - return c.json( - { success: false, error: { code: 'MISSING_REQUIRED_FIELD', message: 'Body must include `deliveryId: string`.' } }, - 400, - ); + let body: any; + try { + body = await c.req.json(); + } catch { + return c.json({ success: false, error: { code: 'INVALID_REQUEST', message: 'Request body must be JSON.' } }, 400); + } + const deliveryId = typeof body?.deliveryId === 'string' ? body.deliveryId.trim() : ''; + if (!deliveryId) { + return c.json( + { success: false, error: { code: 'MISSING_REQUIRED_FIELD', message: 'Body must include `deliveryId: string`.' } }, + 400, + ); + } + try { + // [#10740] `session.session.activeOrganizationId` is the + // canonical spelling of the caller's active organization + // (better-auth's organization plugin; see + // `plugin-auth/auth-schema-config.ts`). Read from the one + // place it lives — a `??` chain over alternative spellings + // would make a MISSING organization indistinguishable from a + // differently-shaped one, and the missing case is the one that + // must stay visible. + const activeOrg = session?.session?.activeOrganizationId; + const tenantId = typeof activeOrg === 'string' && activeOrg.length > 0 + ? activeOrg + : undefined; + const row = await messaging.redeliverHttp(deliveryId, { tenantId }); + ctx.logger.info?.('[webhook-outbox] redelivered', { deliveryId, requestedBy: userId, tenantId }); + return c.json({ success: true, data: { id: row.id, status: row.status } }); + } catch (err: any) { + const code = err?.code; + if (code === 'RESOURCE_NOT_FOUND') { + return c.json({ success: false, error: { code, message: err.message } }, 404); } - try { - // [#10740] `session.session.activeOrganizationId` is the - // canonical spelling of the caller's active organization - // (better-auth's organization plugin; see - // `plugin-auth/auth-schema-config.ts`). Read from the one - // place it lives — a `??` chain over alternative spellings - // would make a MISSING organization indistinguishable from a - // differently-shaped one, and the missing case is the one that - // must stay visible. - const activeOrg = session?.session?.activeOrganizationId; - const tenantId = typeof activeOrg === 'string' && activeOrg.length > 0 - ? activeOrg - : undefined; - const row = await messaging.redeliverHttp(deliveryId, { tenantId }); - ctx.logger.info?.('[webhook-outbox] redelivered', { deliveryId, requestedBy: userId, tenantId }); - return c.json({ success: true, data: { id: row.id, status: row.status } }); - } catch (err: any) { - const code = err?.code; - if (code === 'RESOURCE_NOT_FOUND') { - return c.json({ success: false, error: { code, message: err.message } }, 404); - } - // [#8069] `DELIVERY_NEVER_SENT` is a refusal, not a server - // fault: the row is a parked record of a delivery that was - // never prepared, and re-sending it would be a FIRST delivery — - // unsigned, because the secret that would have signed it is - // what went missing. 409 alongside the eligibility refusal, so - // an operator tool can present both the same way; without this - // arm it would surface as a 500 and read as a transient glitch - // worth retrying. - if (code === 'DELIVERY_NOT_ELIGIBLE' || code === 'DELIVERY_NEVER_SENT') { - return c.json({ success: false, error: { code, message: err.message } }, 409); - } - ctx.logger.error?.('[webhook-outbox] redeliver failed', err as Error); - return c.json( - { success: false, error: { code: 'INTERNAL_ERROR', message: err?.message ?? String(err) } }, - 500, - ); + // [#8069] `DELIVERY_NEVER_SENT` is a refusal, not a server + // fault: the row is a parked record of a delivery that was + // never prepared, and re-sending it would be a FIRST delivery — + // unsigned, because the secret that would have signed it is + // what went missing. 409 alongside the eligibility refusal, so + // an operator tool can present both the same way; without this + // arm it would surface as a 500 and read as a transient glitch + // worth retrying. + if (code === 'DELIVERY_NOT_ELIGIBLE' || code === 'DELIVERY_NEVER_SENT') { + return c.json({ success: false, error: { code, message: err.message } }, 409); } - }); - - ctx.logger.info?.('[webhook-outbox] redeliver endpoint mounted at POST /api/v1/webhooks/redeliver'); + ctx.logger.error?.('[webhook-outbox] redeliver failed', err as Error); + return c.json( + { success: false, error: { code: 'INTERNAL_ERROR', message: err?.message ?? String(err) } }, + 500, + ); + } } /** diff --git a/packages/plugins/plugin-webhooks/src/webhook-redeliver-member.test.ts b/packages/plugins/plugin-webhooks/src/webhook-redeliver-member.test.ts new file mode 100644 index 00000000000..5cbd4c8fa26 --- /dev/null +++ b/packages/plugins/plugin-webhooks/src/webhook-redeliver-member.test.ts @@ -0,0 +1,403 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#22756] `IWebhookService.handleRedeliver`: the webhook redeliver door + * (`POST /api/v1/webhooks/redeliver`) served from a web-standard `Request`, + * through the kernel's `webhooks` slot, for a kernel with no raw app. + * + * Driven on a real `LiteKernel`, with the plugin's REAL boot: its + * `kernel:ready` hook installs the #8069 veto, registers the `webhooks` slot + * and mounts the self-hosted route on a real Hono app when `http.server` + * offers one. The messaging service is the real `MessagingService` over the + * real `MemoryHttpOutbox`, so every refusal below is the outbox's own (the + * tenant scope, the row-local refusals and the veto), not a fixture's. + * + * What this pins: + * 1. the member, read off the started `webhooks` slot on a kernel with no + * `http.server` (the hosted shape), answers every arm of the contract: + * 401 / 400 / 400 / 404 / 409 ×2 / 500 / 200, each with `code` + `status`; + * 2. parity: for the same request the member answers what the mount answers, + * status, headers and body bytes; + * 3. the composition: the route is mounted once where a raw app exists, read + * from `http.server` before the `http-server` alias, and never where none + * does; the slot is registered everywhere the veto is, and nowhere else; + * 4. the veto no longer needs realtime or auto-enqueue: a redelivery of a + * delivery whose subscription is gone is refused by both faces. + */ + +import { randomUUID } from 'node:crypto'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Hono } from 'hono'; +import { LiteKernel } from '@objectstack/core'; +import type { Plugin, PluginContext } from '@objectstack/core'; +import type { IWebhookService } from '@objectstack/spec/contracts'; +import { assertEngineFindOnePredicate } from '@objectstack/metadata-core'; +import { MemoryHttpOutbox, MessagingService } from '@objectstack/service-messaging'; +import { WebhookOutboxPlugin, type WebhookOutboxPluginOptions } from './webhook-outbox-plugin.js'; + +const PATH = '/api/v1/webhooks/redeliver'; +const ORIGIN = 'http://webhooks.test'; +const SUBSCRIPTION = 'wh_live'; +const GONE_SUBSCRIPTION = 'wh_gone'; +const ORG_A = 'org_a'; +const ORG_B = 'org_b'; +const silentLogger = { info() {}, warn() {}, error() {}, debug() {} } as any; + +/** The slice of ObjectQL the plugin's boot and the veto's lookup touch. */ +function fakeEngine(subscriptions: Map>) { + return { + registry: { listItems: () => [], getObject: () => undefined }, + async find() { return []; }, + async findOne(object: string, opts?: any) { + assertEngineFindOnePredicate(object, opts); + return subscriptions.get(opts?.where?.id) ?? null; + }, + registerHook() { /* provenance stamp + headers gate + cleartext gate */ }, + unregisterHooksByPackage() { return 0; }, + }; +} + +/** Realtime that records nothing; present only where a case asks for it. */ +const fakeRealtime = () => ({ + async subscribe() { return `sub_${randomUUID()}`; }, + async unsubscribe() {}, + async publish() {}, +}); + +type ServerShape = 'none' | 'canonical' | 'alias' | 'both' | 'split'; + +interface BootOptions { + server?: ServerShape; + realtime?: boolean; + outbox?: boolean; + /** `false` hands the plugin a messaging surface with no `registerRedeliverGuard`. */ + guardable?: boolean; + plugin?: WebhookOutboxPluginOptions; +} + +/** + * Stands in for the plugin's declared dependency + * (`dependencies = ['com.objectstack.service.messaging']`) and registers the + * collaborators it resolves: the real messaging service over a real outbox, + * an engine holding the subscriptions, an `auth` service whose sessions are + * keyed by the request's `Authorization` header, and the host's server. + */ +class FixturePlugin implements Plugin { + name = 'com.objectstack.service.messaging'; + type = 'standard' as const; + version = '1.0.0'; + readonly subscriptions = new Map>([ + [SUBSCRIPTION, { id: SUBSCRIPTION, name: 'orders_hook', active: true }], + ]); + readonly outbox = new MemoryHttpOutbox(); + readonly messaging = new MessagingService({ logger: silentLogger }); + readonly sessionReads: any[] = []; + readonly canonicalApp = new Hono(); + readonly aliasApp = new Hono(); + + constructor(private readonly options: BootOptions) { + if (options.outbox !== false) this.messaging.setHttpOutbox(this.outbox); + } + + init(ctx: PluginContext): void { + ctx.registerService('manifest', { register: () => {}, list: () => [] }); + ctx.registerService('objectql', fakeEngine(this.subscriptions)); + if (this.options.realtime) ctx.registerService('realtime', fakeRealtime()); + const messaging = this.messaging; + ctx.registerService('messaging', this.options.guardable === false + ? { + enqueueHttp: (input: any) => messaging.enqueueHttp(input), + isHttpDeliveryReady: () => messaging.isHttpDeliveryReady(), + redeliverHttp: (id: string, o: any) => messaging.redeliverHttp(id, o), + } + : messaging); + const reads = this.sessionReads; + ctx.registerService('auth', { + api: { + async getSession(input: { headers: Headers; query?: unknown }) { + reads.push(input); + const auth = input.headers.get('authorization') ?? ''; + const match = /^Bearer (\w+)@(\w+)$/.exec(auth); + if (!match) return null; + return { user: { id: match[1] }, session: { userId: match[1], activeOrganizationId: match[2] } }; + }, + }, + }); + const server = (app: Hono) => ({ getRawApp: () => app }); + const shape = this.options.server ?? 'none'; + if (shape === 'canonical' || shape === 'both' || shape === 'split') { + ctx.registerService('http.server', server(this.canonicalApp)); + } + if (shape === 'alias' || shape === 'both') ctx.registerService('http-server', server(this.canonicalApp)); + if (shape === 'split') ctx.registerService('http-server', server(this.aliasApp)); + } +} + +/** The real plugin, with what it logs during boot observed. */ +class ObservedWebhookPlugin extends WebhookOutboxPlugin { + readonly logs: Array<{ level: string; message: string }> = []; + + override async init(ctx: PluginContext): Promise { + for (const level of ['debug', 'info', 'warn', 'error'] as const) { + vi.spyOn(ctx.logger, level).mockImplementation((message: unknown) => { + this.logs.push({ level, message: String(message) }); + }); + } + await super.init(ctx); + } + + lines(level: string, fragment: string): string[] { + return this.logs.filter((l) => l.level === level && l.message.includes(fragment)).map((l) => l.message); + } +} + +const openKernels: LiteKernel[] = []; + +afterEach(async () => { + while (openKernels.length) { + try { await openKernels.pop()?.shutdown(); } catch { /* already stopped */ } + } + vi.restoreAllMocks(); +}); + +async function boot(options: BootOptions = {}) { + const fixture = new FixturePlugin(options); + const plugin = new ObservedWebhookPlugin(options.plugin); + const kernel = new LiteKernel({ logger: { level: 'silent' } as any }); + openKernels.push(kernel); + kernel.use(fixture); + kernel.use(plugin); + await kernel.bootstrap(); + + const slot = (): IWebhookService | undefined => { + try { return kernel.getService('webhooks'); } catch { return undefined; } + }; + const member = (request: Request): Promise => { + const svc = slot(); + if (typeof svc?.handleRedeliver !== 'function') throw new Error('the webhooks slot holds no handleRedeliver'); + return svc.handleRedeliver(request); + }; + const mounted = (app: Hono) => app.routes.filter((r) => r.path === PATH).map((r) => r.method); + + /** A delivery the dispatcher attempted once and dead-lettered: a row redeliver may replay. */ + const attempted = async (organizationId: string | undefined, refId = SUBSCRIPTION): Promise => { + const id = await fixture.outbox.enqueue({ + source: 'webhook', refId, dedupKey: randomUUID(), url: 'https://receiver.example/hook', + payload: { event: 'record.created' }, organizationId, + }); + const claimed = (await fixture.outbox.claim({ nodeId: 'node_1', limit: 100, claimTtlMs: 60_000 })) + .find((row) => row.id === id)!; + await fixture.outbox.ack( + id, + { success: false, dead: true, httpStatus: 500, error: 'HTTP 500', durationMs: 5 }, + { claimedBy: claimed.claimedBy!, claimedAt: claimed.claimedAt! }, + ); + return id; + }; + /** A delivery still waiting for its first attempt: not finished, so not replayable. */ + const pending = (organizationId: string | undefined) => fixture.outbox.enqueue({ + source: 'webhook', refId: SUBSCRIPTION, dedupKey: randomUUID(), url: 'https://receiver.example/hook', + payload: { event: 'record.created' }, organizationId, + }); + /** A parked record of a delivery that could never be prepared: never sent. */ + const parked = (organizationId: string | undefined) => fixture.messaging.enqueueHttp({ + source: 'webhook', refId: SUBSCRIPTION, dedupKey: randomUUID(), url: 'https://receiver.example/hook', + payload: { event: 'record.created' }, organizationId, undeliverableReason: 'secret unresolvable', + }); + const statusOf = async (id: string) => (await fixture.outbox.list()).find((row) => row.id === id)?.status; + + return { kernel, fixture, plugin, slot, member, mounted, attempted, pending, parked, statusOf }; +} + +/** A redeliver `POST`. `auth` is `user@org`; a string body is sent as it is. */ +const redeliver = (body: unknown, auth: string | null = `u_1@${ORG_A}`, headers: Record = {}) => + new Request(`${ORIGIN}${PATH}`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + ...(auth ? { Authorization: `Bearer ${auth}` } : {}), + ...headers, + }, + body: typeof body === 'string' ? body : JSON.stringify(body), + }); + +/** Everything a response says, the body as bytes. */ +async function snapshot(res: Response) { + return { + status: res.status, + headers: [...res.headers.entries()], + body: new Uint8Array(await res.arrayBuffer()), + }; +} +const json = (s: { body: Uint8Array }) => JSON.parse(new TextDecoder().decode(s.body)); +const JSON_HEADERS = [['content-type', 'application/json']]; + +describe('[#22756] handleRedeliver — the redeliver door from a Request, on a kernel with no http.server', () => { + it('the started kernel holds the member in its webhooks slot, and mounts nothing', async () => { + const { slot, plugin } = await boot(); + expect(typeof slot()?.handleRedeliver).toBe('function'); + expect(plugin.lines('info', 'redeliver endpoint mounted')).toEqual([]); + }); + + it('answers every refusal the contract declares, each with its code and its status', async () => { + const { member, attempted, pending, parked, statusOf } = await boot(); + const crossOrg = await attempted(ORG_B); + const vetoed = await attempted(ORG_A, GONE_SUBSCRIPTION); + const neverSent = await parked(ORG_A); + const notFinished = await pending(ORG_A); + + const cases: Array<[string, Request, number, string]> = [ + ['no signed-in user', redeliver({ deliveryId: crossOrg }, null), 401, 'UNAUTHENTICATED'], + ['a body that is not JSON', redeliver('{deliveryId:'), 400, 'INVALID_REQUEST'], + ['no deliveryId', redeliver({}), 400, 'MISSING_REQUIRED_FIELD'], + ['a deliveryId that is not a string', redeliver({ deliveryId: 42 }), 400, 'MISSING_REQUIRED_FIELD'], + ['a blank deliveryId', redeliver({ deliveryId: ' ' }), 400, 'MISSING_REQUIRED_FIELD'], + ['an unknown delivery', redeliver({ deliveryId: 'no_such_row' }), 404, 'RESOURCE_NOT_FOUND'], + ['another organization\'s delivery', redeliver({ deliveryId: crossOrg }), 404, 'RESOURCE_NOT_FOUND'], + ['a delivery not yet finished', redeliver({ deliveryId: notFinished }), 409, 'DELIVERY_NOT_ELIGIBLE'], + ['a parked delivery, never sent', redeliver({ deliveryId: neverSent }), 409, 'DELIVERY_NEVER_SENT'], + ['a delivery whose subscription is gone (the veto)', redeliver({ deliveryId: vetoed }), 409, 'DELIVERY_NOT_ELIGIBLE'], + ]; + for (const [label, request, status, code] of cases) { + const answer = await snapshot(await member(request)); + expect(answer.status, label).toBe(status); + expect(answer.headers, label).toEqual(JSON_HEADERS); + const body = json(answer); + expect(body, label).toMatchObject({ success: false, error: { code } }); + expect(typeof body.error.message, label).toBe('string'); + } + // Every refusal wrote nothing. + expect(await statusOf(crossOrg)).toBe('dead'); + expect(await statusOf(vetoed)).toBe('dead'); + expect(await statusOf(neverSent)).toBe('dead'); + expect(await statusOf(notFinished)).toBe('pending'); + }); + + it('replays a finished delivery in the caller\'s organization: 200 with its id and new status', async () => { + const { member, attempted, statusOf } = await boot(); + const id = await attempted(ORG_A); + expect(await statusOf(id)).toBe('dead'); + + const answer = await snapshot(await member(redeliver({ deliveryId: ` ${id} ` }))); + expect(answer.status).toBe(200); + expect(answer.headers).toEqual(JSON_HEADERS); + expect(json(answer)).toEqual({ success: true, data: { id, status: 'pending' } }); + expect(await statusOf(id)).toBe('pending'); + }); + + it('answers 500 INTERNAL_ERROR for any other failure — a messaging service with no outbox', async () => { + const { member } = await boot({ outbox: false }); + const answer = await snapshot(await member(redeliver({ deliveryId: 'del_1' }))); + expect(answer.status).toBe(500); + expect(json(answer)).toMatchObject({ success: false, error: { code: 'INTERNAL_ERROR' } }); + }); + + it('reads the caller\'s session by the in-process rule: a cookie request reads without renewal', async () => { + const { member, fixture } = await boot(); + await member(redeliver({ deliveryId: 'no_such_row' }, `u_1@${ORG_A}`, { Cookie: 'better-auth.session_token=tok.sig' })); + await member(redeliver({ deliveryId: 'no_such_row' })); + expect(fixture.sessionReads).toHaveLength(2); + expect(fixture.sessionReads[0].query).toEqual({ disableRefresh: true }); + expect('query' in fixture.sessionReads[1]).toBe(false); + }); +}); + +describe('[#22756] parity — the member answers what the self-hosted mount answers', () => { + it('every refusal: status, headers and body bytes equal, through the member and the mount', async () => { + const { fixture, member, attempted, pending, parked } = await boot({ server: 'both' }); + const app = fixture.canonicalApp; + const crossOrg = await attempted(ORG_B); + const vetoed = await attempted(ORG_A, GONE_SUBSCRIPTION); + const neverSent = await parked(ORG_A); + const notFinished = await pending(ORG_A); + + const requests: Array<[string, () => Request]> = [ + ['401', () => redeliver({ deliveryId: crossOrg }, null)], + ['400 not JSON', () => redeliver('{deliveryId:')], + ['400 empty body', () => redeliver('')], + ['400 no deliveryId', () => redeliver({})], + ['400 not a string', () => redeliver({ deliveryId: 42 })], + ['400 blank', () => redeliver({ deliveryId: ' ' })], + ['404 unknown', () => redeliver({ deliveryId: 'no_such_row' })], + ['404 cross-org', () => redeliver({ deliveryId: crossOrg })], + ['409 not finished', () => redeliver({ deliveryId: notFinished })], + ['409 never sent', () => redeliver({ deliveryId: neverSent })], + ['409 veto', () => redeliver({ deliveryId: vetoed })], + ]; + for (const [label, make] of requests) { + const viaMember = await snapshot(await member(make())); + const viaMount = await snapshot(await app.fetch(make())); + expect(viaMount.status, label).toBeGreaterThanOrEqual(400); + expect(viaMember, label).toEqual(viaMount); + } + }); + + it('a replay: the same status, headers and bytes but for the row id — twin rows, one per face', async () => { + const { fixture, member, attempted, statusOf } = await boot({ server: 'both' }); + const viaMemberId = await attempted(ORG_A); + const viaMountId = await attempted(ORG_A); + + const viaMember = await snapshot(await member(redeliver({ deliveryId: viaMemberId }))); + const viaMount = await snapshot(await fixture.canonicalApp.fetch(redeliver({ deliveryId: viaMountId }))); + expect(viaMount.status).toBe(200); + expect(viaMember.status).toBe(viaMount.status); + expect(viaMember.headers).toEqual(viaMount.headers); + const text = (s: { body: Uint8Array }, id: string) => new TextDecoder().decode(s.body).replace(id, 'ID'); + expect(text(viaMember, viaMemberId)).toBe(text(viaMount, viaMountId)); + expect([await statusOf(viaMemberId), await statusOf(viaMountId)]).toEqual(['pending', 'pending']); + }); +}); + +describe('[#22756] the composition — one mount where a raw app exists, read from http.server first', () => { + it('http.server and its alias on one server (plugin-hono-server\'s shape): mounted once', async () => { + const { fixture, mounted, slot, plugin } = await boot({ server: 'both' }); + expect(mounted(fixture.canonicalApp)).toEqual(['POST']); + expect(typeof slot()?.handleRedeliver).toBe('function'); + expect(plugin.lines('info', 'redeliver endpoint mounted')).toHaveLength(1); + }); + + it('http.server alone (runtime.ts\'s config.server shape, no alias): mounted once', async () => { + const { fixture, mounted } = await boot({ server: 'canonical' }); + expect(mounted(fixture.canonicalApp)).toEqual(['POST']); + }); + + it('the alias alone: still mounted once', async () => { + const { fixture, mounted } = await boot({ server: 'alias' }); + expect(mounted(fixture.canonicalApp)).toEqual(['POST']); + }); + + it('two servers under the two names: the mount goes on http.server\'s, and only there', async () => { + const { fixture, mounted } = await boot({ server: 'split' }); + expect(mounted(fixture.canonicalApp)).toEqual(['POST']); + expect(mounted(fixture.aliasApp)).toEqual([]); + }); +}); + +describe('[#22756] the veto is installed without realtime or auto-enqueue, and the slot only beside it', () => { + for (const [label, options] of [ + ['no realtime', {}], + ['realtime present, autoEnqueue: false', { realtime: true, plugin: { autoEnqueue: false } }], + ] as Array<[string, BootOptions]>) { + it(`${label}: a delivery whose subscription is gone is refused by both faces`, async () => { + const { fixture, member, attempted, statusOf } = await boot({ ...options, server: 'both' }); + const vetoed = await attempted(ORG_A, GONE_SUBSCRIPTION); + // Control: the same kernel replays a delivery whose subscription stands. + const replayable = await attempted(ORG_A); + + for (const face of [member, (r: Request) => fixture.canonicalApp.fetch(r)]) { + const answer = await snapshot(await face(redeliver({ deliveryId: vetoed }))); + expect(answer.status).toBe(409); + expect(json(answer).error.code).toBe('DELIVERY_NOT_ELIGIBLE'); + expect(json(answer).error.message).toContain(GONE_SUBSCRIPTION); + } + expect(await statusOf(vetoed)).toBe('dead'); + expect((await member(redeliver({ deliveryId: replayable }))).status).toBe(200); + }); + } + + it('a messaging service that cannot take the veto: no webhooks slot, and the absence is logged at error', async () => { + const { slot, plugin } = await boot({ guardable: false }); + expect(slot()).toBeUndefined(); + expect(plugin.lines('error', 'exposes no registerRedeliverGuard')).toHaveLength(1); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b983d9712fc..e4bd6bb7f0f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2029,6 +2029,9 @@ importers: '@types/node': specifier: ^26.6.3 version: 26.6.3 + hono: + specifier: ^4.13.5 + version: 4.13.12 tsx: specifier: ^4.23.15 version: 4.23.15(patch_hash=4d1d35da1809be2a945519cb26890e48107810b9b66c6913a6a89abb5ad6a1e6) diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 9c398254490..71d6fa9bd3e 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3656,6 +3656,11 @@ "verb": "update", "pinned": 1 }, + { + "file": "packages/plugins/plugin-webhooks/src/webhook-redeliver-member.test.ts", + "verb": "findOne", + "pinned": 1 + }, { "file": "packages/plugins/plugin-webhooks/src/webhook-signing-secret.test.ts", "verb": "delete",