diff --git a/src/gateway/server-plugin-cron.ts b/src/gateway/server-plugin-cron.ts new file mode 100644 index 000000000..33b3806e9 --- /dev/null +++ b/src/gateway/server-plugin-cron.ts @@ -0,0 +1,159 @@ +/** + * Plugin cron job integration. + * Discovers and registers cron jobs from plugins that expose a `*.getCronJobs` gateway method. + */ + +import type { CronService } from "../cron/service.js"; +import type { CronJobCreate } from "../cron/types.js"; +import type { GatewayRequestHandler } from "./server-methods/types.js"; + +type PluginCronResult = { + jobs?: CronJobCreate[]; +}; + +/** + * Call a plugin gateway handler that returns cron jobs. + * Plugin handlers may either: + * 1. Return the result directly (simpler pattern used by some plugins) + * 2. Call respond() with the result (standard gateway handler pattern) + * + * This helper supports both patterns. + */ +async function callCronJobsHandler(handler: GatewayRequestHandler): Promise { + return new Promise((resolve) => { + let resolved = false; + + // Create a respond function that captures the result + const respond = (ok: boolean, payload?: unknown) => { + if (!resolved) { + resolved = true; + resolve(ok ? (payload as PluginCronResult) : { jobs: [] }); + } + }; + + // Create minimal mock options for the handler + const mockOpts = { + req: { type: "req" as const, method: "getCronJobs", id: "plugin-cron-init" }, + params: {}, + client: null, + isWebchatConnect: () => false, + respond, + context: {} as never, // Handlers that need context won't work, but getCronJobs doesn't need it + }; + + // Call the handler - it may return a value or call respond() + const result = handler(mockOpts); + + // Handle direct return value (async or sync) + if (result instanceof Promise) { + result + .then((value) => { + if (!resolved && value !== undefined) { + resolved = true; + resolve(value as PluginCronResult); + } else if (!resolved) { + // Handler completed without returning or calling respond - return empty + resolved = true; + resolve({ jobs: [] }); + } + }) + .catch(() => { + if (!resolved) { + resolved = true; + resolve({ jobs: [] }); + } + }); + } else if (!resolved && result !== undefined) { + // Sync handler returned a value + resolved = true; + resolve(result as PluginCronResult); + } + + // Fallback timeout in case handler doesn't respond + setTimeout(() => { + if (!resolved) { + resolved = true; + resolve({ jobs: [] }); + } + }, 5000); + }); +} + +/** + * Discovers and registers cron jobs from plugins. + * Looks for gateway methods matching the pattern `*.getCronJobs` and calls them + * to retrieve cron job configurations, then registers those jobs with the cron service. + * + * @param params.cron - The cron service to register jobs with + * @param params.handlers - Plugin gateway handlers to scan for getCronJobs methods + * @param params.log - Logger for status messages + */ +export async function loadPluginCronJobs(params: { + cron: CronService; + handlers: Record; + log: { + info: (msg: string) => void; + warn: (msg: string) => void; + error: (msg: string) => void; + }; +}): Promise { + const { cron, handlers, log } = params; + + // Find all handlers matching *.getCronJobs pattern + const cronMethods = Object.entries(handlers).filter(([method]) => + method.endsWith(".getCronJobs"), + ); + + if (cronMethods.length === 0) { + return; + } + + let totalJobs = 0; + + for (const [method, handler] of cronMethods) { + const pluginId = method.replace(".getCronJobs", ""); + try { + // Call the gateway method to get cron jobs + const result = await callCronJobsHandler(handler); + const jobs = result?.jobs; + + if (!Array.isArray(jobs) || jobs.length === 0) { + continue; + } + + // Register each job with the cron service + for (const job of jobs) { + try { + // Check if job already exists (by name) + const existing = await cron.list(); + const existingJob = existing.find((j) => j.name === job.name); + + if (existingJob) { + // Update existing job schedule if it changed + if ( + JSON.stringify(existingJob.schedule) !== JSON.stringify(job.schedule) || + existingJob.enabled !== job.enabled + ) { + await cron.update(existingJob.id, { + schedule: job.schedule, + enabled: job.enabled, + }); + } + } else { + // Add new job + await cron.add(job); + totalJobs++; + } + } catch (err) { + log.warn(`[${pluginId}] failed to register cron job "${job.name}": ${String(err)}`); + } + } + } catch (err) { + log.error(`[${pluginId}] getCronJobs failed: ${String(err)}`); + } + } + + if (totalJobs > 0) { + log.info(`registered ${totalJobs} cron job(s) from plugins`); + } +} diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index fdf40be61..88c55e802 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -52,6 +52,7 @@ import { createGatewayCloseHandler } from "./server-close.js"; import { buildGatewayCronService } from "./server-cron.js"; import { applyGatewayLaneConcurrency } from "./server-lanes.js"; import { startGatewayMaintenanceTimers } from "./server-maintenance.js"; +import { loadPluginCronJobs } from "./server-plugin-cron.js"; import { coreGatewayHandlers } from "./server-methods.js"; import { GATEWAY_EVENTS, listGatewayMethods } from "./server-methods-list.js"; import { loadGatewayModelCatalog } from "./server-model-catalog.js"; @@ -504,6 +505,18 @@ export async function startGatewayServer( logBrowser, })); + // Load cron jobs from plugins that expose a *.getCronJobs gateway method. + // This runs after plugin services start so storage is initialized. + void loadPluginCronJobs({ + cron, + handlers: pluginRegistry.gatewayHandlers, + log: { + info: (msg) => logCron.info(msg), + warn: (msg) => logCron.warn(msg), + error: (msg) => logCron.error(msg), + }, + }).catch((err) => logCron.error(`failed to load plugin cron jobs: ${String(err)}`)); + const { applyHotReload, requestGatewayRestart } = createGatewayReloadHandlers({ deps, broadcast,