diff --git a/apps/hub/package.json b/apps/hub/package.json index 8b50861412..e7e617dcfa 100644 --- a/apps/hub/package.json +++ b/apps/hub/package.json @@ -37,7 +37,7 @@ "@corbits/last-30-days-research-workflow": "workspace:*", "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#e85b81e4a2e0cc318dd33a4f0f92337b6fac7f38", "@corbits/mcp-tools": "workspace:*", - "@corbits/memory": "github:corbitsdev/corbits-memory#9e6f213fa2c002b531d3f6af1aa0abd737b8afe3", + "@corbits/memory": "github:corbitsdev/corbits-memory#e74da20f148a302dff5400915fe504ee2395e913", "@corbits/morning-brief-workflow": "workspace:*", "@corbits/notify": "workspace:*", "@corbits/pain-point-collateral-workflow": "workspace:*", diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c5b3ba2d45..a16a1a5473 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -33,12 +33,15 @@ import { import { reportError } from "@corbits/error-sink"; import { decodedOrNull } from "@corbits/url-path"; import { + buildMailFrame, createInMemoryMailboxEventBus, createMailboxDb, createMailboxPersist, + generateMailboxMessageId, mountMailbox, } from "@corbits/mailbox"; import { createMemory, loadMemoryConfig } from "@corbits/memory"; +import { applyCronMigrations, createCronTicker, mountCron } from "@corbits/cron"; import { createHubMailboxAuthorizeSender, createHubPersistMailWithSessionEnsure, @@ -546,6 +549,9 @@ export async function createHub(config: HubConfig) { // Package-owned artifacts tables (CL-8188): idempotent, advisory-locked, // safe on every boot of every replica — see @corbits/artifacts' README. await runArtifactMigrations(db); + // `@corbits/cron`'s own schema (CL-8183): idempotent, its own migration + // ledger — see the package's README. + await applyCronMigrations(config.databaseUrl); const artifactContentStore = InlineContentStore; const baseEventCollectors = createEventCollectorRegistry({ db, @@ -881,6 +887,68 @@ export async function createHub(config: HubConfig) { app.route(`${TENANT_PREFIX}/mailbox`, mailboxApp); } + let cronTicker: { start(): void; stop(): void } | undefined; + { + // Cron-to-mail bridge (CL-8183): `@corbits/cron` registers its own + // absolute `/api/tenants/:tenantId/cron` routes, so it gets its own + // `Hono` routed at "/" — matching `mailboxApp` above — + // rather than `TENANT_PREFIX`, which would double the tenant-route + // prefix. `requireTenantMember` reads the tenant middleware's context + // exactly like the mailbox mount's `resolvePrincipal` above. + const cronApp = new Hono(); + mountCron(cronApp, { + db, + requireTenantMember: (ctx, tenantId) => { + const c = ctx as { get(key: "tenant"): { id: string } }; + return c.get("tenant").id === tenantId; + }, + }); + app.route("/", cronApp); + + // The ticker delivers each due schedule as mail through the same + // `lookups.persistMail` every other outbound frame in this hub already + // dispatches through (see the mailbox mount's own `deliver` above). + // `@corbits/cron` hands back only `{to, subject, body, from}`, not a + // raw frame, so this hub builds the RFC 5322 message itself with + // `@corbits/mailbox`'s own `buildMailFrame` — the same builder that + // package uses for every frame it authors. `senderAddressFor` addresses + // the sender as `cron@`, the same domain-scoped shape + // every other mailbox writer in this file uses (see the mailbox + // mount's `senderAddressFor` above), by looking the domain up per + // tenant the way that one does. + cronTicker = createCronTicker({ + db, + intervalMs: 60_000, + senderAddressFor: (tenantId) => `cron@${tenantId}`, + deliver: async (message) => { + // `message.from` is `cron@` from `senderAddressFor` + // above; resolve the real tenant domain to address it from. + 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 lookups.persistMail({ + senderAddress: from, + recipients: message.to, + raw: buildMailFrame({ + from, + to: message.to.join(", "), + subject: message.subject, + body: message.body, + messageId: generateMailboxMessageId(from), + }), + }); + }, + }); + cronTicker.start(); + } + { // Firm memory (CL-8186): `@corbits/memory` registers its own // absolute `/api/tenants/:tenantId/memory/*` routes on whatever @@ -1149,6 +1217,7 @@ export async function createHub(config: HubConfig) { clearTimeout(sidecarAllocationReconciliationTimer); } await closeMailbox(); + cronTicker?.stop(); // The pool end waits on in-flight queries; a query whose socket // died with the process must never stall shutdown, so bound it. // (CL-7584: a fire-and-forget reconcile's request can be cut diff --git a/bun.lock b/bun.lock index c4417b1478..ae48bc7aeb 100644 --- a/bun.lock +++ b/bun.lock @@ -115,7 +115,7 @@ "@corbits/last-30-days-research-workflow": "workspace:*", "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#e85b81e4a2e0cc318dd33a4f0f92337b6fac7f38", "@corbits/mcp-tools": "workspace:*", - "@corbits/memory": "github:corbitsdev/corbits-memory#9e6f213fa2c002b531d3f6af1aa0abd737b8afe3", + "@corbits/memory": "github:corbitsdev/corbits-memory#e74da20f148a302dff5400915fe504ee2395e913", "@corbits/morning-brief-workflow": "workspace:*", "@corbits/notify": "workspace:*", "@corbits/pain-point-collateral-workflow": "workspace:*", @@ -3304,7 +3304,7 @@ "@workbench/hub/@corbits/mailbox": ["@corbits/mailbox@github:corbitsdev/corbits-mailbox#e85b81e", { "dependencies": { "@hono/standard-validator": "0.2.3", "@intx/crypto": "0.3.0", "@standard-community/standard-json": "0.3.5", "@standard-community/standard-openapi": "0.2.9", "arktype": "2.1.29", "hono-openapi": "1.3.1" }, "peerDependencies": { "@intx/log": "^0.2.2", "drizzle-orm": "^0.45.2", "hono": "^4.12.0", "postgres": "^3.4.0" } }, "corbitsdev-corbits-mailbox-e85b81e", "sha512-ATPP/pll1sYG1DLKV1HvtboK8q1tFkqayQqRXdUmRUIsFJx5wqzd8YA1XELjXzLu1P2f2uCMe2CjsJHGv6E90Q=="], - "@workbench/hub/@corbits/memory": ["@corbits/memory@github:corbitsdev/corbits-memory#9e6f213", { "dependencies": { "@intx/agent": "0.2.2", "@intx/authz": "0.2.2", "@intx/hub-api": "0.2.2", "@intx/log": "0.2.2", "@intx/workflow": "0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.1", "hono": "^4.9.0", "hono-openapi": "^1.3.1", "postgres": "^3.4.7" } }, "corbitsdev-corbits-memory-9e6f213", "sha512-utnM4ZT2zmslcPXYWAAqxlDNLcpGsXFiTOtj8h7+OXnhCP0Eaw8yl25+yCTyHpvt3jcdeG4h5uFsSj7ou0BZCA=="], + "@workbench/hub/@corbits/memory": ["@corbits/memory@github:corbitsdev/corbits-memory#e74da20", { "dependencies": { "@intx/agent": "0.2.2", "@intx/authz": "0.2.2", "@intx/hub-api": "0.2.2", "@intx/log": "0.2.2", "@intx/workflow": "0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.1", "hono": "^4.9.0", "hono-openapi": "^1.3.1", "postgres": "^3.4.7" } }, "corbitsdev-corbits-memory-e74da20", "sha512-/uA9xTcF7XwMXgnR8OZamierzkLG4g4mhw3u9Fl3Uj6ait8pH3wcoQ6V32on1nMdaqtPRdYVFnxof5O+4ZacGA=="], "@workbench/sidecar/@corbits/oauth-core": ["@corbits/oauth-core@github:corbitsdev/corbits-oauth-core#e1e69e6", { "dependencies": { "arktype": "2.2.3" } }, "corbitsdev-corbits-oauth-core-e1e69e6", "sha512-wUrD73iVyk/Dtb4yRn3hCh6N8syfsEvAkRz4XYqT0FIB3sYDSMYEWHcDMJx9nq64my/HBOeN1J6PHIOtpjhtPg=="], diff --git a/packages/cron/src/index.ts b/packages/cron/src/index.ts index 21d41d6238..2c3556bb08 100644 --- a/packages/cron/src/index.ts +++ b/packages/cron/src/index.ts @@ -9,7 +9,13 @@ export { type CronField, type ZonedParts, } from "./cron"; -export { cronScheduleTable } from "./schema"; +export { + cronScheduleTable, + cronMigrations, + applyCronMigrations, + type CronMigration, + type ApplyCronMigrationsReport, +} from "./schema"; export { createCronTicker, cronSenderAddress,