From 5bf10d485c802dad7d0e51bb54ddf2a1cbd2011b Mon Sep 17 00:00:00 2001 From: Jiayuan Zhang Date: Sun, 23 Aug 2026 17:37:21 +0800 Subject: [PATCH] MUL-6582: add durable scheduled Plugin hooks (#7448) * MUL-6582 feat(plugins): add durable scheduled hooks Co-authored-by: multica-agent * MUL-6582: add schedule-pulse example plugin Ship a real scheduled plugin that records each unique delivery_id in workspace storage, so a host retry does not look like a second pulse. The cmd/server test installs this example manifest and drives two scheduler replicas through a dropped first attempt. Co-authored-by: multica-agent * MUL-6582 fix(migrations): make schedule indexes retry-safe Co-authored-by: multica-agent --------- Co-authored-by: multica-agent --- examples/plugins/schedule-pulse/README.md | 48 +++ .../schedule-pulse/multica.plugin.json | 48 +++ .../plugins/schedule-pulse/server/handler.mjs | 149 +++++++ examples/plugins/schedule-pulse/ui/main.js | 67 +++ examples/plugins/triage-notify/README.md | 22 +- .../plugins/triage-notify/multica.plugin.json | 23 +- .../plugins/triage-notify/server/handler.mjs | 14 +- packages/core/api/schemas.test.ts | 27 ++ packages/core/api/schemas.ts | 18 + packages/core/types/plugin.ts | 20 +- packages/views/locales/en/settings.json | 14 + packages/views/locales/ja/settings.json | 13 + packages/views/locales/ko/settings.json | 13 + packages/views/locales/zh-Hans/settings.json | 13 + packages/views/plugins/index.ts | 2 +- .../views/plugins/plugin-hook-activity.tsx | 61 ++- .../views/settings/components/plugins-tab.tsx | 94 ++++- server/cmd/migrate/main.go | 2 + server/cmd/server/main.go | 5 + .../server/plugin_hook_schedule_job_test.go | 317 +++++++++++++++ .../plugin_schedule_pulse_example_test.go | 380 ++++++++++++++++++ server/internal/handler/plugin.go | 41 +- server/internal/handler/plugin_hook.go | 28 +- server/internal/handler/plugin_test.go | 169 +++++++- .../handler/workspace_delete_manifest_test.go | 1 + server/internal/scheduler/jobs_plugin_hook.go | 353 ++++++++++++++++ .../scheduler/jobs_plugin_hook_test.go | 127 ++++++ server/internal/service/plugin.go | 33 +- server/internal/service/plugin_hook.go | 80 +++- server/internal/service/plugin_hook_test.go | 46 ++- server/internal/service/plugin_schedule.go | 131 ++++++ .../399_plugin_hook_schedule.down.sql | 9 + .../399_plugin_hook_schedule.up.sql | 35 ++ ...k_schedule_installation_key_index.down.sql | 1 + ...ook_schedule_installation_key_index.up.sql | 2 + ...lugin_hook_schedule_enabled_index.down.sql | 1 + ..._plugin_hook_schedule_enabled_index.up.sql | 3 + ...ocation_schedule_trigger_validate.down.sql | 3 + ...nvocation_schedule_trigger_validate.up.sql | 1 + server/pkg/db/generated/models.go | 17 + server/pkg/db/generated/plugin.sql.go | 328 ++++++++++++++- .../pkg/db/generated/workspace_delete.sql.go | 4 + server/pkg/db/queries/plugin.sql | 74 +++- server/pkg/db/queries/workspace_delete.sql | 4 + server/pkg/plugincontract/capabilities.go | 9 +- server/pkg/plugincontract/manifest.go | 91 ++++- server/pkg/plugincontract/manifest_test.go | 57 ++- 47 files changed, 2917 insertions(+), 81 deletions(-) create mode 100644 examples/plugins/schedule-pulse/README.md create mode 100644 examples/plugins/schedule-pulse/multica.plugin.json create mode 100644 examples/plugins/schedule-pulse/server/handler.mjs create mode 100644 examples/plugins/schedule-pulse/ui/main.js create mode 100644 server/cmd/server/plugin_hook_schedule_job_test.go create mode 100644 server/cmd/server/plugin_schedule_pulse_example_test.go create mode 100644 server/internal/scheduler/jobs_plugin_hook.go create mode 100644 server/internal/scheduler/jobs_plugin_hook_test.go create mode 100644 server/internal/service/plugin_schedule.go create mode 100644 server/migrations/399_plugin_hook_schedule.down.sql create mode 100644 server/migrations/399_plugin_hook_schedule.up.sql create mode 100644 server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql create mode 100644 server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql create mode 100644 server/migrations/401_plugin_hook_schedule_enabled_index.down.sql create mode 100644 server/migrations/401_plugin_hook_schedule_enabled_index.up.sql create mode 100644 server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql create mode 100644 server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql diff --git a/examples/plugins/schedule-pulse/README.md b/examples/plugins/schedule-pulse/README.md new file mode 100644 index 000000000..d37c2daf8 --- /dev/null +++ b/examples/plugins/schedule-pulse/README.md @@ -0,0 +1,48 @@ +# Schedule Pulse + +A real scheduled plugin, not a log line. Multica POSTs the plugin's own HTTPS +endpoint every five minutes. The handler records each unique `delivery_id` in +workspace storage **before** it answers, so a retry of the same planned wake +does not look like a second pulse. + +The issue panel reads that storage row. After install you should see the count +go up about once every five minutes, not on every HTTP attempt. + +## What it verifies + +| Host behavior | What this plugin does | +| --- | --- | +| Durable schedule trigger | Declares `triggers: ["schedule"]` and a five-minute cron | +| At-least-once delivery | Persists `delivery_id` first; a retry with the same id returns `{duplicate: true}` | +| Plugin actor | Writes as the installation through the callback token, into `storage:workspace` | +| Consent | `net:pulse.example.com` is the only outbound host; schedule is not a data scope | + +It does not create issues. That is a later Plugin Action API, not this trigger. + +## Running the handler + +```bash +MULTICA_SIGNING_SECRET=whsec_… node server/handler.mjs +``` + +The signing secret is shown once, next to the install token, when an admin +rotates the plugin's token in workspace settings. + +Point `transport.url` at this process and declare the matching `net:` scope. +Local development also needs: + +```bash +export MULTICA_PLUGIN_DEV_ORIGINS=https://localhost:8787 +``` + +Without that opt-in the host refuses a private address. + +## Why storage, not a log + +A log line cannot prove idempotency. The panel shows `count`, `delivery_id`, +`planned_at`, and `attempt`. If the host retries the same plan, `attempt` +changes and `count` does not. + +The callback token is revoked when this HTTP request returns. All storage +writes happen before the 200. Work that must continue after the response +should use the standing `mpi_` install token instead. diff --git a/examples/plugins/schedule-pulse/multica.plugin.json b/examples/plugins/schedule-pulse/multica.plugin.json new file mode 100644 index 000000000..9fb8c74b3 --- /dev/null +++ b/examples/plugins/schedule-pulse/multica.plugin.json @@ -0,0 +1,48 @@ +{ + "manifest_version": 1, + "key": "ai.multica.schedule-pulse", + "name": "Schedule Pulse", + "description": "A scheduled Hook that Multica wakes every five minutes. Each unique delivery is recorded once in workspace storage, including when the host retries.", + "version": "1.0.0", + "author": { + "name": "multica", + "url": "https://multica.ai" + }, + "scopes": [ + "storage:workspace", + "net:pulse.example.com" + ], + "contributes": { + "surfaces": [ + { + "key": "pulse", + "type": "issue_panel", + "name": "Schedule pulse", + "entry": "ui/main.js", + "platforms": [ + "web", + "desktop" + ] + } + ], + "hooks": [ + { + "key": "pulse", + "name": "Schedule pulse", + "description": "Records one workspace-storage pulse per planned delivery. Retries of the same delivery_id do not increment the count.", + "triggers": [ + "schedule" + ], + "schedule": { + "cron": "*/5 * * * *", + "timezone": "UTC" + }, + "transport": { + "type": "http", + "url": "https://pulse.example.com/hooks/pulse" + }, + "timeout_ms": 8000 + } + ] + } +} diff --git a/examples/plugins/schedule-pulse/server/handler.mjs b/examples/plugins/schedule-pulse/server/handler.mjs new file mode 100644 index 000000000..9208b598b --- /dev/null +++ b/examples/plugins/schedule-pulse/server/handler.mjs @@ -0,0 +1,149 @@ +#!/usr/bin/env node +/** + * Schedule Pulse handler — the plugin author's side of a durable scheduled Hook. + * + * Run: + * MULTICA_SIGNING_SECRET=whsec_... node handler.mjs + */ + +import { createServer as createHTTPServer } from "node:http"; +import { createServer as createHTTPSServer } from "node:https"; +import { createHmac, timingSafeEqual } from "node:crypto"; +import { readFileSync } from "node:fs"; + +const PORT = Number(process.env.PORT ?? 8787); +const SIGNING_SECRET = process.env.MULTICA_SIGNING_SECRET ?? ""; +const TOLERANCE_SECONDS = 5 * 60; +const PULSE_KEY = "last_pulse"; + +if (!SIGNING_SECRET) { + console.error("MULTICA_SIGNING_SECRET is required. Rotate the plugin token in Multica to obtain it."); + process.exit(1); +} + +const seenSignatures = new Map(); + +function rememberSignature(signature, now) { + for (const [seen, at] of seenSignatures) { + if (now - at > TOLERANCE_SECONDS * 1000) seenSignatures.delete(seen); + } + if (seenSignatures.has(signature)) return false; + seenSignatures.set(signature, now); + return true; +} + +function verify(rawBody, headers) { + const timestamp = headers["x-multica-timestamp"]; + const presented = String(headers["x-multica-signature"] ?? "").replace(/^v1=/, ""); + if (!timestamp || !presented) return "missing signature headers"; + + const drift = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp)); + if (!Number.isFinite(drift) || drift > TOLERANCE_SECONDS) return "timestamp outside the accepted window"; + + const secret = Buffer.from(SIGNING_SECRET.replace(/^whsec_/, ""), "hex"); + const expected = createHmac("sha256", secret) + .update(timestamp) + .update(".") + .update(rawBody) + .digest("hex"); + + const expectedBytes = Buffer.from(expected, "utf8"); + const presentedBytes = Buffer.from(presented, "utf8"); + if (expectedBytes.length !== presentedBytes.length) return "signature does not match"; + if (!timingSafeEqual(expectedBytes, presentedBytes)) return "signature does not match"; + if (!rememberSignature(presented, Date.now())) return "this request was already delivered"; + return null; +} + +async function callback(body, method, path, payload) { + const response = await fetch(`${body.callback_url}${path}`, { + method, + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${body.callback_token}`, + }, + body: payload === undefined ? undefined : JSON.stringify(payload), + }); + const text = await response.text(); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`Multica answered ${response.status}: ${text}`); + } + return text ? JSON.parse(text) : null; +} + +async function loadPulse(body) { + const stored = await callback(body, "GET", `/storage/workspace/${PULSE_KEY}`); + if (!stored?.value) return { count: 0, delivery_id: "" }; + try { + return JSON.parse(stored.value); + } catch { + return { count: 0, delivery_id: "" }; + } +} + +const tlsCert = process.env.MULTICA_HOOK_TLS_CERT; +const tlsKey = process.env.MULTICA_HOOK_TLS_KEY; +const createServer = tlsCert && tlsKey + ? (handler) => createHTTPSServer({ cert: readFileSync(tlsCert), key: readFileSync(tlsKey) }, handler) + : createHTTPServer; + +const server = createServer((request, response) => { + if (request.method !== "POST" || request.url !== "/hooks/pulse") { + response.writeHead(404).end(); + return; + } + + const chunks = []; + request.on("data", (chunk) => chunks.push(chunk)); + request.on("end", async () => { + const rawBody = Buffer.concat(chunks); + const failure = verify(rawBody, request.headers); + if (failure) { + console.warn(`rejected: ${failure}`); + response.writeHead(401, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ error: failure })); + return; + } + + const body = JSON.parse(rawBody.toString("utf8")); + if (body.hook_key !== "pulse" || body.trigger !== "schedule") { + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ skipped: "not a scheduled pulse" })); + return; + } + if (!body.delivery_id) { + response.writeHead(400, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ error: "scheduled delivery is missing delivery_id" })); + return; + } + + try { + const previous = await loadPulse(body); + if (previous.delivery_id === body.delivery_id) { + console.log(`duplicate delivery ${body.delivery_id} attempt ${body.attempt}`); + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ duplicate: true, delivery_id: body.delivery_id, count: previous.count })); + return; + } + + const pulse = { + delivery_id: body.delivery_id, + count: (previous.count ?? 0) + 1, + last_planned_at: body.schedule?.planned_at ?? "", + last_occurred_at: body.occurred_at ?? "", + last_attempt: body.attempt ?? 1, + }; + await callback(body, "PUT", `/storage/workspace/${PULSE_KEY}`, { value: JSON.stringify(pulse) }); + console.log(`pulse ${pulse.count} delivery ${pulse.delivery_id} attempt ${pulse.last_attempt}`); + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ accepted: true, ...pulse })); + } catch (error) { + console.error(error); + response.writeHead(502, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ error: "pulse failed" })); + } + }); +}); + +server.listen(PORT, () => console.log(`schedule pulse listening on ${tlsCert ? "https" : "http"}://127.0.0.1:${PORT}`)); diff --git a/examples/plugins/schedule-pulse/ui/main.js b/examples/plugins/schedule-pulse/ui/main.js new file mode 100644 index 000000000..92e34f2d0 --- /dev/null +++ b/examples/plugins/schedule-pulse/ui/main.js @@ -0,0 +1,67 @@ +// Schedule Pulse — reads the workspace-storage row the scheduled Hook writes. +// +// Plain ES module, no build step. The surface has no credential; it talks to +// Multica only through the host bridge. + +const pending = new Map(); +const port = globalThis.__multicaPluginBridgePortV2; +let sequence = 0; + +if (!(port instanceof MessagePort)) throw new Error("Multica surface bridge is unavailable"); +delete globalThis.__multicaPluginBridgePortV2; +port.onmessage = (message) => { + const payload = message.data; + if (payload?.kind === "theme") return applyTheme(payload.theme); + const entry = pending.get(payload?.id); + if (!entry) return; + pending.delete(payload.id); + if (payload.ok) entry.resolve(payload.data); + else entry.reject(new Error(payload.error)); +}; +port.start(); +start(); + +function applyTheme(theme) { + for (const [name, value] of Object.entries(theme ?? {})) { + document.documentElement.style.setProperty(name, value); + } +} + +function call(method, path, body) { + const id = `r${++sequence}`; + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }); + port.postMessage({ id, kind: "action", method, path, body }); + }); +} + +function resize() { + port.postMessage({ id: `resize${Date.now()}`, kind: "ui.resize", height: document.body.scrollHeight + 16 }); +} + +async function start() { + const root = document.getElementById("root"); + root.innerHTML = ` +
+
Schedule pulse
+
Loading…
+ +
`; + + const status = root.querySelector("#status"); + const list = root.querySelector("#pulse"); + try { + const stored = await call("GET", "/storage/workspace/last_pulse"); + const pulse = JSON.parse(stored.value); + status.textContent = "Multica has woken this plugin on a durable five-minute schedule."; + list.style.display = "grid"; + list.innerHTML = ` +
Count
${pulse.count ?? 0}
+
Delivery
${pulse.delivery_id ?? "—"}
+
Planned
${pulse.last_planned_at ?? "—"}
+
Attempt
${pulse.last_attempt ?? "—"}
`; + } catch { + status.textContent = "No pulse yet. After install, Multica calls this plugin every five minutes and writes the first delivery here."; + } + resize(); +} diff --git a/examples/plugins/triage-notify/README.md b/examples/plugins/triage-notify/README.md index d36349f3b..3ec57728f 100644 --- a/examples/plugins/triage-notify/README.md +++ b/examples/plugins/triage-notify/README.md @@ -7,16 +7,18 @@ The three previous examples only ever ran inbound — a sandboxed panel asking t host for things. This one has a backend, which is what makes every check in the hook engine necessary. -## The three triggers +## The four triggers -The same hook, reached three ways. What differs is who caused it, and therefore -whose name the resulting comment carries. +The triage hook is reached three ways; the heartbeat adds a fourth, scheduled +trigger. What differs is who caused each call and therefore whose name any +resulting comment carries. | Trigger | Where it comes from | Blocks? | Comment author | | --- | --- | --- | --- | | `ui` | The **Triage this issue** button in the panel | The panel only | The person who clicked | | `manual` | The **Triage this issue** entry in the issue actions menu | That menu only | The person who picked it | | `event` | Automatically, when an issue is created | **Never** | The plugin itself | +| `schedule` | Every five minutes in UTC | **Never** | The plugin itself | `event` never blocking is not a performance tuning choice. The event bus runs its listeners inline on the goroutine of the request that published, so a hook @@ -58,6 +60,20 @@ it (recorded with `via_plugin_id`), an `event` call writes as the plugin. A handler cannot elect to write as somebody else, which is the reason the callback token exists instead of just handing out the install token. +The callback token is revoked when this HTTP request returns and is held by one +Multica server instance. Use it only for work completed before responding. A +scheduled integration that needs to continue asynchronously or reconcile a +large external backlog should store the `mpi_` install token shown to the admin +at token rotation and use that standing credential instead. + +## Scheduled delivery identity + +The `scheduled_heartbeat` Hook is intentionally side-effect free. It logs the +stable `delivery_id`, per-attempt `invocation_id` and `attempt`, plus +`schedule.planned_at`. A real scheduled Hook must persist `delivery_id` before +performing side effects: network timeouts and stale-lease recovery can deliver +the same planned occurrence more than once. + ## Where a plugin can send data `net:triage.example.com` in the manifest is the whole answer, and the admin sees diff --git a/examples/plugins/triage-notify/multica.plugin.json b/examples/plugins/triage-notify/multica.plugin.json index 582f57241..415813dd5 100644 --- a/examples/plugins/triage-notify/multica.plugin.json +++ b/examples/plugins/triage-notify/multica.plugin.json @@ -2,8 +2,8 @@ "manifest_version": 1, "key": "ai.multica.triage-notify", "name": "Triage Notify", - "description": "Sends an issue to an external triage service and writes its answer back as a comment. Demonstrates all three host-driven hook triggers: a button in the panel, an entry in the issue menu, and automatic dispatch when an issue is created.", - "version": "1.0.0", + "description": "Sends an issue to an external triage service and writes its answer back as a comment. Also demonstrates a no-side-effect Hook that Multica wakes every five minutes.", + "version": "1.1.0", "author": { "name": "multica", "url": "https://multica.ai" @@ -62,7 +62,24 @@ "url": "https://triage.example.com/hooks/triage" }, "timeout_ms": 8000 + }, + { + "key": "scheduled_heartbeat", + "name": "Scheduled heartbeat", + "description": "Acknowledges a durable scheduled delivery without writing to Multica or another service.", + "triggers": [ + "schedule" + ], + "schedule": { + "cron": "*/5 * * * *", + "timezone": "UTC" + }, + "transport": { + "type": "http", + "url": "https://triage.example.com/hooks/heartbeat" + }, + "timeout_ms": 8000 } ] } -} \ No newline at end of file +} diff --git a/examples/plugins/triage-notify/server/handler.mjs b/examples/plugins/triage-notify/server/handler.mjs index 985a2374c..008450a09 100644 --- a/examples/plugins/triage-notify/server/handler.mjs +++ b/examples/plugins/triage-notify/server/handler.mjs @@ -120,7 +120,7 @@ const createServer = tlsCert && tlsKey : createHTTPServer; const server = createServer((request, response) => { - if (request.method !== "POST" || !request.url?.startsWith("/hooks/triage")) { + if (request.method !== "POST" || !request.url?.startsWith("/hooks/")) { response.writeHead(404).end(); return; } @@ -144,6 +144,18 @@ const server = createServer((request, response) => { console.log(`hook ${body.hook_key} via ${body.trigger}${body.event_type ? ` (${body.event_type})` : ""} as ${body.actor?.type}`); try { + // A scheduled delivery has no issue or caller input. delivery_id is + // stable across retries while invocation_id changes per HTTP attempt; a + // real side-effecting handler persists delivery_id before doing work. + if (body.hook_key === "scheduled_heartbeat" && body.trigger === "schedule") { + console.log( + `scheduled delivery ${body.delivery_id} attempt ${body.attempt} planned ${body.schedule?.planned_at}`, + ); + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ received: body.delivery_id })); + return; + } + // The host tells us which issue this is about, having already resolved and // permission-checked it. Preferred over anything in `input`: for an event // trigger there was no client to supply it, and for ui/manual a diff --git a/packages/core/api/schemas.test.ts b/packages/core/api/schemas.test.ts index cf588d76d..6bf2b2752 100644 --- a/packages/core/api/schemas.test.ts +++ b/packages/core/api/schemas.test.ts @@ -1670,6 +1670,33 @@ describe("Plugin schemas", () => { expect(parsed.installed_version).toBe("1.0.0"); expect(parsed.added_scopes).toEqual(["comments:write"]); }); + + it("preserves automatic schedules on both consent and installed Plugin payloads", () => { + const schedule = { cron: "*/5 * * * *", timezone: "Asia/Shanghai" }; + const preview = PluginPreviewSchema.parse({ + manifest: { + key: "com.example.digest", + name: "Digest", + version: "1.0.0", + author: { name: "example" }, + contributes: { + hooks: [{ key: "digest", name: "Digest", triggers: ["schedule"], schedule }], + }, + }, + }); + expect(preview.manifest.contributes?.hooks?.[0]?.schedule).toEqual(schedule); + + const installation = PluginInstallationSchema.parse({ + id: "installation-1", + hooks: [{ + key: "digest", + name: "Digest", + triggers: ["schedule"], + schedule: { ...schedule, next_run_at: "2026-08-23T10:15:00Z" }, + }], + }); + expect(installation.hooks[0]?.schedule?.next_run_at).toBe("2026-08-23T10:15:00Z"); + }); }); // Issue status catalog (MUL-6243). The catalog drives how every status renders, diff --git a/packages/core/api/schemas.ts b/packages/core/api/schemas.ts index 83db10844..85dc600cf 100644 --- a/packages/core/api/schemas.ts +++ b/packages/core/api/schemas.ts @@ -112,6 +112,11 @@ export const PluginHookSchema = z.object({ description: z.string().default(""), triggers: z.array(z.string()).default([]), events: z.array(z.string()).default([]), + schedule: z.object({ + cron: z.string().default(""), + timezone: z.string().default(""), + next_run_at: z.string().optional(), + }).loose().optional(), transport: z.string().default(""), }).loose(); @@ -190,6 +195,8 @@ export const PluginInvocationSchema = z.object({ attempt: z.number().default(1), latency_ms: z.number().default(0), error: z.string().optional(), + delivery_id: z.string().optional(), + planned_at: z.string().optional(), created_at: z.string().default(""), }).loose(); @@ -234,6 +241,17 @@ export const PluginManifestSummarySchema = z.object({ name: z.string().default(""), url: z.string().optional(), }).loose().default({ name: "" }), + contributes: z.object({ + hooks: z.array(z.object({ + key: z.string().default(""), + name: z.string().default(""), + triggers: z.array(z.string()).default([]), + schedule: z.object({ + cron: z.string().default(""), + timezone: z.string().default(""), + }).loose().optional(), + }).loose()).default([]), + }).loose().optional(), }).loose(); export const PluginPreviewSchema = z.object({ diff --git a/packages/core/types/plugin.ts b/packages/core/types/plugin.ts index 134cfb2de..528d4ae33 100644 --- a/packages/core/types/plugin.ts +++ b/packages/core/types/plugin.ts @@ -29,7 +29,14 @@ export interface PluginSurface { platforms?: string[]; } -export type PluginHookTrigger = "ui" | "manual" | "agent" | "event"; +export type PluginHookTrigger = "ui" | "manual" | "agent" | "event" | "schedule"; + +export interface PluginHookSchedule { + cron: string; + timezone: string; + /** Display-only projection. Runtime correctness never depends on this value. */ + next_run_at?: string; +} export interface PluginHook { key: string; @@ -37,6 +44,7 @@ export interface PluginHook { description: string; triggers: (PluginHookTrigger | string)[]; events?: string[]; + schedule?: PluginHookSchedule; transport: string; } @@ -82,6 +90,14 @@ export interface PluginManifestSummary { description?: string; version: string; author: { name: string; url?: string }; + contributes?: { + hooks?: Array<{ + key: string; + name: string; + triggers: (PluginHookTrigger | string)[]; + schedule?: PluginHookSchedule; + }>; + }; } /** @@ -186,6 +202,8 @@ export interface PluginInvocation { attempt: number; latency_ms: number; error?: string; + delivery_id?: string; + planned_at?: string; created_at: string; } diff --git a/packages/views/locales/en/settings.json b/packages/views/locales/en/settings.json index 6e4f11dee..25164ef73 100644 --- a/packages/views/locales/en/settings.json +++ b/packages/views/locales/en/settings.json @@ -368,6 +368,20 @@ "upgrade_from": " · upgrading from v{{version}}", "new_scope": "New" }, + "schedule": { + "title": "Automatic schedules", + "consent_title": "This Plugin will run hooks automatically", + "consent_description": "Installing or upgrading enables these schedules. If several times are missed, only the latest due run is delivered.", + "next_run": "Next: {{time}}", + "activity_title": "Recent deliveries", + "activity_last": "{{status}} · planned {{planned}} · sent {{actual}} · attempt {{attempt}}", + "activity_empty": "Not delivered yet", + "frequency_minutes_one": "Every minute", + "frequency_minutes_other": "Every {{count}} minutes", + "frequency_hourly": "Every hour at minute {{minute}}", + "frequency_daily": "Every day at {{hour}}:{{minute}}", + "frequency_custom": "Custom recurring schedule" + }, "config": { "title": "Configuration", "save": "Save", diff --git a/packages/views/locales/ja/settings.json b/packages/views/locales/ja/settings.json index cefbb4405..b0820f475 100644 --- a/packages/views/locales/ja/settings.json +++ b/packages/views/locales/ja/settings.json @@ -368,6 +368,19 @@ "upgrade_from": " · v{{version}} からアップグレード", "new_scope": "新規" }, + "schedule": { + "title": "自動スケジュール", + "consent_title": "この Plugin はフックを自動実行します", + "consent_description": "インストールまたはアップグレードすると、次のスケジュールが有効になります。複数回分を逃した場合は、直近の実行だけを配信します。", + "next_run": "次回：{{time}}", + "activity_title": "最近の配信", + "activity_last": "{{status}} · 予定 {{planned}} · 送信 {{actual}} · 試行 {{attempt}}", + "activity_empty": "まだ配信されていません", + "frequency_minutes_other": "{{count}} 分ごと", + "frequency_hourly": "毎時 {{minute}} 分", + "frequency_daily": "毎日 {{hour}}:{{minute}}", + "frequency_custom": "カスタム繰り返しスケジュール" + }, "config": { "title": "設定", "save": "保存", diff --git a/packages/views/locales/ko/settings.json b/packages/views/locales/ko/settings.json index 7b7afb866..06f9c3b60 100644 --- a/packages/views/locales/ko/settings.json +++ b/packages/views/locales/ko/settings.json @@ -368,6 +368,19 @@ "upgrade_from": " · v{{version}}에서 업그레이드", "new_scope": "신규" }, + "schedule": { + "title": "자동 스케줄", + "consent_title": "이 Plugin은 훅을 자동 실행합니다", + "consent_description": "설치하거나 업그레이드하면 다음 스케줄이 활성화됩니다. 여러 실행 시점을 놓친 경우 가장 최근 실행만 전달합니다.", + "next_run": "다음 실행: {{time}}", + "activity_title": "최근 전달", + "activity_last": "{{status}} · 예정 {{planned}} · 전송 {{actual}} · 시도 {{attempt}}", + "activity_empty": "아직 전달되지 않음", + "frequency_minutes_other": "{{count}}분마다", + "frequency_hourly": "매시 {{minute}}분", + "frequency_daily": "매일 {{hour}}:{{minute}}", + "frequency_custom": "사용자 지정 반복 스케줄" + }, "config": { "title": "설정", "save": "저장", diff --git a/packages/views/locales/zh-Hans/settings.json b/packages/views/locales/zh-Hans/settings.json index 0d6757cd4..7a2418dbf 100644 --- a/packages/views/locales/zh-Hans/settings.json +++ b/packages/views/locales/zh-Hans/settings.json @@ -368,6 +368,19 @@ "upgrade_from": " · 从 v{{version}} 升级", "new_scope": "新增" }, + "schedule": { + "title": "自动运行计划", + "consent_title": "该 Plugin 将自动运行 hook", + "consent_description": "安装或升级后会启用以下计划。错过多个时间点时，只补跑最近一次。", + "next_run": "下次：{{time}}", + "activity_title": "最近投递", + "activity_last": "{{status}} · 计划 {{planned}} · 发出 {{actual}} · 第 {{attempt}} 次尝试", + "activity_empty": "尚未投递", + "frequency_minutes_other": "每 {{count}} 分钟", + "frequency_hourly": "每小时第 {{minute}} 分钟", + "frequency_daily": "每天 {{hour}}:{{minute}}", + "frequency_custom": "自定义周期" + }, "config": { "title": "配置", "save": "保存", diff --git a/packages/views/plugins/index.ts b/packages/views/plugins/index.ts index 61ac09757..7dad0ed26 100644 --- a/packages/views/plugins/index.ts +++ b/packages/views/plugins/index.ts @@ -17,5 +17,5 @@ export { usePluginModalSurfaces, } from "./plugin-modal-surface"; export type { PluginModalTarget } from "./plugin-modal-surface"; -export { PluginHookActivity, summarizeInvocations } from "./plugin-hook-activity"; +export { PluginHookActivity, PluginScheduleActivity, summarizeInvocations } from "./plugin-hook-activity"; export { PluginMCPApproval, mcpHooks, initialSelection } from "./plugin-mcp-approval"; diff --git a/packages/views/plugins/plugin-hook-activity.tsx b/packages/views/plugins/plugin-hook-activity.tsx index ad3f4635a..014b9b92a 100644 --- a/packages/views/plugins/plugin-hook-activity.tsx +++ b/packages/views/plugins/plugin-hook-activity.tsx @@ -3,9 +3,9 @@ import { useQuery } from "@tanstack/react-query"; import { AlertTriangle } from "lucide-react"; import { pluginInvocationsOptions } from "@multica/core/plugins"; -import type { PluginInvocation } from "@multica/core/types"; +import type { PluginHook, PluginInvocation } from "@multica/core/types"; import { cn } from "@multica/ui/lib/utils"; -import { useT } from "../i18n"; +import { useLocale, useT } from "../i18n"; /** * Why a plugin's hook is failing, where the admin who installed it will look. @@ -77,3 +77,60 @@ export function PluginHookActivity({ wsId, installationId }: { wsId: string; ins ); } + +export function PluginScheduleActivity({ + wsId, + installationId, + hooks, +}: { + wsId: string; + installationId: string; + hooks: readonly PluginHook[]; +}) { + const { t } = useT("settings"); + const locale = useLocale(); + const { data } = useQuery(pluginInvocationsOptions(wsId, installationId)); + const scheduled = hooks.filter((hook) => hook.schedule !== undefined); + if (scheduled.length === 0) return null; + + return ( +
+
{t(($) => $.plugins.schedule.activity_title)}
+
    + {scheduled.map((hook) => { + const last = (data ?? []).find((invocation) => ( + invocation.trigger === "schedule" && invocation.hook_key === hook.key + )); + return ( +
  • + {hook.name} + {last ? ( + + {" · "} + {t(($) => $.plugins.schedule.activity_last, { + status: last.status, + planned: formatInvocationTime(last.planned_at, locale), + actual: formatInvocationTime(last.created_at, locale), + attempt: last.attempt, + })} + + ) : ( + {" · "}{t(($) => $.plugins.schedule.activity_empty)} + )} +
    • + ); + })} +
+
+ ); +} + +function formatInvocationTime(value: string | undefined, locale: string): string { + if (!value) return "—"; + const timestamp = Date.parse(value); + if (Number.isNaN(timestamp)) return "—"; + return new Intl.DateTimeFormat(locale, { + dateStyle: "medium", + timeStyle: "short", + }).format(new Date(timestamp)); +} diff --git a/packages/views/settings/components/plugins-tab.tsx b/packages/views/settings/components/plugins-tab.tsx index 67cbb8817..f28b3557c 100644 --- a/packages/views/settings/components/plugins-tab.tsx +++ b/packages/views/settings/components/plugins-tab.tsx @@ -2,7 +2,7 @@ import { useMemo, useRef, useState } from "react"; import { useQuery } from "@tanstack/react-query"; -import { AlertCircle, Loader2, Trash2, Upload } from "lucide-react"; +import { AlertCircle, CalendarClock, Loader2, Trash2, Upload } from "lucide-react"; import { toast } from "sonner"; import { useCurrentMember } from "@multica/core/permissions"; import { @@ -37,8 +37,8 @@ import { import { Skeleton } from "@multica/ui/components/ui/skeleton"; import { Textarea } from "@multica/ui/components/ui/textarea"; import { Switch } from "@multica/ui/components/ui/switch"; -import { mcpHooks, PluginHookActivity, PluginMCPApproval } from "../../plugins"; -import { useT } from "../../i18n"; +import { mcpHooks, PluginHookActivity, PluginMCPApproval, PluginScheduleActivity } from "../../plugins"; +import { useLocale, useT } from "../../i18n"; import { SettingsCard, SettingsSection, SettingsTab } from "./settings-layout"; /** @@ -65,8 +65,67 @@ function ScopeList({ scopes, highlighted }: { scopes: string[]; highlighted?: st ); } +type ScheduledHook = { + key: string; + name: string; + schedule?: { cron: string; timezone: string; next_run_at?: string }; +}; + +function ScheduleList({ hooks, showNextRun = false }: { hooks: ScheduledHook[]; showNextRun?: boolean }) { + const { t } = useT("settings"); + const locale = useLocale(); + return ( +
    + {hooks.map((hook) => ( +
  • + {hook.name} + {scheduleFrequency(hook.schedule?.cron ?? "", t)} + {hook.schedule?.cron ?? ""} + {hook.schedule?.timezone ?? ""} + {showNextRun && hook.schedule?.next_run_at ? ( + + {t(($) => $.plugins.schedule.next_run, { + time: formatScheduleTime(hook.schedule?.next_run_at, locale), + })} + + ) : null} +
    • + ))} +
+ ); +} + +function formatScheduleTime(value: string | undefined, locale: string): string { + if (!value) return "—"; + const timestamp = Date.parse(value); + if (Number.isNaN(timestamp)) return "—"; + return new Intl.DateTimeFormat(locale, { + dateStyle: "medium", + timeStyle: "short", + }).format(new Date(timestamp)); +} + type Translate = ReturnType>["t"]; +function scheduleFrequency(cron: string, t: Translate): string { + const fields = cron.trim().split(/\s+/); + if (fields.length !== 5) return t(($) => $.plugins.schedule.frequency_custom); + const everyMinutes = /^\*\/(\d+)$/.exec(fields[0] ?? ""); + if (everyMinutes && fields.slice(1).every((field) => field === "*")) { + return t(($) => $.plugins.schedule.frequency_minutes, { count: Number(everyMinutes[1]) }); + } + if (/^\d+$/.test(fields[0] ?? "") && fields.slice(1).every((field) => field === "*")) { + return t(($) => $.plugins.schedule.frequency_hourly, { minute: fields[0] }); + } + if (/^\d+$/.test(fields[0] ?? "") && /^\d+$/.test(fields[1] ?? "") && fields.slice(2).every((field) => field === "*")) { + return t(($) => $.plugins.schedule.frequency_daily, { + hour: String(fields[1]).padStart(2, "0"), + minute: String(fields[0]).padStart(2, "0"), + }); + } + return t(($) => $.plugins.schedule.frequency_custom); +} + function scopeDescription(scope: string, t: Translate): string { if (scope.startsWith("net:")) { return t(($) => $.plugins.scopes.net, { domain: scope.slice("net:".length) }); @@ -270,6 +329,8 @@ function PublishAndInstall({ wsId, canManage }: { wsId: string; canManage: boole const [preview, setPreview] = useState(null); const packages = useMemo(() => data?.packages ?? [], [data]); + const scheduledHooks = (preview?.manifest.contributes?.hooks ?? []) + .filter((hook) => hook.schedule !== undefined); const reportError = (error: unknown) => { toast.error(error instanceof Error ? error.message : t(($) => $.plugins.action_failed)); @@ -386,6 +447,17 @@ function PublishAndInstall({ wsId, canManage }: { wsId: string; canManage: boole + {scheduledHooks.length > 0 ? ( + + + {t(($) => $.plugins.schedule.consent_title)} + + {t(($) => $.plugins.schedule.consent_description)} + + + + ) : null} +
+ {scheduledHooks.length > 0 ? ( +
+
+ + {t(($) => $.plugins.schedule.title)} +
+ + +
+ ) : null} + {mcpContributions.length > 0 ? (
{t(($) => $.plugins.mcp.badge)}
diff --git a/server/cmd/migrate/main.go b/server/cmd/migrate/main.go index f9adc3c4a..61f07f495 100644 --- a/server/cmd/migrate/main.go +++ b/server/cmd/migrate/main.go @@ -252,6 +252,8 @@ var concurrentIndexCleanups = map[string]string{ "396_plugin_package_file_path_index": "idx_plugin_package_file_path", "397_plugin_installation_package_version_index": "idx_plugin_installation_package_version", "398_issue_workspace_status_position_index": "idx_issue_workspace_status_position", + "400_plugin_hook_schedule_installation_key_index": "idx_plugin_hook_schedule_installation_key", + "401_plugin_hook_schedule_enabled_index": "idx_plugin_hook_schedule_enabled", } // concurrentDownIndexCleanups covers every migration whose down direction diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index e23e6c213..f3e67d599 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -666,6 +666,11 @@ func main() { if err := schedulerMgr.Register(scheduler.AutopilotScheduleDispatchJob(pool, queries, autopilotSvc)); err != nil { slog.Warn("scheduler: failed to register autopilot_schedule_dispatch job", "error", err) } + // Manifest-declared Plugin schedules share the same durable lease and retry + // machinery. The job is inert while plugins_v1 is disabled. + if err := schedulerMgr.Register(scheduler.PluginHookScheduleDispatchJob(queries, h.PluginService)); err != nil { + slog.Warn("scheduler: failed to register plugin_hook_schedule_dispatch job", "error", err) + } go func() { _ = schedulerMgr.Run(sweepCtx) }() diff --git a/server/cmd/server/plugin_hook_schedule_job_test.go b/server/cmd/server/plugin_hook_schedule_job_test.go new file mode 100644 index 000000000..6fd5b3fed --- /dev/null +++ b/server/cmd/server/plugin_hook_schedule_job_test.go @@ -0,0 +1,317 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/multica-ai/multica/server/internal/featureflags" + "github.com/multica-ai/multica/server/internal/scheduler" + "github.com/multica-ai/multica/server/internal/service" + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" + "github.com/multica-ai/multica/server/pkg/featureflag" + "github.com/multica-ai/multica/server/pkg/plugincontract" +) + +type receivedScheduledHook struct { + InvocationID string `json:"invocation_id"` + DeliveryID string `json:"delivery_id"` + Attempt int `json:"attempt"` + Trigger string `json:"trigger"` + Callback string `json:"callback_token"` + Actor struct { + Type string `json:"type"` + ID string `json:"id"` + } `json:"actor"` + Schedule struct { + PlannedAt time.Time `json:"planned_at"` + } `json:"schedule"` +} + +// TestPluginHookScheduleTwoReplicasRetryWithStableDelivery is the end-to-end +// acceptance path for durable scheduled Hooks. Two managers race the same plan, +// the endpoint loses the connection after receiving attempt one, and the retry +// is reclaimed by the same DB execution row with the same delivery_id and a +// fresh invocation. +func TestPluginHookScheduleTwoReplicasRetryWithStableDelivery(t *testing.T) { + ctx := context.Background() + queries := db.New(testPool) + + var ( + mu sync.Mutex + received []receivedScheduledHook + verificationErr error + signingSecret string + callbacks = service.NewCallbackTokens() + ) + endpoint := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "read", http.StatusBadRequest) + return + } + mu.Lock() + secret := signingSecret + mu.Unlock() + if err := service.VerifyHookSignature( + secret, + r.Header.Get("X-Multica-Timestamp"), + raw, + r.Header.Get("X-Multica-Signature"), + time.Now(), + ); err != nil { + mu.Lock() + verificationErr = err + mu.Unlock() + http.Error(w, "signature", http.StatusUnauthorized) + return + } + var body receivedScheduledHook + if err := json.Unmarshal(raw, &body); err != nil { + http.Error(w, "json", http.StatusBadRequest) + return + } + grant, err := callbacks.Resolve(body.Callback) + if err != nil || grant.Trigger != plugincontract.TriggerSchedule || + grant.Actor.Type != "plugin" || grant.IssueID.Valid { + mu.Lock() + verificationErr = fmt.Errorf("callback grant mismatch: grant=%+v err=%v", grant, err) + mu.Unlock() + http.Error(w, "callback grant", http.StatusUnauthorized) + return + } + mu.Lock() + received = append(received, body) + attemptIndex := len(received) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if attemptIndex == 1 { + // Model the irreducible uncertain-result case: the Plugin received + // and may have committed the delivery, but the Host sees no response. + connection, _, hijackErr := w.(http.Hijacker).Hijack() + if hijackErr != nil { + http.Error(w, "hijack", http.StatusInternalServerError) + return + } + _ = connection.Close() + return + } + _, _ = w.Write([]byte(`{"accepted":true}`)) + })) + t.Cleanup(endpoint.Close) + + parsedEndpoint, err := url.Parse(endpoint.URL) + if err != nil { + t.Fatalf("parse endpoint: %v", err) + } + _, port, err := net.SplitHostPort(parsedEndpoint.Host) + if err != nil { + t.Fatalf("split endpoint: %v", err) + } + hookOrigin := "https://hooks.example.com:" + port + hookURL := hookOrigin + "/hooks/scheduled" + + client := endpoint.Client() + transport := client.Transport.(*http.Transport).Clone() + transport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, endpoint.Listener.Addr().String()) + } + client.Transport = transport + + provider := featureflag.NewStaticProvider() + provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: true}) + plugins := service.NewPluginService(queries, testPool) + plugins.FeatureFlags = featureflag.NewService(provider) + plugins.DeploymentKey = []byte("0123456789abcdef0123456789abcdef") + plugins.Callbacks = callbacks + plugins.CallbackBaseURL = "https://plugin-api.example.com/v1" + plugins.DevOrigins = []string{hookOrigin} + plugins.HookClient = client + + manifestRaw := []byte(fmt.Sprintf(`{ + "manifest_version":1, + "key":"com.example.schedule-test", + "name":"Schedule Test", + "description":"test", + "version":"1.0.0", + "author":{"name":"test"}, + "scopes":["net:hooks.example.com"], + "contributes":{"hooks":[{ + "key":"scheduled", + "name":"Scheduled", + "description":"Scheduled integration test hook.", + "triggers":["schedule"], + "schedule":{"cron":"*/5 * * * *","timezone":"UTC"}, + "transport":{"type":"http","url":%s}, + "timeout_ms":1000 + }]} + }`, strconv.Quote(hookURL))) + manifest, canonical, err := plugincontract.ParseManifest(manifestRaw) + if err != nil { + t.Fatalf("parse fixture manifest: %v", err) + } + grantedScopes, _ := json.Marshal(manifest.Scopes) + installation, err := queries.CreatePluginInstallation(ctx, db.CreatePluginInstallationParams{ + WorkspaceID: parseUUID(testWorkspaceID), + PluginKey: manifest.Key, + PackageVersionID: pgtype.UUID{Bytes: uuid.New(), Valid: true}, + Version: manifest.Version, + Manifest: canonical, + GrantedScopes: grantedScopes, + InstalledBy: parseUUID(testUserID), + }) + if err != nil { + t.Fatalf("create plugin installation: %v", err) + } + derivedSigningSecret, err := plugins.HookSigningSecret(installation.ID) + if err != nil { + t.Fatalf("derive signing secret: %v", err) + } + mu.Lock() + signingSecret = derivedSigningSecret + mu.Unlock() + scheduleRow, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{ + InstallationID: installation.ID, + WorkspaceID: installation.WorkspaceID, + HookKey: "scheduled", + CronExpression: "*/5 * * * *", + Timezone: "UTC", + Enabled: true, + }) + if err != nil { + t.Fatalf("create plugin schedule: %v", err) + } + if _, err := testPool.Exec(ctx, + `UPDATE plugin_hook_schedule SET activated_at = now() - interval '6 minutes' WHERE id = $1`, + scheduleRow.ID, + ); err != nil { + t.Fatalf("backdate schedule: %v", err) + } + scopeID := util.UUIDToString(scheduleRow.ID) + ":" + util.UUIDToString(scheduleRow.Generation) + t.Cleanup(func() { + cleanupCtx := context.Background() + _, _ = testPool.Exec(cleanupCtx, + `DELETE FROM sys_cron_executions WHERE job_name = $1 AND scope_id = $2`, + scheduler.JobNamePluginHookScheduleDispatch, scopeID, + ) + _ = queries.DeletePluginHookSchedulesByInstallation(cleanupCtx, installation.ID) + _ = queries.DeletePluginInvocationsByInstallation(cleanupCtx, installation.ID) + _ = queries.DeletePluginInstallation(cleanupCtx, installation.ID) + }) + + managerA := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "plugin-schedule-a"}) + managerB := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "plugin-schedule-b"}) + for _, manager := range []*scheduler.Manager{managerA, managerB} { + if err := manager.Register(scheduler.PluginHookScheduleDispatchJob(queries, plugins)); err != nil { + t.Fatalf("register plugin schedule job: %v", err) + } + } + runManagersTogether(t, ctx, managerA, managerB) + + var status string + var attempt int + if err := testPool.QueryRow(ctx, ` + SELECT status, attempt + FROM sys_cron_executions + WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3 + `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil { + t.Fatalf("load failed execution: %v", err) + } + if status != "FAILED" || attempt != 1 { + t.Fatalf("first execution = %s attempt %d, want FAILED attempt 1", status, attempt) + } + if _, err := testPool.Exec(ctx, ` + UPDATE sys_cron_executions + SET next_retry_at = now() - interval '1 second' + WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3 + `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID); err != nil { + t.Fatalf("make retry due: %v", err) + } + runManagersTogether(t, ctx, managerA, managerB) + + if err := testPool.QueryRow(ctx, ` + SELECT status, attempt + FROM sys_cron_executions + WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3 + `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil { + t.Fatalf("load successful retry: %v", err) + } + if status != "SUCCESS" || attempt != 2 { + t.Fatalf("retried execution = %s attempt %d, want SUCCESS attempt 2", status, attempt) + } + + mu.Lock() + requests := append([]receivedScheduledHook(nil), received...) + verifyErr := verificationErr + mu.Unlock() + if verifyErr != nil { + t.Fatalf("endpoint rejected HMAC: %v", verifyErr) + } + if len(requests) != 2 { + t.Fatalf("endpoint calls=%d, want one failed attempt and one retry", len(requests)) + } + if requests[0].DeliveryID == "" || requests[0].DeliveryID != requests[1].DeliveryID { + t.Fatalf("delivery ids changed across retry: %q != %q", requests[0].DeliveryID, requests[1].DeliveryID) + } + if requests[0].InvocationID == requests[1].InvocationID { + t.Fatal("invocation_id must be unique per HTTP attempt") + } + if requests[0].Attempt != 1 || requests[1].Attempt != 2 { + t.Fatalf("attempts=%d,%d, want 1,2", requests[0].Attempt, requests[1].Attempt) + } + if requests[1].Trigger != plugincontract.TriggerSchedule || + requests[1].Actor.Type != "plugin" || requests[1].Actor.ID != util.UUIDToString(installation.ID) { + t.Fatalf("scheduled actor/trigger mismatch: %+v", requests[1]) + } + if requests[0].Callback == "" || requests[1].Callback == "" || requests[0].Callback == requests[1].Callback { + t.Fatal("each attempt must receive a fresh callback token") + } + if _, err := plugins.Callbacks.Resolve(requests[1].Callback); err == nil { + t.Fatal("callback token must be revoked after the endpoint returns") + } + + var invocationCount int + var distinctDeliveries int + if err := testPool.QueryRow(ctx, ` + SELECT count(*), count(DISTINCT delivery_id) + FROM plugin_invocation + WHERE installation_id = $1 AND trigger = 'schedule' + `, installation.ID).Scan(&invocationCount, &distinctDeliveries); err != nil { + t.Fatalf("load invocation audit: %v", err) + } + if invocationCount != 2 || distinctDeliveries != 1 { + t.Fatalf("invocation audit has %d rows/%d deliveries, want 2/1", invocationCount, distinctDeliveries) + } +} + +func runManagersTogether(t *testing.T, ctx context.Context, managers ...*scheduler.Manager) { + t.Helper() + errs := make(chan error, len(managers)) + var wg sync.WaitGroup + for _, manager := range managers { + wg.Add(1) + go func() { + defer wg.Done() + errs <- manager.RunOnce(ctx) + }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("scheduler tick: %v", err) + } + } +} diff --git a/server/cmd/server/plugin_schedule_pulse_example_test.go b/server/cmd/server/plugin_schedule_pulse_example_test.go new file mode 100644 index 000000000..a50d00092 --- /dev/null +++ b/server/cmd/server/plugin_schedule_pulse_example_test.go @@ -0,0 +1,380 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/multica-ai/multica/server/internal/featureflags" + "github.com/multica-ai/multica/server/internal/scheduler" + "github.com/multica-ai/multica/server/internal/service" + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" + "github.com/multica-ai/multica/server/pkg/featureflag" + "github.com/multica-ai/multica/server/pkg/plugincontract" +) + +type schedulePulseState struct { + DeliveryID string `json:"delivery_id"` + Count int `json:"count"` + LastPlannedAt string `json:"last_planned_at"` + LastOccurredAt string `json:"last_occurred_at"` + LastAttempt int `json:"last_attempt"` +} + +type schedulePulseHookBody struct { + DeliveryID string `json:"delivery_id"` + InvocationID string `json:"invocation_id"` + Attempt int `json:"attempt"` + Trigger string `json:"trigger"` + HookKey string `json:"hook_key"` + OccurredAt string `json:"occurred_at"` + CallbackURL string `json:"callback_url"` + Callback string `json:"callback_token"` + Schedule struct { + PlannedAt string `json:"planned_at"` + } `json:"schedule"` +} + +func schedulePulseExampleDir(t *testing.T) string { + t.Helper() + _, thisFile, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot locate this test file") + } + dir := filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "examples", "plugins", "schedule-pulse") + if _, err := os.Stat(filepath.Join(dir, plugincontract.ManifestFilename)); err != nil { + t.Fatalf("the schedule-pulse example is missing: %v", err) + } + return dir +} + +// TestSchedulePulseExampleManifestDrivesDurableRetry is the author-facing +// proof for MUL-6582: the shipped example plugin, not a fixture, is installed +// and woken by two scheduler replicas. The first HTTP attempt writes workspace +// storage and then drops the connection; the retry keeps the same delivery_id +// and does not increment the pulse count. +func TestSchedulePulseExampleManifestDrivesDurableRetry(t *testing.T) { + ctx := context.Background() + queries := db.New(testPool) + + var ( + mu sync.Mutex + hookCalls int + signingSecret string + callbacks = service.NewCallbackTokens() + ) + + plugins := service.NewPluginService(queries, testPool) + provider := featureflag.NewStaticProvider() + provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: true}) + plugins.FeatureFlags = featureflag.NewService(provider) + plugins.DeploymentKey = []byte("0123456789abcdef0123456789abcdef") + plugins.Callbacks = callbacks + + action := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + grant, err := callbacks.Resolve(token) + if err != nil || grant.Trigger != plugincontract.TriggerSchedule || grant.Actor.Type != "plugin" { + http.Error(w, "callback grant", http.StatusUnauthorized) + return + } + scopeID, err := service.ResolveStorageScope(service.PluginStorageWorkspace, grant.WorkspaceID, pgtype.UUID{}) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + key := strings.TrimPrefix(r.URL.Path, "/storage/workspace/") + switch { + case r.Method == http.MethodGet: + value, getErr := plugins.GetStorageValue(r.Context(), grant.InstallationID, service.PluginStorageWorkspace, scopeID, key) + if getErr != nil { + var pluginErr *service.PluginError + if errors.As(getErr, &pluginErr) && pluginErr.Kind == service.PluginErrorNotFound { + http.Error(w, pluginErr.Message, http.StatusNotFound) + return + } + http.Error(w, getErr.Error(), http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"value": value}) + case r.Method == http.MethodPut: + var payload struct { + Value string `json:"value"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, "json", http.StatusBadRequest) + return + } + if err := plugins.SetStorageValue(r.Context(), grant.InstallationID, service.PluginStorageWorkspace, scopeID, key, payload.Value); err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusNoContent) + default: + http.Error(w, "method", http.StatusMethodNotAllowed) + } + })) + t.Cleanup(action.Close) + plugins.CallbackBaseURL = action.URL + + endpoint := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/hooks/pulse" { + http.NotFound(w, r) + return + } + raw, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "read", http.StatusBadRequest) + return + } + mu.Lock() + secret := signingSecret + mu.Unlock() + if err := service.VerifyHookSignature( + secret, + r.Header.Get("X-Multica-Timestamp"), + raw, + r.Header.Get("X-Multica-Signature"), + time.Now(), + ); err != nil { + http.Error(w, "signature", http.StatusUnauthorized) + return + } + var body schedulePulseHookBody + if err := json.Unmarshal(raw, &body); err != nil { + http.Error(w, "json", http.StatusBadRequest) + return + } + if body.HookKey != "pulse" || body.Trigger != plugincontract.TriggerSchedule || body.DeliveryID == "" { + http.Error(w, "not a scheduled pulse", http.StatusBadRequest) + return + } + req, err := http.NewRequest(http.MethodGet, body.CallbackURL+"/storage/workspace/last_pulse", nil) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + req.Header.Set("Authorization", "Bearer "+body.Callback) + stored, err := http.DefaultClient.Do(req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + storedBody, _ := io.ReadAll(stored.Body) + _ = stored.Body.Close() + previous := schedulePulseState{} + if stored.StatusCode == http.StatusOK { + var wrapped struct { + Value string `json:"value"` + } + if err := json.Unmarshal(storedBody, &wrapped); err == nil { + _ = json.Unmarshal([]byte(wrapped.Value), &previous) + } + } else if stored.StatusCode != http.StatusNotFound { + http.Error(w, "storage get", http.StatusBadGateway) + return + } + + mu.Lock() + hookCalls++ + attemptIndex := hookCalls + mu.Unlock() + + w.Header().Set("Content-Type", "application/json") + if previous.DeliveryID == body.DeliveryID { + _, _ = w.Write([]byte(fmt.Sprintf(`{"duplicate":true,"delivery_id":%q,"count":%d}`, body.DeliveryID, previous.Count))) + return + } + pulse := schedulePulseState{ + DeliveryID: body.DeliveryID, + Count: previous.Count + 1, + LastPlannedAt: body.Schedule.PlannedAt, + LastOccurredAt: body.OccurredAt, + LastAttempt: body.Attempt, + } + encoded, _ := json.Marshal(pulse) + put, err := http.NewRequest(http.MethodPut, body.CallbackURL+"/storage/workspace/last_pulse", strings.NewReader(fmt.Sprintf(`{"value":%s}`, strconvQuote(string(encoded))))) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + put.Header.Set("Authorization", "Bearer "+body.Callback) + put.Header.Set("Content-Type", "application/json") + putResp, err := http.DefaultClient.Do(put) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + _ = putResp.Body.Close() + if putResp.StatusCode != http.StatusNoContent { + http.Error(w, "storage put", http.StatusBadGateway) + return + } + if attemptIndex == 1 { + connection, _, hijackErr := w.(http.Hijacker).Hijack() + if hijackErr != nil { + http.Error(w, "hijack", http.StatusInternalServerError) + return + } + _ = connection.Close() + return + } + _, _ = w.Write([]byte(fmt.Sprintf(`{"accepted":true,"delivery_id":%q,"count":%d}`, pulse.DeliveryID, pulse.Count))) + })) + t.Cleanup(endpoint.Close) + + parsedEndpoint, err := url.Parse(endpoint.URL) + if err != nil { + t.Fatalf("parse endpoint: %v", err) + } + _, port, err := net.SplitHostPort(parsedEndpoint.Host) + if err != nil { + t.Fatalf("split endpoint: %v", err) + } + hookOrigin := "https://pulse.example.com:" + port + client := endpoint.Client() + transport := client.Transport.(*http.Transport).Clone() + transport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, endpoint.Listener.Addr().String()) + } + client.Transport = transport + plugins.DevOrigins = []string{hookOrigin} + plugins.HookClient = client + + raw, err := os.ReadFile(filepath.Join(schedulePulseExampleDir(t), plugincontract.ManifestFilename)) + if err != nil { + t.Fatalf("read example manifest: %v", err) + } + rewritten := strings.ReplaceAll(string(raw), "https://pulse.example.com", hookOrigin) + manifest, canonical, err := plugincontract.ParseManifest([]byte(rewritten)) + if err != nil { + t.Fatalf("example manifest does not parse: %v", err) + } + if err := manifest.CheckCapabilities(plugincontract.HostCapabilities()); err != nil { + t.Fatalf("example cannot install on this host: %v", err) + } + grantedScopes, _ := json.Marshal(manifest.Scopes) + installation, err := queries.CreatePluginInstallation(ctx, db.CreatePluginInstallationParams{ + WorkspaceID: parseUUID(testWorkspaceID), + PluginKey: manifest.Key, + PackageVersionID: pgtype.UUID{Bytes: uuid.New(), Valid: true}, + Version: manifest.Version, + Manifest: canonical, + GrantedScopes: grantedScopes, + InstalledBy: parseUUID(testUserID), + }) + if err != nil { + t.Fatalf("create plugin installation: %v", err) + } + derivedSigningSecret, err := plugins.HookSigningSecret(installation.ID) + if err != nil { + t.Fatalf("derive signing secret: %v", err) + } + mu.Lock() + signingSecret = derivedSigningSecret + mu.Unlock() + scheduleRow, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{ + InstallationID: installation.ID, + WorkspaceID: installation.WorkspaceID, + HookKey: "pulse", + CronExpression: "*/5 * * * *", + Timezone: "UTC", + Enabled: true, + }) + if err != nil { + t.Fatalf("create plugin schedule: %v", err) + } + if _, err := testPool.Exec(ctx, + `UPDATE plugin_hook_schedule SET activated_at = now() - interval '6 minutes' WHERE id = $1`, + scheduleRow.ID, + ); err != nil { + t.Fatalf("backdate schedule: %v", err) + } + scopeID := util.UUIDToString(scheduleRow.ID) + ":" + util.UUIDToString(scheduleRow.Generation) + t.Cleanup(func() { + cleanupCtx := context.Background() + _, _ = testPool.Exec(cleanupCtx, + `DELETE FROM sys_cron_executions WHERE job_name = $1 AND scope_id = $2`, + scheduler.JobNamePluginHookScheduleDispatch, scopeID, + ) + _ = queries.DeletePluginHookSchedulesByInstallation(cleanupCtx, installation.ID) + _ = queries.DeletePluginStorageByInstallation(cleanupCtx, installation.ID) + _ = queries.DeletePluginInvocationsByInstallation(cleanupCtx, installation.ID) + _ = queries.DeletePluginInstallation(cleanupCtx, installation.ID) + }) + + managerA := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "schedule-pulse-a"}) + managerB := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "schedule-pulse-b"}) + for _, manager := range []*scheduler.Manager{managerA, managerB} { + if err := manager.Register(scheduler.PluginHookScheduleDispatchJob(queries, plugins)); err != nil { + t.Fatalf("register plugin schedule job: %v", err) + } + } + runManagersTogether(t, ctx, managerA, managerB) + if _, err := testPool.Exec(ctx, ` + UPDATE sys_cron_executions + SET next_retry_at = now() - interval '1 second' + WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3 + `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID); err != nil { + t.Fatalf("make retry due: %v", err) + } + runManagersTogether(t, ctx, managerA, managerB) + + var status string + var attempt int + if err := testPool.QueryRow(ctx, ` + SELECT status, attempt + FROM sys_cron_executions + WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3 + `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil { + t.Fatalf("load execution: %v", err) + } + if status != "SUCCESS" || attempt != 2 { + t.Fatalf("execution = %s attempt %d, want SUCCESS attempt 2", status, attempt) + } + + mu.Lock() + calls := hookCalls + mu.Unlock() + if calls != 2 { + t.Fatalf("hook calls=%d, want 2", calls) + } + + scopeIDUUID, err := service.ResolveStorageScope(service.PluginStorageWorkspace, installation.WorkspaceID, pgtype.UUID{}) + if err != nil { + t.Fatalf("resolve storage scope: %v", err) + } + value, err := plugins.GetStorageValue(ctx, installation.ID, service.PluginStorageWorkspace, scopeIDUUID, "last_pulse") + if err != nil { + t.Fatalf("read pulse storage: %v", err) + } + var pulse schedulePulseState + if err := json.Unmarshal([]byte(value), &pulse); err != nil { + t.Fatalf("decode pulse storage: %v", err) + } + if pulse.Count != 1 || pulse.DeliveryID == "" { + t.Fatalf("pulse storage = %+v, want count 1 with a delivery_id", pulse) + } +} + +func strconvQuote(s string) string { + encoded, _ := json.Marshal(s) + return string(encoded) +} diff --git a/server/internal/handler/plugin.go b/server/internal/handler/plugin.go index 78a8e153f..b40418891 100644 --- a/server/internal/handler/plugin.go +++ b/server/internal/handler/plugin.go @@ -80,12 +80,19 @@ type pluginInstallationResponse struct { // pluginHookResponse omits input_schema: the settings page lists what a hook is // and who may call it, and the raw JSON Schema is noise there. type pluginHookResponse struct { - Key string `json:"key"` - Name string `json:"name"` - Description string `json:"description"` - Triggers []string `json:"triggers"` - Events []string `json:"events,omitempty"` - Transport string `json:"transport"` + Key string `json:"key"` + Name string `json:"name"` + Description string `json:"description"` + Triggers []string `json:"triggers"` + Events []string `json:"events,omitempty"` + Transport string `json:"transport"` + Schedule *pluginHookScheduleResponse `json:"schedule,omitempty"` +} + +type pluginHookScheduleResponse struct { + Cron string `json:"cron"` + Timezone string `json:"timezone"` + NextRunAt string `json:"next_run_at,omitempty"` } func (h *Handler) pluginInstallationPayload(ctx context.Context, installation db.PluginInstallation) (pluginInstallationResponse, error) { @@ -109,17 +116,35 @@ func (h *Handler) pluginInstallationPayload(ctx context.Context, installation db if granted == nil { granted = []string{} } + scheduleRows, err := h.Queries.ListPluginHookSchedulesByInstallation(ctx, installation.ID) + if err != nil { + return pluginInstallationResponse{}, err + } + schedules := make(map[string]db.PluginHookSchedule, len(scheduleRows)) + for _, schedule := range scheduleRows { + schedules[schedule.HookKey] = schedule + } hooks := make([]pluginHookResponse, 0, len(manifest.Contributes.Hooks)) for _, hook := range manifest.Contributes.Hooks { - hooks = append(hooks, pluginHookResponse{ + response := pluginHookResponse{ Key: hook.Key, Name: hook.Name, Description: hook.Description, Triggers: hook.Triggers, Events: hook.Events, Transport: hook.Transport.Type, - }) + } + if schedule, ok := schedules[hook.Key]; ok { + response.Schedule = &pluginHookScheduleResponse{ + Cron: schedule.CronExpression, + Timezone: schedule.Timezone, + } + if schedule.NextRunAt.Valid { + response.Schedule.NextRunAt = schedule.NextRunAt.Time.UTC().Format(timeFormatRFC3339) + } + } + hooks = append(hooks, response) } surfaces := manifest.Contributes.Surfaces if surfaces == nil { diff --git a/server/internal/handler/plugin_hook.go b/server/internal/handler/plugin_hook.go index 0db1fbc19..9869acbdb 100644 --- a/server/internal/handler/plugin_hook.go +++ b/server/internal/handler/plugin_hook.go @@ -105,15 +105,17 @@ func rawOrNil(raw json.RawMessage) any { // --- Installation-scoped hook administration --- type pluginInvocationResponse struct { - ID string `json:"id"` - HookKey string `json:"hook_key"` - Trigger string `json:"trigger"` - Status string `json:"status"` - EventType *string `json:"event_type,omitempty"` - Attempt int `json:"attempt"` - LatencyMs int `json:"latency_ms"` - Error *string `json:"error,omitempty"` - CreatedAt string `json:"created_at"` + ID string `json:"id"` + HookKey string `json:"hook_key"` + Trigger string `json:"trigger"` + Status string `json:"status"` + EventType *string `json:"event_type,omitempty"` + Attempt int `json:"attempt"` + LatencyMs int `json:"latency_ms"` + Error *string `json:"error,omitempty"` + DeliveryID *string `json:"delivery_id,omitempty"` + PlannedAt *string `json:"planned_at,omitempty"` + CreatedAt string `json:"created_at"` } // ListPluginInvocations — GET /api/workspaces/{id}/plugins/{installationId}/invocations @@ -152,6 +154,14 @@ func (h *Handler) ListPluginInvocations(w http.ResponseWriter, r *http.Request) value := row.Error.String item.Error = &value } + if row.DeliveryID.Valid { + value := row.DeliveryID.String + item.DeliveryID = &value + } + if row.PlannedAt.Valid { + value := row.PlannedAt.Time.UTC().Format("2006-01-02T15:04:05Z07:00") + item.PlannedAt = &value + } items = append(items, item) } writeJSON(w, http.StatusOK, map[string]any{"invocations": items}) diff --git a/server/internal/handler/plugin_test.go b/server/internal/handler/plugin_test.go index 7b206becb..5216857be 100644 --- a/server/internal/handler/plugin_test.go +++ b/server/internal/handler/plugin_test.go @@ -59,6 +59,25 @@ const hookOnlyTestManifest = `{ } }` +const scheduleHookTestManifest = `{ + "manifest_version": 1, + "key": "com.example.scheduled", + "name": "Scheduled", + "version": "1.0.0", + "author": { "name": "example" }, + "scopes": ["net:example.com"], + "contributes": { + "hooks": [{ + "key": "heartbeat", + "name": "Heartbeat", + "description": "Send a periodic heartbeat.", + "triggers": ["schedule"], + "schedule": { "cron": "*/5 * * * *", "timezone": "UTC" }, + "transport": { "type": "http", "url": "https://example.com/hooks/heartbeat" } + }] + } +}` + func pluginHandlerRequest(method, path string, body []byte, params map[string]string) *http.Request { request := httptest.NewRequest(method, path, bytes.NewReader(body)) request.Header.Set("X-User-ID", testUserID) @@ -137,7 +156,7 @@ func withLocalPluginSourceIn(t *testing.T, root string, manifest string) string testHandler.PluginService.LocalDir = root testHandler.PluginService.Host = plugincontract.Capabilities{ SurfaceTypes: map[string]bool{plugincontract.SurfaceIssuePanel: true, plugincontract.SurfaceSidebarPanel: true, plugincontract.SurfaceModal: true}, - HookTriggers: map[string]bool{plugincontract.TriggerUI: true, plugincontract.TriggerManual: true, plugincontract.TriggerAgent: true, plugincontract.TriggerEvent: true}, + HookTriggers: map[string]bool{plugincontract.TriggerUI: true, plugincontract.TriggerManual: true, plugincontract.TriggerAgent: true, plugincontract.TriggerEvent: true, plugincontract.TriggerSchedule: true}, HookTransport: map[string]bool{plugincontract.TransportHTTP: true, plugincontract.TransportMCP: true}, ResourceTypes: map[string]bool{plugincontract.ResourceSkill: true}, } @@ -182,6 +201,8 @@ func cleanupPluginInstallations(t *testing.T) { t.Helper() remove := func() { ctx := context.Background() + testPool.Exec(ctx, `DELETE FROM plugin_invocation WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID) + testPool.Exec(ctx, `DELETE FROM plugin_hook_schedule WHERE workspace_id = $1`, testWorkspaceID) testPool.Exec(ctx, `DELETE FROM plugin_storage WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID) testPool.Exec(ctx, `DELETE FROM plugin_secret WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID) testPool.Exec(ctx, `DELETE FROM plugin_installation WHERE workspace_id = $1`, testWorkspaceID) @@ -193,6 +214,152 @@ func cleanupPluginInstallations(t *testing.T) { t.Cleanup(remove) } +func TestPluginScheduleLifecycleReconcilesAtomically(t *testing.T) { + withPluginsV1Flag(t, testHandler, true) + cleanupPluginInstallations(t) + root := t.TempDir() + versionID := withLocalPluginSourceIn(t, root, scheduleHookTestManifest) + + install, _ := json.Marshal(map[string]any{ + "version_id": versionID, + "granted_scopes": []string{"net:example.com"}, + }) + recorder := httptest.NewRecorder() + testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", install, map[string]string{"id": testWorkspaceID})) + if recorder.Code != http.StatusCreated { + t.Fatalf("install status=%d body=%s", recorder.Code, recorder.Body.String()) + } + var installed struct { + ID string `json:"id"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &installed); err != nil { + t.Fatalf("decode installation: %v", err) + } + + loadSchedule := func() (enabled bool, generation string, cron string) { + t.Helper() + if err := testPool.QueryRow(context.Background(), ` + SELECT enabled, generation::text, cron_expression + FROM plugin_hook_schedule + WHERE installation_id = $1 AND hook_key = 'heartbeat'`, installed.ID). + Scan(&enabled, &generation, &cron); err != nil { + t.Fatalf("load schedule: %v", err) + } + return enabled, generation, cron + } + + enabled, generation1, cron := loadSchedule() + if !enabled || cron != "*/5 * * * *" { + t.Fatalf("installed schedule enabled=%v cron=%q", enabled, cron) + } + params := map[string]string{"id": testWorkspaceID, "installationId": installed.ID} + + recorder = httptest.NewRecorder() + testHandler.DisablePlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins/disable", nil, params)) + if recorder.Code != http.StatusOK { + t.Fatalf("disable status=%d body=%s", recorder.Code, recorder.Body.String()) + } + enabled, generationDisabled, _ := loadSchedule() + if enabled || generationDisabled != generation1 { + t.Fatalf("disabled schedule enabled=%v generation=%q, want disabled generation %q", enabled, generationDisabled, generation1) + } + var hasNextRun bool + if err := testPool.QueryRow(context.Background(), ` + SELECT next_run_at IS NOT NULL FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID). + Scan(&hasNextRun); err != nil { + t.Fatalf("load disabled next run: %v", err) + } + if hasNextRun { + t.Fatal("disabled schedule must not advertise a next run") + } + + recorder = httptest.NewRecorder() + testHandler.EnablePlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins/enable", nil, params)) + if recorder.Code != http.StatusOK { + t.Fatalf("enable status=%d body=%s", recorder.Code, recorder.Body.String()) + } + enabled, generation2, _ := loadSchedule() + if !enabled || generation2 == generation1 { + t.Fatalf("reactivated schedule enabled=%v generation=%q, previous %q", enabled, generation2, generation1) + } + + // A code-only upgrade preserves the activation generation and any in-flight + // retry. Changing cron rotates it so the new definition cannot collide with + // an execution claimed under the previous definition. + unchanged := strings.Replace(scheduleHookTestManifest, `"version": "1.0.0"`, `"version": "2.0.0"`, 1) + writeLocalPluginManifest(t, root, unchanged) + upgradeVersionID := publishLocalPlugin(t, "hello") + upgrade, _ := json.Marshal(map[string]any{"version_id": upgradeVersionID, "granted_scopes": []string{"net:example.com"}}) + recorder = httptest.NewRecorder() + testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID})) + if recorder.Code != http.StatusCreated { + t.Fatalf("unchanged upgrade status=%d body=%s", recorder.Code, recorder.Body.String()) + } + _, generationUnchanged, _ := loadSchedule() + if generationUnchanged != generation2 { + t.Fatalf("code-only upgrade rotated generation: got %q want %q", generationUnchanged, generation2) + } + + changed := strings.Replace(unchanged, `"version": "2.0.0"`, `"version": "3.0.0"`, 1) + changed = strings.Replace(changed, `"cron": "*/5 * * * *"`, `"cron": "*/10 * * * *"`, 1) + writeLocalPluginManifest(t, root, changed) + changedVersionID := publishLocalPlugin(t, "hello") + upgrade, _ = json.Marshal(map[string]any{"version_id": changedVersionID, "granted_scopes": []string{"net:example.com"}}) + recorder = httptest.NewRecorder() + testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID})) + if recorder.Code != http.StatusCreated { + t.Fatalf("changed upgrade status=%d body=%s", recorder.Code, recorder.Body.String()) + } + _, generation3, cron := loadSchedule() + if generation3 == generation2 || cron != "*/10 * * * *" { + t.Fatalf("changed schedule generation=%q cron=%q, previous generation %q", generation3, cron, generation2) + } + + removed := strings.Replace(changed, `"version": "3.0.0"`, `"version": "4.0.0"`, 1) + removed = strings.Replace(removed, `"triggers": ["schedule"], + "schedule": { "cron": "*/10 * * * *", "timezone": "UTC" },`, `"triggers": ["manual"],`, 1) + writeLocalPluginManifest(t, root, removed) + removedVersionID := publishLocalPlugin(t, "hello") + upgrade, _ = json.Marshal(map[string]any{"version_id": removedVersionID, "granted_scopes": []string{"net:example.com"}}) + recorder = httptest.NewRecorder() + testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID})) + if recorder.Code != http.StatusCreated { + t.Fatalf("schedule-removing upgrade status=%d body=%s", recorder.Code, recorder.Body.String()) + } + var remaining int + if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID).Scan(&remaining); err != nil { + t.Fatalf("count removed schedules: %v", err) + } + if remaining != 0 { + t.Fatalf("schedule-removing upgrade left %d rows", remaining) + } + + // Restore the schedule so uninstall exercises its own cleanup path instead + // of succeeding only because the previous upgrade already removed the row. + restored := strings.Replace(changed, `"version": "3.0.0"`, `"version": "5.0.0"`, 1) + writeLocalPluginManifest(t, root, restored) + restoredVersionID := publishLocalPlugin(t, "hello") + upgrade, _ = json.Marshal(map[string]any{"version_id": restoredVersionID, "granted_scopes": []string{"net:example.com"}}) + recorder = httptest.NewRecorder() + testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID})) + if recorder.Code != http.StatusCreated { + t.Fatalf("schedule-restoring upgrade status=%d body=%s", recorder.Code, recorder.Body.String()) + } + loadSchedule() + + recorder = httptest.NewRecorder() + testHandler.UninstallPlugin(recorder, pluginHandlerRequest(http.MethodDelete, "/plugins", nil, params)) + if recorder.Code != http.StatusNoContent { + t.Fatalf("uninstall status=%d body=%s", recorder.Code, recorder.Body.String()) + } + if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID).Scan(&remaining); err != nil { + t.Fatalf("count schedules after uninstall: %v", err) + } + if remaining != 0 { + t.Fatalf("uninstall left %d schedule rows", remaining) + } +} + func TestPluginManagementRequiresPluginsV1(t *testing.T) { withPluginsV1Flag(t, testHandler, false) for name, handler := range map[string]http.HandlerFunc{ diff --git a/server/internal/handler/workspace_delete_manifest_test.go b/server/internal/handler/workspace_delete_manifest_test.go index a0e02060e..6858e3704 100644 --- a/server/internal/handler/workspace_delete_manifest_test.go +++ b/server/internal/handler/workspace_delete_manifest_test.go @@ -92,6 +92,7 @@ var workspaceDeletionManifest = map[string]workspaceDeleteAction{ "personal_access_token": workspaceDeleteKeep, "pinned_item": workspaceDelete, "plugin_installation": workspaceDelete, + "plugin_hook_schedule": workspaceDelete, "plugin_invocation": workspaceDelete, "plugin_storage": workspaceDelete, "plugin_secret": workspaceDelete, diff --git a/server/internal/scheduler/jobs_plugin_hook.go b/server/internal/scheduler/jobs_plugin_hook.go new file mode 100644 index 000000000..8ae307d07 --- /dev/null +++ b/server/internal/scheduler/jobs_plugin_hook.go @@ -0,0 +1,353 @@ +package scheduler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/multica-ai/multica/server/internal/featureflags" + "github.com/multica-ai/multica/server/internal/service" + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" + "github.com/multica-ai/multica/server/pkg/plugincontract" +) + +const ( + // JobNamePluginHookScheduleDispatch is persisted in sys_cron_executions and + // must remain stable across releases. + JobNamePluginHookScheduleDispatch = "plugin_hook_schedule_dispatch" + ScopeKindPluginHookSchedule = "plugin_hook_schedule" + pluginHookScheduleAttempts = 3 + pluginHookSchedulePageSize = 1024 +) + +type pluginHookScheduleConfig struct { + ID pgtype.UUID + InstallationID pgtype.UUID + HookKey string + CronExpression string + Timezone string + Generation pgtype.UUID + ActivatedAt time.Time +} + +type pluginHookScheduleCache struct { + mu sync.RWMutex + schedules map[string]pluginHookScheduleConfig + coalesced map[string]int +} + +func newPluginHookScheduleCache() *pluginHookScheduleCache { + return &pluginHookScheduleCache{ + schedules: make(map[string]pluginHookScheduleConfig), + coalesced: make(map[string]int), + } +} + +func (c *pluginHookScheduleCache) replace(schedules map[string]pluginHookScheduleConfig) { + c.mu.Lock() + c.schedules = schedules + // Preserve the count across retry ticks, but discard entries for schedules + // no longer returned by the database. setCoalesced further bounds this to + // one pending entry per active scope, including on a losing replica. + for key := range c.coalesced { + scopeID, _, _ := strings.Cut(key, "/") + if _, active := schedules[scopeID]; !active { + delete(c.coalesced, key) + } + } + c.mu.Unlock() +} + +func (c *pluginHookScheduleCache) get(scopeID string) (pluginHookScheduleConfig, bool) { + c.mu.RLock() + config, ok := c.schedules[scopeID] + c.mu.RUnlock() + return config, ok +} + +func (c *pluginHookScheduleCache) setCoalesced(scopeID string, plan time.Time, count int) { + c.mu.Lock() + for key := range c.coalesced { + if strings.HasPrefix(key, scopeID+"/") { + delete(c.coalesced, key) + } + } + c.coalesced[scopeID+"/"+plan.UTC().Format(time.RFC3339Nano)] = count + c.mu.Unlock() +} + +func (c *pluginHookScheduleCache) takeCoalesced(scopeID string, plan time.Time) int { + key := scopeID + "/" + plan.UTC().Format(time.RFC3339Nano) + c.mu.Lock() + count := c.coalesced[key] + delete(c.coalesced, key) + c.mu.Unlock() + return count +} + +// PluginHookScheduleDispatchJob drives manifest-declared Plugin schedules +// through the shared DB scheduler. Scope includes a generation UUID, so a +// changed/re-enabled schedule starts a fresh serial timeline while old in-flight +// delivery is allowed to finish. +func PluginHookScheduleDispatchJob(queries *db.Queries, plugins *service.PluginService) JobSpec { + cache := newPluginHookScheduleCache() + return JobSpec{ + Name: JobNamePluginHookScheduleDispatch, + CatchUpMode: CatchUpLatestOnly, + RunTimeout: 45 * time.Second, + StaleTimeout: 90 * time.Second, + HeartbeatInterval: 15 * time.Second, + AllowStaleReentry: true, + MaxAttempts: pluginHookScheduleAttempts, + RetryBackoff: []time.Duration{30 * time.Second, 2 * time.Minute}, + MaxPlansPerTick: 1, + Scopes: pluginHookScheduleScopes(queries, plugins, cache), + PlansForScope: pluginHookSchedulePlans(cache), + Handler: pluginHookScheduleHandler(queries, plugins, cache), + } +} + +func pluginHookScheduleScopes( + queries *db.Queries, + plugins *service.PluginService, + cache *pluginHookScheduleCache, +) ScopeProvider { + return func(ctx context.Context, _ time.Time) ([]Scope, error) { + if plugins == nil || !featureflags.PluginsV1Enabled(ctx, plugins.FeatureFlags) { + cache.replace(map[string]pluginHookScheduleConfig{}) + return nil, nil + } + rows, err := queries.ListEnabledPluginHookSchedules(ctx) + if err != nil { + return nil, fmt.Errorf("plugin schedule scope: list enabled schedules: %w", err) + } + configs := make(map[string]pluginHookScheduleConfig, len(rows)) + scopes := make([]Scope, 0, len(rows)) + for _, row := range rows { + scopeID := pluginHookScheduleScopeID(row.ID, row.Generation) + if scopeID == "" || !row.ActivatedAt.Valid { + continue + } + configs[scopeID] = pluginHookScheduleConfig{ + ID: row.ID, + InstallationID: row.InstallationID, + HookKey: row.HookKey, + CronExpression: row.CronExpression, + Timezone: row.Timezone, + Generation: row.Generation, + ActivatedAt: row.ActivatedAt.Time.UTC(), + } + scopes = append(scopes, Scope{Kind: ScopeKindPluginHookSchedule, ID: scopeID}) + } + cache.replace(configs) + return scopes, nil + } +} + +func pluginHookSchedulePlans(cache *pluginHookScheduleCache) func( + context.Context, Scope, time.Time, LatestPlanInfo, +) ([]time.Time, error) { + return func(_ context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) { + config, ok := cache.get(scope.ID) + if !ok { + return nil, nil + } + + // A generation is strictly serial. Do not plan a newer occurrence while + // its latest row is running, nor while that row still owns retry budget. + if latest.Found { + switch latest.Status { + case "RUNNING": + return nil, nil + case "FAILED": + if latest.Attempt < latest.MaxAttempts { + if latest.RetryEligible(now) { + return []time.Time{latest.PlanTime}, nil + } + return nil, nil + } + } + } + + after := config.ActivatedAt + if latest.Found { + after = latest.PlanTime + } + plan, count, err := latestPluginHookOccurrence( + config.CronExpression, config.Timezone, after, now, + ) + if err != nil { + return nil, fmt.Errorf("plugin schedule plans for %s: %w", scope.ID, err) + } + if count == 0 { + return nil, nil + } + cache.setCoalesced(scope.ID, plan, count-1) + return []time.Time{plan}, nil + } +} + +// latestPluginHookOccurrence enumerates in bounded pages so a long outage still +// collapses to the truly latest due occurrence instead of stopping at the first +// 1024 entries returned by service.NextOccurrencesUTC. +func latestPluginHookOccurrence(cronExpression, timezone string, after, until time.Time) (time.Time, int, error) { + var latest time.Time + count := 0 + cursor := after + for { + occurrences, err := service.NextOccurrencesUTC(cronExpression, timezone, cursor, until) + if err != nil { + return time.Time{}, 0, err + } + if len(occurrences) == 0 { + return latest, count, nil + } + count += len(occurrences) + latest = occurrences[len(occurrences)-1] + if len(occurrences) < pluginHookSchedulePageSize || !latest.Before(until) { + return latest, count, nil + } + cursor = latest + } +} + +func pluginHookScheduleHandler( + queries *db.Queries, + plugins *service.PluginService, + cache *pluginHookScheduleCache, +) Handler { + return func(ctx context.Context, in HandlerInput) (HandlerResult, error) { + scheduleID, generation, err := parsePluginHookScheduleScopeID(in.Scope.ID) + if err != nil { + return HandlerResult{}, fmt.Errorf("parse plugin hook schedule scope: %w", err) + } + if plugins == nil || !featureflags.PluginsV1Enabled(ctx, plugins.FeatureFlags) { + return pluginHookScheduleSkipped("feature_disabled"), nil + } + + row, err := queries.GetPluginHookSchedule(ctx, scheduleID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return pluginHookScheduleSkipped("schedule_not_found"), nil + } + return HandlerResult{}, fmt.Errorf("load plugin hook schedule: %w", err) + } + if !row.Enabled || row.Generation != generation { + return pluginHookScheduleSkipped("schedule_generation_changed"), nil + } + + installation, err := queries.GetPluginInstallation(ctx, row.InstallationID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return pluginHookScheduleSkipped("installation_not_found"), nil + } + return HandlerResult{}, fmt.Errorf("load plugin installation: %w", err) + } + if !installation.Enabled { + return pluginHookScheduleSkipped("installation_disabled"), nil + } + hook, err := service.FindHook(installation, row.HookKey) + if err != nil || hook.Schedule == nil || + !service.HookAllowsTrigger(hook, plugincontract.TriggerSchedule) || + hook.Transport.Type != plugincontract.TransportHTTP || + hook.Schedule.Cron != row.CronExpression || hook.Schedule.Timezone != row.Timezone { + return pluginHookScheduleSkipped("manifest_changed"), nil + } + + if plugins.HookBreakerOpen(ctx, installation.ID, hook.Key) { + advancePluginHookNextRun(ctx, queries, row, in.PlanTime) + result := pluginHookScheduleSkipped("circuit_open") + result.Result["coalesced_occurrences"] = cache.takeCoalesced(in.Scope.ID, in.PlanTime) + result.Result["dispatch_lag_ms"] = max(time.Since(in.PlanTime).Milliseconds(), 0) + return result, nil + } + + deliveryID := pluginHookScheduleDeliveryID(installation.ID, hook.Key, row.Generation, in.PlanTime) + _, err = plugins.InvokeHook(ctx, service.HookInvocation{ + Installation: installation, + Hook: hook, + Trigger: plugincontract.TriggerSchedule, + Actor: service.HookActor{Type: "plugin", ID: installation.ID}, + DeliveryID: deliveryID, + PlannedAt: pgtype.Timestamptz{Time: in.PlanTime.UTC(), Valid: true}, + }, in.Attempt) + if err != nil { + if in.Job != nil && in.Attempt >= in.Job.MaxAttempts { + advancePluginHookNextRun(ctx, queries, row, in.PlanTime) + cache.takeCoalesced(in.Scope.ID, in.PlanTime) + } + return HandlerResult{}, fmt.Errorf("deliver scheduled plugin hook: %w", err) + } + + advancePluginHookNextRun(ctx, queries, row, in.PlanTime) + return HandlerResult{RowsAffected: 1, Result: map[string]any{ + "delivery_id": deliveryID, + "coalesced_occurrences": cache.takeCoalesced(in.Scope.ID, in.PlanTime), + "dispatch_lag_ms": max(time.Since(in.PlanTime).Milliseconds(), 0), + }}, nil + } +} + +func pluginHookScheduleSkipped(reason string) HandlerResult { + return HandlerResult{RowsAffected: 0, Result: map[string]any{"skipped_reason": reason}} +} + +func advancePluginHookNextRun(ctx context.Context, queries *db.Queries, row db.PluginHookSchedule, plan time.Time) { + next, err := service.NextOccurrenceAfterUTC(row.CronExpression, row.Timezone, plan) + if err != nil { + return + } + _, _ = queries.UpdatePluginHookScheduleNextRun(ctx, db.UpdatePluginHookScheduleNextRunParams{ + ID: row.ID, + Generation: row.Generation, + NextRunAt: pgtype.Timestamptz{Time: next.UTC(), Valid: true}, + }) +} + +func pluginHookScheduleScopeID(id, generation pgtype.UUID) string { + idString := util.UUIDToString(id) + generationString := util.UUIDToString(generation) + if idString == "" || generationString == "" { + return "" + } + return idString + ":" + generationString +} + +func parsePluginHookScheduleScopeID(scopeID string) (pgtype.UUID, pgtype.UUID, error) { + parts := strings.Split(scopeID, ":") + if len(parts) != 2 { + return pgtype.UUID{}, pgtype.UUID{}, errors.New("expected schedule:generation scope") + } + id, err := util.ParseUUID(parts[0]) + if err != nil { + return pgtype.UUID{}, pgtype.UUID{}, err + } + generation, err := util.ParseUUID(parts[1]) + if err != nil { + return pgtype.UUID{}, pgtype.UUID{}, err + } + return id, generation, nil +} + +func pluginHookScheduleDeliveryID( + installationID pgtype.UUID, + hookKey string, + generation pgtype.UUID, + plan time.Time, +) string { + digest := sha256.Sum256([]byte(strings.Join([]string{ + util.UUIDToString(installationID), + hookKey, + util.UUIDToString(generation), + plan.UTC().Format(time.RFC3339Nano), + }, "\x00"))) + return "psd_" + hex.EncodeToString(digest[:]) +} diff --git a/server/internal/scheduler/jobs_plugin_hook_test.go b/server/internal/scheduler/jobs_plugin_hook_test.go new file mode 100644 index 000000000..a5dce357e --- /dev/null +++ b/server/internal/scheduler/jobs_plugin_hook_test.go @@ -0,0 +1,127 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/multica-ai/multica/server/internal/featureflags" + "github.com/multica-ai/multica/server/internal/service" + "github.com/multica-ai/multica/server/internal/util" + "github.com/multica-ai/multica/server/pkg/featureflag" +) + +func TestPluginHookSchedulePlansCollapseToLatestAndStaySerial(t *testing.T) { + cache := newPluginHookScheduleCache() + activated := time.Date(2026, 8, 23, 10, 0, 0, 0, time.UTC) + scope := Scope{Kind: ScopeKindPluginHookSchedule, ID: "schedule:generation"} + cache.replace(map[string]pluginHookScheduleConfig{ + scope.ID: { + CronExpression: "*/5 * * * *", + Timezone: "UTC", + ActivatedAt: activated, + }, + }) + plans := pluginHookSchedulePlans(cache) + now := activated.Add(16 * time.Minute) + + got, err := plans(context.Background(), scope, now, LatestPlanInfo{}) + if err != nil { + t.Fatalf("plan initial occurrence: %v", err) + } + if len(got) != 1 || !got[0].Equal(activated.Add(15*time.Minute)) { + t.Fatalf("got %v, want only the latest 10:15 occurrence", got) + } + if count := cache.takeCoalesced(scope.ID, got[0]); count != 2 { + t.Fatalf("coalesced=%d, want 2 older due occurrences", count) + } + + latest := LatestPlanInfo{ + Found: true, + PlanTime: activated.Add(5 * time.Minute), + Status: "RUNNING", + Attempt: 1, + MaxAttempts: pluginHookScheduleAttempts, + } + got, err = plans(context.Background(), scope, now, latest) + if err != nil { + t.Fatalf("plan while running: %v", err) + } + if len(got) != 0 { + t.Fatalf("a running latest plan must block newer plans, got %v", got) + } + + latest.Status = "FAILED" + latest.NextRetryAt = now.Add(time.Minute) + got, err = plans(context.Background(), scope, now, latest) + if err != nil { + t.Fatalf("plan before retry: %v", err) + } + if len(got) != 0 { + t.Fatalf("a retryable failure must block newer plans before backoff, got %v", got) + } + + latest.NextRetryAt = now.Add(-time.Second) + got, err = plans(context.Background(), scope, now, latest) + if err != nil { + t.Fatalf("plan due retry: %v", err) + } + if len(got) != 1 || !got[0].Equal(latest.PlanTime) { + t.Fatalf("retry must reuse plan_time %s, got %v", latest.PlanTime, got) + } +} + +func TestLatestPluginHookOccurrencePagesThroughLongOutage(t *testing.T) { + activated := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + now := activated.Add(9 * 24 * time.Hour) + got, count, err := latestPluginHookOccurrence("*/5 * * * *", "UTC", activated, now) + if err != nil { + t.Fatalf("enumerate occurrences: %v", err) + } + if !got.Equal(now) { + t.Fatalf("latest=%s, want %s", got, now) + } + if count != 9*24*12 { + t.Fatalf("count=%d, want %d", count, 9*24*12) + } +} + +func TestPluginHookScheduleDeliveryIDIsStablePerOccurrence(t *testing.T) { + installation := util.MustParseUUID("11111111-1111-4111-8111-111111111111") + generation := util.MustParseUUID("22222222-2222-4222-8222-222222222222") + plan := time.Date(2026, 8, 23, 10, 15, 0, 0, time.UTC) + + first := pluginHookScheduleDeliveryID(installation, "digest", generation, plan) + second := pluginHookScheduleDeliveryID(installation, "digest", generation, plan) + if first != second { + t.Fatalf("same occurrence changed delivery id: %q != %q", first, second) + } + if first == pluginHookScheduleDeliveryID(installation, "digest", generation, plan.Add(5*time.Minute)) { + t.Fatal("different plan_time must produce a different delivery id") + } + if first == pluginHookScheduleDeliveryID( + installation, + "digest", + util.MustParseUUID("33333333-3333-4333-8333-333333333333"), + plan, + ) { + t.Fatal("different generation must produce a different delivery id") + } +} + +func TestPluginHookScheduleHandlerRechecksFeatureFlagAfterClaim(t *testing.T) { + provider := featureflag.NewStaticProvider() + provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: false}) + plugins := &service.PluginService{FeatureFlags: featureflag.NewService(provider)} + handler := pluginHookScheduleHandler(nil, plugins, newPluginHookScheduleCache()) + + result, err := handler(context.Background(), HandlerInput{ + Scope: Scope{ID: "11111111-1111-4111-8111-111111111111:22222222-2222-4222-8222-222222222222"}, + }) + if err != nil { + t.Fatalf("feature-disabled handler: %v", err) + } + if result.Result["skipped_reason"] != "feature_disabled" { + t.Fatalf("skipped_reason=%v, want feature_disabled", result.Result["skipped_reason"]) + } +} diff --git a/server/internal/service/plugin.go b/server/internal/service/plugin.go index c7280907f..c14fe8d42 100644 --- a/server/internal/service/plugin.go +++ b/server/internal/service/plugin.go @@ -402,6 +402,9 @@ func (s *PluginService) InstallPlugin(ctx context.Context, workspaceID, userID p if skillErr := s.InstallSkillResources(ctx, queries, installation, manifest, userID); skillErr != nil { return db.PluginInstallation{}, skillErr } + if scheduleErr := s.reconcilePluginHookSchedules(ctx, queries, installation, manifest); scheduleErr != nil { + return db.PluginInstallation{}, scheduleErr + } if commitErr := tx.Commit(ctx); commitErr != nil { return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit install", Err: commitErr} } @@ -458,6 +461,9 @@ func (s *PluginService) InstallPlugin(ctx context.Context, workspaceID, userID p if err := s.InstallSkillResources(ctx, queries, updated, manifest, userID); err != nil { return db.PluginInstallation{}, err } + if err := s.reconcilePluginHookSchedules(ctx, queries, updated, manifest); err != nil { + return db.PluginInstallation{}, err + } if err := tx.Commit(ctx); err != nil { return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit upgrade", Err: err} } @@ -691,19 +697,35 @@ func (s *PluginService) ConfiguredSecretKeys(ctx context.Context, installationID // SetEnabled toggles an installation. Disabling hides every contribution // immediately but keeps storage and secrets, so re-enabling is not a reinstall. func (s *PluginService) SetEnabled(ctx context.Context, installation db.PluginInstallation, enabled bool) (db.PluginInstallation, error) { - updated, err := s.Queries.SetPluginInstallationEnabled(ctx, db.SetPluginInstallationEnabledParams{ + if installation.Enabled == enabled { + return installation, nil + } + tx, err := s.TxStarter.Begin(ctx) + if err != nil { + return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "begin plugin state update", Err: err} + } + defer func() { _ = tx.Rollback(ctx) }() + queries := s.Queries.WithTx(tx) + + updated, err := queries.SetPluginInstallationEnabled(ctx, db.SetPluginInstallationEnabledParams{ ID: installation.ID, Enabled: enabled, }) if err != nil { return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "update plugin state", Err: err} } + if err := s.setPluginHookSchedulesEnabled(ctx, queries, installation.ID, enabled); err != nil { + return db.PluginInstallation{}, err + } + if err := tx.Commit(ctx); err != nil { + return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit plugin state update", Err: err} + } return updated, nil } -// Uninstall removes the installation with its storage and secrets. There are no -// foreign keys or cascades by repository policy, so the three deletes share one -// transaction: a partial uninstall would strand plugin-owned rows nothing can +// Uninstall removes the installation and all of its application-owned rows. +// There are no foreign keys or cascades by repository policy, so the deletes +// share one transaction: a partial uninstall would strand rows nothing can // reach or clean up later. func (s *PluginService) Uninstall(ctx context.Context, installation db.PluginInstallation) error { tx, err := s.TxStarter.Begin(ctx) @@ -713,6 +735,9 @@ func (s *PluginService) Uninstall(ctx context.Context, installation db.PluginIns defer func() { _ = tx.Rollback(ctx) }() queries := s.Queries.WithTx(tx) + if err := queries.DeletePluginHookSchedulesByInstallation(ctx, installation.ID); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "delete plugin hook schedules", Err: err} + } if err := queries.DeletePluginStorageByInstallation(ctx, installation.ID); err != nil { return &PluginError{Kind: PluginErrorUnavailable, Message: "delete plugin storage", Err: err} } diff --git a/server/internal/service/plugin_hook.go b/server/internal/service/plugin_hook.go index 0b7b5ea4b..166ccfda2 100644 --- a/server/internal/service/plugin_hook.go +++ b/server/internal/service/plugin_hook.go @@ -16,6 +16,7 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/plugincontract" @@ -50,9 +51,10 @@ const ( hookEventAttempts = 3 hookEventBackoff = 2 * time.Second - // Circuit breaker. After this many failures inside the window, event + // Circuit breaker. After this many failures inside the window, background // delivery for the hook stops until the window rolls off — an endpoint that - // has been down for a minute does not need one request per event to notice. + // has been down for a minute does not need one request per event or schedule + // occurrence to notice. hookBreakerThreshold = 5 hookBreakerWindow = 5 * time.Minute @@ -76,9 +78,9 @@ type HookResult struct { // // This is the whole identity model for hooks, and it follows the trigger rather // than the plugin: a ui/manual hook runs because a person pressed something, so -// its writes stay that person's (marked via_plugin_id); an event hook has no -// person behind it and must not borrow the last one who happened to touch the -// issue. +// its writes stay that person's (marked via_plugin_id); event/schedule hooks +// have no person behind them and must not borrow the last one who touched the +// resource. type HookActor struct { Type string ID pgtype.UUID @@ -86,6 +88,10 @@ type HookActor struct { // HookInvocation is one call the engine has been asked to make. type HookInvocation struct { + // ID is allocated before the outbound request. The receiver can log it and + // correlate a response with the durable invocation row even when the call + // itself fails. + ID pgtype.UUID Installation db.PluginInstallation Hook plugincontract.Hook Trigger string @@ -97,18 +103,27 @@ type HookInvocation struct { // grant to reach across the workspace. IssueID pgtype.UUID Input any + // DeliveryID is stable across retries of the same scheduled occurrence. + // PlannedAt is the canonical UTC cron occurrence, not delivery wall time. + DeliveryID string + PlannedAt pgtype.Timestamptz + Attempt int } // hookRequestBody is the wire format. Stable and small on purpose: a hook // handler written against v1 should not have to care what else the host learns // how to send later. type hookRequestBody struct { - Version int `json:"version"` - HookKey string `json:"hook_key"` - Trigger string `json:"trigger"` - EventType string `json:"event_type,omitempty"` - WorkspaceID string `json:"workspace_id"` - Installation string `json:"installation_id"` + Version int `json:"version"` + InvocationID string `json:"invocation_id"` + DeliveryID string `json:"delivery_id,omitempty"` + Attempt int `json:"attempt"` + OccurredAt time.Time `json:"occurred_at"` + HookKey string `json:"hook_key"` + Trigger string `json:"trigger"` + EventType string `json:"event_type,omitempty"` + WorkspaceID string `json:"workspace_id"` + Installation string `json:"installation_id"` // IssueID is the issue this call is about, as resolved and permission-checked // by the host. Sent because the alternative is every handler reading it out // of client-supplied `input` — unvalidated, and absent entirely for the event @@ -130,8 +145,9 @@ type hookRequestBody struct { // minutes it is valid. Narrower than the installation's own token and tied // to this one call, so a handler that leaks it leaks a few minutes of the // scopes it was already using, not standing access. - CallbackToken string `json:"callback_token,omitempty"` - CallbackURL string `json:"callback_url,omitempty"` + CallbackToken string `json:"callback_token,omitempty"` + CallbackURL string `json:"callback_url,omitempty"` + Schedule *hookRequestSchedule `json:"schedule,omitempty"` } type hookRequestActor struct { @@ -139,6 +155,10 @@ type hookRequestActor struct { ID string `json:"id,omitempty"` } +type hookRequestSchedule struct { + PlannedAt time.Time `json:"planned_at"` +} + // FindHook returns the named hook from an installation's consented manifest. // // Read from the installation row, never from the manifest the source URL serves @@ -172,10 +192,16 @@ func HookAllowsTrigger(hook plugincontract.Hook, trigger string) bool { // InvokeHook performs one call and records it. // // Callers pick the blocking behaviour, not this function: ui/manual await the -// result because a person is watching, and the event dispatcher runs it on its -// own goroutine because the host request that produced the event must not wait -// for a third party. +// result because a person is watching, event delivery runs outside the host +// request, and schedule delivery runs under the durable scheduler lease. func (s *PluginService) InvokeHook(ctx context.Context, invocation HookInvocation, attempt int) (HookResult, error) { + if !invocation.ID.Valid { + invocation.ID = pgtype.UUID{Bytes: uuid.New(), Valid: true} + } + if attempt < 1 { + attempt = 1 + } + invocation.Attempt = attempt hook := invocation.Hook result := HookResult{HookKey: hook.Key, Trigger: invocation.Trigger, Attempts: attempt} @@ -355,8 +381,16 @@ func nonSecretConfig(installation db.PluginInstallation) map[string]any { } func (s *PluginService) buildHookBody(ctx context.Context, invocation HookInvocation) (hookRequestBody, error) { + attempt := invocation.Attempt + if attempt < 1 { + attempt = 1 + } body := hookRequestBody{ Version: 1, + InvocationID: uuidString(invocation.ID), + DeliveryID: invocation.DeliveryID, + Attempt: attempt, + OccurredAt: time.Now().UTC(), HookKey: invocation.Hook.Key, Trigger: invocation.Trigger, EventType: invocation.EventType, @@ -364,6 +398,9 @@ func (s *PluginService) buildHookBody(ctx context.Context, invocation HookInvoca Installation: uuidString(invocation.Installation.ID), Actor: hookRequestActor{Type: invocation.Actor.Type, ID: uuidString(invocation.Actor.ID)}, } + if invocation.PlannedAt.Valid { + body.Schedule = &hookRequestSchedule{PlannedAt: invocation.PlannedAt.Time.UTC()} + } if invocation.IssueID.Valid { body.IssueID = uuidString(invocation.IssueID) } @@ -485,8 +522,8 @@ func (s *PluginService) checkHookRate(ctx context.Context, installationID pgtype return nil } -// HookBreakerOpen reports whether event delivery for this hook is currently -// suspended after repeated failures. +// HookBreakerOpen reports whether background delivery for this hook is +// currently suspended after repeated failures. func (s *PluginService) HookBreakerOpen(ctx context.Context, installationID pgtype.UUID, hookKey string) bool { since := pgtype.Timestamptz{Time: time.Now().Add(-hookBreakerWindow), Valid: true} failures, err := s.Queries.CountRecentPluginFailures(ctx, db.CountRecentPluginFailuresParams{ @@ -502,6 +539,7 @@ func (s *PluginService) HookBreakerOpen(ctx context.Context, installationID pgty func (s *PluginService) recordInvocation(ctx context.Context, invocation HookInvocation, status string, attempt, latency int, message string) { params := db.CreatePluginInvocationParams{ + ID: invocation.ID, InstallationID: invocation.Installation.ID, WorkspaceID: invocation.Installation.WorkspaceID, HookKey: invocation.Hook.Key, @@ -510,6 +548,12 @@ func (s *PluginService) recordInvocation(ctx context.Context, invocation HookInv Attempt: int32(attempt), LatencyMs: int32(latency), } + if invocation.DeliveryID != "" { + params.DeliveryID = pgtype.Text{String: invocation.DeliveryID, Valid: true} + } + if invocation.PlannedAt.Valid { + params.PlannedAt = invocation.PlannedAt + } if invocation.EventType != "" { params.EventType = pgtype.Text{String: invocation.EventType, Valid: true} } diff --git a/server/internal/service/plugin_hook_test.go b/server/internal/service/plugin_hook_test.go index d7360bbfa..0b2d62eda 100644 --- a/server/internal/service/plugin_hook_test.go +++ b/server/internal/service/plugin_hook_test.go @@ -1,6 +1,7 @@ package service import ( + "context" "encoding/hex" "strings" "testing" @@ -137,13 +138,56 @@ func TestHookAllowsTriggerOnlyWhenDeclared(t *testing.T) { if !HookAllowsTrigger(hook, plugincontract.TriggerManual) { t.Fatal("a declared trigger must be allowed") } - for _, undeclared := range []string{plugincontract.TriggerUI, plugincontract.TriggerEvent, plugincontract.TriggerAgent} { + for _, undeclared := range []string{plugincontract.TriggerUI, plugincontract.TriggerEvent, plugincontract.TriggerAgent, plugincontract.TriggerSchedule} { if HookAllowsTrigger(hook, undeclared) { t.Fatalf("%s was not declared and must not be allowed", undeclared) } } } +func TestScheduledHookBodyExtendsVersionOneProtocol(t *testing.T) { + installationID := testInstallationID(t) + invocationID := parseUUIDValueForTest(t, "22222222-2222-4222-8222-222222222222") + plannedAt := time.Date(2026, 8, 23, 10, 15, 0, 0, time.UTC) + body, err := (&PluginService{}).buildHookBody(context.Background(), HookInvocation{ + ID: invocationID, + Installation: db.PluginInstallation{ + ID: installationID, + WorkspaceID: parseUUIDValueForTest(t, "33333333-3333-4333-8333-333333333333"), + }, + Hook: plugincontract.Hook{Key: "digest"}, + Trigger: plugincontract.TriggerSchedule, + Actor: HookActor{Type: "plugin", ID: installationID}, + DeliveryID: "psd_stable", + PlannedAt: pgtype.Timestamptz{Time: plannedAt, Valid: true}, + Attempt: 2, + }) + if err != nil { + t.Fatalf("build scheduled hook body: %v", err) + } + if body.Version != 1 { + t.Fatalf("version=%d, want existing protocol version 1", body.Version) + } + if body.InvocationID != uuidString(invocationID) || body.DeliveryID != "psd_stable" || body.Attempt != 2 { + t.Fatalf("unexpected identity fields: %+v", body) + } + if body.OccurredAt.IsZero() { + t.Fatal("occurred_at must be populated per request") + } + if body.Schedule == nil || !body.Schedule.PlannedAt.Equal(plannedAt) { + t.Fatalf("schedule.planned_at=%v, want %s", body.Schedule, plannedAt) + } +} + +func parseUUIDValueForTest(t *testing.T, value string) pgtype.UUID { + t.Helper() + parsed, err := parseUUIDValue(value) + if err != nil { + t.Fatalf("parse uuid: %v", err) + } + return parsed +} + // FindHook reads the CONSENTED manifest on the installation row. A hook that is // not in it does not exist, whatever the source URL serves today. func TestFindHookReadsTheConsentedManifest(t *testing.T) { diff --git a/server/internal/service/plugin_schedule.go b/server/internal/service/plugin_schedule.go new file mode 100644 index 000000000..6017a858c --- /dev/null +++ b/server/internal/service/plugin_schedule.go @@ -0,0 +1,131 @@ +package service + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgtype" + db "github.com/multica-ai/multica/server/pkg/db/generated" + "github.com/multica-ai/multica/server/pkg/plugincontract" +) + +// reconcilePluginHookSchedules makes the durable schedule projection match the +// consented manifest inside the caller's installation/upgrade transaction. +// Unchanged cron/timezone pairs keep their generation: a code-only plugin +// upgrade must not reset the schedule clock or orphan a retryable execution. +func (s *PluginService) reconcilePluginHookSchedules( + ctx context.Context, + queries *db.Queries, + installation db.PluginInstallation, + manifest plugincontract.Manifest, +) error { + existingRows, err := queries.ListPluginHookSchedulesByInstallation(ctx, installation.ID) + if err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "list plugin hook schedules", Err: err} + } + existing := make(map[string]db.PluginHookSchedule, len(existingRows)) + for _, row := range existingRows { + existing[row.HookKey] = row + } + + now := time.Now().UTC() + for _, hook := range manifest.Contributes.Hooks { + if hook.Schedule == nil || !HookAllowsTrigger(hook, plugincontract.TriggerSchedule) { + continue + } + nextRun, err := pluginScheduleNextRun(*hook.Schedule, now) + if err != nil { + // Published manifests were validated already. Surfacing drift here is + // safer than installing a schedule the runtime cannot evaluate. + return &PluginError{Kind: PluginErrorInvalid, Message: fmt.Sprintf("schedule for hook %q is invalid", hook.Key), Err: err} + } + if !installation.Enabled { + // Disabled installations have no meaningful "next" run. Re-enable + // rotates the generation and computes this projection from that time. + nextRun = pgtype.Timestamptz{} + } + + row, found := existing[hook.Key] + if !found { + if _, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{ + InstallationID: installation.ID, + WorkspaceID: installation.WorkspaceID, + HookKey: hook.Key, + CronExpression: hook.Schedule.Cron, + Timezone: hook.Schedule.Timezone, + Enabled: installation.Enabled, + NextRunAt: nextRun, + }); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "create plugin hook schedule", Err: err} + } + continue + } + delete(existing, hook.Key) + if row.CronExpression == hook.Schedule.Cron && row.Timezone == hook.Schedule.Timezone { + continue + } + if _, err := queries.UpdatePluginHookScheduleDefinition(ctx, db.UpdatePluginHookScheduleDefinitionParams{ + ID: row.ID, + CronExpression: hook.Schedule.Cron, + Timezone: hook.Schedule.Timezone, + Enabled: installation.Enabled, + NextRunAt: nextRun, + }); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "update plugin hook schedule", Err: err} + } + } + + for _, orphan := range existing { + if err := queries.DeletePluginHookSchedule(ctx, orphan.ID); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "delete removed plugin hook schedule", Err: err} + } + } + return nil +} + +func pluginScheduleNextRun(schedule plugincontract.HookSchedule, after time.Time) (pgtype.Timestamptz, error) { + next, err := NextOccurrenceAfterUTC(schedule.Cron, schedule.Timezone, after) + if err != nil { + return pgtype.Timestamptz{}, err + } + if next.IsZero() { + return pgtype.Timestamptz{}, nil + } + return pgtype.Timestamptz{Time: next.UTC(), Valid: true}, nil +} + +func (s *PluginService) setPluginHookSchedulesEnabled( + ctx context.Context, + queries *db.Queries, + installationID pgtype.UUID, + enabled bool, +) error { + if !enabled { + if err := queries.DisablePluginHookSchedules(ctx, installationID); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "disable plugin hook schedules", Err: err} + } + return nil + } + + rows, err := queries.ListPluginHookSchedulesByInstallation(ctx, installationID) + if err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "list plugin hook schedules", Err: err} + } + now := time.Now().UTC() + for _, row := range rows { + nextRun, err := pluginScheduleNextRun(plugincontract.HookSchedule{ + Cron: row.CronExpression, Timezone: row.Timezone, + }, now) + if err != nil { + return &PluginError{Kind: PluginErrorInvalid, Message: fmt.Sprintf("schedule for hook %q is invalid", row.HookKey), Err: err} + } + if _, err := queries.ReactivatePluginHookSchedule(ctx, db.ReactivatePluginHookScheduleParams{ + ID: row.ID, + NextRunAt: nextRun, + }); err != nil { + return &PluginError{Kind: PluginErrorUnavailable, Message: "reactivate plugin hook schedule", Err: err} + } + } + return nil +} diff --git a/server/migrations/399_plugin_hook_schedule.down.sql b/server/migrations/399_plugin_hook_schedule.down.sql new file mode 100644 index 000000000..dc8c8fae5 --- /dev/null +++ b/server/migrations/399_plugin_hook_schedule.down.sql @@ -0,0 +1,9 @@ +DELETE FROM plugin_invocation WHERE trigger = 'schedule'; + +ALTER TABLE plugin_invocation DROP CONSTRAINT IF EXISTS plugin_invocation_trigger_check; +ALTER TABLE plugin_invocation ADD CONSTRAINT plugin_invocation_trigger_check + CHECK (trigger IN ('ui', 'manual', 'event', 'agent')) NOT VALID; +ALTER TABLE plugin_invocation DROP COLUMN IF EXISTS planned_at; +ALTER TABLE plugin_invocation DROP COLUMN IF EXISTS delivery_id; + +DROP TABLE IF EXISTS plugin_hook_schedule; diff --git a/server/migrations/399_plugin_hook_schedule.up.sql b/server/migrations/399_plugin_hook_schedule.up.sql new file mode 100644 index 000000000..2323e33eb --- /dev/null +++ b/server/migrations/399_plugin_hook_schedule.up.sql @@ -0,0 +1,35 @@ +-- Durable schedule declarations for Plugin HTTP hooks. The consented manifest +-- remains the public source of truth; this table is the execution projection +-- that gives each installed hook an activation epoch and a stable scheduler +-- scope without reparsing every installation on every tick. +-- +-- Relationships are application-owned by repository policy: install, upgrade, +-- enable/disable, uninstall and workspace teardown reconcile these rows in the +-- same transaction as plugin_installation. No foreign key or cascade is used. +CREATE TABLE plugin_hook_schedule ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + installation_id UUID NOT NULL, + workspace_id UUID NOT NULL, + hook_key TEXT NOT NULL CHECK (char_length(hook_key) BETWEEN 1 AND 128), + cron_expression TEXT NOT NULL CHECK (char_length(cron_expression) BETWEEN 1 AND 255), + timezone TEXT NOT NULL CHECK (char_length(timezone) BETWEEN 1 AND 255), + generation UUID NOT NULL DEFAULT gen_random_uuid(), + activated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Display-only projection. Dispatch correctness is recovered from cron + + -- activated_at + sys_cron_executions even when this value is stale/NULL. + next_run_at TIMESTAMPTZ, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Schedule attempts use the same operational invocation table as every other +-- Hook trigger. delivery_id is stable across retries; planned_at is the cron +-- occurrence, distinct from created_at (the actual attempt time). +ALTER TABLE plugin_invocation + ADD COLUMN delivery_id TEXT CHECK (delivery_id IS NULL OR char_length(delivery_id) BETWEEN 1 AND 128), + ADD COLUMN planned_at TIMESTAMPTZ; + +ALTER TABLE plugin_invocation DROP CONSTRAINT IF EXISTS plugin_invocation_trigger_check; +ALTER TABLE plugin_invocation ADD CONSTRAINT plugin_invocation_trigger_check + CHECK (trigger IN ('ui', 'manual', 'event', 'agent', 'schedule')) NOT VALID; diff --git a/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql b/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql new file mode 100644 index 000000000..b1d1f5ae9 --- /dev/null +++ b/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_installation_key; diff --git a/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql b/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql new file mode 100644 index 000000000..6445015f2 --- /dev/null +++ b/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql @@ -0,0 +1,2 @@ +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_plugin_hook_schedule_installation_key + ON plugin_hook_schedule (installation_id, hook_key); diff --git a/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql b/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql new file mode 100644 index 000000000..d41dad14a --- /dev/null +++ b/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_enabled; diff --git a/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql b/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql new file mode 100644 index 000000000..8110b5024 --- /dev/null +++ b/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql @@ -0,0 +1,3 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_plugin_hook_schedule_enabled + ON plugin_hook_schedule (id) + WHERE enabled; diff --git a/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql b/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql new file mode 100644 index 000000000..6997bc9d6 --- /dev/null +++ b/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql @@ -0,0 +1,3 @@ +-- Validation changes no data or contract shape; migration 399 restores the old +-- constraint during rollback. +SELECT 1; diff --git a/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql b/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql new file mode 100644 index 000000000..a9d612f07 --- /dev/null +++ b/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql @@ -0,0 +1 @@ +ALTER TABLE plugin_invocation VALIDATE CONSTRAINT plugin_invocation_trigger_check; diff --git a/server/pkg/db/generated/models.go b/server/pkg/db/generated/models.go index be5748aab..dbde55fdd 100644 --- a/server/pkg/db/generated/models.go +++ b/server/pkg/db/generated/models.go @@ -981,6 +981,21 @@ type PinnedItem struct { CreatedAt pgtype.Timestamptz `json:"created_at"` } +type PluginHookSchedule struct { + ID pgtype.UUID `json:"id"` + InstallationID pgtype.UUID `json:"installation_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + HookKey string `json:"hook_key"` + CronExpression string `json:"cron_expression"` + Timezone string `json:"timezone"` + Generation pgtype.UUID `json:"generation"` + ActivatedAt pgtype.Timestamptz `json:"activated_at"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` + Enabled bool `json:"enabled"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type PluginInstallation struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` @@ -1011,6 +1026,8 @@ type PluginInvocation struct { LatencyMs int32 `json:"latency_ms"` Error pgtype.Text `json:"error"` CreatedAt pgtype.Timestamptz `json:"created_at"` + DeliveryID pgtype.Text `json:"delivery_id"` + PlannedAt pgtype.Timestamptz `json:"planned_at"` } type PluginPackage struct { diff --git a/server/pkg/db/generated/plugin.sql.go b/server/pkg/db/generated/plugin.sql.go index e4ed4ad36..257671039 100644 --- a/server/pkg/db/generated/plugin.sql.go +++ b/server/pkg/db/generated/plugin.sql.go @@ -67,6 +67,52 @@ func (q *Queries) CountRecentPluginInvocations(ctx context.Context, arg CountRec return count, err } +const createPluginHookSchedule = `-- name: CreatePluginHookSchedule :one +INSERT INTO plugin_hook_schedule ( + installation_id, workspace_id, hook_key, cron_expression, timezone, + next_run_at, enabled +) VALUES ($1, $2, $3, $4, $5, $7, $6) +RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at +` + +type CreatePluginHookScheduleParams struct { + InstallationID pgtype.UUID `json:"installation_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + HookKey string `json:"hook_key"` + CronExpression string `json:"cron_expression"` + Timezone string `json:"timezone"` + Enabled bool `json:"enabled"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` +} + +func (q *Queries) CreatePluginHookSchedule(ctx context.Context, arg CreatePluginHookScheduleParams) (PluginHookSchedule, error) { + row := q.db.QueryRow(ctx, createPluginHookSchedule, + arg.InstallationID, + arg.WorkspaceID, + arg.HookKey, + arg.CronExpression, + arg.Timezone, + arg.Enabled, + arg.NextRunAt, + ) + var i PluginHookSchedule + err := row.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + const createPluginInstallation = `-- name: CreatePluginInstallation :one INSERT INTO plugin_installation ( workspace_id, plugin_key, package_version_id, version, manifest, granted_scopes, installed_by @@ -117,21 +163,29 @@ func (q *Queries) CreatePluginInstallation(ctx context.Context, arg CreatePlugin const createPluginInvocation = `-- name: CreatePluginInvocation :one INSERT INTO plugin_invocation ( - installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error -) VALUES ($1, $2, $3, $4, $5, $8, $6, $7, $9) -RETURNING id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at + id, installation_id, workspace_id, hook_key, trigger, status, event_type, + delivery_id, planned_at, attempt, latency_ms, error +) VALUES ( + COALESCE($8::uuid, gen_random_uuid()), + $1, $2, $3, $4, $5, $9, $10, + $11, $6, $7, $12 +) +RETURNING id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at, delivery_id, planned_at ` type CreatePluginInvocationParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - HookKey string `json:"hook_key"` - Trigger string `json:"trigger"` - Status string `json:"status"` - Attempt int32 `json:"attempt"` - LatencyMs int32 `json:"latency_ms"` - EventType pgtype.Text `json:"event_type"` - Error pgtype.Text `json:"error"` + InstallationID pgtype.UUID `json:"installation_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + HookKey string `json:"hook_key"` + Trigger string `json:"trigger"` + Status string `json:"status"` + Attempt int32 `json:"attempt"` + LatencyMs int32 `json:"latency_ms"` + ID pgtype.UUID `json:"id"` + EventType pgtype.Text `json:"event_type"` + DeliveryID pgtype.Text `json:"delivery_id"` + PlannedAt pgtype.Timestamptz `json:"planned_at"` + Error pgtype.Text `json:"error"` } func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginInvocationParams) (PluginInvocation, error) { @@ -143,7 +197,10 @@ func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginIn arg.Status, arg.Attempt, arg.LatencyMs, + arg.ID, arg.EventType, + arg.DeliveryID, + arg.PlannedAt, arg.Error, ) var i PluginInvocation @@ -159,6 +216,8 @@ func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginIn &i.LatencyMs, &i.Error, &i.CreatedAt, + &i.DeliveryID, + &i.PlannedAt, ) return i, err } @@ -278,6 +337,24 @@ func (q *Queries) DeleteExpiredPluginInvocations(ctx context.Context, createdAt return result.RowsAffected(), nil } +const deletePluginHookSchedule = `-- name: DeletePluginHookSchedule :exec +DELETE FROM plugin_hook_schedule WHERE id = $1 +` + +func (q *Queries) DeletePluginHookSchedule(ctx context.Context, id pgtype.UUID) error { + _, err := q.db.Exec(ctx, deletePluginHookSchedule, id) + return err +} + +const deletePluginHookSchedulesByInstallation = `-- name: DeletePluginHookSchedulesByInstallation :exec +DELETE FROM plugin_hook_schedule WHERE installation_id = $1 +` + +func (q *Queries) DeletePluginHookSchedulesByInstallation(ctx context.Context, installationID pgtype.UUID) error { + _, err := q.db.Exec(ctx, deletePluginHookSchedulesByInstallation, installationID) + return err +} + const deletePluginInstallation = `-- name: DeletePluginInstallation :exec DELETE FROM plugin_installation WHERE id = $1 ` @@ -410,6 +487,41 @@ func (q *Queries) DeletePluginStorageValue(ctx context.Context, arg DeletePlugin return result.RowsAffected(), nil } +const disablePluginHookSchedules = `-- name: DisablePluginHookSchedules :exec +UPDATE plugin_hook_schedule +SET enabled = FALSE, next_run_at = NULL, updated_at = now() +WHERE installation_id = $1 +` + +func (q *Queries) DisablePluginHookSchedules(ctx context.Context, installationID pgtype.UUID) error { + _, err := q.db.Exec(ctx, disablePluginHookSchedules, installationID) + return err +} + +const getPluginHookSchedule = `-- name: GetPluginHookSchedule :one +SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule WHERE id = $1 +` + +func (q *Queries) GetPluginHookSchedule(ctx context.Context, id pgtype.UUID) (PluginHookSchedule, error) { + row := q.db.QueryRow(ctx, getPluginHookSchedule, id) + var i PluginHookSchedule + err := row.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + const getPluginInstallation = `-- name: GetPluginInstallation :one SELECT id, workspace_id, plugin_key, version, manifest, granted_scopes, config, enabled, installed_by, created_at, updated_at, token_hash, token_rotated_at, mcp_approvals, package_version_id FROM plugin_installation WHERE id = $1 ` @@ -725,8 +837,88 @@ func (q *Queries) GetWorkspacePluginPackageVersion(ctx context.Context, arg GetW return i, err } +const listEnabledPluginHookSchedules = `-- name: ListEnabledPluginHookSchedules :many +SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule +WHERE enabled +ORDER BY id ASC +` + +// Do not filter by next_run_at: it is display-only, while retries and recovery +// derive eligibility from cron + activated_at + sys_cron_executions. +func (q *Queries) ListEnabledPluginHookSchedules(ctx context.Context) ([]PluginHookSchedule, error) { + rows, err := q.db.Query(ctx, listEnabledPluginHookSchedules) + if err != nil { + return nil, err + } + defer rows.Close() + items := []PluginHookSchedule{} + for rows.Next() { + var i PluginHookSchedule + if err := rows.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPluginHookSchedulesByInstallation = `-- name: ListPluginHookSchedulesByInstallation :many +SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule +WHERE installation_id = $1 +ORDER BY hook_key ASC +` + +func (q *Queries) ListPluginHookSchedulesByInstallation(ctx context.Context, installationID pgtype.UUID) ([]PluginHookSchedule, error) { + rows, err := q.db.Query(ctx, listPluginHookSchedulesByInstallation, installationID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []PluginHookSchedule{} + for rows.Next() { + var i PluginHookSchedule + if err := rows.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listPluginInvocations = `-- name: ListPluginInvocations :many -SELECT id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at FROM plugin_invocation +SELECT id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at, delivery_id, planned_at FROM plugin_invocation WHERE installation_id = $1 ORDER BY created_at DESC LIMIT $2 @@ -758,6 +950,8 @@ func (q *Queries) ListPluginInvocations(ctx context.Context, arg ListPluginInvoc &i.LatencyMs, &i.Error, &i.CreatedAt, + &i.DeliveryID, + &i.PlannedAt, ); err != nil { return nil, err } @@ -1041,6 +1235,45 @@ func (q *Queries) LockPluginPackageKey(ctx context.Context, dollar_1 string) err return err } +const reactivatePluginHookSchedule = `-- name: ReactivatePluginHookSchedule :one +UPDATE plugin_hook_schedule +SET enabled = TRUE, + generation = gen_random_uuid(), + activated_at = now(), + next_run_at = $2, + updated_at = now() +WHERE id = $1 +RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at +` + +type ReactivatePluginHookScheduleParams struct { + ID pgtype.UUID `json:"id"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` +} + +// Re-enable starts a new epoch so occurrences while the installation was off +// are never caught up. An already-sent request from the old generation may +// finish, but no unstarted old plan survives the generation check. +func (q *Queries) ReactivatePluginHookSchedule(ctx context.Context, arg ReactivatePluginHookScheduleParams) (PluginHookSchedule, error) { + row := q.db.QueryRow(ctx, reactivatePluginHookSchedule, arg.ID, arg.NextRunAt) + var i PluginHookSchedule + err := row.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + const setPluginInstallationEnabled = `-- name: SetPluginInstallationEnabled :one UPDATE plugin_installation SET enabled = $2, @@ -1128,6 +1361,75 @@ func (q *Queries) SetPluginMCPApprovals(ctx context.Context, arg SetPluginMCPApp return i, err } +const updatePluginHookScheduleDefinition = `-- name: UpdatePluginHookScheduleDefinition :one +UPDATE plugin_hook_schedule +SET cron_expression = $2, + timezone = $3, + generation = gen_random_uuid(), + activated_at = now(), + next_run_at = $5, + enabled = $4, + updated_at = now() +WHERE id = $1 +RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at +` + +type UpdatePluginHookScheduleDefinitionParams struct { + ID pgtype.UUID `json:"id"` + CronExpression string `json:"cron_expression"` + Timezone string `json:"timezone"` + Enabled bool `json:"enabled"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` +} + +// A cron/timezone change creates a new scheduler generation. The old +// sys_cron_executions rows remain immutable history under the prior scope id. +func (q *Queries) UpdatePluginHookScheduleDefinition(ctx context.Context, arg UpdatePluginHookScheduleDefinitionParams) (PluginHookSchedule, error) { + row := q.db.QueryRow(ctx, updatePluginHookScheduleDefinition, + arg.ID, + arg.CronExpression, + arg.Timezone, + arg.Enabled, + arg.NextRunAt, + ) + var i PluginHookSchedule + err := row.Scan( + &i.ID, + &i.InstallationID, + &i.WorkspaceID, + &i.HookKey, + &i.CronExpression, + &i.Timezone, + &i.Generation, + &i.ActivatedAt, + &i.NextRunAt, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const updatePluginHookScheduleNextRun = `-- name: UpdatePluginHookScheduleNextRun :execrows +UPDATE plugin_hook_schedule +SET next_run_at = $3, updated_at = now() +WHERE id = $1 AND generation = $2 AND enabled +` + +type UpdatePluginHookScheduleNextRunParams struct { + ID pgtype.UUID `json:"id"` + Generation pgtype.UUID `json:"generation"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` +} + +func (q *Queries) UpdatePluginHookScheduleNextRun(ctx context.Context, arg UpdatePluginHookScheduleNextRunParams) (int64, error) { + result, err := q.db.Exec(ctx, updatePluginHookScheduleNextRun, arg.ID, arg.Generation, arg.NextRunAt) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const updatePluginInstallationConfig = `-- name: UpdatePluginInstallationConfig :one UPDATE plugin_installation SET config = $2, diff --git a/server/pkg/db/generated/workspace_delete.sql.go b/server/pkg/db/generated/workspace_delete.sql.go index f75106277..d39617c78 100644 --- a/server/pkg/db/generated/workspace_delete.sql.go +++ b/server/pkg/db/generated/workspace_delete.sql.go @@ -503,6 +503,10 @@ deleted_secrets AS ( DELETE FROM plugin_secret WHERE installation_id IN (SELECT id FROM installations) ), +deleted_hook_schedules AS ( + DELETE FROM plugin_hook_schedule + WHERE installation_id IN (SELECT id FROM installations) +), deleted_invocations AS ( DELETE FROM plugin_invocation WHERE workspace_id = $1 diff --git a/server/pkg/db/queries/plugin.sql b/server/pkg/db/queries/plugin.sql index e4653f6bc..19e38595b 100644 --- a/server/pkg/db/queries/plugin.sql +++ b/server/pkg/db/queries/plugin.sql @@ -143,6 +143,71 @@ ORDER BY created_at ASC; -- name: DeletePluginInstallation :exec DELETE FROM plugin_installation WHERE id = $1; +-- name: CreatePluginHookSchedule :one +INSERT INTO plugin_hook_schedule ( + installation_id, workspace_id, hook_key, cron_expression, timezone, + next_run_at, enabled +) VALUES ($1, $2, $3, $4, $5, sqlc.narg(next_run_at), $6) +RETURNING *; + +-- name: GetPluginHookSchedule :one +SELECT * FROM plugin_hook_schedule WHERE id = $1; + +-- name: ListPluginHookSchedulesByInstallation :many +SELECT * FROM plugin_hook_schedule +WHERE installation_id = $1 +ORDER BY hook_key ASC; + +-- name: ListEnabledPluginHookSchedules :many +-- Do not filter by next_run_at: it is display-only, while retries and recovery +-- derive eligibility from cron + activated_at + sys_cron_executions. +SELECT * FROM plugin_hook_schedule +WHERE enabled +ORDER BY id ASC; + +-- name: UpdatePluginHookScheduleDefinition :one +-- A cron/timezone change creates a new scheduler generation. The old +-- sys_cron_executions rows remain immutable history under the prior scope id. +UPDATE plugin_hook_schedule +SET cron_expression = $2, + timezone = $3, + generation = gen_random_uuid(), + activated_at = now(), + next_run_at = sqlc.narg(next_run_at), + enabled = $4, + updated_at = now() +WHERE id = $1 +RETURNING *; + +-- name: DisablePluginHookSchedules :exec +UPDATE plugin_hook_schedule +SET enabled = FALSE, next_run_at = NULL, updated_at = now() +WHERE installation_id = $1; + +-- name: ReactivatePluginHookSchedule :one +-- Re-enable starts a new epoch so occurrences while the installation was off +-- are never caught up. An already-sent request from the old generation may +-- finish, but no unstarted old plan survives the generation check. +UPDATE plugin_hook_schedule +SET enabled = TRUE, + generation = gen_random_uuid(), + activated_at = now(), + next_run_at = sqlc.narg(next_run_at), + updated_at = now() +WHERE id = $1 +RETURNING *; + +-- name: UpdatePluginHookScheduleNextRun :execrows +UPDATE plugin_hook_schedule +SET next_run_at = sqlc.narg(next_run_at), updated_at = now() +WHERE id = $1 AND generation = $2 AND enabled; + +-- name: DeletePluginHookSchedule :exec +DELETE FROM plugin_hook_schedule WHERE id = $1; + +-- name: DeletePluginHookSchedulesByInstallation :exec +DELETE FROM plugin_hook_schedule WHERE installation_id = $1; + -- name: UpsertPluginStorageValue :one INSERT INTO plugin_storage (installation_id, scope_type, scope_id, key, value) VALUES ($1, $2, $3, $4, $5) @@ -211,8 +276,13 @@ SELECT * FROM plugin_installation WHERE token_hash = $1; -- name: CreatePluginInvocation :one INSERT INTO plugin_invocation ( - installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error -) VALUES ($1, $2, $3, $4, $5, sqlc.narg(event_type), $6, $7, sqlc.narg(error)) + id, installation_id, workspace_id, hook_key, trigger, status, event_type, + delivery_id, planned_at, attempt, latency_ms, error +) VALUES ( + COALESCE(sqlc.narg('id')::uuid, gen_random_uuid()), + $1, $2, $3, $4, $5, sqlc.narg(event_type), sqlc.narg(delivery_id), + sqlc.narg(planned_at), $6, $7, sqlc.narg(error) +) RETURNING *; -- name: ListPluginInvocations :many diff --git a/server/pkg/db/queries/workspace_delete.sql b/server/pkg/db/queries/workspace_delete.sql index 42006eacb..f26a837c0 100644 --- a/server/pkg/db/queries/workspace_delete.sql +++ b/server/pkg/db/queries/workspace_delete.sql @@ -602,6 +602,10 @@ deleted_secrets AS ( DELETE FROM plugin_secret WHERE installation_id IN (SELECT id FROM installations) ), +deleted_hook_schedules AS ( + DELETE FROM plugin_hook_schedule + WHERE installation_id IN (SELECT id FROM installations) +), -- Hook call records are workspace-scoped in their own right, so this deletes by -- workspace rather than through the installation ids: a row whose installation -- was already uninstalled would otherwise survive the workspace it described. diff --git a/server/pkg/plugincontract/capabilities.go b/server/pkg/plugincontract/capabilities.go index f2cea286e..76c4275f1 100644 --- a/server/pkg/plugincontract/capabilities.go +++ b/server/pkg/plugincontract/capabilities.go @@ -39,10 +39,11 @@ func HostCapabilities() Capabilities { // offered to an agent as an MCP tool and the agent decides. The // daemon-side server that renders it is daemon/plugin_hook_mcp.go. HookTriggers: map[string]bool{ - TriggerUI: true, - TriggerManual: true, - TriggerEvent: true, - TriggerAgent: true, + TriggerUI: true, + TriggerManual: true, + TriggerEvent: true, + TriggerAgent: true, + TriggerSchedule: true, }, // http calls one declared endpoint; mcp adopts an external MCP server's // tools, which is why it ships with an approval step that pins them by diff --git a/server/pkg/plugincontract/manifest.go b/server/pkg/plugincontract/manifest.go index c78007d6f..c1b6ce2c5 100644 --- a/server/pkg/plugincontract/manifest.go +++ b/server/pkg/plugincontract/manifest.go @@ -20,6 +20,9 @@ import ( "net/url" "regexp" "strings" + "time" + + "github.com/robfig/cron/v3" ) const ( @@ -44,14 +47,24 @@ const ( ) // Hook triggers. Declaring a trigger says who may invoke the hook, never what -// the hook does. Only `event` is asynchronous, and it never blocks the host. +// the hook does. `event` and `schedule` are asynchronous, and neither blocks a +// user-facing host request. const ( - TriggerUI = "ui" - TriggerManual = "manual" - TriggerAgent = "agent" - TriggerEvent = "event" + TriggerUI = "ui" + TriggerManual = "manual" + TriggerAgent = "agent" + TriggerEvent = "event" + TriggerSchedule = "schedule" ) +// Scheduled hooks are a polling primitive, not a high-frequency timer. The +// floor keeps one installation from turning its manifest into an unbounded +// source of outbound traffic and leaves enough room for the HTTP timeout plus +// durable retries before the next plan becomes due. +const MinimumScheduleInterval = 5 * time.Minute + +var scheduleCronParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) + // Hook transports. const ( TransportHTTP = "http" @@ -203,10 +216,19 @@ type Hook struct { InputSchema json.RawMessage `json:"input_schema,omitempty"` Triggers []string `json:"triggers"` Events []string `json:"events,omitempty"` + Schedule *HookSchedule `json:"schedule,omitempty"` Transport HookTransport `json:"transport"` TimeoutMs int `json:"timeout_ms,omitempty"` } +// HookSchedule declares the one automatic cadence attached to a hook. Multiple +// cadences are represented by multiple hook keys, which keeps the durable +// execution identity unambiguous. +type HookSchedule struct { + Cron string `json:"cron"` + Timezone string `json:"timezone"` +} + type HookTransport struct { Type string `json:"type"` URL string `json:"url"` @@ -605,7 +627,7 @@ func (m Manifest) validateContributions() error { triggerSeen := map[string]bool{} for _, trigger := range hook.Triggers { switch trigger { - case TriggerUI, TriggerManual, TriggerAgent, TriggerEvent: + case TriggerUI, TriggerManual, TriggerAgent, TriggerEvent, TriggerSchedule: default: return fmt.Errorf("%s.triggers contains unsupported trigger %q", field, trigger) } @@ -640,6 +662,19 @@ func (m Manifest) validateContributions() error { } else if len(hook.Events) > 0 { return fmt.Errorf("%s.events requires the event trigger", field) } + if triggerSeen[TriggerSchedule] { + if hook.Schedule == nil { + return fmt.Errorf("%s.schedule is required when the schedule trigger is declared", field) + } + if hook.Transport.Type != TransportHTTP { + return fmt.Errorf("%s.schedule only supports the http transport", field) + } + if err := validateHookSchedule(field+".schedule", *hook.Schedule); err != nil { + return err + } + } else if hook.Schedule != nil { + return fmt.Errorf("%s.schedule requires the schedule trigger", field) + } if err := m.validateHookTransport(field, hook.Transport); err != nil { return err } @@ -671,6 +706,50 @@ func (m Manifest) validateContributions() error { return nil } +func validateHookSchedule(field string, schedule HookSchedule) error { + expression := strings.TrimSpace(schedule.Cron) + if expression == "" { + return fmt.Errorf("%s.cron must not be empty", field) + } + // Timezone belongs in its own field. Besides keeping the public shape + // singular, rejecting an inline prefix avoids robfig/cron's malformed-prefix + // panic path and prevents two timezone declarations from disagreeing. + if strings.HasPrefix(expression, "TZ=") || strings.HasPrefix(expression, "CRON_TZ=") { + return fmt.Errorf("%s.cron must not contain an inline timezone", field) + } + parsed, err := scheduleCronParser.Parse(expression) + if err != nil { + return fmt.Errorf("%s.cron must be a standard five-field cron expression: %w", field, err) + } + if strings.TrimSpace(schedule.Timezone) == "" { + return fmt.Errorf("%s.timezone must not be empty", field) + } + location, err := time.LoadLocation(schedule.Timezone) + if err != nil { + return fmt.Errorf("%s.timezone is invalid: %w", field, err) + } + + // A 400-day window covers every month plus a DST cycle. High-frequency + // expressions fail after their first two occurrences; compliant five-minute + // expressions stay below 116k iterations, bounded work at publish/preview + // time rather than on every scheduler tick. A monthly/yearly expression with + // zero or one occurrence in the window is valid by definition. + start := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC).In(location) + end := start.AddDate(0, 0, 400) + previous := parsed.Next(start) + for !previous.IsZero() && !previous.After(end) { + next := parsed.Next(previous) + if next.IsZero() || next.After(end) { + break + } + if next.Sub(previous) < MinimumScheduleInterval { + return fmt.Errorf("%s.cron must not run more often than every five minutes", field) + } + previous = next + } + return nil +} + // validateHookTransport enforces that a hook can only reach a host the manifest // declared through a `net:` scope. Without this the consent screen would not // describe where plugin data actually goes. diff --git a/server/pkg/plugincontract/manifest_test.go b/server/pkg/plugincontract/manifest_test.go index 295391199..71388c448 100644 --- a/server/pkg/plugincontract/manifest_test.go +++ b/server/pkg/plugincontract/manifest_test.go @@ -77,6 +77,61 @@ func TestParseManifestAcceptsReferenceDocument(t *testing.T) { } } +func TestParseManifestAcceptsScheduledHook(t *testing.T) { + raw := mutate(t, func(doc map[string]any) { + contributes := doc["contributes"].(map[string]any) + hook := contributes["hooks"].([]any)[0].(map[string]any) + hook["triggers"] = []any{"schedule", "manual"} + hook["schedule"] = map[string]any{"cron": "*/5 * * * *", "timezone": "Asia/Shanghai"} + }) + manifest, _, err := ParseManifest(raw) + if err != nil { + t.Fatalf("ParseManifest: %v", err) + } + hook := manifest.Contributes.Hooks[0] + if hook.Schedule == nil || hook.Schedule.Cron != "*/5 * * * *" || hook.Schedule.Timezone != "Asia/Shanghai" { + t.Fatalf("schedule = %+v", hook.Schedule) + } +} + +func TestParseManifestValidatesScheduledHookContract(t *testing.T) { + tests := []struct { + name string + triggers []any + schedule any + transport any + want string + }{ + {name: "missing schedule", triggers: []any{"schedule"}, want: "schedule is required"}, + {name: "schedule without trigger", triggers: []any{"manual"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "UTC"}, want: "requires the schedule trigger"}, + {name: "seconds field", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "0 */5 * * * *", "timezone": "UTC"}, want: "five-field"}, + {name: "inline timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "CRON_TZ=UTC */5 * * * *", "timezone": "UTC"}, want: "inline timezone"}, + {name: "missing timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": ""}, want: "timezone must not be empty"}, + {name: "invalid timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "Mars/Olympus"}, want: "timezone is invalid"}, + {name: "too frequent", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/4 * * * *", "timezone": "UTC"}, want: "every five minutes"}, + {name: "mcp transport", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "UTC"}, transport: map[string]any{"type": "mcp", "url": "https://example.com/mcp"}, want: "only supports the http transport"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + raw := mutate(t, func(doc map[string]any) { + contributes := doc["contributes"].(map[string]any) + hook := contributes["hooks"].([]any)[0].(map[string]any) + hook["triggers"] = tc.triggers + if tc.schedule != nil { + hook["schedule"] = tc.schedule + } + if tc.transport != nil { + hook["transport"] = tc.transport + } + }) + _, _, err := ParseManifest(raw) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("error = %v, want it to mention %q", err, tc.want) + } + }) + } +} + func TestConfigSchemaPreservesDeclarationOrder(t *testing.T) { manifest, canonical, err := ParseManifest([]byte(validManifest)) if err != nil { @@ -440,7 +495,7 @@ func TestCheckCapabilitiesReportsEveryUnavailableContribution(t *testing.T) { full := Capabilities{ SurfaceTypes: map[string]bool{SurfaceIssuePanel: true, SurfaceSidebarPanel: true, SurfaceModal: true}, - HookTriggers: map[string]bool{TriggerUI: true, TriggerManual: true, TriggerAgent: true, TriggerEvent: true}, + HookTriggers: map[string]bool{TriggerUI: true, TriggerManual: true, TriggerAgent: true, TriggerEvent: true, TriggerSchedule: true}, HookTransport: map[string]bool{TransportHTTP: true, TransportMCP: true}, ResourceTypes: map[string]bool{ResourceSkill: true}, }