From e55c292133c0e7be979c3d9ded1a12c18a8c2a87 Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 20 Oct 2026 07:36:00 -0700 Subject: [PATCH] build(jobs): upgrade OpenWorkflow to 0.11.0 --- .changeset/openworkflow-engine.md | 7 + .gitignore | 1 + apps/web/package.json | 2 +- .../routes/(auth)/admin/page.server.test.ts | 6 +- apps/worker/README.md | 42 +- apps/worker/package.json | 2 +- apps/worker/src/index.ts | 6 +- .../worker/src/services/engine-logger.test.ts | 91 +++++ apps/worker/src/services/engine-logger.ts | 28 ++ apps/worker/src/services/metrics.ts | 8 +- .../workflows/children.integration.test.ts | 374 ++++++++++++++++++ packages/jobs/README.md | 48 ++- packages/jobs/package.json | 2 +- packages/jobs/src/index.ts | 1 + packages/jobs/src/openworkflow.test.ts | 274 ++++++++----- packages/jobs/src/openworkflow.ts | 35 +- pnpm-lock.yaml | 41 +- 17 files changed, 790 insertions(+), 178 deletions(-) create mode 100644 .changeset/openworkflow-engine.md create mode 100644 apps/worker/src/services/engine-logger.test.ts create mode 100644 apps/worker/src/services/engine-logger.ts create mode 100644 apps/worker/src/workflows/children.integration.test.ts diff --git a/.changeset/openworkflow-engine.md b/.changeset/openworkflow-engine.md new file mode 100644 index 00000000..27d80e1a --- /dev/null +++ b/.changeset/openworkflow-engine.md @@ -0,0 +1,7 @@ +--- +'@when/web': patch +'@when/worker': patch +'@when/cli': patch +--- + +Upgrade OpenWorkflow to 0.11.0 with single-slot child-workflow regression coverage. Route worker engine diagnostics through structured logging and expose explicit queue connection shutdown. Existing appointment workflows and maintenance timers are unchanged in this step. diff --git a/.gitignore b/.gitignore index 41085a85..c64f5421 100644 --- a/.gitignore +++ b/.gitignore @@ -33,6 +33,7 @@ blob-report/ playwright/.cache/ # Per-user / local tool state +/.plans/ /.claude/ /.antigravitycli/ .nono.json diff --git a/apps/web/package.json b/apps/web/package.json index 1546ef92..c7fe9f5a 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -61,7 +61,7 @@ "bits-ui": "2.18.1", "kysely": "0.28.17", "mrmime": "2.0.1", - "openworkflow": "0.9.0", + "openworkflow": "0.11.0", "pino": "10.3.1", "prom-client": "15.1.3", "takumi-js": "2.2.0", diff --git a/apps/web/src/routes/(auth)/admin/page.server.test.ts b/apps/web/src/routes/(auth)/admin/page.server.test.ts index 56e5ac8d..182d3d20 100644 --- a/apps/web/src/routes/(auth)/admin/page.server.test.ts +++ b/apps/web/src/routes/(auth)/admin/page.server.test.ts @@ -1,6 +1,7 @@ -import { describe, expect, test, vi } from 'vitest'; +import { afterEach, describe, expect, test, vi } from 'vitest'; import { openDb, runMigrations } from '@when/db'; import { validConfig } from '$lib/server/__fixtures__/valid-config'; +import { systemClock } from '$lib/server/clock'; let mockDb: Awaited> | null = null; vi.mock('$lib/server/state', () => ({ @@ -38,6 +39,8 @@ function loadEvent(): Parameters[0] { } describe('admin +page.server load', () => { + afterEach(() => vi.restoreAllMocks()); + test('returns purgedCount: 0 when no purged appointments exist', async () => { mockDb = await makeDb(); await mockDb @@ -74,6 +77,7 @@ describe('admin +page.server load', () => { }); test('calculates lifetimeMinutes accurately without floating-point truncation', async () => { + vi.spyOn(systemClock, 'now').mockReturnValue(new Date('2026-09-10T12:00:00Z')); mockDb = await makeDb(); await mockDb .insertInto('appointments') diff --git a/apps/worker/README.md b/apps/worker/README.md index cdfab9fb..ddffa5af 100644 --- a/apps/worker/README.md +++ b/apps/worker/README.md @@ -1,16 +1,16 @@ # @when/worker -The `when` worker — a long-running Node service that runs jobs produced by the -web app (and, later, jobs it schedules itself). Built on -[openworkflow](https://openworkflow.dev) over `node:sqlite`, so it shares the -`when.sqlite` app DB and owns the `queue.sqlite` job DB. +The `when` worker is a long-running Node service for appointment side effects, +calendar maintenance, and on-demand service probes. Built on +[OpenWorkflow](https://openworkflow.dev) `0.11.0` over `node:sqlite`, it shares the +`when.sqlite` app DB and the `openworkflow.sqlite` queue with producers. -Its job is sending appointment emails: web enqueues a `send-appointment-email` run when an -appointment is created, confirmed, cancelled, rescheduled, or declined, and the -worker renders the email(s), sends them over SMTP, and records the per-channel -outcome on the appointment (`email_notification_status`). On boot it loads -config, migrates the app DB, connects the queue, registers its workflows, starts -polling, and serves `/healthz`. +Web enqueues `reconcile-appointment` when an appointment changes. The worker +coordinates calendar/video updates, renders emails, sends them over SMTP, and +records outcomes in the appointment action log and service status. It also handles +purge, calendar-repair requests, video-link generation, and service probes. On boot +it loads config, migrates the app DB, connects the queue, registers workflows, +starts polling with one execution slot, and serves `/healthz`. ## Run it @@ -55,12 +55,18 @@ Relative paths resolve against the config file's directory, so web and worker - `pnpm start` — `node dist/src/index.js`. - `pnpm check` / `pnpm lint` / `pnpm test` / `pnpm test:coverage`. -## Recurring jobs (future) +## Engine integration -openworkflow has no native cron yet. When the worker needs periodic work — e.g. a -calendar refresh — schedule it itself: a `setInterval` that calls -`getOpenWorkflow().runWorkflow(spec, input, { idempotencyKey })` with a -time-bucketed key (e.g. `calendar-refresh:2026-06-05T12`), so a given bucket runs -at most once even if the interval double-fires or the process restarts (the same -pattern luzzle uses for its purge job). Swap to native cron once openworkflow -ships it. +Engine polling, schedule, heartbeat, and execution diagnostics go through the +worker's Pino logger. The adapter emits operational categories without forwarding +arbitrary engine error text or payloads that may contain credentials or guest links. + +The existing job metrics describe **handler executions**, including replays. An +escaped handler execution may mean durable suspension rather than a terminal job +failure. Email metrics count physical SMTP attempt outcomes, not an exactly-once +delivery ledger. + +OpenWorkflow `0.11.0` supports persisted cron schedules. This engine-upgrade step +still uses the existing process timers for refresh, expiry, and repair; migrating +those tasks to awaited cron workflows is a separate application change. Likewise, +upgrading the engine alone does not change the current workflow error handling. diff --git a/apps/worker/package.json b/apps/worker/package.json index 2dec0422..52bc1d0c 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -27,7 +27,7 @@ "eta": "4.6.0", "kysely": "0.28.17", "nodemailer": "9.0.3", - "openworkflow": "0.9.0", + "openworkflow": "0.11.0", "pino": "10.3.1", "prom-client": "15.1.3", "ts-ics": "2.4.4" diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 1146c0d8..c4413e98 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -7,6 +7,7 @@ import { applyConfig } from './services/config-reload.js'; import { setWorkerContext, type WorkerContext } from './services/context.js'; import { createHealthServer } from './services/health.js'; import { logger } from './services/logger.js'; +import { createEngineLogger } from './services/engine-logger.js'; import { registerWorkflows } from './workflows/index.js'; import { registerSyncCalendarsWorkflow } from './workflows/sync-calendars.js'; import { refreshCycle } from './calendar/refresh.js'; @@ -48,7 +49,10 @@ async function main(): Promise { } }, CALENDAR_SYNC_FLOOR_MS); - const client = initOpenWorkflow({ dbPath: config.database.queue }); + const client = initOpenWorkflow({ + dbPath: config.database.queue, + logger: createEngineLogger(logger) + }); registerWorkflows(); registerSyncCalendarsWorkflow(calendarSync); const worker = client.newWorker(); diff --git a/apps/worker/src/services/engine-logger.test.ts b/apps/worker/src/services/engine-logger.test.ts new file mode 100644 index 00000000..f6707b7a --- /dev/null +++ b/apps/worker/src/services/engine-logger.test.ts @@ -0,0 +1,91 @@ +import pino from 'pino'; +import { describe, expect, test } from 'vitest'; +import { createEngineLogger } from './engine-logger.js'; + +function captureLogs() { + let output = ''; + const log = pino( + { level: 'debug', base: null, timestamp: false }, + { + write(chunk: string) { + output += chunk; + } + } + ); + return { + engine: createEngineLogger(log), + output: () => output, + entries: () => + output + .trim() + .split('\n') + .map((line) => JSON.parse(line)) + }; +} + +const secret = 'SECRET_CREDENTIAL_SENTINEL'; +const capabilityUrl = 'https://example.test/appointment?token=CAPABILITY_SENTINEL'; +const unsafeError = new Error(`${secret} ${capabilityUrl}`, { + cause: new Error(`nested cause ${secret}`) +}); +unsafeError.name = secret; +unsafeError.stack = `unsafe stack ${capabilityUrl}`; +const unsafeArgs = [ + secret, + capabilityUrl, + unsafeError, + { nested: { password: secret, envelope: { html: capabilityUrl }, error: unsafeError } } +]; + +function expectNoSecrets(output: string) { + expect(output).not.toContain(secret); + expect(output).not.toContain('CAPABILITY_SENTINEL'); +} + +describe('createEngineLogger', () => { + test.each([ + ['Worker tick failed:', 'Workflow engine worker poll failed'], + ['Worker schedule tick failed:', 'Workflow engine schedule poll failed'], + ['Heartbeat failed:', 'Workflow engine heartbeat failed'], + [ + 'Critical error during workflow execution for run 936e818f-c6ab-4a1d-8af6-19d349482b50:', + 'Workflow engine execution failed' + ] + ])('classifies %s without forwarding any input', (message, expected) => { + const logs = captureLogs(); + logs.engine.error(`${message} ${secret} ${capabilityUrl}`, ...unsafeArgs); + expect(logs.entries()).toEqual([{ level: 50, msg: expected }]); + expectNoSecrets(logs.output()); + }); + + test.each([ + ['debug', 20], + ['info', 30], + ['warn', 40], + ['error', 50] + ] as const)('maps unknown %s output to fixed text at the same level', (method, level) => { + const logs = captureLogs(); + logs.engine[method](`unrecognized ${secret} ${capabilityUrl}`, ...unsafeArgs); + expect(logs.entries()).toEqual([{ level, msg: `Workflow engine ${method}` }]); + expectNoSecrets(logs.output()); + }); + + test('does not inspect or serialize unexpected message objects or arguments', () => { + const logs = captureLogs(); + const hostile = { + toString() { + throw new Error('must not stringify'); + }, + toJSON() { + throw new Error('must not serialize'); + } + }; + Reflect.apply(logs.engine.info, logs.engine, [hostile, ...unsafeArgs]); + logs.engine.warn('Heartbeat failed:', hostile, ...unsafeArgs); + expect(logs.entries()).toEqual([ + { level: 30, msg: 'Workflow engine info' }, + { level: 40, msg: 'Workflow engine heartbeat failed' } + ]); + expectNoSecrets(logs.output()); + }); +}); diff --git a/apps/worker/src/services/engine-logger.ts b/apps/worker/src/services/engine-logger.ts new file mode 100644 index 00000000..9078426b --- /dev/null +++ b/apps/worker/src/services/engine-logger.ts @@ -0,0 +1,28 @@ +import type { Logger as EngineLogger } from 'openworkflow'; +import type { Logger as PinoLogger } from './logger.js'; + +const operationalMessages = [ + ['Worker tick failed:', 'Workflow engine worker poll failed'], + ['Worker schedule tick failed:', 'Workflow engine schedule poll failed'], + ['Heartbeat failed:', 'Workflow engine heartbeat failed'], + ['Critical error during workflow execution for run ', 'Workflow engine execution failed'] +] as const; + +function safeMessage(message: string, level: 'debug' | 'info' | 'warn' | 'error'): string { + if (typeof message === 'string') { + for (const [prefix, fixedMessage] of operationalMessages) { + if (message.startsWith(prefix)) return fixedMessage; + } + } + return `Workflow engine ${level}`; +} + +/** Keep operational categories, never engine text or arguments that may contain secrets. */ +export function createEngineLogger(log: PinoLogger): EngineLogger { + return { + debug: (message) => log.debug(safeMessage(message, 'debug')), + info: (message) => log.info(safeMessage(message, 'info')), + warn: (message) => log.warn(safeMessage(message, 'warn')), + error: (message) => log.error(safeMessage(message, 'error')) + }; +} diff --git a/apps/worker/src/services/metrics.ts b/apps/worker/src/services/metrics.ts index 3288a3ba..43c085e9 100644 --- a/apps/worker/src/services/metrics.ts +++ b/apps/worker/src/services/metrics.ts @@ -44,28 +44,28 @@ for (const metric of metricsToPrune) { export const jobsTotal = new Counter({ name: 'when_jobs_total', - help: 'Total number of background jobs processed.', + help: 'Workflow handler executions by return (success) or escape (failure, including durable suspension), not terminal job outcomes.', labelNames: ['job_name', 'status'], registers: [register] }); export const jobDuration = new Histogram({ name: 'when_job_duration_seconds', - help: 'Execution duration of background jobs.', + help: 'Duration of each workflow handler execution, including executions that durably suspend.', labelNames: ['job_name'], registers: [register] }); export const jobsActive = new Gauge({ name: 'when_jobs_active', - help: 'Number of background jobs currently running.', + help: 'Number of active workflow handler executions; durably suspended runs are excluded.', labelNames: ['job_name'], registers: [register] }); export const emailsTotal = new Counter({ name: 'when_emails_total', - help: 'Total number of emails sent.', + help: 'Physical SMTP send attempts by outcome, not unique email deliveries.', labelNames: ['recipient_type', 'status', 'email_kind'], registers: [register] }); diff --git a/apps/worker/src/workflows/children.integration.test.ts b/apps/worker/src/workflows/children.integration.test.ts new file mode 100644 index 00000000..61846ae2 --- /dev/null +++ b/apps/worker/src/workflows/children.integration.test.ts @@ -0,0 +1,374 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { OpenWorkflow, defineWorkflowSpec, type RetryPolicy, type Worker } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { reconcileAppointment } from '@when/jobs'; +import { describe, expect, test, vi } from 'vitest'; + +type Role = 'guest' | 'host'; +const roles: Role[] = ['guest', 'host']; + +// Keep the real budgets; only shorten native delays, not the number of attempts. +const deliveryRetry: RetryPolicy = { + maximumAttempts: 5, + initialInterval: '10ms', + backoffCoefficient: 2, + maximumInterval: '20ms' +}; + +interface Delivery { + send(role: Role): Promise; + status(role: Role): Promise; + parkRole?: Role; + waitTimeout?: number; +} + +function openQueue(path: string, delivery: Delivery) { + const backend = BackendSqlite.connect(path); + const client = new OpenWorkflow({ backend }); + const child = defineWorkflowSpec<{ role: Role }, 'sent'>({ + name: 'delivery-child', + retryPolicy: { maximumAttempts: 3 } + }); + const parent = defineWorkflowSpec, 'sent'[]>({ + name: 'delivery-parent', + retryPolicy: { + ...reconcileAppointment.retryPolicy, + initialInterval: '10ms', + maximumInterval: '20ms' + } + }); + client.implementWorkflow(child, async ({ input, step }) => { + if (input.role === delivery.parkRole) { + await step.waitForSignal({ name: 'release', signal: `release:${input.role}` }); + } + await step.run({ name: 'smtp', retryPolicy: deliveryRetry }, async () => { + await delivery.send(input.role); + return 'sent'; + }); + await step.run({ name: 'status', retryPolicy: deliveryRetry }, async () => { + await delivery.status(input.role); + return 'recorded'; + }); + return 'sent'; + }); + client.implementWorkflow(parent, ({ step }) => + Promise.all( + roles.map((role) => + step.runWorkflow(child, { role }, { name: role, timeout: delivery.waitTimeout }) + ) + ) + ); + const worker = client.newWorker({ concurrency: 1 }); + return { + backend, + client, + worker, + parent, + async close() { + try { + await worker.stop(); + } finally { + await backend.stop(); + } + } + }; +} + +type Queue = ReturnType; + +async function runs(queue: Queue) { + const page = await queue.backend.listWorkflowRuns({ limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data; +} + +async function attempts(queue: Queue, workflowRunId: string, stepName?: string) { + const page = await queue.backend.listStepAttempts({ workflowRunId, limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data.filter((attempt) => stepName === undefined || attempt.stepName === stepName); +} + +async function readRun(queue: Queue, workflowRunId: string) { + const run = await queue.backend.getWorkflowRun({ workflowRunId }); + expect(run).not.toBeNull(); + return run!; +} + +async function driveUntil(worker: Worker, assertion: () => Promise) { + await vi.waitFor( + async () => { + await worker.tick(); + // Observe a finished/parked pass, not an intermediate database write. + await worker.stop(); + await assertion(); + }, + { timeout: 3000, interval: 5 } + ); +} + +// No background poll loop: each tick admits work only when the assertion needs it. +async function admitted(queue: Queue, parentId: string) { + expect(await queue.worker.tick()).toBe(1); + await queue.worker.stop(); + await vi.waitFor( + async () => { + const parent = await readRun(queue, parentId); + expect(parent).toMatchObject({ status: 'running', workerId: null }); + const children = (await runs(queue)).filter((run) => run.id !== parentId); + expect(children).toHaveLength(2); + expect(children.map((run) => run.input)).toEqual( + expect.arrayContaining(roles.map((role) => ({ role }))) + ); + const links = await attempts(queue, parentId); + expect(links).toHaveLength(2); + expect(links.map((attempt) => attempt.childWorkflowRunId).sort()).toEqual( + children.map((run) => run.id).sort() + ); + }, + { timeout: 3000, interval: 5 } + ); +} + +async function childFor(queue: Queue, role: Role) { + const children = (await runs(queue)).filter((run) => run.workflowName === 'delivery-child'); + expect(children).toHaveLength(2); + const child = children.find((run) => JSON.stringify(run.input) === JSON.stringify({ role })); + expect(child).toBeDefined(); + return child!; +} + +function fakeDelivery( + failures: Record, + statusFailures: Partial> = {} +) { + const sends = { guest: 0, host: 0 }; + const statuses = { guest: 0, host: 0 }; + return { + sends, + statuses, + send: vi.fn(async (role: Role) => { + sends[role] += 1; + if (sends[role] <= failures[role]) throw new Error(`SMTP ${role} unavailable`); + }), + status: vi.fn(async (role: Role) => { + statuses[role] += 1; + if (statuses[role] <= (statusFailures[role] ?? 0)) { + throw new Error(`status ${role} unavailable`); + } + }) + }; +} + +describe('native parallel child workflows at concurrency one', () => { + test('admits both children, parks the parent and gives each child its full native budget', async () => { + const delivery = fakeDelivery({ guest: 2, host: 4 }); + const queue = openQueue(':memory:', delivery); + try { + expect(queue.parent.retryPolicy?.maximumAttempts).toBe(3); + const parent = await queue.client.runWorkflow(queue.parent, {}); + await admitted(queue, parent.workflowRun.id); + await driveUntil(queue.worker, async () => { + expect(await readRun(queue, parent.workflowRun.id)).toMatchObject({ + status: 'completed', + output: ['sent', 'sent'] + }); + }); + expect(await parent.result({ timeoutMs: 500 })).toEqual(['sent', 'sent']); + expect(delivery.sends).toEqual({ guest: 3, host: 5 }); + for (const role of roles) { + const child = await childFor(queue, role); + expect(child).toMatchObject({ status: 'completed', output: 'sent' }); + const smtp = await attempts(queue, child.id, 'smtp'); + expect(smtp.filter((attempt) => attempt.status === 'failed')).toHaveLength( + role === 'guest' ? 2 : 4 + ); + expect(smtp.filter((attempt) => attempt.status === 'completed')).toHaveLength(1); + } + expect(await runs(queue)).toHaveLength(3); + expect(await attempts(queue, parent.workflowRun.id)).toHaveLength(2); + } finally { + await queue.close(); + } + }); + + test.each(roles)( + '%s exhaustion leaves its sibling independent and eventually fails the parent', + async (failedRole) => { + const siblingRole = failedRole === 'guest' ? 'host' : 'guest'; + const delivery = fakeDelivery({ [failedRole]: 5, [siblingRole]: 4 } as Record); + const queue = openQueue(':memory:', { ...delivery, parkRole: siblingRole }); + try { + const parent = await queue.client.runWorkflow(queue.parent, {}); + await admitted(queue, parent.workflowRun.id); + await driveUntil(queue.worker, async () => { + expect(await childFor(queue, failedRole)).toMatchObject({ + status: 'failed', + output: null, + error: { message: `SMTP ${failedRole} unavailable` } + }); + expect(await childFor(queue, siblingRole)).toMatchObject({ + status: 'running', + workerId: null + }); + }); + const failed = await childFor(queue, failedRole); + expect(failed.status).toBe('failed'); + expect(await attempts(queue, failed.id, 'smtp')).toHaveLength(5); + expect( + (await attempts(queue, failed.id, 'smtp')).every((attempt) => attempt.status === 'failed') + ).toBe(true); + expect(delivery.sends[failedRole]).toBe(5); + expect(delivery.sends[siblingRole]).toBe(0); + // Failure propagation can wait for the sibling; it must not cancel it. + expect((await readRun(queue, parent.workflowRun.id)).status).not.toBe('completed'); + const sibling = await childFor(queue, siblingRole); + expect(sibling).toMatchObject({ status: 'running', workerId: null }); + expect(await attempts(queue, sibling.id, 'release')).toHaveLength(1); + expect(await queue.client.sendSignal({ signal: `release:${siblingRole}` })).toEqual({ + workflowRunIds: [sibling.id] + }); + await driveUntil(queue.worker, async () => { + expect(await readRun(queue, sibling.id)).toMatchObject({ + status: 'completed', + output: 'sent' + }); + expect(await readRun(queue, parent.workflowRun.id)).toMatchObject({ + status: 'failed', + output: null, + error: { message: `SMTP ${failedRole} unavailable` } + }); + }); + await expect(parent.result({ timeoutMs: 500 })).rejects.toThrow( + `SMTP ${failedRole} unavailable` + ); + expect(delivery.sends).toEqual({ guest: 5, host: 5 }); + expect(await attempts(queue, sibling.id, 'smtp')).toHaveLength(5); + expect((await readRun(queue, parent.workflowRun.id)).status).toBe('failed'); + expect(await runs(queue)).toHaveLength(3); + } finally { + await queue.close(); + } + } + ); + + test('both children can exhaust five attempts independently', async () => { + const delivery = fakeDelivery({ guest: 5, host: 5 }); + const queue = openQueue(':memory:', delivery); + try { + const parent = await queue.client.runWorkflow(queue.parent, {}); + await admitted(queue, parent.workflowRun.id); + await driveUntil(queue.worker, async () => { + expect((await runs(queue)).map((run) => run.status)).toEqual([ + 'failed', + 'failed', + 'failed' + ]); + }); + expect(delivery.sends).toEqual({ guest: 5, host: 5 }); + for (const role of roles) { + expect(await attempts(queue, (await childFor(queue, role)).id, 'smtp')).toHaveLength(5); + } + } finally { + await queue.close(); + } + }); + + test('reopens a file-backed queue during child retry without duplicating children or saved SMTP success', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-children-')); + const delivery = fakeDelivery({ guest: 0, host: 1 }, { host: 2 }); + let queue = openQueue(join(directory, 'queue.sqlite'), delivery); + try { + const parent = await queue.client.runWorkflow(queue.parent, {}); + const parentId = parent.workflowRun.id; + await admitted(queue, parentId); + await driveUntil(queue.worker, async () => { + const host = await childFor(queue, 'host'); + expect(host).toMatchObject({ status: 'pending', workerId: null }); + expect(await attempts(queue, host.id, 'smtp')).toHaveLength(1); + }); + const childIds = (await runs(queue)).map((run) => run.id).sort(); + await queue.close(); + queue = openQueue(join(directory, 'queue.sqlite'), delivery); + await driveUntil(queue.worker, async () => { + const host = await childFor(queue, 'host'); + expect(host).toMatchObject({ status: 'pending', workerId: null }); + expect(await attempts(queue, host.id, 'status')).toHaveLength(1); + expect(await attempts(queue, host.id, 'smtp')).toHaveLength(2); + expect((await childFor(queue, 'guest')).status).toBe('completed'); + }); + const host = await childFor(queue, 'host'); + const savedSmtp = await attempts(queue, host.id, 'smtp'); + const savedGuest = await attempts(queue, (await childFor(queue, 'guest')).id); + await queue.close(); + queue = openQueue(join(directory, 'queue.sqlite'), delivery); + await driveUntil(queue.worker, async () => { + expect(await readRun(queue, parentId)).toMatchObject({ + status: 'completed', + output: ['sent', 'sent'] + }); + }); + expect((await runs(queue)).map((run) => run.id).sort()).toEqual(childIds); + expect(await attempts(queue, parentId)).toHaveLength(2); + expect(await attempts(queue, host.id, 'smtp')).toEqual(savedSmtp); + expect(await attempts(queue, (await childFor(queue, 'guest')).id)).toEqual(savedGuest); + expect((await attempts(queue, host.id, 'status')).map((attempt) => attempt.status)).toEqual([ + 'failed', + 'failed', + 'completed' + ]); + expect(delivery.sends).toEqual({ guest: 1, host: 2 }); + expect(delivery.statuses).toEqual({ guest: 1, host: 3 }); + } finally { + try { + await queue.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + } + }); + + test.each(['cancel', 'timeout'] as const)( + 'parent %s leaves its already-started child alive', + async (mode) => { + const delivery = fakeDelivery({ guest: 0, host: 0 }); + const queue = openQueue(':memory:', { + ...delivery, + parkRole: 'host', + waitTimeout: mode === 'timeout' ? 50 : undefined + }); + try { + const parent = await queue.client.runWorkflow(queue.parent, {}); + await admitted(queue, parent.workflowRun.id); + await driveUntil(queue.worker, async () => { + const host = await childFor(queue, 'host'); + expect(host).toMatchObject({ status: 'running', workerId: null }); + expect(await attempts(queue, host.id, 'release')).toHaveLength(1); + }); + if (mode === 'cancel') await parent.cancel(); + await driveUntil(queue.worker, async () => { + expect((await readRun(queue, parent.workflowRun.id)).status).toBe( + mode === 'cancel' ? 'canceled' : 'failed' + ); + }); + await expect(parent.result({ timeoutMs: 500 })).rejects.toThrow( + mode === 'cancel' ? /cancel/i : /Timed out waiting for child workflow/ + ); + const host = await childFor(queue, 'host'); + expect(host.status).toBe('running'); + await queue.client.sendSignal({ signal: 'release:host' }); + await driveUntil(queue.worker, async () => { + expect(await readRun(queue, host.id)).toMatchObject({ + status: 'completed', + output: 'sent' + }); + }); + expect(delivery.sends.host).toBe(1); + expect(await runs(queue)).toHaveLength(3); + } finally { + await queue.close(); + } + } + ); +}); diff --git a/packages/jobs/README.md b/packages/jobs/README.md index 61ad98ec..de1750fd 100644 --- a/packages/jobs/README.md +++ b/packages/jobs/README.md @@ -1,23 +1,43 @@ # @when/jobs -Shared job/workflow contract for the `when` workspace, built on -[openworkflow](https://openworkflow.dev) (durable execution on `node:sqlite`, -zero native addons). +Shared workflow contracts for the `when` workspace, built on +[OpenWorkflow](https://openworkflow.dev) `0.11.0` over `node:sqlite`. -This package is the single source of truth for **what** jobs exist and their -input/output shapes. The producer (`@when/web`) triggers runs from a spec; the -worker (`@when/worker`) provides the implementation. Keeping the spec here lets -both sides share types and resolve the workflow by name without web depending on -the worker. +The producer (`@when/web`, or `@when/cli` for service probes) enqueues work from a +shared spec; `@when/worker` registers the implementation. Web never imports worker +implementations. Initializing a producer client does not start a worker or register +recurring schedules. ## API -- `sendAppointmentEmail` — the `send-appointment-email` workflow spec. -- Types: `AppointmentEmailKind`, `SendAppointmentEmailInput`, `SendAppointmentEmailResult`. -- `createBackend(path, opts?)` / `createClient(path, opts?)` — connect a - `node:sqlite` openworkflow backend / build an `OpenWorkflow` client over it. +- `reconcileAppointment` — coordinate appointment calendar/video updates and email. +- `purgeAppointment` — clean up purged appointment chains. +- `syncCalendars` — request the worker's calendar repair scan. +- `testEmail`, `testProvider`, `listProviderCalendars`, `testCalendar`, + `generateVideoChat` — on-demand worker operations. +- `initOpenWorkflow({ dbPath, logger? })` — open and migrate the queue database and + initialize the process-wide client. The first call wins, including its logger. +- `getOpenWorkflow()` — return that client; throws before initialization or after + closing. +- `closeOpenWorkflow()` — idempotently close the queue connection, **after stopping + its worker**. Closing an initialized client is terminal for that process. Calling + this before initialization is a no-op. +- `getWorkflowRun(db, id)`, `getStepAttempts(db, id)` — read queue history through a + caller-owned `DatabaseSync` connection. + +Specs and their input/output types live in `src/specs.ts`. The `./specs` entrypoint +exports contracts without the process-wide client helpers. + +Queue migrations run on connection, including producer connections. Existing step +results survive the engine upgrade, but that does not make a changed application +workflow graph safe to replay: drain old appointment work before deploying changed +handlers. Back up both SQLite databases consistently before an engine upgrade. ## Scripts -- `pnpm build` — typecheck (`tsc --noEmit`). -- `pnpm test` / `pnpm test:coverage` / `pnpm lint`. +- `build` — typecheck and compile to `dist/`. +- `check` — typecheck without emitting. +- `test` / `test:coverage` — unit and real-engine integration tests. +- `lint` — Prettier and ESLint. + +Tests use temporary queues, never a personal deployment database. diff --git a/packages/jobs/package.json b/packages/jobs/package.json index 34ce9c16..dfe1b1d2 100644 --- a/packages/jobs/package.json +++ b/packages/jobs/package.json @@ -30,7 +30,7 @@ "dependencies": { "@when/config": "workspace:*", "@when/db": "workspace:*", - "openworkflow": "0.9.0" + "openworkflow": "0.11.0" }, "devDependencies": { "@eslint/js": "catalog:", diff --git a/packages/jobs/src/index.ts b/packages/jobs/src/index.ts index c0eeabdc..4d7b04f6 100644 --- a/packages/jobs/src/index.ts +++ b/packages/jobs/src/index.ts @@ -31,6 +31,7 @@ export type { } from './specs.js'; export { initOpenWorkflow, + closeOpenWorkflow, getOpenWorkflow, getWorkflowRun, getStepAttempts diff --git a/packages/jobs/src/openworkflow.test.ts b/packages/jobs/src/openworkflow.test.ts index 74204d9f..d9920919 100644 --- a/packages/jobs/src/openworkflow.test.ts +++ b/packages/jobs/src/openworkflow.test.ts @@ -1,128 +1,186 @@ -import { describe, expect, test } from 'vitest'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; -import { - initOpenWorkflow, - getOpenWorkflow, - getWorkflowRun, - getStepAttempts -} from './openworkflow.js'; +import { OpenWorkflow, defineWorkflowSpec } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { getWorkflowRun, getStepAttempts } from './openworkflow.js'; import { reconcileAppointment } from './specs.js'; -import type { ReconcileAppointmentInput } from './specs.js'; describe('client singleton', () => { - test('getOpenWorkflow throws before initialization', () => { - expect(() => getOpenWorkflow()).toThrow(/has not been initialized/); + let api: typeof import('./index.js'); + + beforeEach(async () => { + vi.resetModules(); + api = await import('./index.js'); }); - test('initOpenWorkflow returns a reusable singleton', () => { - // :memory: backend auto-migrates on connect. - const client = initOpenWorkflow({ dbPath: ':memory:' }); - expect(client).toBeDefined(); - expect(getOpenWorkflow()).toBe(client); - // Idempotent: a second init keeps the first instance. - expect(initOpenWorkflow({ dbPath: ':memory:' })).toBe(client); + afterEach(async () => { + await api.closeOpenWorkflow(); }); - test('the client can enqueue a run (producer-only path)', async () => { - const input: ReconcileAppointmentInput = { - appointmentId: 'appt-1', + test('get throws before initialization; closing before init does not poison later init', async () => { + expect(() => api.getOpenWorkflow()).toThrow(/has not been initialized/); + await api.closeOpenWorkflow(); + const client = api.initOpenWorkflow({ dbPath: ':memory:' }); + expect(api.getOpenWorkflow()).toBe(client); + }); + + test('first initialization wins and a producer can enqueue without worker registration', async () => { + const client = api.initOpenWorkflow({ dbPath: ':memory:' }); + expect(api.initOpenWorkflow({ dbPath: ':memory:' })).toBe(client); + expect(api.getOpenWorkflow()).toBe(client); + const handle = await client.runWorkflow(reconcileAppointment, { + appointmentId: 'synthetic-appointment', emailKind: 'confirmed' - }; - - const handle = await getOpenWorkflow().runWorkflow(reconcileAppointment, input); - + }); expect(handle.workflowRun.id).toBeTruthy(); expect(handle.workflowRun.status).toBe('pending'); }); + + test('public close is shared, idempotent and terminal for init/get and the retained client', async () => { + const client = api.initOpenWorkflow({ dbPath: ':memory:' }); + const closing = api.closeOpenWorkflow(); + expect(api.closeOpenWorkflow()).toBe(closing); + expect(() => api.getOpenWorkflow()).toThrow(/has been closed/); + expect(() => api.initOpenWorkflow({ dbPath: ':memory:' })).toThrow(/has been closed/); + await closing; + expect(api.closeOpenWorkflow()).toBe(closing); + await expect( + client.runWorkflow(reconcileAppointment, { appointmentId: 'after-close' }) + ).rejects.toThrow(); + }); }); -describe('read helpers', () => { - function seedQueueDb(): DatabaseSync { - const db = new DatabaseSync(':memory:'); - db.exec(` - CREATE TABLE workflow_runs ( - id TEXT PRIMARY KEY, - workflow_name TEXT NOT NULL, - status TEXT NOT NULL, - error TEXT, - input TEXT, - output TEXT, - finished_at TEXT, - created_at TEXT NOT NULL - ); - CREATE TABLE step_attempts ( - id TEXT PRIMARY KEY, - workflow_run_id TEXT NOT NULL, - step_name TEXT NOT NULL, - status TEXT NOT NULL, - started_at TEXT, - finished_at TEXT, - error TEXT, - created_at TEXT NOT NULL - ); - `); - db.prepare( - `INSERT INTO workflow_runs (id, workflow_name, status, error, input, output, finished_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - ).run( - 'run-1', - 'reconcile-appointment', - 'completed', - null, - '{}', - '"reconciled"', - null, - '2026-01-01T10:00:00Z' - ); - db.prepare( - `INSERT INTO step_attempts (id, workflow_run_id, step_name, status, started_at, finished_at, error, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - ).run( - 'step-1', - 'run-1', - 'smtp:jane@example.com', - 'completed', - '2026-01-01T10:00:01Z', - '2026-01-01T10:00:02Z', - null, - '2026-01-01T10:00:01Z' - ); - db.prepare( - `INSERT INTO step_attempts (id, workflow_run_id, step_name, status, started_at, finished_at, error, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - ).run( - 'step-2', - 'run-1', - 'status', - 'completed', - '2026-01-01T10:00:03Z', - '2026-01-01T10:00:04Z', - null, - '2026-01-01T10:00:03Z' - ); - return db; - } +describe('real SQLite engine integration', () => { + let directory: string; + let raw: DatabaseSync; + let backend: BackendSqlite; + let client: OpenWorkflow; + let worker: ReturnType | undefined; - test('getWorkflowRun returns a run by id, or null when missing', () => { - const db = seedQueueDb(); - const run = getWorkflowRun(db, 'run-1'); - expect(run).not.toBeNull(); - expect(run!.workflow_name).toBe('reconcile-appointment'); - expect(run!.status).toBe('completed'); - expect(getWorkflowRun(db, 'missing')).toBeNull(); + beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'when-queue-read-')); + const path = join(directory, 'queue.sqlite'); + backend = BackendSqlite.connect(path); + client = new OpenWorkflow({ backend }); + raw = new DatabaseSync(path); }); - test("getStepAttempts lists a run's steps in chronological order", () => { - const db = seedQueueDb(); - const steps = getStepAttempts(db, 'run-1'); - expect(steps).toHaveLength(2); - expect(steps[0]).toEqual({ - phase: 'smtp:jane@example.com', - status: 'completed', - started_at: '2026-01-01T10:00:01Z', - finished_at: '2026-01-01T10:00:02Z', - message: null + afterEach(async () => { + try { + await worker?.stop(); + await backend.stop(); + } finally { + worker = undefined; + vi.restoreAllMocks(); + vi.useRealTimers(); + raw.close(); + await rm(directory, { recursive: true, force: true }); + } + }); + + test('raw readers expose real run and chronological step rows, including missing ids', async () => { + const workflow = client.defineWorkflow({ name: 'reader-probe' }, async ({ step }) => { + await step.run({ name: 'first' }, async () => { + await new Promise((resolve) => setTimeout(resolve, 2)); + return 'saved'; + }); + return step.run({ name: 'second' }, () => 'done'); }); - expect(steps[1].phase).toBe('status'); + const handle = await workflow.run({}); + expect(getWorkflowRun(raw, handle.workflowRun.id)).toMatchObject({ status: 'pending' }); + expect(getStepAttempts(raw, handle.workflowRun.id)).toEqual([]); + worker = client.newWorker({ concurrency: 1 }); + expect(await worker.tick()).toBe(1); + await worker.stop(); + const run = getWorkflowRun(raw, handle.workflowRun.id)!; + expect(run).toMatchObject({ + workflow_name: 'reader-probe', + status: 'completed', + input: '{}', + output: '"done"', + error: null + }); + expect(run.finished_at).toEqual(expect.any(String)); + const steps = getStepAttempts(raw, handle.workflowRun.id); + expect(steps).toEqual( + ['first', 'second'].map((phase) => ({ + phase, + status: 'completed', + message: null, + started_at: expect.any(String), + finished_at: expect.any(String) + })) + ); + expect(steps[0].started_at! <= steps[1].started_at!).toBe(true); + expect(getWorkflowRun(raw, 'missing')).toBeNull(); + expect(getStepAttempts(raw, 'missing')).toEqual([]); + }); + + test.each([ + ['completed', 10], + ['completed', 11], + ['failed', 10], + ['failed', 11], + ['canceled', 10], + ['canceled', 11] + ] as const)( + 'result() observes %s when its read crosses the %ims timeout boundary', + async (status, elapsed) => { + vi.useFakeTimers({ toFake: ['Date'] }); + const start = new Date('2026-02-01T00:00:00.000Z'); + vi.setSystemTime(start); + const workflow = client.defineWorkflow( + { name: 'boundary-probe', retryPolicy: { maximumAttempts: 1 } }, + () => { + if (status === 'failed') throw new Error('synthetic boundary failure'); + return 'finished'; + } + ); + const handle = await workflow.run({}); + if (status === 'canceled') { + await handle.cancel(); + } else { + worker = client.newWorker({ concurrency: 1 }); + expect(await worker.tick()).toBe(1); + await worker.stop(); + } + expect(getWorkflowRun(raw, handle.workflowRun.id)!.status).toBe(status); + // Model a backend read completing at/after the deadline. Exercise the actual + // returned handle, not a local reproduction of its polling algorithm. + const read = backend.getWorkflowRun.bind(backend); + vi.spyOn(backend, 'getWorkflowRun').mockImplementation(async (params) => { + const run = await read(params); + vi.setSystemTime(start.getTime() + elapsed); + return run; + }); + if (status === 'completed') { + await expect(handle.result({ timeoutMs: 10 })).resolves.toBe('finished'); + } else { + await expect(handle.result({ timeoutMs: 10 })).rejects.toThrow( + status === 'failed' ? /synthetic boundary failure/ : /canceled/ + ); + } + } + ); + + test('result() still times out if a read crosses the boundary with a nonterminal run', async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + const start = new Date('2026-02-01T00:00:00.000Z'); + vi.setSystemTime(start); + const handle = await client.runWorkflow( + defineWorkflowSpec({ name: 'unregistered-producer' }), + {} + ); + const read = backend.getWorkflowRun.bind(backend); + vi.spyOn(backend, 'getWorkflowRun').mockImplementation(async (params) => { + const run = await read(params); + vi.setSystemTime(start.getTime() + 11); + return run; + }); + await expect(handle.result({ timeoutMs: 10 })).rejects.toThrow(/Timed out/); + expect(getWorkflowRun(raw, handle.workflowRun.id)!.status).toBe('pending'); }); }); diff --git a/packages/jobs/src/openworkflow.ts b/packages/jobs/src/openworkflow.ts index eb8f3d13..7be1a00c 100644 --- a/packages/jobs/src/openworkflow.ts +++ b/packages/jobs/src/openworkflow.ts @@ -1,29 +1,34 @@ -import { OpenWorkflow } from 'openworkflow'; +import { OpenWorkflow, type Logger } from 'openworkflow'; import { BackendSqlite } from 'openworkflow/sqlite'; import type { DatabaseSync } from 'node:sqlite'; let instance: OpenWorkflow | null = null; +let backend: BackendSqlite | null = null; +let closePromise: Promise | null = null; export interface InitOpenWorkflowOptions { - /** Path to the openworkflow `node:sqlite` queue DB (or `:memory:`). */ dbPath: string; + logger?: Logger; } -/** - * Initialize the process-wide openworkflow client over a `node:sqlite` backend. - * Idempotent: the first call wins; later calls return the same instance. The - * worker calls this at boot; the web producer keeps its own client (it lives in - * a separate process), so the singleton is per-process either way. - */ -export function initOpenWorkflow({ dbPath }: InitOpenWorkflowOptions): OpenWorkflow { +export function initOpenWorkflow({ dbPath, logger }: InitOpenWorkflowOptions): OpenWorkflow { + if (closePromise) throw new Error('openworkflow client has been closed'); if (!instance) { - instance = new OpenWorkflow({ backend: BackendSqlite.connect(dbPath) }); + backend = BackendSqlite.connect(dbPath); + instance = new OpenWorkflow({ backend, logger }); } return instance; } -/** The initialized client. Throws if `initOpenWorkflow` hasn't been called. */ +// Stop the worker before closing its database connection. +export function closeOpenWorkflow(): Promise { + if (!backend) return Promise.resolve(); + closePromise ??= backend.stop(); + return closePromise; +} + export function getOpenWorkflow(): OpenWorkflow { + if (closePromise) throw new Error('openworkflow client has been closed'); if (!instance) { throw new Error( 'openworkflow client has not been initialized; call initOpenWorkflow({ dbPath }) first' @@ -32,7 +37,6 @@ export function getOpenWorkflow(): OpenWorkflow { return instance; } -/** A row from openworkflow's `workflow_runs` table (the columns web cares about). */ export interface WorkflowRunRow { id: string; workflow_name: string; @@ -44,7 +48,6 @@ export interface WorkflowRunRow { created_at: string; } -/** A flattened step attempt, for surfacing run progress in the UI. */ export interface StepAttemptRow { phase: string; status: string; @@ -53,11 +56,6 @@ export interface StepAttemptRow { message: string | null; } -/** - * Look up a single workflow run by id. Read-only over a raw `DatabaseSync` - * handle to the queue DB, so consumers (web) can surface run status without - * pulling in the openworkflow engine. - */ export function getWorkflowRun(db: DatabaseSync, id: string): WorkflowRunRow | null { const row = db .prepare( @@ -68,7 +66,6 @@ export function getWorkflowRun(db: DatabaseSync, id: string): WorkflowRunRow | n return row ? (row as unknown as WorkflowRunRow) : null; } -/** List a run's step attempts in chronological order. */ export function getStepAttempts(db: DatabaseSync, workflowRunId: string): StepAttemptRow[] { const rows = db .prepare( diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 604f8a69..330663f8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -159,8 +159,8 @@ importers: specifier: 2.0.1 version: 2.0.1 openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.11.0 + version: 0.11.0(@opentelemetry/api@1.9.1) pino: specifier: 10.3.1 version: 10.3.1 @@ -277,8 +277,8 @@ importers: specifier: 9.0.3 version: 9.0.3 openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.11.0 + version: 0.11.0(@opentelemetry/api@1.9.1) pino: specifier: 10.3.1 version: 10.3.1 @@ -470,8 +470,8 @@ importers: specifier: workspace:* version: link:../db openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.11.0 + version: 0.11.0(@opentelemetry/api@1.9.1) devDependencies: '@eslint/js': specifier: 'catalog:' @@ -1608,6 +1608,10 @@ packages: resolution: {integrity: sha512-U71cyTamuh1CRNCfpGY6to28lxvNwPG4Guz/EVjgf3Jmzv0vlDp1atT9eS5dDjMYHucpHbWns6Lwf3BKz6svdw==} engines: {node: '>= 0.6'} + cron-parser@5.10.1: + resolution: {integrity: sha512-pKRrRagItwk9rGIStcUyMxFX34x56zoX3KDf9HJ4ttwkXIK1qxK63EvXvEDmlxI9AdPJ+Y7TEKRftRlW5eSS7A==} + engines: {node: '>=18'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -2048,6 +2052,10 @@ packages: resolution: {integrity: sha512-iPZK6eYjbxRu3uB4/WZ3EsEIMJFMqAoopl3R+zuq0UjcAm/MO6KCweDgPfP3elTztoKP3KtnVHxTn2NHBSDVUw==} engines: {node: '>=10'} + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + lz-string@1.5.0: resolution: {integrity: sha512-h5bgJWpxJNswbU7qCrV0tIKQCaS3blPDrqKWx+QxzuzL1zGUzij9XCWLrSLsJPu5t+eWA/ycetzYAO5IOMcWAQ==} hasBin: true @@ -2108,12 +2116,15 @@ packages: once@1.4.0: resolution: {integrity: sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==} - openworkflow@0.9.0: - resolution: {integrity: sha512-Ii6qkAh4ozbTHH5w7PdjjNYGF7u8xV2rsvTaBleo21rRlS8sEzrvDdxWsHawiY1VLfI1nfuRf8I3NDerlrugtg==} - engines: {node: '>=20'} + openworkflow@0.11.0: + resolution: {integrity: sha512-fdDfRjuNMLV+p3svZShrV5i/ZPbBS8GruWLlthk8Ezt3ueYPwVKk0mcD+knAmQ4J9RASBsY5JqiatiLtEcwgVw==} + engines: {node: '>=22.22.0'} peerDependencies: + '@opentelemetry/api': '1' postgres: ^3.4.9 peerDependenciesMeta: + '@opentelemetry/api': + optional: true postgres: optional: true @@ -3549,6 +3560,10 @@ snapshots: cookie@0.6.0: {} + cron-parser@5.10.1: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -3944,6 +3959,8 @@ snapshots: dependencies: p-locate: 5.0.0 + luxon@3.7.2: {} + lz-string@1.5.0: {} magic-string@0.30.21: @@ -3995,7 +4012,11 @@ snapshots: dependencies: wrappy: 1.0.2 - openworkflow@0.9.0: {} + openworkflow@0.11.0(@opentelemetry/api@1.9.1): + dependencies: + cron-parser: 5.10.1 + optionalDependencies: + '@opentelemetry/api': 1.9.1 optionator@0.9.4: dependencies: -- 2.54.0