diff --git a/.changeset/approvals-sla-escalation.md b/.changeset/approvals-sla-escalation.md new file mode 100644 index 0000000000..5ee781d65c --- /dev/null +++ b/.changeset/approvals-sla-escalation.md @@ -0,0 +1,9 @@ +--- +"@objectstack/spec": minor +"@objectstack/plugin-approvals": minor +"@objectstack/trigger-record-change": minor +"@objectstack/trigger-schedule": minor +"@objectstack/cli": patch +--- + +ADR-0042 SLA auto-escalation + ADR-0041 mechanical landing. plugin-approvals now owns a jobs-backed escalation scanner (`runEscalations`, interval job `approvals-sla-escalation` + boot catch-up): overdue pending requests escalate **at most once** (the `escalate` audit row is the idempotency marker, written audit-first) executing the node's `escalation.action` — notify / reassign-to-`escalateTo` / auto_approve / auto_reject as the reserved actor `system:sla`. The trigger packages drop their `plugin-` prefix (`@objectstack/trigger-record-change`, `@objectstack/trigger-schedule`) per ADR-0041, and `ActionDescriptor` gains an optional `maturity: 'ga' | 'beta' | 'reserved'` field so designers can grey out contract-ahead-of-runtime surfaces. diff --git a/docs/adr/0042-approval-sla-escalation.md b/docs/adr/0042-approval-sla-escalation.md new file mode 100644 index 0000000000..06b0824610 --- /dev/null +++ b/docs/adr/0042-approval-sla-escalation.md @@ -0,0 +1,146 @@ +# ADR-0042: Approval SLA escalation — a jobs-backed scanner with audit-row idempotency + +**Status**: Proposed (2026-06-12) +**Deciders**: ObjectStack Protocol Architects +**Builds on**: [ADR-0019](./0019-approval-as-flow-node.md) (approval as flow node), [ADR-0041](./0041-flow-trigger-family.md) (triggers vs jobs vs hooks — this is the canonical "jobs, not trigger" case), thread interactions (#1740) +**Closes**: [#1742](https://github.com/objectstack-ai/framework/issues/1742) +**Consumers**: `@objectstack/plugin-approvals` (scanner + execution), `@objectstack/service-job` (the clock), Console inbox (timeline rendering) + +--- + +## TL;DR + +`ApprovalEscalationSchema` (`timeoutHours`, `action: reassign | auto_approve | +auto_reject | notify`, `escalateTo`, `notifySubmitter`) has been contract- +complete for a while, and #1740 made the deadline *visible* (`sla_due_at` +chips). Nothing *executes* it: a breached SLA changes a chip color and +nothing else. + +**Decision**: plugin-approvals owns a periodic scanner registered directly +on the `job` service (`IJobService`) — per ADR-0041's boundary, a plugin- +internal clock is a job, not a trigger. The scanner escalates each overdue +pending request **at most once** (single-shot), using the `escalate` audit +row itself as the idempotency marker — no schema migration. Machine +decisions are recorded under the reserved actor id **`system:sla`**. The +SLA clock runs from `created_at` and does **not** pause during +`request_info` round-trips in v1 (recorded as a future option). + +## Context (verified 2026-06-12) + +- Execution primitives all exist after #1740: `decide()` (with + `context.isSystem` bypassing the approver check), approver-slot + rewriting (reassign), `notify()` via the optional messaging surface, and + the `escalate` member of the audit-action enum (both contract and + `sys_approval_action` select). +- `sla_due_at = created_at + escalation.timeoutHours` is already computed + by the row mapper and rendered by the Console. +- The runtime auto-loads `service-job` (`JobServicePlugin`, registered as + service **`job`**) with interval/cron/once adapters. +- The request table has no escalation bookkeeping column, and the + append-only `sys_approval_action` table is already the audit source of + truth. + +## The three policy decisions + +### 1. Idempotency & multi-instance safety → single-shot, audit-row marker + +An escalation fires **at most once per request, ever**. The marker is the +`escalate` audit row, written **before** any mutation (the audit-first +ordering #1740 established for `reassign`): + +``` +scan: overdue && no prior 'escalate' action → escalate +``` + +Rationale: + +- **No schema migration.** An `escalated_at` column would need a column add + across drivers; the audit row is already durable, queryable, and + tenant-scoped. +- **Crash-safe.** If the process dies between the audit insert and the + mutation, the next scan skips the request (marker present) — the failure + mode is "escalation logged but not executed", which an operator can see + on the timeline, rather than a double `auto_approve`. +- **Multi-instance**: the in-process job adapters run one scheduler per + node. Two nodes could race between the marker check and the insert; the + window is milliseconds and the worst case is a duplicate *notify* (the + mutating actions are guarded again by `decide()`'s pending-status check, + which makes the second decision throw `INVALID_STATE`). A distributed + lock is deliberately out of scope until clustered deployments demand it + (`service-cluster` exists when that day comes). + +Repeat/laddered escalation (re-escalate every N hours, multi-level chains) +is a recorded non-goal for v1 — single-shot covers the dominant "don't let +it rot silently" need. + +### 2. Machine decisions → reserved actor `system:sla` + +`auto_approve` / `auto_reject` call `decide()` with +`actorId: 'system:sla'` under a system context. Consequences: + +- The audit trail shows two rows: `escalate` (what policy fired, with the + configured action in the comment) followed by `approve`/`reject` **by + `system:sla`** — a reader can always distinguish a human decision from a + policy decision. +- Clients render `system:sla` as a localized display name ("System (SLA)") + — it is a reserved identity, never a `sys_user` row. +- Deployments that must forbid machine decisions simply don't author + `auto_approve`/`auto_reject` in their flows; a platform-level kill switch + is deferred until a compliance requirement actually arrives. + +### 3. The clock → `created_at`, no pause in v1 + +The deadline stays `created_at + timeoutHours` — exactly what the Console +already shows. A `request_info` round-trip does **not** pause or re-arm the +clock: the approver who needs more material can see the deadline coming and +the submitter is incentivised to answer fast. ServiceNow-style SLA pause +("clock stops while waiting on the requester") is a future option that +would require per-request clock bookkeeping (schema change) and is exactly +the kind of speculative state this ADR avoids; it gets built when a real +tenant asks. + +## Mechanics + +- **Wiring**: `ApprovalsServicePlugin.start()` resolves the `job` service + (optional, like `messaging`); when present it schedules an interval job + `approvals-sla-escalation` (default every 5 min, + `escalationScanIntervalMs` plugin option) and fires one **catch-up scan + at boot** so restarts don't extend a breach by a scan period. `stop()` + cancels the job. No job service → SLA stays display-only, exactly as + today. +- **Scan** (`runEscalations()`): list pending requests (the same capped + read the inbox uses), keep rows whose node config declares + `escalation.timeoutHours` and whose deadline has passed, skip rows with a + prior `escalate` action, then per action: + - `notify` (default): message pending approvers + `escalateTo`; + - `reassign`: replace `pending_approvers` with `escalateTo` (falls back + to `notify` when `escalateTo` is missing) and notify the new approver; + - `auto_approve` / `auto_reject`: `decide()` as `system:sla` — the owning + flow resumes down the matching branch like any human decision; + - in all cases, notify the submitter when `notifySubmitter !== false`. +- **Failure isolation**: per-request try/catch; one bad row never stops the + sweep. The scan logs `{scanned, escalated}`. + +## Consequences + +- A breached SLA now *does* something, closing the "red chip, no action" + gap — and the timeline explains exactly what and why. +- No migrations, no new services, no new packages: one service method, one + optional wiring block, UI timeline rendering for `escalate` + + `system:sla`. +- The single-shot + audit-marker discipline gives clustered deployments a + bounded, documented race (duplicate notify at worst) instead of a silent + double-decision hazard. +- Future work has clean seams: laddered escalation = relax the marker rule; + SLA pause = add clock bookkeeping; compliance kill-switch = plugin + option. + +## References + +- Schema: `ApprovalEscalationSchema` + (`packages/spec/src/automation/approval.zod.ts`) +- Primitives: `decide` / `notify` / audit-first ordering in + `packages/plugins/plugin-approvals/src/approval-service.ts` (#1740) +- Clock: `IJobService` (`packages/spec/src/contracts/job-service.ts`), + registered as `job` by `@objectstack/service-job` +- Boundary rationale: ADR-0041 §1 (plugin clocks are jobs, not triggers) diff --git a/examples/app-showcase/src/flows/index.ts b/examples/app-showcase/src/flows/index.ts index 70632ce2bb..cc80c99580 100644 --- a/examples/app-showcase/src/flows/index.ts +++ b/examples/app-showcase/src/flows/index.ts @@ -271,7 +271,7 @@ export const TaskCompletedSlackFlow = defineFlow({ * * A `type: 'schedule'` flow whose start node carries an interval descriptor. * The automation engine parses that into a schedule binding; the schedule - * trigger plugin (`@objectstack/plugin-trigger-schedule`, paired with the job + * trigger plugin (`@objectstack/trigger-schedule`, paired with the job * service) registers a job that fires this flow every interval. Each tick runs * the `notify` node, dropping a fresh `sys_inbox_message` row — so the * scheduled fire is observable end-to-end with no manual `engine.execute()`. diff --git a/packages/cli/package.json b/packages/cli/package.json index d83ae497d1..a3fb047b00 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -65,8 +65,8 @@ "@objectstack/plugin-reports": "workspace:*", "@objectstack/plugin-security": "workspace:*", "@objectstack/plugin-sharing": "workspace:*", - "@objectstack/plugin-trigger-record-change": "workspace:*", - "@objectstack/plugin-trigger-schedule": "workspace:*", + "@objectstack/trigger-record-change": "workspace:*", + "@objectstack/trigger-schedule": "workspace:*", "@objectstack/plugin-webhooks": "workspace:*", "@objectstack/rest": "workspace:*", "@objectstack/runtime": "workspace:^", diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 729db77cfb..f73f36e234 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -1422,12 +1422,12 @@ export default class Serve extends Command { // the `FlowTrigger` wiring; these plugins are the concrete triggers: // record-change (ObjectQL lifecycle hooks) + schedule (cron/interval // via the job service — so pair `triggers` with `job`). - pkg: '@objectstack/plugin-trigger-record-change', + pkg: '@objectstack/trigger-record-change', export: 'RecordChangeTriggerPlugin', nameMatch: ['trigger-record-change', 'RecordChangeTriggerPlugin'], extras: [ { - pkg: '@objectstack/plugin-trigger-schedule', + pkg: '@objectstack/trigger-schedule', export: 'ScheduleTriggerPlugin', nameMatch: ['trigger-schedule', 'ScheduleTriggerPlugin'], }, diff --git a/packages/plugins/plugin-approvals/src/approval-service.test.ts b/packages/plugins/plugin-approvals/src/approval-service.test.ts index ccf2519df4..d178479c00 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.test.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.test.ts @@ -473,6 +473,82 @@ describe('ApprovalService (node era)', () => { expect(actions.filter(a => a.action === 'comment')).toHaveLength(2); }); + // ── SLA escalation (ADR-0042) ─────────────────────────────────── + + function makeOverdue(reqId: string) { + // Push created_at into the past so a small timeoutHours is breached. + const row = engine._tables['sys_approval_request'].find(r => r.id === reqId)!; + row.created_at = new Date(baseTime - 10 * 3600_000).toISOString(); + } + + it('runEscalations: notify action messages approvers + escalateTo + submitter, once', async () => { + const emitted: any[] = []; + svc.attachMessaging({ async emit(input) { emitted.push(input); } }); + const req = await svc.openNodeRequest( + openInput(['u9'], {}, { escalation: { timeoutHours: 2, action: 'notify', escalateTo: 'boss', notifySubmitter: true } }), CTX, + ); + makeOverdue(req.id); + const first = await svc.runEscalations(); + expect(first.escalated).toBe(1); + expect(emitted.map(e => e.topic)).toEqual(['approval.sla_breached', 'approval.sla_breached']); + expect(emitted[0].audience).toEqual(['u9', 'boss']); + expect(emitted[1].audience).toEqual(['u1']); // submitter + const actions = await svc.listActions(req.id, SYS); + expect(actions.at(-1)).toMatchObject({ action: 'escalate', actor_id: 'system:sla', comment: 'notify → boss' }); + // Single-shot: second sweep is a no-op. + const second = await svc.runEscalations(); + expect(second.escalated).toBe(0); + expect(emitted).toHaveLength(2); + }); + + it('runEscalations: auto_approve decides as system:sla and resumes the flow', async () => { + const resumed: any[] = []; + svc.attachAutomation({ async resume(runId, signal) { resumed.push({ runId, signal }); } }); + const req = await svc.openNodeRequest( + openInput(['u9'], {}, { escalation: { timeoutHours: 1, action: 'auto_approve', notifySubmitter: false } }), CTX, + ); + makeOverdue(req.id); + const out = await svc.runEscalations(); + expect(out.escalated).toBe(1); + const fresh = await svc.getRequest(req.id, SYS); + expect(fresh?.status).toBe('approved'); + expect(resumed[0]).toMatchObject({ runId: 'run_1', signal: { branchLabel: 'approve' } }); + const actions = await svc.listActions(req.id, SYS); + expect(actions.map(a => a.action)).toEqual(['submit', 'escalate', 'approve']); + expect(actions.at(-1)?.actor_id).toBe('system:sla'); + }); + + it('runEscalations: auto_reject decides as system:sla', async () => { + const req = await svc.openNodeRequest( + openInput(['u9'], {}, { escalation: { timeoutHours: 1, action: 'auto_reject', notifySubmitter: false } }), CTX, + ); + makeOverdue(req.id); + await svc.runEscalations(); + const fresh = await svc.getRequest(req.id, SYS); + expect(fresh?.status).toBe('rejected'); + }); + + it('runEscalations: reassign replaces the approver set with escalateTo', async () => { + const req = await svc.openNodeRequest( + openInput(['u9', 'u2'], {}, { escalation: { timeoutHours: 1, action: 'reassign', escalateTo: 'boss', notifySubmitter: false } }), CTX, + ); + makeOverdue(req.id); + await svc.runEscalations(); + const fresh = await svc.getRequest(req.id, SYS); + expect(fresh?.status).toBe('pending'); + expect(fresh?.pending_approvers).toEqual(['boss']); + }); + + it('runEscalations: skips requests that are not yet due or have no SLA', async () => { + await svc.openNodeRequest( + openInput(['u9'], {}, { escalation: { timeoutHours: 1000, action: 'auto_approve' } }), CTX, + ); + await svc.openNodeRequest(openInput(['u9'], { recordId: 'opp2', record: { id: 'opp2' } }), CTX); + const out = await svc.runEscalations(); + expect(out.scanned).toBe(2); + expect(out.escalated).toBe(0); + }); + // ── SLA + flow steps ──────────────────────────────────────────── it('rows expose sla_due_at when the node declares escalation.timeoutHours', async () => { diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index e3ee47aebc..2e94504497 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -71,6 +71,13 @@ export interface ApprovalMessagingSurface { /** Minimum time between submitter reminders on one request. */ export const REMIND_COOLDOWN_MS = 4 * 60 * 60 * 1000; +/** Named job under which the SLA escalation scan is registered (ADR-0042). */ +export const ESCALATION_JOB_NAME = 'approvals-sla-escalation'; +/** Default interval between SLA escalation scans. */ +export const ESCALATION_SCAN_INTERVAL_MS = 5 * 60 * 1000; +/** Reserved actor id for machine decisions made by the SLA scanner. */ +export const SLA_ACTOR_ID = 'system:sla'; + const SYSTEM_CTX = { isSystem: true, roles: [], permissions: [] } as const; function uid(prefix: string): string { @@ -827,6 +834,125 @@ export class ApprovalService implements IApprovalService { return { request: fresh! }; } + // ── SLA escalation (ADR-0042) ───────────────────────────────── + + /** + * One escalation sweep: every *pending* request whose node config declares + * `escalation.timeoutHours` and whose deadline has passed is escalated + * **at most once, ever** — the `escalate` audit row is the idempotency + * marker, written before any mutation (audit-first, like reassign). One + * bad row never stops the sweep. + */ + async runEscalations(): Promise<{ scanned: number; escalated: number }> { + let rows: any[] = []; + try { + rows = await this.engine.find('sys_approval_request', { + where: { status: 'pending' }, limit: 500, context: SYSTEM_CTX, + }) ?? []; + } catch (err: any) { + this.logger?.warn?.('[approvals] escalation scan failed to list requests', { + error: err?.message ?? String(err), + }); + return { scanned: 0, escalated: 0 }; + } + + let escalated = 0; + for (const raw of rows) { + try { + const cfg = parseJson(raw.node_config_json, undefined); + const esc = cfg?.escalation; + if (!esc || typeof esc.timeoutHours !== 'number' || esc.timeoutHours <= 0) continue; + const due = slaDueAt(raw.created_at, cfg); + if (!due || Date.parse(due) > this.clock.now().getTime()) continue; + + // Single-shot: a prior 'escalate' action means this request is done. + const prior = await this.engine.find('sys_approval_action', { + where: { request_id: raw.id, action: 'escalate' }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(prior) && prior[0]) continue; + + await this.escalateRequest(raw, esc); + escalated++; + } catch (err: any) { + this.logger?.warn?.('[approvals] escalation failed for request', { + request: raw?.id, error: err?.message ?? String(err), + }); + } + } + if (escalated > 0) { + this.logger?.info?.('[approvals] SLA escalation sweep', { scanned: rows.length, escalated }); + } + return { scanned: rows.length, escalated }; + } + + /** Execute the configured escalation action for one overdue request. */ + private async escalateRequest(raw: any, esc: any): Promise { + const action: string = esc.action ?? 'notify'; + const escalateTo: string | undefined = + typeof esc.escalateTo === 'string' && esc.escalateTo.trim() ? esc.escalateTo.trim() : undefined; + const now = this.clock.now().toISOString(); + const pending = csvSplit(raw.pending_approvers); + + // Audit first — this row IS the idempotency marker (ADR-0042 §1). + await this.engine.insert('sys_approval_action', { + id: uid('aact'), request_id: raw.id, organization_id: raw.organization_id ?? null, + step_name: raw.flow_node_id ?? raw.current_step ?? null, step_index: 0, action: 'escalate', + actor_id: SLA_ACTOR_ID, + comment: `${action}${escalateTo ? ` → ${escalateTo}` : ''}`, + created_at: now, + }, { context: SYSTEM_CTX }); + + if (action === 'reassign' && escalateTo) { + await this.engine.update('sys_approval_request', { + id: raw.id, pending_approvers: escalateTo, updated_at: now, + }, { context: SYSTEM_CTX }); + await this.notify({ + topic: 'approval.escalated', + audience: [escalateTo], + actorId: SLA_ACTOR_ID, + source: { object: 'sys_approval_request', id: raw.id }, + payload: { + title: 'Approval escalated to you', + message: `An overdue approval on ${raw.object_name}/${raw.record_id} was escalated to you.`, + actionUrl: '/system/approvals', + }, + }); + } else if (action === 'auto_approve' || action === 'auto_reject') { + await this.decide(raw.id, { + decision: action === 'auto_approve' ? 'approve' : 'reject', + actorId: SLA_ACTOR_ID, + comment: 'SLA escalation', + }, SYSTEM_CTX as unknown as SharingExecutionContext); + } else { + // 'notify' (and the reassign-without-target fallback) + await this.notify({ + topic: 'approval.sla_breached', + audience: [...pending, ...(escalateTo ? [escalateTo] : [])], + actorId: SLA_ACTOR_ID, + source: { object: 'sys_approval_request', id: raw.id }, + payload: { + title: 'Approval SLA breached', + message: `A decision on ${raw.object_name}/${raw.record_id} is overdue.`, + actionUrl: '/system/approvals', + }, + }); + } + + if (esc.notifySubmitter !== false && raw.submitter_id) { + await this.notify({ + topic: 'approval.sla_breached', + audience: [String(raw.submitter_id)], + actorId: SLA_ACTOR_ID, + source: { object: 'sys_approval_request', id: raw.id }, + payload: { + title: 'Your approval request breached its SLA', + message: `${raw.object_name}/${raw.record_id}: escalation action '${action}' was taken.`, + actionUrl: '/system/approvals', + }, + }); + } + } + // ── Display enrichment ─────────────────────────────────────── /** diff --git a/packages/plugins/plugin-approvals/src/approvals-plugin.ts b/packages/plugins/plugin-approvals/src/approvals-plugin.ts index ca73e1ee38..a56836f294 100644 --- a/packages/plugins/plugin-approvals/src/approvals-plugin.ts +++ b/packages/plugins/plugin-approvals/src/approvals-plugin.ts @@ -3,13 +3,24 @@ import type { Plugin, PluginContext } from '@objectstack/core'; import { SysApprovalRequest } from './sys-approval-request.object.js'; import { SysApprovalAction } from './sys-approval-action.object.js'; -import { ApprovalService, type ApprovalEngine } from './approval-service.js'; +import { + ApprovalService, + ESCALATION_JOB_NAME, + ESCALATION_SCAN_INTERVAL_MS, + type ApprovalEngine, +} from './approval-service.js'; import { bindApprovalLockHook, unbindAllHooks } from './lifecycle-hooks.js'; import { registerApprovalNode, type ApprovalAutomationSurface } from './approval-node.js'; export interface ApprovalsPluginOptions { /** Disable runtime registration (schemas still register). */ disableService?: boolean; + /** + * Interval between SLA escalation scans (ADR-0042). Defaults to + * {@link ESCALATION_SCAN_INTERVAL_MS} (5 min). Only takes effect when a + * `job` service is installed; without one, SLA stays display-only. + */ + escalationScanIntervalMs?: number; /** * Disable the record-lock hook. Schema + service stay intact; only the * engine-level lock wiring is suppressed. Useful when a caller wants the @@ -36,6 +47,7 @@ export class ApprovalsServicePlugin implements Plugin { private readonly options: ApprovalsPluginOptions; private service?: ApprovalService; private engine?: any; + private escalationJobScheduled = false; constructor(options: ApprovalsPluginOptions = {}) { this.options = options; @@ -123,6 +135,33 @@ export class ApprovalsServicePlugin implements Plugin { } } catch { /* messaging not installed */ } + // SLA escalation clock (ADR-0042): a plugin-internal job, deliberately + // NOT a flow trigger (ADR-0041 §1). Interval sweep + one catch-up scan at + // boot so a restart doesn't extend a breach by a scan period. Wired on + // kernel:ready — the job service may start after this plugin. No `job` + // service → SLA stays display-only. + const wireEscalationClock = async () => { + try { + const jobs = ctx.getService('job'); + if (!jobs || typeof jobs.schedule !== 'function' || !this.service) return; + const svc = this.service; + const intervalMs = this.options.escalationScanIntervalMs ?? ESCALATION_SCAN_INTERVAL_MS; + await jobs.schedule(ESCALATION_JOB_NAME, { type: 'interval', intervalMs }, async () => { + await svc.runEscalations(); + }); + this.escalationJobScheduled = true; + void svc.runEscalations().catch((err: any) => { + ctx.logger.warn?.('[approvals] boot escalation sweep failed', { error: err?.message }); + }); + ctx.logger.info('ApprovalsServicePlugin: SLA escalation scan scheduled', { intervalMs }); + } catch { /* job service not installed */ } + }; + if (typeof (ctx as any).hook === 'function') { + (ctx as any).hook('kernel:ready', wireEscalationClock); + } else { + await wireEscalationClock(); + } + // ADR-0019: contribute the `approval` node to the flow engine when one is // present. The node lets a flow suspend on an approval and resume on // decision; the service is wired to the same engine so `decide()` can @@ -138,7 +177,14 @@ export class ApprovalsServicePlugin implements Plugin { } } - async stop(_ctx: PluginContext): Promise { + async stop(ctx: PluginContext): Promise { + if (this.escalationJobScheduled) { + try { + const jobs = ctx.getService('job'); + await jobs?.cancel?.(ESCALATION_JOB_NAME); + } catch { /* ignore */ } + this.escalationJobScheduled = false; + } if (this.engine) { try { unbindAllHooks(this.engine); } catch { /* ignore */ } } diff --git a/packages/plugins/trigger-record-change/CHANGELOG.md b/packages/plugins/trigger-record-change/CHANGELOG.md new file mode 100644 index 0000000000..b53c9dde91 --- /dev/null +++ b/packages/plugins/trigger-record-change/CHANGELOG.md @@ -0,0 +1,178 @@ +# @objectstack/plugin-trigger-record-change + +## 9.2.0 + +### Patch Changes + +- Updated dependencies [2f57b75] +- Updated dependencies [2f57b75] + - @objectstack/spec@9.2.0 + - @objectstack/core@9.2.0 + +## 9.1.0 + +### Patch Changes + +- Updated dependencies [b9062c9] + - @objectstack/spec@9.1.0 + - @objectstack/core@9.1.0 + +## 9.0.1 + +### Patch Changes + +- Updated dependencies [1817845] + - @objectstack/spec@9.0.1 + - @objectstack/core@9.0.1 + +## 9.0.0 + +### Patch Changes + +- Updated dependencies [4c3f693] +- Updated dependencies [0bf39f1] +- Updated dependencies [f533f42] +- Updated dependencies [1c83ee8] + - @objectstack/spec@9.0.0 + - @objectstack/core@9.0.0 + +## 8.0.1 + +### Patch Changes + +- @objectstack/spec@8.0.1 +- @objectstack/core@8.0.1 + +## 8.0.0 + +### Patch Changes + +- Updated dependencies [a46c017] +- Updated dependencies [b990b89] +- Updated dependencies [99111ec] +- Updated dependencies [d5a8161] +- Updated dependencies [5cf1f1b] +- Updated dependencies [9ef89d4] +- Updated dependencies [3306d2f] +- Updated dependencies [c262301] +- Updated dependencies [bc44195] +- Updated dependencies [9e2e229] + - @objectstack/spec@8.0.0 + - @objectstack/core@8.0.0 + +## 7.9.0 + +### Patch Changes + +- @objectstack/spec@7.9.0 +- @objectstack/core@7.9.0 + +## 7.8.0 + +### Patch Changes + +- Updated dependencies [06f2bbb] +- Updated dependencies [36719db] +- Updated dependencies [424ab26] + - @objectstack/spec@7.8.0 + - @objectstack/core@7.8.0 + +## 7.7.0 + +### Patch Changes + +- Updated dependencies [b391955] +- Updated dependencies [f06b64e] +- Updated dependencies [023bf93] + - @objectstack/spec@7.7.0 + - @objectstack/core@7.7.0 + +## 7.6.0 + +### Patch Changes + +- Updated dependencies [955d4c8] +- Updated dependencies [c4a4cbd] +- Updated dependencies [b046ec2] +- Updated dependencies [2170ad9] +- Updated dependencies [02d6359] +- Updated dependencies [7648242] +- Updated dependencies [8fa1e7f] +- Updated dependencies [55866f5] +- Updated dependencies [60f9c45] + - @objectstack/spec@7.6.0 + - @objectstack/core@7.6.0 + +## 7.5.0 + +### Patch Changes + +- @objectstack/spec@7.5.0 +- @objectstack/core@7.5.0 + +## 7.4.1 + +### Patch Changes + +- @objectstack/spec@7.4.1 +- @objectstack/core@7.4.1 + +## 7.4.0 + +### Minor Changes + +- 13d8653: Record-change flow trigger — auto-launch flows on data mutations. + + Completes the automation engine's `FlowTrigger` extension point so flows whose + `start` node declares a record-change trigger (`config: { objectName, +triggerType: 'record-after-update', condition }`) actually fire on the matching + mutation. Previously the slot was dead — nothing called `trigger.start` — so + such flows could only run via a manual `engine.execute()`. + + **Engine baseline (`@objectstack/service-automation`)** + + - Redefines `FlowTrigger` around a parsed `FlowTriggerBinding` (flowName, + object, event, condition, schedule, raw config). The engine parses the start + node and hands the trigger a normalized binding, keeping trigger plugins + decoupled from flow-definition internals (mirrors `connector_action` ↔ + `connector-rest`). + - Ordering-independent, bidirectional wiring: `registerFlow`/`toggleFlow` + activate bindings; `registerTrigger` retro-binds already-registered flows (a + trigger plugin wires up on `kernel:ready`, after flows are pulled in); + `unregisterFlow`/`unregisterTrigger`/disable tear them down. + - Centralized start-condition gate in `execute()`: the start node's `condition` + (e.g. `status == 'done' && previous.status != 'done'`) is evaluated once for + every trigger type and manual runs; false ⇒ `{ skipped: true }`. + - Seeds `record`, flattened record fields, and `previous` into flow variables. + - New `getActiveTriggerBindings()` getter + exports `FlowTriggerBinding`. + + **Spec (`@objectstack/spec`)** + + - Adds `previous?` to `AutomationContext` — the pre-update "old" row, so flows + can gate on transitions. + + **New package (`@objectstack/plugin-trigger-record-change`)** + + - The concrete trigger: subscribes to ObjectQL lifecycle hooks + (`record-after-update` → `afterUpdate`, etc.), builds an `AutomationContext` + from the new/old record, and runs the flow. Error-isolated (a flow failure + never breaks the CRUD write); graceful degrade when the automation service or + ObjectQL engine is absent (mirrors `plugin-audit`). + + The `schedule` trigger (ticker/cron + `sys_job` lifecycle) is a follow-up. + +### Patch Changes + +- Updated dependencies [23c7107] +- Updated dependencies [c72daad] +- Updated dependencies [f115182] +- Updated dependencies [2faf9f2] +- Updated dependencies [2faf9f2] +- Updated dependencies [2faf9f2] +- Updated dependencies [58b450b] +- Updated dependencies [82eb6cf] +- Updated dependencies [13d8653] +- Updated dependencies [ff3d006] +- Updated dependencies [5e831de] + - @objectstack/spec@7.4.0 + - @objectstack/core@7.4.0 diff --git a/packages/plugins/trigger-record-change/README.md b/packages/plugins/trigger-record-change/README.md new file mode 100644 index 0000000000..afc4248ae8 --- /dev/null +++ b/packages/plugins/trigger-record-change/README.md @@ -0,0 +1,63 @@ +# @objectstack/plugin-trigger-record-change + +Auto-launch ObjectStack flows on record changes. + +The automation engine ships the `FlowTrigger` extension point and the wiring +that turns a flow's `start` node into a normalized trigger binding — but the +*concrete* record-change trigger lives here, as a plugin. This mirrors the +connector split (engine baseline + `connector-rest` plugin) and reuses the same +`kernel:ready` → `getService('objectql')` pattern `plugin-audit` uses to reach +the data engine's lifecycle-hook surface. + +## What it does + +A flow whose `start` node declares a record-change trigger: + +```ts +{ + type: 'start', + config: { + objectName: 'showcase_task', + triggerType: 'record-after-update', + condition: "status == 'done' && previous.status != 'done'", // optional + }, +} +``` + +auto-launches whenever a matching mutation happens — no manual +`engine.execute()`. The engine evaluates the optional `condition` (the start +node gate) before running the flow, with `record` (the new row) and `previous` +(the old row) in scope. + +### Trigger event → hook mapping + +| `triggerType` | ObjectQL hook | +| ------------------------ | -------------- | +| `record-after-create` | `afterInsert` | +| `record-after-update` | `afterUpdate` | +| `record-after-delete` | `afterDelete` | +| `record-before-create` | `beforeInsert` | +| `record-before-update` | `beforeUpdate` | +| `record-before-delete` | `beforeDelete` | + +## Usage + +```ts +import { AutomationServicePlugin } from '@objectstack/service-automation'; +import { MessagingServicePlugin } from '@objectstack/service-messaging'; +import { RecordChangeTriggerPlugin } from '@objectstack/plugin-trigger-record-change'; + +kernel + .use(new AutomationServicePlugin()) // engine + flows + .use(new MessagingServicePlugin()) // notify channels (optional) + .use(new RecordChangeTriggerPlugin()); // ← makes record-change flows live +``` + +Requires the ObjectQL engine (`com.objectstack.engine.objectql`). If either the +automation service or the data engine is unavailable at `kernel:ready`, the +plugin logs a warning and no-ops rather than failing startup. + +## Error isolation + +A flow that throws during a triggered run is logged and swallowed — it never +breaks the CRUD write that triggered it. diff --git a/packages/plugins/trigger-record-change/package.json b/packages/plugins/trigger-record-change/package.json new file mode 100644 index 0000000000..31f0cd8e57 --- /dev/null +++ b/packages/plugins/trigger-record-change/package.json @@ -0,0 +1,54 @@ +{ + "name": "@objectstack/trigger-record-change", + "version": "9.2.0", + "license": "Apache-2.0", + "description": "Record-change flow trigger for ObjectStack — auto-launches flows on object insert/update/delete via ObjectQL lifecycle hooks (ADR-0018)", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.mjs", + "require": "./dist/index.js" + } + }, + "scripts": { + "build": "tsup --config ../../../tsup.config.ts", + "test": "vitest run --passWithNoTests" + }, + "dependencies": { + "@objectstack/core": "workspace:*", + "@objectstack/spec": "workspace:*" + }, + "devDependencies": { + "@types/node": "^25.9.2", + "typescript": "^6.0.3", + "vitest": "^4.1.8" + }, + "keywords": [ + "objectstack", + "plugin", + "automation", + "flow", + "trigger", + "record-change" + ], + "author": "ObjectStack", + "repository": { + "type": "git", + "url": "https://github.com/objectstack-ai/framework.git", + "directory": "packages/plugins/plugin-trigger-record-change" + }, + "homepage": "https://objectstack.ai/docs", + "bugs": "https://github.com/objectstack-ai/framework/issues", + "publishConfig": { + "access": "public" + }, + "files": [ + "dist", + "README.md" + ], + "engines": { + "node": ">=18.0.0" + } +} diff --git a/packages/plugins/trigger-record-change/src/index.ts b/packages/plugins/trigger-record-change/src/index.ts new file mode 100644 index 0000000000..831d477a91 --- /dev/null +++ b/packages/plugins/trigger-record-change/src/index.ts @@ -0,0 +1,13 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +export { RecordChangeTriggerPlugin } from './plugin.js'; +export { + RecordChangeTrigger, + triggerTypeToHookEvent, +} from './record-change-trigger.js'; +export type { + FlowTrigger, + FlowTriggerBinding, + RecordChangeDataEngine, + TriggerLogger, +} from './record-change-trigger.js'; diff --git a/packages/plugins/trigger-record-change/src/plugin.ts b/packages/plugins/trigger-record-change/src/plugin.ts new file mode 100644 index 0000000000..239721423c --- /dev/null +++ b/packages/plugins/trigger-record-change/src/plugin.ts @@ -0,0 +1,85 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { Plugin, PluginContext } from '@objectstack/core'; +import { RecordChangeTrigger } from './record-change-trigger.js'; +import type { FlowTrigger, RecordChangeDataEngine } from './record-change-trigger.js'; + +/** + * The slice of the automation engine this plugin needs: register a trigger on + * its `FlowTrigger` extension point. Declared structurally so the plugin does + * not take a build dependency on `@objectstack/service-automation`. + */ +interface AutomationTriggerRegistry { + registerTrigger(trigger: FlowTrigger): void; + unregisterTrigger?(type: string): void; +} + +/** + * RecordChangeTriggerPlugin + * + * Makes record-change-triggered flows actually fire. The automation engine + * ships the `FlowTrigger` wiring (it parses each flow's start node into a + * binding and calls `trigger.start(...)`), but the *concrete* record-change + * trigger — the one that subscribes to ObjectQL lifecycle hooks — lives here as + * a plugin. This mirrors the connector split (engine baseline + connector-rest + * plugin) and reuses plugin-audit's `kernel:ready` → `getService('objectql')` + * pattern to reach the data engine's hook surface. + * + * With this plugin installed, a flow whose start node declares + * `config: { objectName, triggerType: 'record-after-update', condition }` + * auto-launches on the matching mutation — no manual `engine.execute()`. + */ +export class RecordChangeTriggerPlugin implements Plugin { + name = 'com.objectstack.trigger.record-change'; + type = 'standard'; + version = '7.3.0'; + dependencies = ['com.objectstack.engine.objectql']; + + async init(ctx: PluginContext): Promise { + ctx.logger.info('Record-change trigger plugin initialized'); + } + + async start(ctx: PluginContext): Promise { + // ObjectQL engine + the automation service are only resolvable once the + // kernel is ready (kernel:ready fires after AutomationServicePlugin.start() + // has pulled flows into the engine, so binding order is correct). + ctx.hook('kernel:ready', async () => { + const automation = this.resolveService(ctx, 'automation'); + if (!automation || typeof automation.registerTrigger !== 'function') { + ctx.logger.warn( + 'RecordChangeTriggerPlugin: automation service not available — record-change trigger NOT installed', + ); + return; + } + + const engine = this.resolveDataEngine(ctx); + if (!engine || typeof engine.registerHook !== 'function') { + ctx.logger.warn( + 'RecordChangeTriggerPlugin: ObjectQL engine not available — record-change trigger NOT installed', + ); + return; + } + + const trigger = new RecordChangeTrigger(engine, ctx.logger); + automation.registerTrigger(trigger); + ctx.logger.info('RecordChangeTriggerPlugin: record-change trigger registered'); + }); + } + + private resolveService(ctx: PluginContext, name: string): T | null { + try { + return ctx.getService(name) ?? null; + } catch { + return null; + } + } + + private resolveDataEngine(ctx: PluginContext): RecordChangeDataEngine | null { + // Primary alias 'objectql', fallback 'data' (some kernels register the + // engine under both) — same lookup plugin-audit uses. + return ( + this.resolveService(ctx, 'objectql') ?? + this.resolveService(ctx, 'data') + ); + } +} diff --git a/packages/plugins/trigger-record-change/src/record-change-trigger.test.ts b/packages/plugins/trigger-record-change/src/record-change-trigger.test.ts new file mode 100644 index 0000000000..2322691384 --- /dev/null +++ b/packages/plugins/trigger-record-change/src/record-change-trigger.test.ts @@ -0,0 +1,293 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, vi } from 'vitest'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { HookContext } from '@objectstack/spec/data'; +import { + RecordChangeTrigger, + triggerTypeToHookEvent, + type FlowTriggerBinding, + type RecordChangeDataEngine, + type TriggerLogger, +} from './record-change-trigger.js'; +import { RecordChangeTriggerPlugin } from './plugin.js'; + +// ─── Test doubles ─────────────────────────────────────────────────── + +interface RegisteredHook { + event: string; + handler: (ctx: HookContext) => unknown | Promise; + object?: string | string[]; + packageId?: string; +} + +/** Fake ObjectQL engine: records registerHook calls + supports unregister. */ +function fakeEngine() { + const hooks: RegisteredHook[] = []; + const engine: RecordChangeDataEngine = { + registerHook(event, handler, options) { + hooks.push({ event, handler, object: options?.object, packageId: options?.packageId }); + }, + unregisterHooksByPackage(packageId: string) { + const before = hooks.length; + for (let i = hooks.length - 1; i >= 0; i--) { + if (hooks[i].packageId === packageId) hooks.splice(i, 1); + } + return before - hooks.length; + }, + }; + return { engine, hooks }; +} + +function silentLogger(): TriggerLogger { + return { info: () => {}, warn: () => {}, debug: () => {} }; +} + +function binding(overrides: Partial = {}): FlowTriggerBinding { + return { + flowName: 'task_assigned_notify', + object: 'showcase_task', + event: 'record-after-update', + ...overrides, + }; +} + +function hookCtx(overrides: Partial = {}): HookContext { + return { + object: 'showcase_task', + event: 'afterUpdate', + input: { id: 't1', doc: { status: 'done' } }, + result: { _id: 't1', status: 'done', assignee: 'u2' }, + previous: { _id: 't1', status: 'open', assignee: 'u1' }, + session: { userId: 'u9' }, + ql: {}, + ...overrides, + } as HookContext; +} + +// ─── triggerTypeToHookEvent ───────────────────────────────────────── + +describe('triggerTypeToHookEvent', () => { + it('maps after-create / after-update / after-delete', () => { + expect(triggerTypeToHookEvent('record-after-create')).toBe('afterInsert'); + expect(triggerTypeToHookEvent('record-after-update')).toBe('afterUpdate'); + expect(triggerTypeToHookEvent('record-after-delete')).toBe('afterDelete'); + }); + + it('maps before-* variants', () => { + expect(triggerTypeToHookEvent('record-before-create')).toBe('beforeInsert'); + expect(triggerTypeToHookEvent('record-before-update')).toBe('beforeUpdate'); + expect(triggerTypeToHookEvent('record-before-delete')).toBe('beforeDelete'); + }); + + it('treats record-*-insert as Insert too', () => { + expect(triggerTypeToHookEvent('record-after-insert')).toBe('afterInsert'); + }); + + it('returns null for unsupported / missing tokens', () => { + expect(triggerTypeToHookEvent(undefined)).toBeNull(); + expect(triggerTypeToHookEvent('schedule')).toBeNull(); + expect(triggerTypeToHookEvent('record-after-frobnicate')).toBeNull(); + expect(triggerTypeToHookEvent('on_update')).toBeNull(); + }); +}); + +// ─── RecordChangeTrigger ──────────────────────────────────────────── + +describe('RecordChangeTrigger', () => { + it('registers a hook for the mapped event, filtered to the object', () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + + trigger.start(binding(), async () => {}); + + expect(hooks).toHaveLength(1); + expect(hooks[0].event).toBe('afterUpdate'); + expect(hooks[0].object).toBe('showcase_task'); + expect(hooks[0].packageId).toBe('com.objectstack.trigger.record-change:task_assigned_notify'); + }); + + it('does not register a hook for an unsupported trigger event', () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + + trigger.start(binding({ event: 'schedule' }), async () => {}); + + expect(hooks).toHaveLength(0); + }); + + it('fires the callback with a record context built from the hook ctx', async () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + const seen: AutomationContext[] = []; + + trigger.start(binding(), async (ctx) => { + seen.push(ctx); + }); + + await hooks[0].handler(hookCtx()); + + expect(seen).toHaveLength(1); + const ctx = seen[0]; + expect(ctx.object).toBe('showcase_task'); + expect(ctx.event).toBe('record-after-update'); + expect(ctx.userId).toBe('u9'); + // new record = ctx.result + expect(ctx.record).toEqual({ _id: 't1', status: 'done', assignee: 'u2' }); + // old record = ctx.previous + expect(ctx.previous).toEqual({ _id: 't1', status: 'open', assignee: 'u1' }); + // record exposed as params too + expect(ctx.params).toEqual(ctx.record); + }); + + it('falls back to input.doc when result is absent (e.g. before-hooks)', async () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + let captured: AutomationContext | undefined; + + trigger.start(binding({ event: 'record-before-update' }), async (ctx) => { + captured = ctx; + }); + + await hooks[0].handler(hookCtx({ event: 'beforeUpdate', result: undefined })); + + expect(captured?.record).toEqual({ status: 'done' }); + }); + + it('reads the __previous stash when ctx.previous is absent', async () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + let captured: AutomationContext | undefined; + + trigger.start(binding(), async (ctx) => { + captured = ctx; + }); + + const ctx = hookCtx({ previous: undefined }); + (ctx as unknown as { __previous: Record }).__previous = { status: 'old' }; + await hooks[0].handler(ctx); + + expect(captured?.previous).toEqual({ status: 'old' }); + }); + + it('isolates flow errors so the CRUD write is never broken', async () => { + const { engine, hooks } = fakeEngine(); + const warn = vi.fn(); + const trigger = new RecordChangeTrigger(engine, { info: () => {}, warn, debug: () => {} }); + + trigger.start(binding(), async () => { + throw new Error('flow blew up'); + }); + + // Must resolve, not reject. + await expect(hooks[0].handler(hookCtx())).resolves.toBeUndefined(); + expect(warn).toHaveBeenCalled(); + }); + + it('stop() unregisters exactly that flow\'s hook', () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + + trigger.start(binding({ flowName: 'flow_a' }), async () => {}); + trigger.start(binding({ flowName: 'flow_b' }), async () => {}); + expect(hooks).toHaveLength(2); + + trigger.stop('flow_a'); + expect(hooks).toHaveLength(1); + expect(hooks[0].packageId).toBe('com.objectstack.trigger.record-change:flow_b'); + }); + + it('re-binding the same flow is idempotent (no duplicate hooks)', () => { + const { engine, hooks } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + + trigger.start(binding(), async () => {}); + trigger.start(binding(), async () => {}); + + expect(hooks).toHaveLength(1); + }); + + it('stop() on an unknown flow is a no-op', () => { + const { engine } = fakeEngine(); + const trigger = new RecordChangeTrigger(engine, silentLogger()); + expect(() => trigger.stop('never_bound')).not.toThrow(); + }); +}); + +// ─── RecordChangeTriggerPlugin ────────────────────────────────────── + +describe('RecordChangeTriggerPlugin', () => { + interface FakeCtx { + services: Record; + readyHandlers: Array<() => Promise | void>; + ctx: { + logger: TriggerLogger; + getService: (name: string) => T; + hook: (event: string, handler: () => Promise | void) => void; + }; + } + + function fakePluginCtx(services: Record): FakeCtx { + const readyHandlers: Array<() => Promise | void> = []; + return { + services, + readyHandlers, + ctx: { + logger: silentLogger() as TriggerLogger, + getService(name: string): T { + if (!(name in services)) throw new Error(`no service '${name}'`); + return services[name] as T; + }, + hook(event: string, handler: () => Promise | void) { + if (event === 'kernel:ready') readyHandlers.push(handler); + }, + }, + }; + } + + it('registers the trigger on the automation service when both services exist', async () => { + const registerTrigger = vi.fn(); + const { engine } = fakeEngine(); + const fake = fakePluginCtx({ automation: { registerTrigger }, objectql: engine }); + + const plugin = new RecordChangeTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + expect(registerTrigger).toHaveBeenCalledTimes(1); + const trigger = registerTrigger.mock.calls[0][0] as RecordChangeTrigger; + expect(trigger.type).toBe('record_change'); + }); + + it('skips gracefully when the automation service is absent', async () => { + const { engine } = fakeEngine(); + const fake = fakePluginCtx({ objectql: engine }); + + const plugin = new RecordChangeTriggerPlugin(); + await plugin.start(fake.ctx as never); + await expect(fake.readyHandlers[0]()).resolves.toBeUndefined(); + }); + + it('skips gracefully when the ObjectQL engine is absent', async () => { + const registerTrigger = vi.fn(); + const fake = fakePluginCtx({ automation: { registerTrigger } }); + + const plugin = new RecordChangeTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + expect(registerTrigger).not.toHaveBeenCalled(); + }); + + it('falls back to the "data" engine alias', async () => { + const registerTrigger = vi.fn(); + const { engine } = fakeEngine(); + const fake = fakePluginCtx({ automation: { registerTrigger }, data: engine }); + + const plugin = new RecordChangeTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + expect(registerTrigger).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/plugins/trigger-record-change/src/record-change-trigger.ts b/packages/plugins/trigger-record-change/src/record-change-trigger.ts new file mode 100644 index 0000000000..36a005d77e --- /dev/null +++ b/packages/plugins/trigger-record-change/src/record-change-trigger.ts @@ -0,0 +1,186 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { HookContext } from '@objectstack/spec/data'; + +/** + * Structural mirror of the automation engine's `FlowTriggerBinding` + * (service-automation/src/engine.ts). Declared locally so this trigger plugin + * stays decoupled from the automation package — same pattern the connector / + * messaging integrations use to avoid a hard build edge. The engine parses the + * flow's start node and hands us one of these per activated flow. + */ +export interface FlowTriggerBinding { + readonly flowName: string; + readonly object?: string; + readonly event?: string; + readonly condition?: string | { dialect?: string; source?: string; ast?: unknown }; + readonly schedule?: unknown; + readonly config?: Record; +} + +/** + * Structural mirror of the engine's `FlowTrigger` extension point. The engine + * calls {@link start} with a parsed binding + a callback that runs the flow, + * and {@link stop} when the flow is unregistered/disabled. + */ +export interface FlowTrigger { + readonly type: string; + start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise): void; + stop(flowName: string): void; +} + +/** + * The slice of the ObjectQL data engine this trigger needs: subscribe to a + * lifecycle hook, and (for teardown) drop all hooks owned by a packageId. + * Typed structurally because `IDataEngine` (the public contract) doesn't model + * the hook surface, but the concrete engine implements both. + */ +export interface RecordChangeDataEngine { + registerHook( + event: string, + handler: (ctx: HookContext) => unknown | Promise, + options?: { object?: string | string[]; priority?: number; packageId?: string }, + ): void; + unregisterHooksByPackage?(packageId: string): number; +} + +/** Minimal logger surface (matches core's `ctx.logger`). */ +export interface TriggerLogger { + info(msg: string, ...args: unknown[]): void; + warn(msg: string, ...args: unknown[]): void; + debug?(msg: string, ...args: unknown[]): void; +} + +const TRIGGER_PREFIX = 'com.objectstack.trigger.record-change'; + +/** + * Map a flow start node's `triggerType` (e.g. `record-after-update`) to an + * ObjectQL `HookEvent` (e.g. `afterUpdate`). Returns `null` for anything that + * isn't a `record-(before|after)-(create|insert|update|delete)` token. + */ +export function triggerTypeToHookEvent(triggerType: string | undefined): string | null { + if (!triggerType) return null; + const m = /^record-(before|after)-(create|insert|update|delete)$/.exec(triggerType.trim()); + if (!m) return null; + const phase = m[1]; // 'before' | 'after' + const op = m[2]; // create|insert|update|delete + const verb = op === 'create' || op === 'insert' ? 'Insert' : op.charAt(0).toUpperCase() + op.slice(1); + return `${phase}${verb}`; // e.g. 'afterUpdate', 'beforeDelete' +} + +/** + * RecordChangeTrigger + * + * Bridges the automation engine's {@link FlowTrigger} extension point to + * ObjectQL lifecycle hooks. For each flow the engine activates, it subscribes + * to the matching hook event (filtered to the flow's target object) and, when + * the hook fires, builds an {@link AutomationContext} from the new/old record + * and invokes the engine-supplied callback (which runs the flow — the engine + * owns the start-node condition gate, so we don't re-evaluate it here). + * + * Each flow's hooks are registered under a per-flow packageId so {@link stop} + * can tear exactly that flow's subscription down via + * `unregisterHooksByPackage`, without touching other flows or audit hooks. + */ +export class RecordChangeTrigger implements FlowTrigger { + readonly type = 'record_change'; + + private readonly engine: RecordChangeDataEngine; + private readonly logger: TriggerLogger; + /** flowName → packageId used for its hook(s), so stop() can unregister it. */ + private readonly bound = new Map(); + + constructor(engine: RecordChangeDataEngine, logger: TriggerLogger) { + this.engine = engine; + this.logger = logger; + } + + start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise): void { + const hookEvent = triggerTypeToHookEvent(binding.event); + if (!hookEvent) { + this.logger.warn( + `[record-change] flow '${binding.flowName}' has unsupported trigger event '${binding.event ?? '(none)'}' — not bound`, + ); + return; + } + + // Idempotent: drop any prior subscription for this flow before re-binding + // (covers disable→enable cycles and hot reload). + this.stop(binding.flowName); + + const packageId = `${TRIGGER_PREFIX}:${binding.flowName}`; + + const handler = async (ctx: HookContext): Promise => { + try { + const automationCtx = this.buildContext(binding, ctx); + await callback(automationCtx); + } catch (err) { + // Error isolation: a flow failure must NEVER break the CRUD write + // that triggered it. Log and swallow. + this.logger.warn( + `[record-change] flow '${binding.flowName}' execution failed: ${(err as Error)?.message ?? String(err)}`, + ); + } + }; + + this.engine.registerHook(hookEvent, handler, { + object: binding.object, + packageId, + }); + this.bound.set(binding.flowName, packageId); + this.logger.info( + `[record-change] bound flow '${binding.flowName}' → ${hookEvent}${binding.object ? ` on '${binding.object}'` : ''}`, + ); + } + + stop(flowName: string): void { + const packageId = this.bound.get(flowName); + if (!packageId) return; + try { + this.engine.unregisterHooksByPackage?.(packageId); + } catch (err) { + this.logger.warn( + `[record-change] failed to unbind flow '${flowName}': ${(err as Error)?.message ?? String(err)}`, + ); + } + this.bound.delete(flowName); + this.logger.debug?.(`[record-change] unbound flow '${flowName}'`); + } + + /** + * Build the flow execution context from an ObjectQL hook context. The new + * record comes from `ctx.result` (after-hooks) or falls back to the + * mutation input doc / previous row; the old record from `ctx.previous` + * (with the `__previous` stash audit also uses as a fallback). + */ + private buildContext(binding: FlowTriggerBinding, ctx: HookContext): AutomationContext { + const input = (ctx.input ?? {}) as { doc?: Record; id?: unknown }; + const after = ctx.result as Record | undefined; + const previous = + (ctx.previous as Record | undefined) ?? + ((ctx as unknown as { __previous?: Record }).__previous ?? undefined); + + const record: Record = + after && typeof after === 'object' + ? after + : input.doc && typeof input.doc === 'object' + ? input.doc + : previous && typeof previous === 'object' + ? previous + : {}; + + const session = (ctx.session ?? {}) as { userId?: string }; + + return { + record, + previous, + object: binding.object ?? ctx.object, + event: binding.event, + userId: session.userId, + // Expose the record as params too, so flows with named `isInput` + // variables matching record fields get them seeded. + params: record, + }; + } +} diff --git a/packages/plugins/trigger-record-change/tsconfig.json b/packages/plugins/trigger-record-change/tsconfig.json new file mode 100644 index 0000000000..f6a1e8bad5 --- /dev/null +++ b/packages/plugins/trigger-record-change/tsconfig.json @@ -0,0 +1,18 @@ +{ + "extends": "../../../tsconfig.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src", + "types": [ + "node" + ] + }, + "include": [ + "src/**/*" + ], + "exclude": [ + "dist", + "node_modules", + "**/*.test.ts" + ] +} diff --git a/packages/plugins/trigger-schedule/CHANGELOG.md b/packages/plugins/trigger-schedule/CHANGELOG.md new file mode 100644 index 0000000000..72a9f32f9a --- /dev/null +++ b/packages/plugins/trigger-schedule/CHANGELOG.md @@ -0,0 +1,163 @@ +# @objectstack/plugin-trigger-schedule + +## 9.2.0 + +### Patch Changes + +- Updated dependencies [2f57b75] +- Updated dependencies [2f57b75] + - @objectstack/spec@9.2.0 + - @objectstack/core@9.2.0 + +## 9.1.0 + +### Patch Changes + +- Updated dependencies [b9062c9] + - @objectstack/spec@9.1.0 + - @objectstack/core@9.1.0 + +## 9.0.1 + +### Patch Changes + +- Updated dependencies [1817845] + - @objectstack/spec@9.0.1 + - @objectstack/core@9.0.1 + +## 9.0.0 + +### Patch Changes + +- Updated dependencies [4c3f693] +- Updated dependencies [0bf39f1] +- Updated dependencies [f533f42] +- Updated dependencies [1c83ee8] + - @objectstack/spec@9.0.0 + - @objectstack/core@9.0.0 + +## 8.0.1 + +### Patch Changes + +- @objectstack/spec@8.0.1 +- @objectstack/core@8.0.1 + +## 8.0.0 + +### Patch Changes + +- Updated dependencies [a46c017] +- Updated dependencies [b990b89] +- Updated dependencies [99111ec] +- Updated dependencies [d5a8161] +- Updated dependencies [5cf1f1b] +- Updated dependencies [9ef89d4] +- Updated dependencies [3306d2f] +- Updated dependencies [c262301] +- Updated dependencies [bc44195] +- Updated dependencies [9e2e229] + - @objectstack/spec@8.0.0 + - @objectstack/core@8.0.0 + +## 7.9.0 + +### Patch Changes + +- @objectstack/spec@7.9.0 +- @objectstack/core@7.9.0 + +## 7.8.0 + +### Patch Changes + +- Updated dependencies [06f2bbb] +- Updated dependencies [36719db] +- Updated dependencies [424ab26] + - @objectstack/spec@7.8.0 + - @objectstack/core@7.8.0 + +## 7.7.0 + +### Patch Changes + +- Updated dependencies [b391955] +- Updated dependencies [f06b64e] +- Updated dependencies [023bf93] + - @objectstack/spec@7.7.0 + - @objectstack/core@7.7.0 + +## 7.6.0 + +### Patch Changes + +- Updated dependencies [955d4c8] +- Updated dependencies [c4a4cbd] +- Updated dependencies [b046ec2] +- Updated dependencies [2170ad9] +- Updated dependencies [02d6359] +- Updated dependencies [7648242] +- Updated dependencies [8fa1e7f] +- Updated dependencies [55866f5] +- Updated dependencies [60f9c45] + - @objectstack/spec@7.6.0 + - @objectstack/core@7.6.0 + +## 7.5.0 + +### Patch Changes + +- @objectstack/spec@7.5.0 +- @objectstack/core@7.5.0 + +## 7.4.1 + +### Patch Changes + +- @objectstack/spec@7.4.1 +- @objectstack/core@7.4.1 + +## 7.4.0 + +### Minor Changes + +- 03fd7f0: Schedule flow trigger — auto-launch flows on a cron/interval/once schedule. + + The sibling of `@objectstack/plugin-trigger-record-change`: it completes the + _time-based_ arm of the automation engine's `FlowTrigger` extension point. The + engine already parses a flow's start node into a `schedule` binding + (`flow.type === 'schedule'` or a start-node `config.schedule` descriptor); this + plugin registers the concrete `schedule` trigger and delegates timing to the + platform `IJobService` (the `'job'` service), so it stays adapter-agnostic — the + job service selects a cron-capable adapter (durable `DbJobAdapter` / + `CronJobAdapter`) for cron schedules and the interval adapter otherwise. + + - `normalizeSchedule` accepts the canonical `JobSchedule` plus shorthands (a + bare cron string, `{ cron }` / `{ expression }`, `{ every }` / `{ intervalMs }`, + `{ at }`). + - When a job fires, the flow runs with `event: 'schedule'` and + `params: { jobId, flowName, schedule }`; the engine's start-condition gate + still applies. + - Error-isolated (a flow failure never crashes the job runner); per-flow job + name so `stop()` cancels exactly one flow; the job service is resolved lazily + per bind so adapter upgrades are picked up; graceful degrade when the + automation or job service is absent. + + No engine change required — the `schedule` binding shipped with the + record-change trigger PR. + +### Patch Changes + +- Updated dependencies [23c7107] +- Updated dependencies [c72daad] +- Updated dependencies [f115182] +- Updated dependencies [2faf9f2] +- Updated dependencies [2faf9f2] +- Updated dependencies [2faf9f2] +- Updated dependencies [58b450b] +- Updated dependencies [82eb6cf] +- Updated dependencies [13d8653] +- Updated dependencies [ff3d006] +- Updated dependencies [5e831de] + - @objectstack/spec@7.4.0 + - @objectstack/core@7.4.0 diff --git a/packages/plugins/trigger-schedule/README.md b/packages/plugins/trigger-schedule/README.md new file mode 100644 index 0000000000..d426d30351 --- /dev/null +++ b/packages/plugins/trigger-schedule/README.md @@ -0,0 +1,68 @@ +# @objectstack/plugin-trigger-schedule + +Auto-launch ObjectStack flows on a schedule (cron / interval / once). + +The automation engine ships the `FlowTrigger` extension point and the wiring +that turns a flow's `start` node into a normalized trigger binding — but the +*concrete* schedule trigger lives here, as a plugin. It delegates timing to the +platform `IJobService` (the `'job'` service), so it stays adapter-agnostic: the +job service selects a cron-capable adapter (e.g. the durable `DbJobAdapter` or +`CronJobAdapter`) for cron schedules and the interval adapter for the rest. + +This is the sibling of `@objectstack/plugin-trigger-record-change` — same +engine baseline, a different event source. + +## What it does + +A flow whose `start` node declares a schedule: + +```ts +{ + type: 'start', + config: { + schedule: { type: 'cron', expression: '0 1 * * *', timezone: 'UTC' }, + condition: "...", // optional start-condition gate + }, +} +// or simply: a flow with `type: 'schedule'` and a start-node schedule descriptor +``` + +auto-launches on that schedule — no manual `engine.execute()`. When it fires, +the flow runs with `event: 'schedule'` and `params: { jobId, flowName, schedule }` +in its context. + +### Schedule shapes + +`normalizeSchedule` accepts the canonical `JobSchedule` plus shorthands: + +| Input | Normalized | +| ---------------------------------------------- | ---------------------------------------- | +| `{ type: 'cron', expression, timezone? }` | cron | +| `'0 1 * * *'` (bare string) | `{ type: 'cron', expression: '0 1 * * *' }` | +| `{ cron }` / `{ expression }` | cron | +| `{ type: 'interval', intervalMs }` / `{ every }` | interval | +| `{ type: 'once', at }` / `{ at }` | once | + +## Usage + +```ts +import { AutomationServicePlugin } from '@objectstack/service-automation'; +import { JobServicePlugin } from '@objectstack/service-job'; +import { ScheduleTriggerPlugin } from '@objectstack/plugin-trigger-schedule'; + +kernel + .use(new AutomationServicePlugin()) // engine + flows + .use(new JobServicePlugin()) // the 'job' service (cron/interval/db) + .use(new ScheduleTriggerPlugin()); // ← makes schedule flows live +``` + +Depends on the job service plugin (`com.objectstack.service.job`) so its +`kernel:ready` adapter upgrade runs first; the job service is nonetheless +resolved lazily per bind, so adapter upgrades are always picked up. If the +automation or job service is unavailable, the plugin logs a warning and no-ops +rather than failing startup. + +## Error isolation + +A flow that throws during a scheduled run is logged and swallowed — it never +crashes the job runner. diff --git a/packages/plugins/trigger-schedule/package.json b/packages/plugins/trigger-schedule/package.json new file mode 100644 index 0000000000..7f3a387991 --- /dev/null +++ b/packages/plugins/trigger-schedule/package.json @@ -0,0 +1,55 @@ +{ + "name": "@objectstack/trigger-schedule", + "version": "9.2.0", + "license": "Apache-2.0", + "description": "Schedule flow trigger for ObjectStack — auto-launches flows on a cron/interval/once schedule via the IJobService (ADR-0018)", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.mjs", + "require": "./dist/index.js" + } + }, + "scripts": { + "build": "tsup --config ../../../tsup.config.ts", + "test": "vitest run --passWithNoTests" + }, + "dependencies": { + "@objectstack/core": "workspace:*", + "@objectstack/spec": "workspace:*" + }, + "devDependencies": { + "@types/node": "^25.9.2", + "typescript": "^6.0.3", + "vitest": "^4.1.8" + }, + "keywords": [ + "objectstack", + "plugin", + "automation", + "flow", + "trigger", + "schedule", + "cron" + ], + "author": "ObjectStack", + "repository": { + "type": "git", + "url": "https://github.com/objectstack-ai/framework.git", + "directory": "packages/plugins/plugin-trigger-schedule" + }, + "homepage": "https://objectstack.ai/docs", + "bugs": "https://github.com/objectstack-ai/framework/issues", + "publishConfig": { + "access": "public" + }, + "files": [ + "dist", + "README.md" + ], + "engines": { + "node": ">=18.0.0" + } +} diff --git a/packages/plugins/trigger-schedule/src/index.ts b/packages/plugins/trigger-schedule/src/index.ts new file mode 100644 index 0000000000..a639fe7ca7 --- /dev/null +++ b/packages/plugins/trigger-schedule/src/index.ts @@ -0,0 +1,10 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +export { ScheduleTriggerPlugin } from './plugin.js'; +export { ScheduleTrigger, normalizeSchedule } from './schedule-trigger.js'; +export type { + FlowTrigger, + FlowTriggerBinding, + JobServiceSurface, + TriggerLogger, +} from './schedule-trigger.js'; diff --git a/packages/plugins/trigger-schedule/src/plugin.ts b/packages/plugins/trigger-schedule/src/plugin.ts new file mode 100644 index 0000000000..89cf794c35 --- /dev/null +++ b/packages/plugins/trigger-schedule/src/plugin.ts @@ -0,0 +1,83 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { Plugin, PluginContext } from '@objectstack/core'; +import { ScheduleTrigger } from './schedule-trigger.js'; +import type { FlowTrigger, JobServiceSurface } from './schedule-trigger.js'; + +/** + * The slice of the automation engine this plugin needs: register a trigger on + * its `FlowTrigger` extension point. Declared structurally so the plugin does + * not take a build dependency on `@objectstack/service-automation`. + */ +interface AutomationTriggerRegistry { + registerTrigger(trigger: FlowTrigger): void; + unregisterTrigger?(type: string): void; +} + +/** + * ScheduleTriggerPlugin + * + * Makes schedule-triggered flows actually fire. The automation engine ships the + * `FlowTrigger` wiring (it parses each flow's start node — `flow.type === + * 'schedule'` or a start-node `config.schedule` descriptor — into a binding and + * calls `trigger.start(...)`), but the *concrete* schedule trigger lives here as + * a plugin and delegates timing to the platform `IJobService` (the `'job'` + * service). This mirrors the connector / record-change split (engine baseline + + * trigger plugin). + * + * With this plugin (and a job service) installed, a flow whose start node + * declares `config: { schedule: { type: 'cron', expression: '0 1 * * *' } }` + * auto-launches on that schedule — no manual `engine.execute()`. + * + * Depends on the job service plugin so its `kernel:ready` upgrade (to the + * durable DbJobAdapter) runs before ours; the job service is nonetheless + * resolved lazily per `start()` so we always use its current adapter. + */ +export class ScheduleTriggerPlugin implements Plugin { + name = 'com.objectstack.trigger.schedule'; + type = 'standard'; + version = '7.3.0'; + dependencies = ['com.objectstack.service.job']; + + async init(ctx: PluginContext): Promise { + ctx.logger.info('Schedule trigger plugin initialized'); + } + + async start(ctx: PluginContext): Promise { + // The automation service + job service are resolvable once the kernel is + // ready (kernel:ready fires after AutomationServicePlugin.start() has + // pulled flows in and after the job service upgrades its adapter). + ctx.hook('kernel:ready', async () => { + const automation = this.resolveService(ctx, 'automation'); + if (!automation || typeof automation.registerTrigger !== 'function') { + ctx.logger.warn( + 'ScheduleTriggerPlugin: automation service not available — schedule trigger NOT installed', + ); + return; + } + + // Probe once for a clear startup warning; the trigger re-resolves + // lazily on each start() so adapter upgrades are always picked up. + if (!this.resolveService(ctx, 'job')) { + ctx.logger.warn( + 'ScheduleTriggerPlugin: job service not available — scheduled flows will not run until one is registered', + ); + } + + const trigger = new ScheduleTrigger( + () => this.resolveService(ctx, 'job'), + ctx.logger, + ); + automation.registerTrigger(trigger); + ctx.logger.info('ScheduleTriggerPlugin: schedule trigger registered'); + }); + } + + private resolveService(ctx: PluginContext, name: string): T | null { + try { + return ctx.getService(name) ?? null; + } catch { + return null; + } + } +} diff --git a/packages/plugins/trigger-schedule/src/schedule-trigger.test.ts b/packages/plugins/trigger-schedule/src/schedule-trigger.test.ts new file mode 100644 index 0000000000..502245cd7f --- /dev/null +++ b/packages/plugins/trigger-schedule/src/schedule-trigger.test.ts @@ -0,0 +1,295 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, vi } from 'vitest'; +import type { AutomationContext, JobSchedule, JobHandler } from '@objectstack/spec/contracts'; +import { + ScheduleTrigger, + normalizeSchedule, + type FlowTriggerBinding, + type JobServiceSurface, + type TriggerLogger, +} from './schedule-trigger.js'; +import { ScheduleTriggerPlugin } from './plugin.js'; + +// ─── Test doubles ─────────────────────────────────────────────────── + +interface ScheduledJob { + name: string; + schedule: JobSchedule; + handler: JobHandler; +} + +/** Fake IJobService slice: records schedule()/cancel() and can fire a job. */ +function fakeJobService() { + const jobs = new Map(); + const service: JobServiceSurface = { + async schedule(name, schedule, handler) { + jobs.set(name, { name, schedule, handler }); + }, + async cancel(name) { + jobs.delete(name); + }, + }; + return { + service, + jobs, + async fire(name: string, jobId = 'run1') { + await jobs.get(name)?.handler({ jobId }); + }, + }; +} + +function silentLogger(): TriggerLogger { + return { info: () => {}, warn: () => {}, debug: () => {} }; +} + +function binding(overrides: Partial = {}): FlowTriggerBinding { + return { + flowName: 'nightly_health_sweep', + schedule: { type: 'cron', expression: '0 1 * * *', timezone: 'UTC' }, + ...overrides, + }; +} + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +// ─── normalizeSchedule ────────────────────────────────────────────── + +describe('normalizeSchedule', () => { + it('passes through canonical cron/interval/once shapes', () => { + expect(normalizeSchedule({ type: 'cron', expression: '* * * * *', timezone: 'UTC' })).toEqual({ + type: 'cron', + expression: '* * * * *', + timezone: 'UTC', + }); + expect(normalizeSchedule({ type: 'interval', intervalMs: 5000 })).toEqual({ + type: 'interval', + intervalMs: 5000, + }); + expect(normalizeSchedule({ type: 'once', at: '2026-01-01T00:00:00Z' })).toEqual({ + type: 'once', + at: '2026-01-01T00:00:00Z', + }); + }); + + it('treats a bare string as a cron expression', () => { + expect(normalizeSchedule('0 1 * * *')).toEqual({ type: 'cron', expression: '0 1 * * *' }); + }); + + it('accepts shorthands { cron } / { expression } / { every } / { at }', () => { + expect(normalizeSchedule({ cron: '*/5 * * * *' })).toEqual({ type: 'cron', expression: '*/5 * * * *' }); + expect(normalizeSchedule({ expression: '0 0 * * *' })).toEqual({ type: 'cron', expression: '0 0 * * *' }); + expect(normalizeSchedule({ every: 1000 })).toEqual({ type: 'interval', intervalMs: 1000 }); + expect(normalizeSchedule({ at: '2026-06-01T00:00:00Z' })).toEqual({ + type: 'once', + at: '2026-06-01T00:00:00Z', + }); + }); + + it('returns null for missing / unusable descriptors', () => { + expect(normalizeSchedule(undefined)).toBeNull(); + expect(normalizeSchedule(null)).toBeNull(); + expect(normalizeSchedule('')).toBeNull(); + expect(normalizeSchedule({ type: 'cron' })).toBeNull(); // no expression + expect(normalizeSchedule({ type: 'interval', intervalMs: 0 })).toBeNull(); + expect(normalizeSchedule({ type: 'once' })).toBeNull(); // no at + expect(normalizeSchedule(42)).toBeNull(); + }); +}); + +// ─── ScheduleTrigger ──────────────────────────────────────────────── + +describe('ScheduleTrigger', () => { + it('schedules a job for the flow with the normalized schedule', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + + trigger.start(binding(), async () => {}); + await flush(); + + expect(job.jobs.size).toBe(1); + const scheduled = job.jobs.get('flow-schedule:nightly_health_sweep'); + expect(scheduled?.schedule).toEqual({ type: 'cron', expression: '0 1 * * *', timezone: 'UTC' }); + }); + + it('reads schedule from binding.config.schedule as a fallback', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + + trigger.start( + binding({ schedule: undefined, config: { schedule: { type: 'interval', intervalMs: 2000 } } }), + async () => {}, + ); + await flush(); + + expect(job.jobs.get('flow-schedule:nightly_health_sweep')?.schedule).toEqual({ + type: 'interval', + intervalMs: 2000, + }); + }); + + it('does not schedule when no schedule descriptor is present', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + + trigger.start(binding({ schedule: undefined }), async () => {}); + await flush(); + + expect(job.jobs.size).toBe(0); + }); + + it('does not schedule when the job service is unavailable', async () => { + const trigger = new ScheduleTrigger(() => null, silentLogger()); + expect(() => trigger.start(binding(), async () => {})).not.toThrow(); + }); + + it('fires the callback with a schedule context when the job runs', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + const seen: AutomationContext[] = []; + + trigger.start(binding(), async (ctx) => { + seen.push(ctx); + }); + await flush(); + await job.fire('flow-schedule:nightly_health_sweep', 'run42'); + + expect(seen).toHaveLength(1); + expect(seen[0].event).toBe('schedule'); + expect(seen[0].params).toMatchObject({ jobId: 'run42', flowName: 'nightly_health_sweep' }); + }); + + it('isolates flow errors so the job runner is never broken', async () => { + const job = fakeJobService(); + const warn = vi.fn(); + const trigger = new ScheduleTrigger(() => job.service, { info: () => {}, warn, debug: () => {} }); + + trigger.start(binding(), async () => { + throw new Error('flow blew up'); + }); + await flush(); + + await expect(job.fire('flow-schedule:nightly_health_sweep')).resolves.toBeUndefined(); + expect(warn).toHaveBeenCalled(); + }); + + it('stop() cancels the flow\'s job', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + + trigger.start(binding(), async () => {}); + await flush(); + expect(job.jobs.size).toBe(1); + + trigger.stop('nightly_health_sweep'); + await flush(); + expect(job.jobs.size).toBe(0); + }); + + it('re-binding the same flow is idempotent (one job)', async () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + + trigger.start(binding(), async () => {}); + await flush(); + trigger.start(binding({ schedule: { type: 'interval', intervalMs: 9000 } }), async () => {}); + await flush(); + + expect(job.jobs.size).toBe(1); + expect(job.jobs.get('flow-schedule:nightly_health_sweep')?.schedule).toEqual({ + type: 'interval', + intervalMs: 9000, + }); + }); + + it('stop() on an unknown flow is a no-op', () => { + const job = fakeJobService(); + const trigger = new ScheduleTrigger(() => job.service, silentLogger()); + expect(() => trigger.stop('never_bound')).not.toThrow(); + }); +}); + +// ─── ScheduleTriggerPlugin ────────────────────────────────────────── + +describe('ScheduleTriggerPlugin', () => { + interface FakeCtx { + readyHandlers: Array<() => Promise | void>; + ctx: { + logger: TriggerLogger; + getService: (name: string) => T; + hook: (event: string, handler: () => Promise | void) => void; + }; + } + + function fakePluginCtx(services: Record): FakeCtx { + const readyHandlers: Array<() => Promise | void> = []; + return { + readyHandlers, + ctx: { + logger: silentLogger() as TriggerLogger, + getService(name: string): T { + if (!(name in services)) throw new Error(`no service '${name}'`); + return services[name] as T; + }, + hook(event: string, handler: () => Promise | void) { + if (event === 'kernel:ready') readyHandlers.push(handler); + }, + }, + }; + } + + it('registers the trigger when automation + job services exist', async () => { + const registerTrigger = vi.fn(); + const job = fakeJobService(); + const fake = fakePluginCtx({ automation: { registerTrigger }, job: job.service }); + + const plugin = new ScheduleTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + expect(registerTrigger).toHaveBeenCalledTimes(1); + expect((registerTrigger.mock.calls[0][0] as ScheduleTrigger).type).toBe('schedule'); + }); + + it('still registers the trigger when the job service is missing (warns)', async () => { + const registerTrigger = vi.fn(); + const fake = fakePluginCtx({ automation: { registerTrigger } }); + + const plugin = new ScheduleTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + // Registered so it can lazily pick up a job service later. + expect(registerTrigger).toHaveBeenCalledTimes(1); + }); + + it('skips gracefully when the automation service is absent', async () => { + const job = fakeJobService(); + const fake = fakePluginCtx({ job: job.service }); + + const plugin = new ScheduleTriggerPlugin(); + await plugin.start(fake.ctx as never); + await expect(fake.readyHandlers[0]()).resolves.toBeUndefined(); + }); + + it('lazily resolves the job service at fire time (adapter upgrade)', async () => { + const registerTrigger = vi.fn(); + const job = fakeJobService(); + // Job service appears AFTER the trigger is registered. + const services: Record = { automation: { registerTrigger } }; + const fake = fakePluginCtx(services); + + const plugin = new ScheduleTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + // Now the job service becomes available. + services.job = job.service; + + const trigger = registerTrigger.mock.calls[0][0] as ScheduleTrigger; + trigger.start(binding(), async () => {}); + await flush(); + + expect(job.jobs.size).toBe(1); + }); +}); diff --git a/packages/plugins/trigger-schedule/src/schedule-trigger.ts b/packages/plugins/trigger-schedule/src/schedule-trigger.ts new file mode 100644 index 0000000000..e95355291b --- /dev/null +++ b/packages/plugins/trigger-schedule/src/schedule-trigger.ts @@ -0,0 +1,208 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { JobSchedule, JobHandler } from '@objectstack/spec/contracts'; + +/** + * Structural mirror of the automation engine's `FlowTriggerBinding` + * (service-automation/src/engine.ts). Declared locally so this trigger plugin + * stays decoupled from the automation package — same pattern the record-change + * trigger and the connector / messaging integrations use. The engine parses the + * flow's start node and hands us a binding whose `schedule` carries the + * cron/interval/once descriptor. + */ +export interface FlowTriggerBinding { + readonly flowName: string; + readonly object?: string; + readonly event?: string; + readonly condition?: string | { dialect?: string; source?: string; ast?: unknown }; + readonly schedule?: unknown; + readonly config?: Record; +} + +/** + * Structural mirror of the engine's `FlowTrigger` extension point. The engine + * calls {@link start} with a parsed binding + a callback that runs the flow, + * and {@link stop} when the flow is unregistered/disabled. + */ +export interface FlowTrigger { + readonly type: string; + start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise): void; + stop(flowName: string): void; +} + +/** + * The slice of `IJobService` this trigger needs: schedule a named job and + * cancel it. Typed structurally so the plugin depends on the spec contract + * shape, not a concrete adapter. + */ +export interface JobServiceSurface { + schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise; + cancel(name: string): Promise; +} + +/** Minimal logger surface (matches core's `ctx.logger`). */ +export interface TriggerLogger { + info(msg: string, ...args: unknown[]): void; + warn(msg: string, ...args: unknown[]): void; + debug?(msg: string, ...args: unknown[]): void; +} + +const JOB_PREFIX = 'flow-schedule'; + +/** + * Normalize a flow's raw `schedule` descriptor into a {@link JobSchedule}, or + * `null` if it can't be understood. Accepts the canonical + * `{ type: 'cron'|'interval'|'once', ... }` shape plus a few ergonomic + * shorthands (a bare cron string, `{ cron }`, `{ expression }`, `{ every }` / + * `{ intervalMs }`, `{ at }`). + */ +export function normalizeSchedule(raw: unknown): JobSchedule | null { + if (raw == null) return null; + + // Bare cron string, e.g. '0 1 * * *'. + if (typeof raw === 'string') { + const expr = raw.trim(); + return expr ? { type: 'cron', expression: expr } : null; + } + + if (typeof raw !== 'object') return null; + const s = raw as Record; + + const type = typeof s.type === 'string' ? s.type : undefined; + + if (type === 'cron' || (!type && (typeof s.cron === 'string' || typeof s.expression === 'string'))) { + const expression = + (typeof s.expression === 'string' && s.expression) || + (typeof s.cron === 'string' && s.cron) || + undefined; + if (!expression) return null; + const out: JobSchedule = { type: 'cron', expression }; + if (typeof s.timezone === 'string') out.timezone = s.timezone; + return out; + } + + if (type === 'interval' || (!type && (typeof s.intervalMs === 'number' || typeof s.every === 'number'))) { + const intervalMs = + (typeof s.intervalMs === 'number' && s.intervalMs) || + (typeof s.every === 'number' && s.every) || + undefined; + if (!intervalMs || intervalMs <= 0) return null; + return { type: 'interval', intervalMs }; + } + + if (type === 'once' || (!type && typeof s.at === 'string')) { + const at = typeof s.at === 'string' ? s.at : undefined; + if (!at) return null; + return { type: 'once', at }; + } + + return null; +} + +/** + * ScheduleTrigger + * + * Bridges the automation engine's {@link FlowTrigger} extension point to the + * platform {@link JobServiceSurface}. For each schedule-triggered flow the + * engine activates, it registers a job whose handler runs the flow; the job + * service owns the actual cron/interval/once timing (so this trigger stays + * adapter-agnostic — cron schedules need a cron-capable adapter, which the + * job service selects). + * + * The job service is resolved lazily (per `start()`) via the supplied accessor, + * so we always pick up the job service's *upgraded* adapter (e.g. the durable + * DbJobAdapter that replaces the bootstrap interval adapter on `kernel:ready`). + */ +export class ScheduleTrigger implements FlowTrigger { + readonly type = 'schedule'; + + private readonly getJobService: () => JobServiceSurface | null; + private readonly logger: TriggerLogger; + /** flowName → job name registered for it, so stop() can cancel it. */ + private readonly bound = new Map(); + + constructor(getJobService: () => JobServiceSurface | null, logger: TriggerLogger) { + this.getJobService = getJobService; + this.logger = logger; + } + + start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise): void { + const raw = binding.schedule ?? (binding.config as Record | undefined)?.schedule; + const schedule = normalizeSchedule(raw); + if (!schedule) { + this.logger.warn( + `[schedule] flow '${binding.flowName}' has no recognizable schedule descriptor — not bound`, + ); + return; + } + + const jobService = this.getJobService(); + if (!jobService || typeof jobService.schedule !== 'function') { + this.logger.warn( + `[schedule] job service unavailable — flow '${binding.flowName}' not scheduled`, + ); + return; + } + + // Idempotent: drop any prior schedule for this flow before re-binding + // (covers disable→enable cycles and hot reload). + this.stop(binding.flowName); + + const jobName = `${JOB_PREFIX}:${binding.flowName}`; + + const handler: JobHandler = async ({ jobId }) => { + try { + const ctx: AutomationContext = { + event: 'schedule', + params: { + jobId, + flowName: binding.flowName, + schedule, + }, + }; + await callback(ctx); + } catch (err) { + // Error isolation: a scheduled flow failure must not crash the + // job runner / ticker. Log and swallow. + this.logger.warn( + `[schedule] flow '${binding.flowName}' execution failed: ${(err as Error)?.message ?? String(err)}`, + ); + } + }; + + this.bound.set(binding.flowName, jobName); + // FlowTrigger.start is sync; the job service's schedule() is async. + // Fire-and-forget with error logging. + void Promise.resolve(jobService.schedule(jobName, schedule, handler)) + .then(() => { + this.logger.info( + `[schedule] bound flow '${binding.flowName}' → ${schedule.type}` + + (schedule.expression ? ` '${schedule.expression}'` : '') + + (schedule.intervalMs ? ` every ${schedule.intervalMs}ms` : '') + + (schedule.at ? ` at ${schedule.at}` : ''), + ); + }) + .catch((err) => { + this.bound.delete(binding.flowName); + this.logger.warn( + `[schedule] failed to schedule flow '${binding.flowName}': ${(err as Error)?.message ?? String(err)}`, + ); + }); + } + + stop(flowName: string): void { + const jobName = this.bound.get(flowName); + if (!jobName) return; + this.bound.delete(flowName); + const jobService = this.getJobService(); + if (!jobService || typeof jobService.cancel !== 'function') return; + void Promise.resolve(jobService.cancel(jobName)) + .then(() => this.logger.debug?.(`[schedule] unbound flow '${flowName}'`)) + .catch((err) => { + this.logger.warn( + `[schedule] failed to unbind flow '${flowName}': ${(err as Error)?.message ?? String(err)}`, + ); + }); + } +} diff --git a/packages/plugins/trigger-schedule/tsconfig.json b/packages/plugins/trigger-schedule/tsconfig.json new file mode 100644 index 0000000000..f6a1e8bad5 --- /dev/null +++ b/packages/plugins/trigger-schedule/tsconfig.json @@ -0,0 +1,18 @@ +{ + "extends": "../../../tsconfig.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src", + "types": [ + "node" + ] + }, + "include": [ + "src/**/*" + ], + "exclude": [ + "dist", + "node_modules", + "**/*.test.ts" + ] +} diff --git a/packages/spec/src/automation/node-executor.zod.ts b/packages/spec/src/automation/node-executor.zod.ts index 98b2cd7a5e..1968e2d7fa 100644 --- a/packages/spec/src/automation/node-executor.zod.ts +++ b/packages/spec/src/automation/node-executor.zod.ts @@ -259,6 +259,16 @@ export const ActionDescriptorSchema = lazySchema(() => z.object({ isAsync: z.boolean().default(false) .describe('Suspends the flow awaiting an external reply'), + /** + * Runtime maturity of the capability behind this descriptor (ADR-0041 §4). + * The platform routinely ships contracts ahead of runtimes; `reserved` + * marks a surface whose runtime has NOT shipped, so designers (Studio) + * can render it visible-but-disabled instead of letting authors build on + * something that silently never runs. + */ + maturity: z.enum(['ga', 'beta', 'reserved']).default('ga') + .describe('Runtime maturity: ga (shipped), beta, or reserved (contract only — designers grey this out)'), + // ── provenance ──────────────────────────────────────────────────── /** * Whether this action ships with the platform (`builtin`, seeded by the diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 83c857a892..76a9198a98 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -527,12 +527,6 @@ importers: '@objectstack/plugin-sharing': specifier: workspace:* version: link:../plugins/plugin-sharing - '@objectstack/plugin-trigger-record-change': - specifier: workspace:* - version: link:../plugins/plugin-trigger-record-change - '@objectstack/plugin-trigger-schedule': - specifier: workspace:* - version: link:../plugins/plugin-trigger-schedule '@objectstack/plugin-webhooks': specifier: workspace:* version: link:../plugins/plugin-webhooks @@ -584,6 +578,12 @@ importers: '@objectstack/spec': specifier: workspace:* version: link:../spec + '@objectstack/trigger-record-change': + specifier: workspace:* + version: link:../plugins/trigger-record-change + '@objectstack/trigger-schedule': + specifier: workspace:* + version: link:../plugins/trigger-schedule '@objectstack/types': specifier: workspace:* version: link:../types @@ -1563,11 +1563,14 @@ importers: specifier: ^4.1.8 version: 4.1.8(@opentelemetry/api@1.9.1)(@types/node@25.9.2)(@vitest/coverage-v8@4.1.8)(happy-dom@20.10.2)(msw@2.14.6(@types/node@25.9.2)(typescript@6.0.3))(vite@8.0.16(@types/node@25.9.2)(esbuild@0.28.0)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) - packages/plugins/plugin-trigger-record-change: + packages/plugins/plugin-webhooks: dependencies: '@objectstack/core': specifier: workspace:* version: link:../../core + '@objectstack/service-messaging': + specifier: workspace:* + version: link:../../services/service-messaging '@objectstack/spec': specifier: workspace:* version: link:../../spec @@ -1582,7 +1585,7 @@ importers: specifier: ^4.1.8 version: 4.1.8(@opentelemetry/api@1.9.1)(@types/node@25.9.2)(@vitest/coverage-v8@4.1.8)(happy-dom@20.10.2)(msw@2.14.6(@types/node@25.9.2)(typescript@6.0.3))(vite@8.0.16(@types/node@25.9.2)(esbuild@0.28.0)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) - packages/plugins/plugin-trigger-schedule: + packages/plugins/trigger-record-change: dependencies: '@objectstack/core': specifier: workspace:* @@ -1601,14 +1604,11 @@ importers: specifier: ^4.1.8 version: 4.1.8(@opentelemetry/api@1.9.1)(@types/node@25.9.2)(@vitest/coverage-v8@4.1.8)(happy-dom@20.10.2)(msw@2.14.6(@types/node@25.9.2)(typescript@6.0.3))(vite@8.0.16(@types/node@25.9.2)(esbuild@0.28.0)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) - packages/plugins/plugin-webhooks: + packages/plugins/trigger-schedule: dependencies: '@objectstack/core': specifier: workspace:* version: link:../../core - '@objectstack/service-messaging': - specifier: workspace:* - version: link:../../services/service-messaging '@objectstack/spec': specifier: workspace:* version: link:../../spec