From 050c12a101e50713a2a25d48d2528e7f1e045b07 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 18 Sep 2026 08:44:16 -0700 Subject: [PATCH 1/2] fix(hub): cron ticker delivers as an authorized mailbox sender (CL-8533) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A due cron schedule was handed to the mailbox persist path with a synthesized `cron@` sender. That path authorizes a sender against a live routable endpoint — only an open workflow run address resolves — so the mailbox write was refused and the hub's own persist threw for the same reason. Nothing was delivered, and the run was never triggered: persisting mail records does not fire a run. A schedule firing with nobody signed in is a trigger, not a person's mail. It now takes the route an inbound webhook already takes: materialize the run's mail-triggered grants, hand them to the sidecar, and route a signed frame in which the run is the authenticated sender of its own trigger mail. No authorization check is skipped; the frame never enters the mailbox sender path. The ticker no longer invents a sender address — it names the tenant and leaves the identity to the host — and one schedule's failed delivery is now reported and isolated instead of aborting the whole tick. --- apps/hub/src/cron-deliver.test.ts | 82 ++++++++++++++++++++ apps/hub/src/cron-deliver.ts | 122 ++++++++++++++++++++++++++++++ apps/hub/src/server.ts | 82 ++++++++++---------- packages/cron/src/index.ts | 9 +-- packages/cron/src/ticker.ts | 39 ++++++---- 5 files changed, 270 insertions(+), 64 deletions(-) create mode 100644 apps/hub/src/cron-deliver.test.ts create mode 100644 apps/hub/src/cron-deliver.ts diff --git a/apps/hub/src/cron-deliver.test.ts b/apps/hub/src/cron-deliver.test.ts new file mode 100644 index 000000000..42a17399b --- /dev/null +++ b/apps/hub/src/cron-deliver.test.ts @@ -0,0 +1,82 @@ +// The bug this covers: a due schedule used to be handed to the mailbox +// persist path, whose sender authorization has no live endpoint for +// `cron@` and drops the delivery. Cron must take the run-trigger +// route instead, with the run as the authenticated sender. +import { describe, expect, test } from "bun:test"; + +import { createCronDeliver } from "./cron-deliver"; + +const RUN_ADDRESS = "run_abc123@alice.localhost"; + +function routerSpy() { + const routed: { address: string; authenticatedSender: string }[] = []; + const granted: string[] = []; + return { + routed, + granted, + router: { + routeMail: (address: string, _raw: string, authenticatedSender: string) => { + routed.push({ address, authenticatedSender }); + return true; + }, + sendRunGrants: (address: string) => { + granted.push(address); + return true; + }, + }, + }; +} + +describe("createCronDeliver", () => { + test("routes a due schedule as the run's own authenticated sender", async () => { + const spy = routerSpy(); + const deliver = createCronDeliver({ + router: spy.router, + materialize: async () => ({ outcome: "materialized", stepGrants: [] }), + tenantDomain: async () => "alice.localhost", + }); + + await deliver({ + to: [RUN_ADDRESS], + subject: "standup", + body: "time to check in", + tenantId: "tenant-1", + }); + + expect(spy.granted).toEqual([RUN_ADDRESS]); + expect(spy.routed).toEqual([{ address: RUN_ADDRESS, authenticatedSender: RUN_ADDRESS }]); + }); + + test("refuses an address that is not a live run", async () => { + const spy = routerSpy(); + const deliver = createCronDeliver({ + router: spy.router, + materialize: async () => ({ outcome: "materialized", stepGrants: [] }), + tenantDomain: async () => "alice.localhost", + }); + + await expect( + deliver({ + to: ["alice@alice.localhost"], + subject: "s", + body: "b", + tenantId: "tenant-1", + }), + ).rejects.toThrow("not a live run address"); + expect(spy.routed).toEqual([]); + }); + + test("refuses when the deployment has no materialized grants", async () => { + const spy = routerSpy(); + const deliver = createCronDeliver({ + router: spy.router, + materialize: async () => ({ outcome: "rejected", message: "run is terminal" }), + tenantDomain: async () => "alice.localhost", + }); + + await expect( + deliver({ to: [RUN_ADDRESS], subject: "s", body: "b", tenantId: "tenant-1" }), + ).rejects.toThrow("run is terminal"); + expect(spy.routed).toEqual([]); + }); +}); diff --git a/apps/hub/src/cron-deliver.ts b/apps/hub/src/cron-deliver.ts new file mode 100644 index 000000000..652b6ab07 --- /dev/null +++ b/apps/hub/src/cron-deliver.ts @@ -0,0 +1,122 @@ +// A cron schedule fires with nobody signed in, so it cannot ride the +// mailbox persist path: that path authorizes its sender against a live +// routable endpoint, and `cron@` is not one. A due schedule is a +// trigger, exactly like an inbound webhook, so it takes the same route an +// inbound webhook takes: materialize the run's mail-triggered grants, hand +// them to the sidecar, then route a signed trigger frame in which the run +// is the authenticated sender of its own trigger mail. No authorization +// check is skipped -- the frame never enters the mailbox sender path at +// all. + +import { createEd25519Crypto, generateKeyPair } from "@intx/crypto"; +import { + assembleMessage, + assembleSignedContent, + createDetachedSignatureFromProvider, +} from "@intx/mime"; +import { base64Encode, deriveWorkflowRunId, isRunAddress } from "@intx/types"; + +export type CronMailRouter = { + routeMail: ( + address: string, + rawMessage: string, + authenticatedSender: string, + messageId?: string, + ) => boolean; + sendRunGrants: ( + address: string, + runId: string, + stepGrants: unknown, + senderIdentities: unknown, + ) => boolean; +}; + +export type CronMaterializeRunGrants = (args: { agentAddress: string; runId: string }) => Promise<{ + outcome: string; + stepGrants?: unknown; + code?: string; + message?: string; +}>; + +export type CreateCronDeliverOpts = { + router: CronMailRouter; + materialize: CronMaterializeRunGrants; + tenantDomain: (tenantId: string) => Promise; +}; + +export type CronMessage = { + to: string[]; + subject: string; + body: string; + tenantId: string; +}; + +export function createCronDeliver( + opts: CreateCronDeliverOpts, +): (message: CronMessage) => Promise { + return async (message) => { + const domain = await opts.tenantDomain(message.tenantId); + for (const address of message.to) { + if (!isRunAddress(address)) { + throw new Error(`cron schedule address "${address}" is not a live run address`); + } + const runId = deriveWorkflowRunId(address); + const grants = await opts.materialize({ agentAddress: address, runId }); + if (grants.outcome !== "materialized" || grants.stepGrants === undefined) { + throw new Error(grants.message ?? grants.code ?? `no live deployment at "${address}"`); + } + // A cron trigger carries no inbound sender, so there is no sender key + // to co-deliver on this barrier. + if (!opts.router.sendRunGrants(address, runId, grants.stepGrants, undefined)) { + throw new Error(`run grants not routable for "${address}"`); + } + const raw = await assembleCronMail({ + address, + subject: message.subject, + body: message.body, + tenantId: message.tenantId, + domain, + }); + if (!opts.router.routeMail(address, raw.base64, address, raw.messageId)) { + throw new Error(`run mail not routable for "${address}"`); + } + } + }; +} + +async function assembleCronMail(opts: { + address: string; + subject: string; + body: string; + tenantId: string; + domain: string; +}): Promise<{ base64: string; messageId: string }> { + const cryptoProvider = createEd25519Crypto(await generateKeyPair()); + const messageId = `<${crypto.randomUUID()}@${opts.domain}>`; + const signedContent = assembleSignedContent({ kind: "conversation", text: opts.body }); + const rawMessage = assembleMessage( + { + from: `cron@${opts.domain}`, + to: [opts.address], + cc: undefined, + date: new Date(), + messageId, + subject: opts.subject, + inReplyTo: undefined, + references: undefined, + mimeVersion: "1.0" as const, + interchangeType: "conversation.message" as const, + interchangeCorrelationId: undefined, + interchangeAgentId: undefined, + interchangeSessionId: undefined, + interchangeOfferingId: undefined, + interchangeSchemaVersion: undefined, + interchangeTenantId: opts.tenantId, + traceparent: undefined, + tracestate: undefined, + }, + signedContent, + await createDetachedSignatureFromProvider(signedContent, cryptoProvider), + ); + return { base64: base64Encode(rawMessage), messageId }; +} diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index f53d6f662..719325719 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -51,11 +51,9 @@ import { Hono } from "hono"; // Everything above this line is upstream Interchange's server.ts, verbatim; // see AGENTS.md's "plain Interchange tenant" ruling. import { - buildMailFrame, createInMemoryMailboxEventBus, createMailboxDb, createMailboxPersist, - generateMailboxMessageId, mountMailbox, } from "@corbits/mailbox"; import { createMemory, loadMemoryConfig } from "@corbits/memory"; @@ -65,6 +63,8 @@ import { createHubPersistMailWithSessionEnsure, } from "./mailbox-persist"; import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send"; +import { createCronDeliver } from "./cron-deliver"; +import { reportError } from "@corbits/error-sink"; import { installWebhooks, type HookMailRouter } from "@corbits/webhooks"; import { createProcessSidecarProvisioner, @@ -546,6 +546,22 @@ export async function createHubServer({ app.route(`${TENANT_PREFIX}/mailbox`, mailboxApp); } + // The router every system-originated trigger (webhook, cron) goes + // through. `HookMailRouter` types its payloads as `unknown` at the + // package boundary; this just narrows them back to `sidecarRouter`'s own + // types on the way through, with no behavior change. + const systemTriggerMailRouter: HookMailRouter = { + routeMail: (address, rawMessage, authenticatedSender, messageId) => + sidecarRouter.routeMail(address, rawMessage, authenticatedSender, messageId), + sendRunGrants: (address, runId, stepGrants, senderIdentities) => + sidecarRouter.sendRunGrants( + address, + runId, + stepGrants as Parameters[2], + senderIdentities as Parameters[3], + ), + }; + let cronTicker: { start(): void; stop(): void } | undefined; { const cronApp = new Hono(); @@ -561,28 +577,29 @@ export async function createHubServer({ cronTicker = createCronTicker({ db, intervalMs: 60_000, - senderAddressFor: (tenantId) => `cron@${tenantId}`, - deliver: async (message) => { - const tenantId = message.from.slice("cron@".length); - const [tenantRow] = await db - .select({ domain: tenantTable.domain }) - .from(tenantTable) - .where(eq(tenantTable.id, tenantId)) - .limit(1); - if (tenantRow === undefined) { - throw new Error(`no tenant "${tenantId}" to address cron mail from`); - } - const from = `cron@${tenantRow.domain}`; - await mailboxLookups.persistMail({ - senderAddress: from, - recipients: message.to, - raw: buildMailFrame({ - from, - to: message.to.join(", "), - subject: message.subject, - body: message.body, - messageId: generateMailboxMessageId(from), - }), + deliver: createCronDeliver({ + router: systemTriggerMailRouter, + materialize: createMailTriggeredRunGrantsMaterializer({ + db, + principalKeyStore, + grantStore, + }), + tenantDomain: async (tenantId) => { + const [tenantRow] = await db + .select({ domain: tenantTable.domain }) + .from(tenantTable) + .where(eq(tenantTable.id, tenantId)) + .limit(1); + if (tenantRow === undefined) { + throw new Error(`no tenant "${tenantId}" to address cron mail from`); + } + return tenantRow.domain; + }, + }), + onDeliveryError: (error, schedule) => { + reportError(error, { + operation: "hub.cron.deliver", + extra: { scheduleId: schedule.id, tenantId: schedule.tenantId }, }); }, }); @@ -600,27 +617,12 @@ export async function createHubServer({ app.route("/", memoryApp); } - // `HookMailRouter` types its payloads as `unknown` at the package - // boundary; this just narrows them back to `sidecarRouter`'s own types - // on the way through, with no behavior change. - const webhookMailRouter: HookMailRouter = { - routeMail: (address, rawMessage, authenticatedSender, messageId) => - sidecarRouter.routeMail(address, rawMessage, authenticatedSender, messageId), - sendRunGrants: (address, runId, stepGrants, senderIdentities) => - sidecarRouter.sendRunGrants( - address, - runId, - stepGrants as Parameters[2], - senderIdentities as Parameters[3], - ), - }; - await installWebhooks({ app, db, credentialCipher, principalKeyStore, - router: webhookMailRouter, + router: systemTriggerMailRouter, }); // End of Corbits mount block. diff --git a/packages/cron/src/index.ts b/packages/cron/src/index.ts index 23615e03e..d6fbba33e 100644 --- a/packages/cron/src/index.ts +++ b/packages/cron/src/index.ts @@ -10,12 +10,5 @@ export { type ZonedParts, } from "./cron"; export { cronScheduleTable, applyCronMigrations } from "./schema"; -export { - createCronTicker, - cronSenderAddress, - type CronDb, - type CronSenderAddress, - type CronTicker, - type DeliverCronMail, -} from "./ticker"; +export { createCronTicker, type CronDb, type CronTicker, type DeliverCronMail } from "./ticker"; export { mountCron, type MountCronOpts, type RequireTenantMember } from "./mount"; diff --git a/packages/cron/src/ticker.ts b/packages/cron/src/ticker.ts index 5a3657915..a3638ea87 100644 --- a/packages/cron/src/ticker.ts +++ b/packages/cron/src/ticker.ts @@ -11,25 +11,25 @@ import { cronScheduleTable } from "./schema"; export type CronDb = Record> = PostgresJsDatabase; +/** A due schedule handed to the host. The sender identity is the host's to + * decide: only the host knows which addresses its mail transport + * authorizes, so this package names the tenant and never invents an + * address for it. */ export type DeliverCronMail = (message: { to: string[]; subject: string; body: string; - from: string; + tenantId: string; }) => Promise | void; -export type CronSenderAddress = (tenantId: string) => string; - -/** The system sender address a tenant's cron mail comes from. */ -export const cronSenderAddress: CronSenderAddress = (tenantId) => `cron@${tenantId}.internal`; - export type CreateCronTickerOpts< TSchema extends Record = Record, > = { db: CronDb; deliver: DeliverCronMail; intervalMs: number; - senderAddressFor?: CronSenderAddress; + /** Told about a delivery that failed, so the host can report it. */ + onDeliveryError?: (error: unknown, schedule: { id: string; tenantId: string }) => void; }; export type CronTicker = { @@ -56,7 +56,7 @@ function isDue( async function tick>( db: CronDb, deliver: DeliverCronMail, - senderAddressFor: CronSenderAddress, + onDeliveryError: (error: unknown, schedule: { id: string; tenantId: string }) => void, ) { await db.transaction(async (tx) => { const now = new Date(); @@ -70,12 +70,19 @@ async function tick>( .for("update", { skipLocked: true }); for (const row of candidates.filter((row) => isDue(row, now))) { - await deliver({ - to: [row.toAddress], - subject: row.subject, - body: row.body, - from: senderAddressFor(row.tenantId), - }); + // One schedule's failed delivery is its own: the tick still advances + // every due row, so a permanently undeliverable schedule cannot block + // the rest of the table or re-fire every minute forever. + try { + await deliver({ + to: [row.toAddress], + subject: row.subject, + body: row.body, + tenantId: row.tenantId, + }); + } catch (error) { + onDeliveryError(error, { id: row.id, tenantId: row.tenantId }); + } await tx .update(cronScheduleTable) .set({ lastFiredAt: now }) @@ -88,13 +95,13 @@ async function tick>( export function createCronTicker>( opts: CreateCronTickerOpts, ): CronTicker { - const senderAddressFor = opts.senderAddressFor ?? cronSenderAddress; + const onDeliveryError = opts.onDeliveryError ?? (() => undefined); let timer: ReturnType | undefined; let inFlight: Promise | undefined; const runTick = () => { if (inFlight !== undefined) return; - inFlight = tick(opts.db, opts.deliver, senderAddressFor).finally(() => { + inFlight = tick(opts.db, opts.deliver, onDeliveryError).finally(() => { inFlight = undefined; }); }; From 90dc73e1a51a79eace0b1901bbf76f82747a7cc9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 18 Sep 2026 08:50:53 -0700 Subject: [PATCH 2/2] refactor(hub): cron delivers through the webhooks run-trigger deliverer The hub is upstream server.ts plus one Corbits block, so the run-trigger frame assembly does not belong in a hub file. @corbits/webhooks now exports the deliverer it already had internally, with a caller-supplied sender local part; @corbits/cron fans a due schedule's recipients out over it. --- apps/hub/package.json | 2 +- apps/hub/src/cron-deliver.test.ts | 82 -------------------- apps/hub/src/cron-deliver.ts | 122 ------------------------------ apps/hub/src/server.ts | 49 +++++++----- bun.lock | 4 +- packages/cron/src/deliver.test.ts | 31 ++++++++ packages/cron/src/deliver.ts | 21 +++++ packages/cron/src/index.ts | 1 + 8 files changed, 84 insertions(+), 228 deletions(-) delete mode 100644 apps/hub/src/cron-deliver.test.ts delete mode 100644 apps/hub/src/cron-deliver.ts create mode 100644 packages/cron/src/deliver.test.ts create mode 100644 packages/cron/src/deliver.ts diff --git a/apps/hub/package.json b/apps/hub/package.json index 563da3b4d..ee9656570 100644 --- a/apps/hub/package.json +++ b/apps/hub/package.json @@ -21,7 +21,7 @@ "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#65590a85fa143251b3ac20ba2eca92fc35e70e51", "@corbits/memory": "github:corbitsdev/corbits-memory#e74da20f148a302dff5400915fe504ee2395e913", "@corbits/url-path": "workspace:*", - "@corbits/webhooks": "github:corbitsdev/webhooks#570bd52688920c0ce018a8fbcd046199fa4125dd", + "@corbits/webhooks": "github:corbitsdev/webhooks#5c9e7d8fad13ebfbc747d01981ed4b0b44b810cd", "@corbits/workflows": "workspace:*", "@corbits/xai-provider": "github:corbitsdev/corbits-xai-provider#9f2d4bac40ea075df092fef807404a13c638cf14", "@intx/authz": "0.3.0", diff --git a/apps/hub/src/cron-deliver.test.ts b/apps/hub/src/cron-deliver.test.ts deleted file mode 100644 index 42a17399b..000000000 --- a/apps/hub/src/cron-deliver.test.ts +++ /dev/null @@ -1,82 +0,0 @@ -// The bug this covers: a due schedule used to be handed to the mailbox -// persist path, whose sender authorization has no live endpoint for -// `cron@` and drops the delivery. Cron must take the run-trigger -// route instead, with the run as the authenticated sender. -import { describe, expect, test } from "bun:test"; - -import { createCronDeliver } from "./cron-deliver"; - -const RUN_ADDRESS = "run_abc123@alice.localhost"; - -function routerSpy() { - const routed: { address: string; authenticatedSender: string }[] = []; - const granted: string[] = []; - return { - routed, - granted, - router: { - routeMail: (address: string, _raw: string, authenticatedSender: string) => { - routed.push({ address, authenticatedSender }); - return true; - }, - sendRunGrants: (address: string) => { - granted.push(address); - return true; - }, - }, - }; -} - -describe("createCronDeliver", () => { - test("routes a due schedule as the run's own authenticated sender", async () => { - const spy = routerSpy(); - const deliver = createCronDeliver({ - router: spy.router, - materialize: async () => ({ outcome: "materialized", stepGrants: [] }), - tenantDomain: async () => "alice.localhost", - }); - - await deliver({ - to: [RUN_ADDRESS], - subject: "standup", - body: "time to check in", - tenantId: "tenant-1", - }); - - expect(spy.granted).toEqual([RUN_ADDRESS]); - expect(spy.routed).toEqual([{ address: RUN_ADDRESS, authenticatedSender: RUN_ADDRESS }]); - }); - - test("refuses an address that is not a live run", async () => { - const spy = routerSpy(); - const deliver = createCronDeliver({ - router: spy.router, - materialize: async () => ({ outcome: "materialized", stepGrants: [] }), - tenantDomain: async () => "alice.localhost", - }); - - await expect( - deliver({ - to: ["alice@alice.localhost"], - subject: "s", - body: "b", - tenantId: "tenant-1", - }), - ).rejects.toThrow("not a live run address"); - expect(spy.routed).toEqual([]); - }); - - test("refuses when the deployment has no materialized grants", async () => { - const spy = routerSpy(); - const deliver = createCronDeliver({ - router: spy.router, - materialize: async () => ({ outcome: "rejected", message: "run is terminal" }), - tenantDomain: async () => "alice.localhost", - }); - - await expect( - deliver({ to: [RUN_ADDRESS], subject: "s", body: "b", tenantId: "tenant-1" }), - ).rejects.toThrow("run is terminal"); - expect(spy.routed).toEqual([]); - }); -}); diff --git a/apps/hub/src/cron-deliver.ts b/apps/hub/src/cron-deliver.ts deleted file mode 100644 index 652b6ab07..000000000 --- a/apps/hub/src/cron-deliver.ts +++ /dev/null @@ -1,122 +0,0 @@ -// A cron schedule fires with nobody signed in, so it cannot ride the -// mailbox persist path: that path authorizes its sender against a live -// routable endpoint, and `cron@` is not one. A due schedule is a -// trigger, exactly like an inbound webhook, so it takes the same route an -// inbound webhook takes: materialize the run's mail-triggered grants, hand -// them to the sidecar, then route a signed trigger frame in which the run -// is the authenticated sender of its own trigger mail. No authorization -// check is skipped -- the frame never enters the mailbox sender path at -// all. - -import { createEd25519Crypto, generateKeyPair } from "@intx/crypto"; -import { - assembleMessage, - assembleSignedContent, - createDetachedSignatureFromProvider, -} from "@intx/mime"; -import { base64Encode, deriveWorkflowRunId, isRunAddress } from "@intx/types"; - -export type CronMailRouter = { - routeMail: ( - address: string, - rawMessage: string, - authenticatedSender: string, - messageId?: string, - ) => boolean; - sendRunGrants: ( - address: string, - runId: string, - stepGrants: unknown, - senderIdentities: unknown, - ) => boolean; -}; - -export type CronMaterializeRunGrants = (args: { agentAddress: string; runId: string }) => Promise<{ - outcome: string; - stepGrants?: unknown; - code?: string; - message?: string; -}>; - -export type CreateCronDeliverOpts = { - router: CronMailRouter; - materialize: CronMaterializeRunGrants; - tenantDomain: (tenantId: string) => Promise; -}; - -export type CronMessage = { - to: string[]; - subject: string; - body: string; - tenantId: string; -}; - -export function createCronDeliver( - opts: CreateCronDeliverOpts, -): (message: CronMessage) => Promise { - return async (message) => { - const domain = await opts.tenantDomain(message.tenantId); - for (const address of message.to) { - if (!isRunAddress(address)) { - throw new Error(`cron schedule address "${address}" is not a live run address`); - } - const runId = deriveWorkflowRunId(address); - const grants = await opts.materialize({ agentAddress: address, runId }); - if (grants.outcome !== "materialized" || grants.stepGrants === undefined) { - throw new Error(grants.message ?? grants.code ?? `no live deployment at "${address}"`); - } - // A cron trigger carries no inbound sender, so there is no sender key - // to co-deliver on this barrier. - if (!opts.router.sendRunGrants(address, runId, grants.stepGrants, undefined)) { - throw new Error(`run grants not routable for "${address}"`); - } - const raw = await assembleCronMail({ - address, - subject: message.subject, - body: message.body, - tenantId: message.tenantId, - domain, - }); - if (!opts.router.routeMail(address, raw.base64, address, raw.messageId)) { - throw new Error(`run mail not routable for "${address}"`); - } - } - }; -} - -async function assembleCronMail(opts: { - address: string; - subject: string; - body: string; - tenantId: string; - domain: string; -}): Promise<{ base64: string; messageId: string }> { - const cryptoProvider = createEd25519Crypto(await generateKeyPair()); - const messageId = `<${crypto.randomUUID()}@${opts.domain}>`; - const signedContent = assembleSignedContent({ kind: "conversation", text: opts.body }); - const rawMessage = assembleMessage( - { - from: `cron@${opts.domain}`, - to: [opts.address], - cc: undefined, - date: new Date(), - messageId, - subject: opts.subject, - inReplyTo: undefined, - references: undefined, - mimeVersion: "1.0" as const, - interchangeType: "conversation.message" as const, - interchangeCorrelationId: undefined, - interchangeAgentId: undefined, - interchangeSessionId: undefined, - interchangeOfferingId: undefined, - interchangeSchemaVersion: undefined, - interchangeTenantId: opts.tenantId, - traceparent: undefined, - tracestate: undefined, - }, - signedContent, - await createDetachedSignatureFromProvider(signedContent, cryptoProvider), - ); - return { base64: base64Encode(rawMessage), messageId }; -} diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index 719325719..4e68962dd 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -57,15 +57,14 @@ import { mountMailbox, } from "@corbits/mailbox"; import { createMemory, loadMemoryConfig } from "@corbits/memory"; -import { createCronTicker, mountCron } from "@corbits/cron"; +import { createCronTicker, createRunTriggerCronDeliver, mountCron } from "@corbits/cron"; import { createHubMailboxAuthorizeSender, createHubPersistMailWithSessionEnsure, } from "./mailbox-persist"; import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send"; -import { createCronDeliver } from "./cron-deliver"; import { reportError } from "@corbits/error-sink"; -import { installWebhooks, type HookMailRouter } from "@corbits/webhooks"; +import { createRunTriggerDeliverer, installWebhooks, type HookMailRouter } from "@corbits/webhooks"; import { createProcessSidecarProvisioner, readProcessProvisionerConfig, @@ -577,25 +576,33 @@ export async function createHubServer({ cronTicker = createCronTicker({ db, intervalMs: 60_000, - deliver: createCronDeliver({ - router: systemTriggerMailRouter, - materialize: createMailTriggeredRunGrantsMaterializer({ - db, - principalKeyStore, - grantStore, + // A due schedule fires with nobody signed in, so it cannot ride the + // mailbox persist path, which authorizes its sender against a live + // routable endpoint. It is a system trigger like an inbound webhook, + // so it takes the same route: the run is the authenticated sender of + // its own signed trigger mail, with its grants materialized first. + deliver: createRunTriggerCronDeliver( + createRunTriggerDeliverer({ + router: systemTriggerMailRouter, + materialize: createMailTriggeredRunGrantsMaterializer({ + db, + principalKeyStore, + grantStore, + }), + tenantDomain: async (tenantId) => { + const [tenantRow] = await db + .select({ domain: tenantTable.domain }) + .from(tenantTable) + .where(eq(tenantTable.id, tenantId)) + .limit(1); + if (tenantRow === undefined) { + throw new Error(`no tenant "${tenantId}" to address cron mail from`); + } + return tenantRow.domain; + }, + senderLocalPart: "cron", }), - tenantDomain: async (tenantId) => { - const [tenantRow] = await db - .select({ domain: tenantTable.domain }) - .from(tenantTable) - .where(eq(tenantTable.id, tenantId)) - .limit(1); - if (tenantRow === undefined) { - throw new Error(`no tenant "${tenantId}" to address cron mail from`); - } - return tenantRow.domain; - }, - }), + ), onDeliveryError: (error, schedule) => { reportError(error, { operation: "hub.cron.deliver", diff --git a/bun.lock b/bun.lock index cecd11b78..a24cb6462 100644 --- a/bun.lock +++ b/bun.lock @@ -62,7 +62,7 @@ "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#65590a85fa143251b3ac20ba2eca92fc35e70e51", "@corbits/memory": "github:corbitsdev/corbits-memory#e74da20f148a302dff5400915fe504ee2395e913", "@corbits/url-path": "workspace:*", - "@corbits/webhooks": "github:corbitsdev/webhooks#570bd52688920c0ce018a8fbcd046199fa4125dd", + "@corbits/webhooks": "github:corbitsdev/webhooks#5c9e7d8fad13ebfbc747d01981ed4b0b44b810cd", "@corbits/workflows": "workspace:*", "@corbits/xai-provider": "github:corbitsdev/corbits-xai-provider#9f2d4bac40ea075df092fef807404a13c638cf14", "@intx/authz": "0.3.0", @@ -645,7 +645,7 @@ "@corbits/url-path": ["@corbits/url-path@workspace:packages/url-path"], - "@corbits/webhooks": ["@corbits/webhooks@github:corbitsdev/webhooks#570bd52", { "dependencies": { "@intx/crypto": "^0.3.0", "@intx/db": "^0.3.0", "@intx/hub-api": "^0.3.0", "@intx/mime": "^0.3.0", "@intx/types": "^0.3.0", "hono": "4.11.9" } }, "corbitsdev-webhooks-570bd52", "sha512-GWduublrOUDgiH20a5VHFcI/zcf4mcFlQL9IEesyXKXJLekqpLFbacs4V490h0jl0oVG9KQ5dVb+CjRfTcajZA=="], + "@corbits/webhooks": ["@corbits/webhooks@github:corbitsdev/webhooks#5c9e7d8", { "dependencies": { "@intx/crypto": "^0.3.0", "@intx/db": "^0.3.0", "@intx/hub-api": "^0.3.0", "@intx/mime": "^0.3.0", "@intx/types": "^0.3.0", "hono": "4.11.9" } }, "corbitsdev-webhooks-5c9e7d8", "sha512-Zk3f6nhXwJ8NWrmMLd12+ypDJyAXfuR9/c14UT+n49MC3BTVavzzppcZK6TPQ5XnLtypfYDmOoHTWC+EwgNjDA=="], "@corbits/workflows": ["@corbits/workflows@workspace:packages/workflows"], diff --git a/packages/cron/src/deliver.test.ts b/packages/cron/src/deliver.test.ts new file mode 100644 index 000000000..aaaa9a6bf --- /dev/null +++ b/packages/cron/src/deliver.test.ts @@ -0,0 +1,31 @@ +import { expect, test } from "bun:test"; + +import { createRunTriggerCronDeliver } from "./deliver"; + +test("every recipient of a due schedule gets its own trigger", async () => { + const calls: Array<[string, string, string, string | undefined]> = []; + const deliver = createRunTriggerCronDeliver({ + to: async (address, content, tenantId, subject) => { + calls.push([address, content, tenantId, subject]); + }, + }); + + await deliver({ to: ["run-a@x", "run-b@x"], subject: "Daily", body: "go", tenantId: "t1" }); + + expect(calls).toEqual([ + ["run-a@x", "go", "t1", "Daily"], + ["run-b@x", "go", "t1", "Daily"], + ]); +}); + +test("a failed recipient stops the fan-out so the ticker can report it", async () => { + const deliver = createRunTriggerCronDeliver({ + to: async (address) => { + if (address === "run-a@x") throw new Error("not routable"); + }, + }); + + expect( + deliver({ to: ["run-a@x", "run-b@x"], subject: "s", body: "b", tenantId: "t1" }), + ).rejects.toThrow("not routable"); +}); diff --git a/packages/cron/src/deliver.ts b/packages/cron/src/deliver.ts new file mode 100644 index 000000000..a8a301146 --- /dev/null +++ b/packages/cron/src/deliver.ts @@ -0,0 +1,21 @@ +import type { DeliverCronMail } from "./ticker"; + +/** A system-trigger deliverer, shaped like `@corbits/webhooks`'s + * `MailDeliverer`. Structural so this package stays free of it. */ +export type RunTriggerDeliverer = { + to: ( + address: string, + content: string, + tenantId: string, + subject: string | undefined, + ) => Promise; +}; + +/** Fan a due schedule's recipients out over a run-trigger deliverer. */ +export function createRunTriggerCronDeliver(deliverer: RunTriggerDeliverer): DeliverCronMail { + return async (message) => { + for (const address of message.to) { + await deliverer.to(address, message.body, message.tenantId, message.subject); + } + }; +} diff --git a/packages/cron/src/index.ts b/packages/cron/src/index.ts index d6fbba33e..db2400824 100644 --- a/packages/cron/src/index.ts +++ b/packages/cron/src/index.ts @@ -12,3 +12,4 @@ export { export { cronScheduleTable, applyCronMigrations } from "./schema"; export { createCronTicker, type CronDb, type CronTicker, type DeliverCronMail } from "./ticker"; export { mountCron, type MountCronOpts, type RequireTenantMember } from "./mount"; +export { createRunTriggerCronDeliver, type RunTriggerDeliverer } from "./deliver";