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/server.ts b/apps/hub/src/server.ts index f53d6f662..4e68962dd 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -51,21 +51,20 @@ 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"; -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 { installWebhooks, type HookMailRouter } from "@corbits/webhooks"; +import { reportError } from "@corbits/error-sink"; +import { createRunTriggerDeliverer, installWebhooks, type HookMailRouter } from "@corbits/webhooks"; import { createProcessSidecarProvisioner, readProcessProvisionerConfig, @@ -546,6 +545,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 +576,37 @@ 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), + // 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", + }), + ), + onDeliveryError: (error, schedule) => { + reportError(error, { + operation: "hub.cron.deliver", + extra: { scheduleId: schedule.id, tenantId: schedule.tenantId }, }); }, }); @@ -600,27 +624,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/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 23615e03e..db2400824 100644 --- a/packages/cron/src/index.ts +++ b/packages/cron/src/index.ts @@ -10,12 +10,6 @@ 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"; +export { createRunTriggerCronDeliver, type RunTriggerDeliverer } from "./deliver"; 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; }); };