diff --git a/.changeset/publish-admission-guard.md b/.changeset/publish-admission-guard.md new file mode 100644 index 00000000..ce807822 --- /dev/null +++ b/.changeset/publish-admission-guard.md @@ -0,0 +1,5 @@ +--- +"@luzzle/web": patch +--- + +Upgrade OpenWorkflow to 0.9.2 and serialize publish and audit check-and-enqueue requests within one web process. Check the latest five runs per workflow through the SDK, retain active-run conflict responses and audit checks, and release admission on enqueue failure. Older active runs outside that window are not detected. diff --git a/.changeset/sdk-workflow-reads.md b/.changeset/sdk-workflow-reads.md new file mode 100644 index 00000000..a209826b --- /dev/null +++ b/.changeset/sdk-workflow-reads.md @@ -0,0 +1,5 @@ +--- +"@luzzle/web": patch +--- + +Use OpenWorkflow SDK APIs for workflow metadata reads in publish/preview pages, dashboard, connectivity checks and progress streams. Preserve existing payload shapes and step-attempt ordering, remove the unused numeric job-ID lookup and the web's raw queue connection, and keep retention cleanup SQL unchanged because the SDK has no purge API. diff --git a/apps/web/package.json b/apps/web/package.json index 265755ba..2e58332a 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -90,6 +90,6 @@ "@replit/codemirror-vim": "6.4.0", "eventsource-parser": "4.1.1", "kysely": "0.28.17", - "openworkflow": "0.9.0" + "openworkflow": "0.9.2" } } diff --git a/apps/web/src/lib/server/health/connectivity.test.ts b/apps/web/src/lib/server/health/connectivity.test.ts index 01d1ae9f..413136df 100644 --- a/apps/web/src/lib/server/health/connectivity.test.ts +++ b/apps/web/src/lib/server/health/connectivity.test.ts @@ -1,11 +1,11 @@ import { describe, test, expect, vi, beforeEach } from 'vitest' -import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js' +import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js' import { getWorkflowRun } from '@luzzle/web.jobs' import { testConnectivity } from './connectivity.js' vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflow: vi.fn(), - getOpenWorkflowDb: vi.fn() + getOpenWorkflowBackend: vi.fn() })) vi.mock('@luzzle/web.jobs', () => ({ @@ -18,7 +18,7 @@ vi.mock('@luzzle/web.jobs/specs', () => ({ const mocks = { getOpenWorkflow: vi.mocked(getOpenWorkflow), - getOpenWorkflowDb: vi.mocked(getOpenWorkflowDb), + getOpenWorkflowBackend: vi.mocked(getOpenWorkflowBackend), getWorkflowRun: vi.mocked(getWorkflowRun) } @@ -27,16 +27,18 @@ const runWorkflow = vi.fn() beforeEach(() => { vi.clearAllMocks() mocks.getOpenWorkflow.mockReturnValue({ runWorkflow } as never) - mocks.getOpenWorkflowDb.mockReturnValue({} as never) + mocks.getOpenWorkflowBackend.mockReturnValue({} as never) runWorkflow.mockResolvedValue({ workflowRun: { id: 'run-1' } }) }) describe('testConnectivity', () => { test('triggers the workflow and polls until it completes', async () => { - mocks.getWorkflowRun.mockReturnValueOnce({ status: 'running' } as never).mockReturnValueOnce({ - status: 'completed', - output: JSON.stringify({ ok: true }) - } as never) + mocks.getWorkflowRun + .mockResolvedValueOnce({ status: 'running' } as never) + .mockResolvedValueOnce({ + status: 'completed', + output: JSON.stringify({ ok: true }) + } as never) const result = await testConnectivity('archive', { pollIntervalMs: 1 }) @@ -48,7 +50,7 @@ describe('testConnectivity', () => { }) test('returns not-ok when the run fails', async () => { - mocks.getWorkflowRun.mockReturnValue({ status: 'failed', error: 'boom' } as never) + mocks.getWorkflowRun.mockResolvedValue({ status: 'failed', error: 'boom' } as never) const result = await testConnectivity('cdn', { pollIntervalMs: 1 }) @@ -56,7 +58,7 @@ describe('testConnectivity', () => { }) test('returns a generic reason when a failed run has no error message', async () => { - mocks.getWorkflowRun.mockReturnValue({ status: 'canceled', error: null } as never) + mocks.getWorkflowRun.mockResolvedValue({ status: 'canceled', error: null } as never) const result = await testConnectivity('cdn', { pollIntervalMs: 1 }) @@ -64,7 +66,7 @@ describe('testConnectivity', () => { }) test('returns not-ok when the output cannot be parsed', async () => { - mocks.getWorkflowRun.mockReturnValue({ status: 'completed', output: 'not json' } as never) + mocks.getWorkflowRun.mockResolvedValue({ status: 'completed', output: 'not json' } as never) const result = await testConnectivity('archive', { pollIntervalMs: 1 }) @@ -72,7 +74,7 @@ describe('testConnectivity', () => { }) test('times out waiting for the worker', async () => { - mocks.getWorkflowRun.mockReturnValue({ status: 'running' } as never) + mocks.getWorkflowRun.mockResolvedValue({ status: 'running' } as never) const result = await testConnectivity('archive', { pollIntervalMs: 5, timeoutMs: 20 }) diff --git a/apps/web/src/lib/server/health/connectivity.ts b/apps/web/src/lib/server/health/connectivity.ts index 41daa9a5..65d07a9a 100644 --- a/apps/web/src/lib/server/health/connectivity.ts +++ b/apps/web/src/lib/server/health/connectivity.ts @@ -1,4 +1,4 @@ -import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js' +import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js' import { getWorkflowRun, type TestConnectivityResult } from '@luzzle/web.jobs' import { testConnectivitySpec } from '@luzzle/web.jobs/specs' @@ -20,11 +20,11 @@ export async function testConnectivity( const openWorkflow = getOpenWorkflow() const handle = await openWorkflow.runWorkflow(testConnectivitySpec, { target }) const runId = handle.workflowRun.id - const openWorkflowDb = getOpenWorkflowDb() + const backend = getOpenWorkflowBackend() const deadline = Date.now() + timeoutMs while (Date.now() < deadline) { - const run = getWorkflowRun(openWorkflowDb, runId) + const run = await getWorkflowRun(backend, runId) if (run && TERMINAL_STATES.has(run.status)) { if (run.status === 'failed' || run.status === 'canceled') { return { ok: false, reason: run.error ?? `workflow ${run.status}` } diff --git a/apps/web/src/lib/server/load/dashboard.test.ts b/apps/web/src/lib/server/load/dashboard.test.ts index d48611cc..6e590d1a 100644 --- a/apps/web/src/lib/server/load/dashboard.test.ts +++ b/apps/web/src/lib/server/load/dashboard.test.ts @@ -2,7 +2,7 @@ import { describe, test, expect, vi, beforeEach } from 'vitest' import { config } from '$lib/server/config' import { getPieceStats, getAssetStats } from '../dashboardStats.js' import { getRecentlyEditedPieces } from '../pieces.js' -import { getOpenWorkflowDb } from '../workflow/index.js' +import { getOpenWorkflowBackend } from '../workflow/index.js' import { getLatestWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs' import { makeConfig, credentialsAuth, oidcAuth } from '../config.fixture' import { loadDashboardPage } from './dashboard.js' @@ -36,7 +36,7 @@ vi.mock('../pieces.js', () => ({ })) vi.mock('../workflow/index.js', () => ({ - getOpenWorkflowDb: vi.fn() + getOpenWorkflowBackend: vi.fn() })) vi.mock('@luzzle/web.jobs', () => ({ @@ -47,7 +47,7 @@ const mocks = { getPieceStats: vi.mocked(getPieceStats), getAssetStats: vi.mocked(getAssetStats), getRecentlyEditedPieces: vi.mocked(getRecentlyEditedPieces), - getOpenWorkflowDb: vi.mocked(getOpenWorkflowDb), + getOpenWorkflowBackend: vi.mocked(getOpenWorkflowBackend), getLatestWorkflowRun: vi.mocked(getLatestWorkflowRun) } @@ -70,8 +70,8 @@ beforeEach(() => { mocks.getPieceStats.mockResolvedValue({ total: 3, byType: [{ type: 'article', count: 3 }] }) mocks.getAssetStats.mockResolvedValue({ total: 2 }) mocks.getRecentlyEditedPieces.mockResolvedValue([]) - mocks.getOpenWorkflowDb.mockReturnValue({} as never) - mocks.getLatestWorkflowRun.mockReturnValue(null) + mocks.getOpenWorkflowBackend.mockReturnValue({} as never) + mocks.getLatestWorkflowRun.mockResolvedValue(null) config.auth = oidcAuth const defaults = makeConfig() config.sync = { @@ -82,7 +82,7 @@ beforeEach(() => { describe('loadDashboardPage', () => { test('assembles stats, setup config, and publish state into the page shape', async () => { - mocks.getLatestWorkflowRun.mockReturnValue(makeRun()) + mocks.getLatestWorkflowRun.mockResolvedValue(makeRun()) const result = await loadDashboardPage() @@ -103,7 +103,7 @@ describe('loadDashboardPage', () => { }) test('reports the latest publish attempt even if it failed', async () => { - mocks.getLatestWorkflowRun.mockReturnValue(makeRun({ status: 'failed' })) + mocks.getLatestWorkflowRun.mockResolvedValue(makeRun({ status: 'failed' })) const result = await loadDashboardPage() @@ -127,7 +127,7 @@ describe('loadDashboardPage', () => { }) test('falls back to null publish state when OpenWorkflow queries fail', async () => { - mocks.getOpenWorkflowDb.mockImplementation(() => { + mocks.getOpenWorkflowBackend.mockImplementation(() => { throw new Error('db unavailable') }) diff --git a/apps/web/src/lib/server/load/dashboard.ts b/apps/web/src/lib/server/load/dashboard.ts index dec5d194..a41fdb7a 100644 --- a/apps/web/src/lib/server/load/dashboard.ts +++ b/apps/web/src/lib/server/load/dashboard.ts @@ -2,7 +2,7 @@ import { db } from '$lib/server/database' import { config } from '$lib/server/config' import { getPieceStats, getAssetStats, type PieceStats } from '../dashboardStats.js' import { getRecentlyEditedPieces } from '../pieces.js' -import { getOpenWorkflowDb } from '../workflow/index.js' +import { getOpenWorkflowBackend } from '../workflow/index.js' import { getLatestWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs' import type { WebPieces } from '@luzzle/web.db' @@ -33,8 +33,7 @@ export async function loadDashboardPage(): Promise { let lastPublish: WorkflowRunRow | null = null try { - const openWorkflowDb = getOpenWorkflowDb() - lastPublish = getLatestWorkflowRun(openWorkflowDb, 'Publish') + lastPublish = await getLatestWorkflowRun(getOpenWorkflowBackend(), 'Publish') } catch (err) { console.error('Failed to query OpenWorkflow runs in dashboard loader:', err) } diff --git a/apps/web/src/lib/server/workflow/index.test.ts b/apps/web/src/lib/server/workflow/index.test.ts new file mode 100644 index 00000000..8dd202cf --- /dev/null +++ b/apps/web/src/lib/server/workflow/index.test.ts @@ -0,0 +1,31 @@ +import { expect, test, vi } from 'vitest' +import { publishSpec } from '@luzzle/web.jobs/specs' +import { config } from '$lib/server/config.js' +import { getOpenWorkflow, getOpenWorkflowBackend } from './index.js' +import { createWorkflowQueue, closeWorkflowQueue } from './publish.fixture.js' + +vi.mock('$lib/server/config.js', async () => { + const { makeConfig } = await import('$lib/server/config.fixture.js') + return { config: makeConfig() } +}) + +test('exposes the singleton SDK backend used by the web workflow client', async () => { + const queue = createWorkflowQueue() + config.worker.queue.path = `${queue.directory}/queue.sqlite` + const backend = getOpenWorkflowBackend() + try { + expect(getOpenWorkflowBackend()).toBe(backend) + const client = getOpenWorkflow() + expect(getOpenWorkflow()).toBe(client) + const handle = await client.runWorkflow(publishSpec, { bisync: false }) + const runs = await backend.listWorkflowRuns({ + workflowName: 'Publish', + status: 'pending', + limit: 1 + }) + expect(runs.data.map((run) => run.id)).toEqual([handle.workflowRun.id]) + } finally { + await backend.stop() + await closeWorkflowQueue(queue) + } +}) diff --git a/apps/web/src/lib/server/workflow/index.ts b/apps/web/src/lib/server/workflow/index.ts index 032201b4..3848786f 100644 --- a/apps/web/src/lib/server/workflow/index.ts +++ b/apps/web/src/lib/server/workflow/index.ts @@ -1,11 +1,10 @@ import path from 'node:path' -import { DatabaseSync } from 'node:sqlite' import { OpenWorkflow } from 'openworkflow' import { BackendSqlite } from 'openworkflow/sqlite' import { config } from '$lib/server/config.js' let openWorkflowInstance: OpenWorkflow | null = null -let openWorkflowDbInstance: DatabaseSync | null = null +let openWorkflowBackendInstance: BackendSqlite | null = null function resolveOpenWorkflowDbPath(): string { const queuePath = config.worker?.queue?.path || './data/sidequest.sqlite' @@ -13,22 +12,16 @@ function resolveOpenWorkflowDbPath(): string { return path.resolve(process.cwd(), openWorkflowPath) } +export function getOpenWorkflowBackend(): BackendSqlite { + if (!openWorkflowBackendInstance) { + openWorkflowBackendInstance = BackendSqlite.connect(resolveOpenWorkflowDbPath()) + } + return openWorkflowBackendInstance +} + export function getOpenWorkflow(): OpenWorkflow { if (!openWorkflowInstance) { - const dbPath = resolveOpenWorkflowDbPath() - const backend = BackendSqlite.connect(dbPath) - openWorkflowInstance = new OpenWorkflow({ backend }) + openWorkflowInstance = new OpenWorkflow({ backend: getOpenWorkflowBackend() }) } return openWorkflowInstance } - -export function getOpenWorkflowDb(): DatabaseSync { - if (!openWorkflowDbInstance) { - const db = new DatabaseSync(resolveOpenWorkflowDbPath()) - db.exec('PRAGMA journal_mode = WAL;') - db.exec('PRAGMA synchronous = NORMAL;') - db.exec('PRAGMA busy_timeout = 5000;') - openWorkflowDbInstance = db - } - return openWorkflowDbInstance -} diff --git a/apps/web/src/lib/server/workflow/publish.fixture.ts b/apps/web/src/lib/server/workflow/publish.fixture.ts new file mode 100644 index 00000000..31760248 --- /dev/null +++ b/apps/web/src/lib/server/workflow/publish.fixture.ts @@ -0,0 +1,62 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { BackendSqlite } from 'openworkflow/sqlite' + +export function createWorkflowQueue() { + const directory = mkdtempSync(path.join(tmpdir(), 'luzzle-workflow-')) + const filename = path.join(directory, 'queue.sqlite') + const backend = BackendSqlite.connect(filename) + return { directory, backend } +} + +export type WorkflowQueue = ReturnType + +export async function closeWorkflowQueue(queue: WorkflowQueue): Promise { + await queue.backend.stop() + rmSync(queue.directory, { recursive: true, force: true }) +} + +export function createPendingRun( + backend: BackendSqlite, + workflowName: string, + input: Record = {} +) { + return backend.createWorkflowRun({ + workflowName, + version: null, + idempotencyKey: null, + config: {}, + context: null, + input, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + availableAt: null, + deadlineAt: null + }) +} + +export async function claimNextRun(backend: BackendSqlite) { + const run = await backend.claimWorkflowRun({ + workerId: 'fixture-worker', + leaseDurationMs: 60_000 + }) + if (!run) throw new Error('Expected a pending workflow run') + return run +} + +export async function completeNextRun(backend: BackendSqlite) { + const run = await claimNextRun(backend) + return backend.completeWorkflowRun({ + workflowRunId: run.id, + workerId: 'fixture-worker', + output: null + }) +} + +export async function createCompletedRun(backend: BackendSqlite, workflowName: string) { + const run = await createPendingRun(backend, workflowName) + const completed = await completeNextRun(backend) + if (completed.id !== run.id) throw new Error('Fixture completed a different pending run') + return completed +} diff --git a/apps/web/src/lib/server/workflow/publish.test.ts b/apps/web/src/lib/server/workflow/publish.test.ts index 1ec04a7c..03c6d4ce 100644 --- a/apps/web/src/lib/server/workflow/publish.test.ts +++ b/apps/web/src/lib/server/workflow/publish.test.ts @@ -1,4 +1,12 @@ -import { describe, test, expect, vi, beforeEach } from 'vitest' +import { describe, test, expect, vi, beforeEach, afterEach } from 'vitest' +import { + createWorkflowQueue, + closeWorkflowQueue, + createPendingRun, + createCompletedRun, + claimNextRun, + type WorkflowQueue +} from './publish.fixture.js' import { getLatestWorkflowRun, getWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs' import { findInFlightPublishRun, validateAuditForPublish, parsePiecesDiff } from './publish.js' @@ -12,7 +20,7 @@ const mocks = { getWorkflowRun: vi.mocked(getWorkflowRun) } -const db = {} as never +const backend = {} as never function makeRun(overrides: Partial = {}): WorkflowRunRow { return { @@ -33,104 +41,202 @@ beforeEach(() => { }) describe('findInFlightPublishRun', () => { - test('returns the in-flight Publish run', () => { - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => - name === 'Publish' - ? makeRun({ id: 'pub-1', workflow_name: 'Publish', status: 'running' }) - : null - ) - expect(findInFlightPublishRun(db)?.id).toBe('pub-1') + let queue: WorkflowQueue + + beforeEach(() => { + vi.setSystemTime(new Date('2026-06-20T00:00:00Z')) + queue = createWorkflowQueue() }) - test('returns the in-flight PublishAudit run', () => { - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => - name === 'PublishAudit' ? makeRun({ id: 'aud-1', status: 'pending' }) : null - ) - expect(findInFlightPublishRun(db)?.id).toBe('aud-1') + afterEach(async () => { + vi.useRealTimers() + await closeWorkflowQueue(queue) }) - test('returns null when neither is in-flight', () => { - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => - makeRun({ workflow_name: name, status: 'completed' }) - ) - expect(findInFlightPublishRun(db)).toBeNull() + test.each([ + { name: 'Publish', status: 'pending' }, + { name: 'Publish', status: 'running' }, + { name: 'PublishAudit', status: 'pending' }, + { name: 'PublishAudit', status: 'running' } + ])('finds an older $status $name behind a newer completed run', async ({ name, status }) => { + const old = await createPendingRun(queue.backend, name) + await claimNextRun(queue.backend) + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + await createCompletedRun(queue.backend, name) + if (status === 'pending') { + await queue.backend.rescheduleWorkflowRunAfterFailedStepAttempt({ + workflowRunId: old.id, + workerId: 'fixture-worker', + error: { message: 'retry' }, + availableAt: new Date() + }) + } + expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: old.id }) }) - test('returns null when there are no runs', () => { - mocks.getLatestWorkflowRun.mockReturnValue(null) - expect(findInFlightPublishRun(db)).toBeNull() + test('returns an active run when both workflows have pending work', async () => { + const audit = await createPendingRun(queue.backend, 'PublishAudit') + const publish = await createPendingRun(queue.backend, 'Publish') + const active = await findInFlightPublishRun(queue.backend) + expect([audit.id, publish.id]).toContain(active?.id) + }) + + test('ignores finished runs and unrelated active workflows', async () => { + await createCompletedRun(queue.backend, 'PublishAudit') + await createPendingRun(queue.backend, 'Publish') + const failed = await claimNextRun(queue.backend) + await queue.backend.failWorkflowRun({ + workflowRunId: failed.id, + workerId: 'fixture-worker', + error: { message: 'failed' }, + retryPolicy: { + maximumAttempts: 1, + initialInterval: '1s', + backoffCoefficient: 2, + maximumInterval: '1s' + } + }) + await createPendingRun(queue.backend, 'Preview') + expect(await findInFlightPublishRun(queue.backend)).toBeNull() + }) + + test.each(['Publish', 'PublishAudit'])( + 'finds an active %s in the fifth-newest position', + async (name) => { + const old = await createPendingRun(queue.backend, name) + await claimNextRun(queue.backend) + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + for (let i = 0; i < 4; i++) await createCompletedRun(queue.backend, name) + expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: old.id }) + } + ) + + test.each(['Publish', 'PublishAudit'])( + 'accepts the known limit: a sixth-newest active %s is not detected', + async (name) => { + const old = await createPendingRun(queue.backend, name) + await claimNextRun(queue.backend) + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + for (let i = 0; i < 5; i++) await createCompletedRun(queue.backend, name) + expect(await findInFlightPublishRun(queue.backend)).toBeNull() + expect((await queue.backend.getWorkflowRun({ workflowRunId: old.id }))?.status).toBe( + 'running' + ) + } + ) + + test('checks only the latest five per workflow without paging, status filters or counts', async () => { + for (let i = 0; i < 5; i++) { + await createCompletedRun(queue.backend, 'Publish') + await createCompletedRun(queue.backend, 'PublishAudit') + } + await createPendingRun(queue.backend, 'Preview') + const list = vi.spyOn(queue.backend, 'listWorkflowRuns') + const count = vi.spyOn(queue.backend, 'countWorkflowRuns') + expect(await findInFlightPublishRun(queue.backend)).toBeNull() + expect(list.mock.calls.map(([params]) => params)).toEqual([ + { workflowName: 'Publish', limit: 5 }, + { workflowName: 'PublishAudit', limit: 5 } + ]) + expect(count).not.toHaveBeenCalled() + }) + + test('does not miss a running run rescheduled to pending during lookup', async () => { + const run = await createPendingRun(queue.backend, 'Publish') + await claimNextRun(queue.backend) + const original = queue.backend.listWorkflowRuns.bind(queue.backend) + let rescheduled = false + vi.spyOn(queue.backend, 'listWorkflowRuns').mockImplementation(async (params) => { + const page = await original(params) + if (params.workflowName === 'Publish' && !rescheduled) { + await queue.backend.rescheduleWorkflowRunAfterFailedStepAttempt({ + workflowRunId: run.id, + workerId: 'fixture-worker', + error: { message: 'retry' }, + availableAt: new Date() + }) + rescheduled = true + } + return page + }) + expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: run.id }) + expect((await queue.backend.getWorkflowRun({ workflowRunId: run.id }))?.status).toBe('pending') + }) + + test('returns null when there are no runs', async () => { + expect(await findInFlightPublishRun(queue.backend)).toBeNull() }) }) describe('validateAuditForPublish', () => { - test('rejects a missing or non-string id', () => { - expect(validateAuditForPublish(db, undefined).ok).toBe(false) - expect(validateAuditForPublish(db, '').ok).toBe(false) + test('rejects a missing or non-string id', async () => { + expect((await validateAuditForPublish(backend, undefined)).ok).toBe(false) + expect((await validateAuditForPublish(backend, '')).ok).toBe(false) }) - test('rejects when the run is not found', () => { - mocks.getWorkflowRun.mockReturnValue(null) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ + test('rejects when the run is not found', async () => { + mocks.getWorkflowRun.mockResolvedValue(null) + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: false, reason: 'audit run not found' }) }) - test('rejects when the run is not a PublishAudit', () => { - mocks.getWorkflowRun.mockReturnValue(makeRun({ workflow_name: 'Publish' })) - expect(validateAuditForPublish(db, 'audit-1').ok).toBe(false) + test('rejects when the run is not a PublishAudit', async () => { + mocks.getWorkflowRun.mockResolvedValue(makeRun({ workflow_name: 'Publish' })) + expect((await validateAuditForPublish(backend, 'audit-1')).ok).toBe(false) }) - test('rejects when the audit has not completed', () => { - mocks.getWorkflowRun.mockReturnValue(makeRun({ status: 'running' })) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ + test('rejects when the audit has not completed', async () => { + mocks.getWorkflowRun.mockResolvedValue(makeRun({ status: 'running' })) + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: false, reason: 'audit has not completed' }) }) - test('rejects when a newer audit has run since', () => { - mocks.getWorkflowRun.mockReturnValue(makeRun({ id: 'audit-1' })) - mocks.getLatestWorkflowRun.mockReturnValue(makeRun({ id: 'audit-2' })) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ + test('rejects when a newer audit has run since', async () => { + mocks.getWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-1' })) + mocks.getLatestWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-2' })) + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: false, reason: 'a newer audit has run; re-check before publishing' }) }) - test('rejects when a publish ran after the audit', () => { - mocks.getWorkflowRun.mockReturnValue( + test('rejects when a publish ran after the audit', async () => { + mocks.getWorkflowRun.mockResolvedValue( makeRun({ id: 'audit-1', created_at: '2026-06-20T00:00:00Z' }) ) - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => name === 'PublishAudit' ? makeRun({ id: 'audit-1', created_at: '2026-06-20T00:00:00Z' }) : makeRun({ id: 'pub-1', workflow_name: 'Publish', created_at: '2026-06-20T01:00:00Z' }) ) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: false, reason: 'changes were published after this check; re-check before publishing' }) }) - test('accepts a completed audit that is the latest with no later publish', () => { - mocks.getWorkflowRun.mockReturnValue(makeRun({ id: 'audit-1' })) - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => + test('accepts a completed audit that is the latest with no later publish', async () => { + mocks.getWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-1' })) + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => name === 'PublishAudit' ? makeRun({ id: 'audit-1' }) : null ) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ ok: true }) + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true }) }) - test('accepts a succeeded audit newer than the last publish', () => { - mocks.getWorkflowRun.mockReturnValue( + test('accepts a succeeded audit newer than the last publish', async () => { + mocks.getWorkflowRun.mockResolvedValue( makeRun({ id: 'audit-1', status: 'succeeded', created_at: '2026-06-20T02:00:00Z' }) ) - mocks.getLatestWorkflowRun.mockImplementation((_db, name) => + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => name === 'PublishAudit' ? makeRun({ id: 'audit-1', status: 'succeeded', created_at: '2026-06-20T02:00:00Z' }) : makeRun({ id: 'pub-1', workflow_name: 'Publish', created_at: '2026-06-20T01:00:00Z' }) ) - expect(validateAuditForPublish(db, 'audit-1')).toEqual({ ok: true }) + expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true }) }) }) diff --git a/apps/web/src/lib/server/workflow/publish.ts b/apps/web/src/lib/server/workflow/publish.ts index 54db0574..ba3f5b25 100644 --- a/apps/web/src/lib/server/workflow/publish.ts +++ b/apps/web/src/lib/server/workflow/publish.ts @@ -1,30 +1,46 @@ -import type { DatabaseSync } from 'node:sqlite' -import { getLatestWorkflowRun, getWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs' +import type { BackendSqlite } from 'openworkflow/sqlite' +import { getLatestWorkflowRun, getWorkflowRun } from '@luzzle/web.jobs' import type { PiecesDiff } from '@luzzle/core' -const PUBLISH_WORKFLOWS = ['Publish', 'PublishAudit'] as const -const IN_FLIGHT_STATES = new Set(['pending', 'running']) const COMPLETED_STATES = new Set(['completed', 'succeeded']) +let publishAdmission: Promise = Promise.resolve() -// Publish and PublishAudit are mutually exclusive: an audit's bisync mutates the archive a publish reads. -export function findInFlightPublishRun(db: DatabaseSync): WorkflowRunRow | null { - for (const name of PUBLISH_WORKFLOWS) { - const run = getLatestWorkflowRun(db, name) - if (run && IN_FLIGHT_STATES.has(run.status)) { - return run - } +export async function withPublishAdmission(operation: () => Promise): Promise { + const previous = publishAdmission + let release!: () => void + publishAdmission = new Promise((resolve) => { + release = resolve + }) + await previous + try { + return await operation() + } finally { + release() + } +} + +export async function findInFlightPublishRun( + backend: Pick +): Promise<{ id: string } | null> { + for (const workflowName of ['Publish', 'PublishAudit']) { + const { data } = await backend.listWorkflowRuns({ workflowName, limit: 5 }) + const active = data.find((run) => run.status === 'pending' || run.status === 'running') + if (active) return { id: active.id } } return null } export type AuditGuard = { ok: true } | { ok: false; reason: string } -export function validateAuditForPublish(db: DatabaseSync, auditRunId: unknown): AuditGuard { +export async function validateAuditForPublish( + backend: Pick, + auditRunId: unknown +): Promise { if (typeof auditRunId !== 'string' || auditRunId.length === 0) { return { ok: false, reason: 'no audit run provided; check for changes before publishing' } } - const run = getWorkflowRun(db, auditRunId) + const run = await getWorkflowRun(backend, auditRunId) if (!run || run.workflow_name !== 'PublishAudit') { return { ok: false, reason: 'audit run not found' } } @@ -32,13 +48,13 @@ export function validateAuditForPublish(db: DatabaseSync, auditRunId: unknown): return { ok: false, reason: 'audit has not completed' } } - const latest = getLatestWorkflowRun(db, 'PublishAudit') + const latest = await getLatestWorkflowRun(backend, 'PublishAudit') if (!latest || latest.id !== auditRunId) { return { ok: false, reason: 'a newer audit has run; re-check before publishing' } } // a publish consumes its audit; created_at is ISO so string compare is chronological - const lastPublish = getLatestWorkflowRun(db, 'Publish') + const lastPublish = await getLatestWorkflowRun(backend, 'Publish') if (lastPublish && lastPublish.created_at > run.created_at) { return { ok: false, diff --git a/apps/web/src/lib/server/workflow/stream.test.ts b/apps/web/src/lib/server/workflow/stream.test.ts index 45c3d216..75a07cd8 100644 --- a/apps/web/src/lib/server/workflow/stream.test.ts +++ b/apps/web/src/lib/server/workflow/stream.test.ts @@ -12,7 +12,7 @@ const mocks = vi.hoisted(() => { return { query, selectFrom: vi.fn(() => query), - getOpenWorkflowDb: vi.fn(() => ({})), + getOpenWorkflowBackend: vi.fn(() => ({})), getOpenWorkflow: vi.fn(), getWorkflowRun: vi.fn(), getStepAttempts: vi.fn() @@ -23,7 +23,7 @@ vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: mocks.selectFrom } })) vi.mock('./index.js', () => ({ - getOpenWorkflowDb: mocks.getOpenWorkflowDb, + getOpenWorkflowBackend: mocks.getOpenWorkflowBackend, getOpenWorkflow: mocks.getOpenWorkflow })) vi.mock('@luzzle/web.jobs', () => ({ @@ -100,9 +100,9 @@ beforeEach(() => { mocks.query.selectAll.mockReturnThis() mocks.query.where.mockReturnThis() mocks.query.orderBy.mockReturnThis() - mocks.getOpenWorkflowDb.mockReturnValue({}) - mocks.getWorkflowRun.mockReturnValue(run) - mocks.getStepAttempts.mockReturnValue([step]) + mocks.getOpenWorkflowBackend.mockReturnValue({}) + mocks.getWorkflowRun.mockResolvedValue(run) + mocks.getStepAttempts.mockResolvedValue([step]) mocks.query.execute.mockResolvedValue([]) }) @@ -342,6 +342,39 @@ describe('streamJobProgress', () => { } ) + test.each([ + { read: 'run', disconnect: 'request abort' }, + { read: 'run', disconnect: 'reader cancel' }, + { read: 'steps', disconnect: 'request abort' }, + { read: 'steps', disconnect: 'reader cancel' } + ])( + '$disconnect during an SDK $read read does not start later queries', + async ({ read, disconnect }) => { + vi.useFakeTimers() + let finishRead!: () => void + const pending = new Promise((resolve) => { + finishRead = () => resolve(read === 'run' ? run : [step]) + }) + if (read === 'run') mocks.getWorkflowRun.mockReturnValueOnce(pending) + else mocks.getStepAttempts.mockReturnValueOnce(pending) + const abort = new AbortController() + const reader = openStream({ signal: abort.signal }).body!.getReader() + if (read === 'steps') await readFrame(reader) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1) + if (disconnect === 'request abort') abort.abort() + else await reader.cancel() + expect(await reader.read()).toEqual({ done: true, value: undefined }) + expect(vi.getTimerCount()).toBe(0) + + finishRead() + await vi.advanceTimersByTimeAsync(0) + expect(mocks.getStepAttempts).toHaveBeenCalledTimes(read === 'run' ? 0 : 1) + expect(mocks.query.execute).not.toHaveBeenCalled() + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + } + ) + test('an already-aborted request never polls, emits, or installs timers', async () => { vi.useFakeTimers() const abort = new AbortController() @@ -350,7 +383,7 @@ describe('streamJobProgress', () => { expect(await readChunks(openStream({ signal: abort.signal }))).toEqual([]) expect(vi.getTimerCount()).toBe(0) await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) - expect(mocks.getOpenWorkflowDb).not.toHaveBeenCalled() + expect(mocks.getOpenWorkflowBackend).not.toHaveBeenCalled() expect(mocks.getWorkflowRun).not.toHaveBeenCalled() expect(mocks.getStepAttempts).not.toHaveBeenCalled() expect(mocks.query.execute).not.toHaveBeenCalled() diff --git a/apps/web/src/lib/server/workflow/stream.ts b/apps/web/src/lib/server/workflow/stream.ts index dfecca97..f2c549da 100644 --- a/apps/web/src/lib/server/workflow/stream.ts +++ b/apps/web/src/lib/server/workflow/stream.ts @@ -1,7 +1,7 @@ import { db, type JobProgressRow } from '$lib/server/database/index.js' import { createEventStream } from '../sse.js' import { WORKFLOW_POLL_INTERVAL_MS } from '../constants.js' -import { getOpenWorkflowDb } from './index.js' +import { getOpenWorkflowBackend } from './index.js' import { getWorkflowRun, getStepAttempts } from '@luzzle/web.jobs' const TERMINAL_STATES = new Set(['completed', 'failed', 'canceled']) @@ -56,7 +56,8 @@ async function pollOnce( jobId: string, jobClass: string | string[], cursors: Cursors, - emit: Emit + emit: Emit, + signal: AbortSignal ): Promise { try { let job: { class: string; state: string; result: unknown; errors: unknown } | null = null @@ -64,8 +65,8 @@ async function pollOnce( // Query OpenWorkflow try { - const openWorkflowDb = getOpenWorkflowDb() - const run = getWorkflowRun(openWorkflowDb, jobId) + const run = await getWorkflowRun(getOpenWorkflowBackend(), jobId) + if (signal.aborted) return true if (run) { let state = 'waiting' if (run.status === 'running') state = 'running' @@ -100,9 +101,10 @@ async function pollOnce( let phases: JobProgressRow[] = [] if (runId) { + if (signal.aborted) return true try { - const openWorkflowDb = getOpenWorkflowDb() - const rows = getStepAttempts(openWorkflowDb, runId) + const rows = await getStepAttempts(getOpenWorkflowBackend(), runId) + if (signal.aborted) return true phases = rows.map((r) => { let status = 'waiting' if (r.status === 'running') status = 'running' @@ -129,7 +131,9 @@ async function pollOnce( let hasNewLogs = false for (const phase of phases) { + if (signal.aborted) return true const newLogs = await fetchNewLogs(jobId, phase.phase, cursors[phase.phase] ?? 0) + if (signal.aborted) return true if (newLogs.length > 0) { hasNewLogs = true emit('log', newLogs) @@ -167,7 +171,7 @@ export function streamJobProgress({ async function poll() { try { while (!events.signal.aborted) { - if (await pollOnce(jobId, jobClass, cursors, events.emit)) break + if (await pollOnce(jobId, jobClass, cursors, events.emit, events.signal)) break await waitForNextPoll(events.signal) } } catch (error) { diff --git a/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/+layout.server.ts b/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/+layout.server.ts index 30f7d310..2b421dc3 100644 --- a/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/+layout.server.ts +++ b/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/+layout.server.ts @@ -1,7 +1,7 @@ import { error } from '@sveltejs/kit' import { config } from '$lib/server/config' import { assemblePreview, type PreviewWorkerResult } from '$lib/pieces/preview/assemble.server.js' -import { getOpenWorkflowDb } from '$lib/server/workflow/index.js' +import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js' import { getWorkflowRun, getStepAttempts } from '@luzzle/web.jobs' import { db, type JobProgressLogsRow, type JobProgressRow } from '$lib/server/database/index.js' import type { LayoutServerLoad } from './$types' @@ -9,11 +9,10 @@ import type { LayoutServerLoad } from './$types' export const load: LayoutServerLoad = async ({ params }) => { const runId = params.jobId - let run: ReturnType | undefined + let run: Awaited> | undefined try { - const openWorkflowDb = getOpenWorkflowDb() - run = getWorkflowRun(openWorkflowDb, runId) + run = await getWorkflowRun(getOpenWorkflowBackend(), runId) } catch (err) { console.error('Failed to query OpenWorkflow preview run:', err) } @@ -30,8 +29,7 @@ export const load: LayoutServerLoad = async ({ params }) => { let phases: JobProgressRow[] = [] try { - const openWorkflowDb = getOpenWorkflowDb() - const rows = getStepAttempts(openWorkflowDb, runId) + const rows = await getStepAttempts(getOpenWorkflowBackend(), runId) phases = rows.map((r) => { let status = 'waiting' if (r.status === 'running') status = 'running' diff --git a/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/layout.server.test.ts b/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/layout.server.test.ts new file mode 100644 index 00000000..a26bee92 --- /dev/null +++ b/apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/layout.server.test.ts @@ -0,0 +1,91 @@ +import { afterEach, beforeEach, expect, test, vi } from 'vitest' +import { getWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs' +import { load } from './+layout.server.js' + +const { assemblePreview, query } = vi.hoisted(() => ({ + assemblePreview: vi.fn(() => ({ html: 'rendered-preview' })), + query: { + selectAll: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + orderBy: vi.fn().mockReturnThis(), + execute: vi.fn() + } +})) +vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: () => query } })) +vi.mock('$lib/server/config', async () => { + const { makeConfig } = await import('$lib/server/config.fixture.js') + return { + config: makeConfig({ + pieces: [{ type: 'book', fields: { title: 'title', date_consumed: 'read_on' } }] + }) + } +}) +vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflowBackend: () => ({}) })) +vi.mock('$lib/pieces/preview/assemble.server.js', () => ({ assemblePreview })) +vi.mock('@luzzle/web.jobs', () => ({ getWorkflowRun: vi.fn(), getStepAttempts: vi.fn() })) + +const preview = { + filePath: 'example.book.md', + type: 'book', + slug: 'example', + pieceKey: 'piece-key', + sanitizedFrontmatter: { title: 'Draft' }, + note: 'note', + pathToKey: {}, + transforms: [] +} +function run(overrides: Partial = {}): WorkflowRunRow { + return { + id: 'preview-id', + workflow_name: 'Preview', + status: 'completed', + error: null, + input: '{}', + output: JSON.stringify(preview), + finished_at: '2026-06-21T00:00:00Z', + created_at: '2026-06-20T00:00:00Z', + ...overrides + } +} +function event() { + return { params: { jobId: 'preview-id', path: 'example.book.md' } } as Parameters[0] +} + +beforeEach(() => { + vi.clearAllMocks() + vi.setSystemTime(new Date('2026-06-22T00:00:00Z')) + query.execute.mockResolvedValue([]) + vi.mocked(getWorkflowRun).mockResolvedValue(run()) + vi.mocked(getStepAttempts).mockResolvedValue([]) +}) +afterEach(() => vi.useRealTimers()) + +test('awaits SDK-backed metadata and preserves completed preview assembly', async () => { + expect(await load(event())).toMatchObject({ + file: 'example.book.md', + status: 'completed', + jobId: 'preview-id', + phases: [], + logs: [], + html: 'rendered-preview' + }) + expect(assemblePreview).toHaveBeenCalledWith(preview, expect.objectContaining({ type: 'book' })) +}) + +test('preserves failed preview error data', async () => { + vi.mocked(getWorkflowRun).mockResolvedValue(run({ status: 'failed', error: 'preview failed' })) + expect(await load(event())).toMatchObject({ status: 'failed', errorMessage: 'preview failed' }) + expect(assemblePreview).not.toHaveBeenCalled() +}) + +test('preserves expiration based on the serialized SDK finish time', async () => { + vi.mocked(getWorkflowRun).mockResolvedValue(run({ finished_at: '2026-06-19T00:00:00Z' })) + expect(await load(event())).toMatchObject({ status: 'expired', jobId: 'preview-id' }) + expect(assemblePreview).not.toHaveBeenCalled() +}) + +test('keeps missing SDK runs as a 404', async () => { + vi.mocked(getWorkflowRun).mockResolvedValue(null) + await expect(load(event())).rejects.toMatchObject({ status: 404 }) + expect(getStepAttempts).not.toHaveBeenCalled() +}) diff --git a/apps/web/src/routes/admin/publish/+page.server.ts b/apps/web/src/routes/admin/publish/+page.server.ts index 267d7d35..349f0635 100644 --- a/apps/web/src/routes/admin/publish/+page.server.ts +++ b/apps/web/src/routes/admin/publish/+page.server.ts @@ -1,6 +1,6 @@ import { config } from '$lib/server/config' import { db, type JobProgressRow, type JobProgressLogsRow } from '$lib/server/database/index.js' -import { getOpenWorkflowDb } from '$lib/server/workflow/index.js' +import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js' import { parsePiecesDiff } from '$lib/server/workflow/publish.js' import { getLatestWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs' import type { PiecesDiff } from '@luzzle/core' @@ -24,8 +24,8 @@ function mapState(status: string): string { return 'waiting' } -function mapPhases(jobId: string): JobProgressRow[] { - const rows = getStepAttempts(getOpenWorkflowDb(), jobId) +async function mapPhases(jobId: string): Promise { + const rows = await getStepAttempts(getOpenWorkflowBackend(), jobId) return rows.map((r) => ({ job_id: jobId, phase: r.phase, @@ -51,7 +51,7 @@ async function buildRunView(run: WorkflowRunRow | null): Promise let phases: JobProgressRow[] = [] let logs: JobProgressLogsRow[] = [] try { - phases = mapPhases(run.id) + phases = await mapPhases(run.id) logs = await loadLogs(run.id) } catch (err) { console.error('Failed to load run progress in publish loader:', err) @@ -74,9 +74,9 @@ export const load: PageServerLoad = async () => { let publish: RunView | null = null try { - const openWorkflowDb = getOpenWorkflowDb() - audit = await buildRunView(getLatestWorkflowRun(openWorkflowDb, 'PublishAudit')) - publish = await buildRunView(getLatestWorkflowRun(openWorkflowDb, 'Publish')) + const backend = getOpenWorkflowBackend() + audit = await buildRunView(await getLatestWorkflowRun(backend, 'PublishAudit')) + publish = await buildRunView(await getLatestWorkflowRun(backend, 'Publish')) } catch (err) { console.error('Failed to query OpenWorkflow runs in publish loader:', err) } diff --git a/apps/web/src/routes/admin/publish/page.server.test.ts b/apps/web/src/routes/admin/publish/page.server.test.ts new file mode 100644 index 00000000..9b80a255 --- /dev/null +++ b/apps/web/src/routes/admin/publish/page.server.test.ts @@ -0,0 +1,89 @@ +import { beforeEach, expect, test, vi } from 'vitest' +import { getLatestWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs' +import { load } from './+page.server.js' + +const { query } = vi.hoisted(() => ({ + query: { + selectAll: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + orderBy: vi.fn().mockReturnThis(), + execute: vi.fn() + } +})) +vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: () => query } })) +vi.mock('$lib/server/config', async () => { + const { makeConfig } = await import('$lib/server/config.fixture.js') + return { config: makeConfig() } +}) +vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflowBackend: () => ({}) })) +vi.mock('@luzzle/web.jobs', () => ({ getLatestWorkflowRun: vi.fn(), getStepAttempts: vi.fn() })) + +const diff = { + schemas: { added: [], updated: [], pruned: [] }, + pieces: { added: ['new.book.md'], updated: [], pruned: [] } +} +function run(name: string): WorkflowRunRow { + return { + id: `${name}-id`, + workflow_name: name, + status: 'completed', + error: null, + input: '{}', + output: JSON.stringify(diff), + finished_at: '2026-06-20T00:01:00Z', + created_at: '2026-06-20T00:00:00Z' + } +} + +beforeEach(() => { + vi.clearAllMocks() + query.execute.mockResolvedValue([]) + vi.mocked(getLatestWorkflowRun).mockImplementation(async (_backend, name) => run(name)) + vi.mocked(getStepAttempts).mockResolvedValue([ + { + phase: 'sync', + status: 'completed', + started_at: '2026-06-20T00:00:00Z', + finished_at: '2026-06-20T00:01:00Z', + message: null + } + ]) +}) + +test('awaits SDK-backed run and step reads while preserving publish page data', async () => { + const result = await load({} as Parameters[0]) + if (!result) throw new Error('Expected publish page data') + for (const [view, name] of [ + [result.audit, 'PublishAudit'], + [result.publish, 'Publish'] + ] as const) { + expect(view).toEqual({ + jobId: `${name}-id`, + state: 'completed', + errors: null, + logs: [], + diff, + phases: [ + { + job_id: `${name}-id`, + phase: 'sync', + status: 'completed', + started_at: Date.parse('2026-06-20T00:00:00Z'), + finished_at: Date.parse('2026-06-20T00:01:00Z'), + message: null + } + ] + }) + } + expect(query.where).toHaveBeenCalledWith('job_id', '=', 'PublishAudit-id') + expect(query.where).toHaveBeenCalledWith('job_id', '=', 'Publish-id') +}) + +test('keeps absent run views null', async () => { + vi.mocked(getLatestWorkflowRun).mockResolvedValue(null) + const result = await load({} as Parameters[0]) + if (!result) throw new Error('Expected publish page data') + expect(result.audit).toBeNull() + expect(result.publish).toBeNull() + expect(getStepAttempts).not.toHaveBeenCalled() +}) diff --git a/apps/web/src/routes/api/admin/publish/+server.ts b/apps/web/src/routes/api/admin/publish/+server.ts index 77f204df..ef16f76d 100644 --- a/apps/web/src/routes/api/admin/publish/+server.ts +++ b/apps/web/src/routes/api/admin/publish/+server.ts @@ -1,34 +1,38 @@ import { json } from '@sveltejs/kit' import type { RequestHandler } from './$types' -import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js' -import { findInFlightPublishRun, validateAuditForPublish } from '$lib/server/workflow/publish.js' +import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js' +import { + findInFlightPublishRun, + validateAuditForPublish, + withPublishAdmission +} from '$lib/server/workflow/publish.js' import { publishSpec } from '@luzzle/web.jobs/specs' export const POST: RequestHandler = async ({ request }) => { try { - const db = getOpenWorkflowDb() - - const inFlight = findInFlightPublishRun(db) - if (inFlight) { - return json({ jobId: inFlight.id }, { status: 409 }) - } - const body = await request.json().catch(() => ({})) - const auditRunId = body?.auditRunId - const bisync = body?.bisync === true - - if (auditRunId) { - const guard = validateAuditForPublish(db, auditRunId) - if (!guard.ok) { - return json({ message: guard.reason }, { status: 412 }) + return await withPublishAdmission(async () => { + const inFlight = await findInFlightPublishRun(getOpenWorkflowBackend()) + if (inFlight) { + return json({ jobId: inFlight.id }, { status: 409 }) } - } - const openWorkflow = getOpenWorkflow() - const handle = await openWorkflow.runWorkflow(publishSpec, { - bisync: auditRunId ? false : bisync + const auditRunId = body?.auditRunId + const bisync = body?.bisync === true + + if (auditRunId) { + const guard = await validateAuditForPublish(getOpenWorkflowBackend(), auditRunId) + if (!guard.ok) { + return json({ message: guard.reason }, { status: 412 }) + } + } + + const openWorkflow = getOpenWorkflow() + const handle = await openWorkflow.runWorkflow(publishSpec, { + bisync: auditRunId ? false : bisync + }) + return json({ jobId: handle.workflowRun.id }) }) - return json({ jobId: handle.workflowRun.id }) } catch (error) { const message = error instanceof Error ? error.message : String(error) return new Response(`Internal server error: ${message}`, { status: 500 }) diff --git a/apps/web/src/routes/api/admin/publish/admission.test.ts b/apps/web/src/routes/api/admin/publish/admission.test.ts new file mode 100644 index 00000000..e84a8be0 --- /dev/null +++ b/apps/web/src/routes/api/admin/publish/admission.test.ts @@ -0,0 +1,240 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js' +import { + createWorkflowQueue, + closeWorkflowQueue, + createPendingRun, + createCompletedRun, + claimNextRun, + completeNextRun, + type WorkflowQueue +} from '$lib/server/workflow/publish.fixture.js' +import { POST as publish } from './+server.js' +import { POST as audit } from './audit/+server.js' + +const { runWorkflow } = vi.hoisted(() => ({ runWorkflow: vi.fn() })) +vi.mock('$lib/server/workflow/index.js', () => ({ + getOpenWorkflow: () => ({ runWorkflow }), + getOpenWorkflowBackend: vi.fn() +})) + +function event(body: unknown = {}) { + return { + request: new Request('http://localhost/api/admin/publish', { + method: 'POST', + body: JSON.stringify(body) + }) + } as Parameters[0] & Parameters[0] +} + +describe('publish/audit admission', () => { + let queue: WorkflowQueue + + async function enqueue(spec: { name: string }, input: { bisync: boolean }) { + vi.setSystemTime(new Date('2026-06-21T00:00:00Z')) + const workflowRun = await createPendingRun(queue.backend, spec.name, input) + return { workflowRun } + } + + beforeEach(() => { + vi.resetAllMocks() + vi.setSystemTime(new Date('2026-06-20T00:00:00Z')) + queue = createWorkflowQueue() + vi.mocked(getOpenWorkflowBackend).mockReturnValue(queue.backend) + runWorkflow.mockImplementation(async (spec, input) => enqueue(spec, input)) + }) + + afterEach(async () => { + vi.useRealTimers() + await closeWorkflowQueue(queue) + }) + + test.each([ + { firstName: 'Publish', firstRoute: publish, otherName: 'Publish', otherRoute: publish }, + { firstName: 'Publish', firstRoute: publish, otherName: 'PublishAudit', otherRoute: audit }, + { firstName: 'PublishAudit', firstRoute: audit, otherName: 'Publish', otherRoute: publish }, + { firstName: 'PublishAudit', firstRoute: audit, otherName: 'PublishAudit', otherRoute: audit } + ])( + 'admits only one overlapping $firstName/$otherName request', + async ({ firstRoute, otherRoute }) => { + let release!: () => void + let entered!: () => void + const hold = new Promise((resolve) => { + release = resolve + }) + const started = new Promise((resolve) => { + entered = resolve + }) + runWorkflow.mockImplementationOnce(async (spec, input) => { + entered() + await hold + return enqueue(spec, input) + }) + + const first = Promise.resolve(firstRoute(event())) + try { + await Promise.race([ + started, + first.then(() => { + throw new Error('First request finished before enqueue') + }) + ]) + const other = otherRoute(event()) + release() + const [accepted, rejected] = await Promise.all([first, other]) + expect(accepted.status).toBe(200) + expect(rejected.status).toBe(409) + expect(await rejected.json()).toEqual(await accepted.json()) + expect(runWorkflow).toHaveBeenCalledTimes(1) + expect((await queue.backend.countWorkflowRuns()).pending).toBe(1) + } finally { + release() + await first + } + } + ) + + test.each([ + { slowName: 'Publish', slowRoute: publish, otherRoute: audit }, + { slowName: 'PublishAudit', slowRoute: audit, otherRoute: publish } + ])( + 'does not let a slow $slowName body block other admissions', + async ({ slowRoute, otherRoute }) => { + let releaseBody!: () => void + const body = new Promise((resolve) => { + releaseBody = () => resolve({}) + }) + const request = { json: vi.fn(() => body) } as unknown as Request + const slow = Promise.resolve(slowRoute({ ...event(), request })) + const other = Promise.resolve(otherRoute(event())) + try { + await expect.poll(() => runWorkflow.mock.calls.length).toBe(1) + expect((await other).status).toBe(200) + } finally { + releaseBody() + await Promise.all([slow, other]) + } + expect((await slow).status).toBe(409) + expect(await (await slow).json()).toEqual(await (await other).json()) + expect(runWorkflow).toHaveBeenCalledTimes(1) + } + ) + + test('releases admission after enqueue failure so a waiting request can succeed', async () => { + let release!: () => void + let entered!: () => void + const hold = new Promise((resolve) => { + release = resolve + }) + const started = new Promise((resolve) => { + entered = resolve + }) + runWorkflow.mockImplementationOnce(async () => { + entered() + await hold + throw new Error('enqueue failed') + }) + + const first = Promise.resolve(publish(event())) + try { + await Promise.race([ + started, + first.then(() => { + throw new Error('First request finished before enqueue') + }) + ]) + const other = audit(event({ bisync: true })) + release() + const [failed, accepted] = await Promise.all([first, other]) + expect(failed.status).toBe(500) + expect(await failed.text()).toContain('enqueue failed') + expect(accepted.status).toBe(200) + expect(runWorkflow).toHaveBeenCalledTimes(2) + expect(runWorkflow).toHaveBeenLastCalledWith( + expect.objectContaining({ name: 'PublishAudit' }), + { bisync: true } + ) + } finally { + release() + await first + } + }) + + test('keeps detecting persisted work when enqueue writes the run before rejecting', async () => { + let persistedRunId: string | undefined + runWorkflow.mockImplementationOnce(async (spec, input) => { + persistedRunId = (await enqueue(spec, input)).workflowRun.id + throw new Error('enqueue response failed') + }) + expect((await publish(event())).status).toBe(500) + const rejected = await audit(event()) + expect(rejected.status).toBe(409) + expect(await rejected.json()).toEqual({ jobId: persistedRunId }) + expect(runWorkflow).toHaveBeenCalledTimes(1) + }) + + test.each([publish, audit])( + 'uses persisted active runs even without a local submission', + async (route) => { + const active = await createPendingRun(queue.backend, 'Publish') + await claimNextRun(queue.backend) + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + await createCompletedRun(queue.backend, 'Publish') + const response = await route(event()) + expect(response.status).toBe(409) + expect(await response.json()).toEqual({ jobId: active.id }) + expect(runWorkflow).not.toHaveBeenCalled() + } + ) + + test('retains audit freshness checks and releases admission after rejecting an audit', async () => { + const oldAudit = await createCompletedRun(queue.backend, 'PublishAudit') + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + const newAudit = await createCompletedRun(queue.backend, 'PublishAudit') + const rejected = await publish(event({ auditRunId: oldAudit.id })) + expect(rejected.status).toBe(412) + expect(await rejected.json()).toEqual({ + message: 'a newer audit has run; re-check before publishing' + }) + expect(runWorkflow).not.toHaveBeenCalled() + + const accepted = await publish(event({ auditRunId: newAudit.id, bisync: true })) + expect(accepted.status).toBe(200) + expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), { + bisync: false + }) + }) + + test('does not reuse a consumed audit after the admitted publish finishes', async () => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit') + const accepted = await publish(event({ auditRunId: auditRun.id })) + expect(accepted.status).toBe(200) + const { jobId } = await accepted.json() + expect((await completeNextRun(queue.backend)).id).toBe(jobId) + + const rejected = await publish(event({ auditRunId: auditRun.id })) + expect(rejected.status).toBe(412) + expect(await rejected.json()).toEqual({ + message: 'changes were published after this check; re-check before publishing' + }) + expect(runWorkflow).toHaveBeenCalledTimes(1) + }) + + test('retains unchecked publish input and tolerates malformed request JSON', async () => { + const accepted = await publish(event({ bisync: true })) + expect(accepted.status).toBe(200) + expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), { + bisync: true + }) + await completeNextRun(queue.backend) + const request = new Request('http://localhost/api/admin/publish/audit', { + method: 'POST', + body: '{' + }) + expect((await audit({ request } as Parameters[0])).status).toBe(200) + expect(runWorkflow).toHaveBeenLastCalledWith( + expect.objectContaining({ name: 'PublishAudit' }), + { bisync: false } + ) + }) +}) diff --git a/apps/web/src/routes/api/admin/publish/audit/+server.ts b/apps/web/src/routes/api/admin/publish/audit/+server.ts index ac0c7ca7..c899fa4f 100644 --- a/apps/web/src/routes/api/admin/publish/audit/+server.ts +++ b/apps/web/src/routes/api/admin/publish/audit/+server.ts @@ -1,22 +1,24 @@ import { json } from '@sveltejs/kit' import type { RequestHandler } from './$types' -import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js' -import { findInFlightPublishRun } from '$lib/server/workflow/publish.js' +import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js' +import { findInFlightPublishRun, withPublishAdmission } from '$lib/server/workflow/publish.js' import { publishAuditSpec } from '@luzzle/web.jobs/specs' export const POST: RequestHandler = async ({ request }) => { try { - const inFlight = findInFlightPublishRun(getOpenWorkflowDb()) - if (inFlight) { - return json({ jobId: inFlight.id }, { status: 409 }) - } - const body = await request.json().catch(() => ({})) - const bisync = body?.bisync === true + return await withPublishAdmission(async () => { + const inFlight = await findInFlightPublishRun(getOpenWorkflowBackend()) + if (inFlight) { + return json({ jobId: inFlight.id }, { status: 409 }) + } - const openWorkflow = getOpenWorkflow() - const handle = await openWorkflow.runWorkflow(publishAuditSpec, { bisync }) - return json({ jobId: handle.workflowRun.id }) + const bisync = body?.bisync === true + + const openWorkflow = getOpenWorkflow() + const handle = await openWorkflow.runWorkflow(publishAuditSpec, { bisync }) + return json({ jobId: handle.workflowRun.id }) + }) } catch (error) { const message = error instanceof Error ? error.message : String(error) return new Response(`Internal server error: ${message}`, { status: 500 }) diff --git a/packages/web/jobs/README.md b/packages/web/jobs/README.md index a20be2d9..3f7164ae 100644 --- a/packages/web/jobs/README.md +++ b/packages/web/jobs/README.md @@ -6,7 +6,7 @@ job queue instantiations for Luzzle. ## Purpose 🧠 This library defines the schemas, payload interfaces, and results for tasks -processed by the Sidequest worker queue: +processed by the OpenWorkflow worker queue: - **`publish`** — Triggers CDN asset generation, cache purging, and remote cloud synchronization. diff --git a/packages/web/jobs/package.json b/packages/web/jobs/package.json index 5cbc9a35..5f833525 100644 --- a/packages/web/jobs/package.json +++ b/packages/web/jobs/package.json @@ -49,7 +49,7 @@ }, "dependencies": { "@luzzle/core": "workspace:*", - "openworkflow": "0.9.0" + "openworkflow": "0.9.2" }, "type": "module" } diff --git a/packages/web/jobs/src/index.ts b/packages/web/jobs/src/index.ts index 617cd441..f9fd5730 100644 --- a/packages/web/jobs/src/index.ts +++ b/packages/web/jobs/src/index.ts @@ -2,7 +2,6 @@ export { initOpenWorkflow, getOpenWorkflow, getLatestWorkflowRun, - getWorkflowRunByJobId, getWorkflowRun, getStepAttempts, purgeExpiredWorkflowRuns, diff --git a/packages/web/jobs/src/openworkflow.test.ts b/packages/web/jobs/src/openworkflow.test.ts index 8a809fe6..bdac93bb 100644 --- a/packages/web/jobs/src/openworkflow.test.ts +++ b/packages/web/jobs/src/openworkflow.test.ts @@ -1,10 +1,9 @@ -import { describe, expect, test } from 'vitest' -import { DatabaseSync } from 'node:sqlite' +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { BackendSqlite } from 'openworkflow/sqlite' import { initOpenWorkflow, getOpenWorkflow, getLatestWorkflowRun, - getWorkflowRunByJobId, getWorkflowRun, getStepAttempts, jobProgressPurgeSpec, @@ -21,103 +20,269 @@ describe('openworkflow client initialization', () => { const client = initOpenWorkflow({ dbPath: ':memory:' }) expect(client).toBeDefined() expect(getOpenWorkflow()).toBe(client) + expect(initOpenWorkflow({ dbPath: ':memory:' })).toBe(client) }) }) -describe('database helpers', () => { - test('queries workflow runs and step attempts', () => { - const db = new DatabaseSync(':memory:') +type CreateRunParams = Parameters[0] +type RunError = Parameters[0]['error'] +type StepAttempt = Awaited> - // Create tables - db.exec(` - CREATE TABLE workflow_runs ( - id TEXT PRIMARY KEY, - workflow_name TEXT NOT NULL, - status TEXT NOT NULL, - error TEXT, - input TEXT NOT NULL, - output TEXT, - finished_at TEXT, - created_at TEXT NOT NULL - ) - `) +const workerId = 'read-helper-test' +const initialTime = new Date('2026-06-02T05:00:00.000Z') - db.exec(` - 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 - ) - `) +describe('SDK read helpers', () => { + let backend: BackendSqlite - // 1. Test getLatestWorkflowRun when no runs exist - expect(getLatestWorkflowRun(db, 'Publish')).toBeNull() + beforeEach(() => { + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(initialTime) + backend = BackendSqlite.connect(':memory:') + }) - // Insert dummy runs - db.prepare(` - INSERT INTO workflow_runs (id, workflow_name, status, error, input, output, finished_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - `).run('run-1', 'Publish', 'completed', null, '{}', 'ok', null, '2026-06-02T05:00:00Z') + afterEach(async () => { + vi.restoreAllMocks() + vi.useRealTimers() + await backend.stop() + }) - db.prepare(` - INSERT INTO workflow_runs (id, workflow_name, status, error, input, output, finished_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - `).run('run-2', 'Publish', 'failed', 'some error', '{"jobId":123}', null, null, '2026-06-02T05:10:00Z') + function createRun(overrides: Partial = {}) { + return backend.createWorkflowRun({ + workflowName: 'Publish', + version: null, + idempotencyKey: null, + config: {}, + context: null, + input: { jobId: 123 }, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + availableAt: null, + deadlineAt: null, + ...overrides, + }) + } - // Test getLatestWorkflowRun returns the latest by created_at DESC - const latest = getLatestWorkflowRun(db, 'Publish') - expect(latest).not.toBeNull() - expect(latest!.id).toBe('run-2') - expect(latest!.error).toBe('some error') + async function claimRun(id: string) { + const claimed = await backend.claimWorkflowRun({ workerId, leaseDurationMs: 60_000 }) + expect(claimed?.id).toBe(id) + } - // 2. Test getWorkflowRunByJobId - const runByJobId = getWorkflowRunByJobId(db, 123) - expect(runByJobId).not.toBeNull() - expect(runByJobId!.id).toBe('run-2') + function createStep(workflowRunId: string, stepName: string) { + return backend.createStepAttempt({ + workflowRunId, + workerId, + stepName, + kind: 'function', + config: {}, + context: null, + }) + } - const runByNonexistentJobId = getWorkflowRunByJobId(db, 999) - expect(runByNonexistentJobId).toBeNull() + test('returns null or an empty list for missing runs', async () => { + expect(await getLatestWorkflowRun(backend, 'Publish')).toBeNull() + expect(await getWorkflowRun(backend, 'missing')).toBeNull() + expect(await getStepAttempts(backend, 'missing')).toEqual([]) + }) - // 3. Test getWorkflowRun - const run = getWorkflowRun(db, 'run-1') - expect(run).not.toBeNull() - expect(run!.id).toBe('run-1') + test('returns the newest matching workflow as the existing DTO, not an SDK run', async () => { + const first = await createRun() + vi.setSystemTime(new Date('2026-06-02T05:10:00.000Z')) + const latest = await createRun({ input: { jobId: 456, assets: ['cover'] } }) + vi.setSystemTime(new Date('2026-06-02T05:20:00.000Z')) + await createRun({ workflowName: 'Preview' }) - const nonexistentRun = getWorkflowRun(db, 'nonexistent') - expect(nonexistentRun).toBeNull() + expect(await getLatestWorkflowRun(backend, 'Publish')).toEqual({ + id: latest.id, + workflow_name: 'Publish', + status: 'pending', + error: null, + input: '{"jobId":456,"assets":["cover"]}', + output: null, + finished_at: null, + created_at: '2026-06-02T05:10:00.000Z', + }) + expect(await getWorkflowRun(backend, latest.id)).toEqual( + await getLatestWorkflowRun(backend, 'Publish') + ) + expect((await getWorkflowRun(backend, first.id))?.id).toBe(first.id) + expect(await getLatestWorkflowRun(backend, 'Unknown')).toBeNull() + }) - // 4. Test getStepAttempts - 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', 'step-parse', 'completed', '2026-06-02T05:01:00Z', '2026-06-02T05:02:00Z', null, '2026-06-02T05:01:00Z') + test.each([ + { input: { jobId: 123 }, output: { published: ['cover'], count: 1 } }, + { input: 'input string', output: 'ok' }, + { input: '', output: '' }, + { input: false, output: false }, + { input: 0, output: 0 }, + { input: null, output: null }, + ])('serializes input/output JSON and completed ISO timestamps: %j', async ({ input, output }) => { + const run = await createRun({ input }) + await claimRun(run.id) + vi.setSystemTime(new Date('2026-06-02T05:01:00.000Z')) + await backend.completeWorkflowRun({ workflowRunId: run.id, workerId, output }) - 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', 'step-transform', 'failed', '2026-06-02T05:03:00Z', '2026-06-02T05:04:00Z', 'transform error', '2026-06-02T05:03:00Z') - - const steps = getStepAttempts(db, 'run-1') - expect(steps).toHaveLength(2) - expect(steps[0]).toEqual({ - phase: 'step-parse', + expect(await getWorkflowRun(backend, run.id)).toEqual({ + id: run.id, + workflow_name: 'Publish', status: 'completed', - started_at: '2026-06-02T05:01:00Z', - finished_at: '2026-06-02T05:02:00Z', - message: null, + error: null, + input: JSON.stringify(input), + output: output === null ? null : JSON.stringify(output), + finished_at: '2026-06-02T05:01:00.000Z', + created_at: initialTime.toISOString(), }) - expect(steps[1]).toEqual({ - phase: 'step-transform', + }) + + test.each([ + { name: 'PublishError', message: 'upload failed', stack: 'stack', detail: { code: 503 } }, + 'legacy failure', + '', + ])('preserves workflow error JSON encoding: %j', async error => { + const run = await createRun() + await claimRun(run.id) + vi.setSystemTime(new Date('2026-06-02T05:02:00.000Z')) + await backend.failWorkflowRun({ + workflowRunId: run.id, + workerId, + // The public mutation type requires an object, but SQLite also stores + // string JSON errors. Exercise that compatibility through the real API. + error: error as unknown as RunError, + retryPolicy: { + maximumAttempts: 1, + initialInterval: '1s', + maximumInterval: '1s', + backoffCoefficient: 1, + }, + }) + + const expected = { + id: run.id, + workflow_name: 'Publish', status: 'failed', - started_at: '2026-06-02T05:03:00Z', - finished_at: '2026-06-02T05:04:00Z', - message: 'transform error', + error: JSON.stringify(error), + input: '{"jobId":123}', + output: null, + finished_at: '2026-06-02T05:02:00.000Z', + created_at: initialTime.toISOString(), + } + expect(await getWorkflowRun(backend, run.id)).toEqual(expected) + expect(await getLatestWorkflowRun(backend, 'Publish')).toEqual(expected) + }) + + test('maps completed and running step attempts, omitting SDK internals', async () => { + const run = await createRun() + await claimRun(run.id) + expect(await getStepAttempts(backend, run.id)).toEqual([]) + const completed = await createStep(run.id, 'parse') + vi.setSystemTime(new Date('2026-06-02T05:01:00.000Z')) + await backend.completeStepAttempt({ + workflowRunId: run.id, + stepAttemptId: completed.id, + workerId, + output: { internalStepResult: true }, }) + await createStep(run.id, 'transform') + + expect(await getStepAttempts(backend, run.id)).toEqual([ + { + phase: 'parse', + status: 'completed', + started_at: initialTime.toISOString(), + finished_at: '2026-06-02T05:01:00.000Z', + message: null, + }, + { + phase: 'transform', + status: 'running', + started_at: '2026-06-02T05:01:00.000Z', + finished_at: null, + message: null, + }, + ]) + }) + + test('maps an absent SDK step start time to the existing nullable DTO field', async () => { + const run = await createRun() + await claimRun(run.id) + await createStep(run.id, 'wait') + const original = backend.listStepAttempts.bind(backend) + vi.spyOn(backend, 'listStepAttempts').mockImplementation(async params => { + const page = await original(params) + return { ...page, data: page.data.map(attempt => ({ ...attempt, startedAt: null })) } + }) + expect(await getStepAttempts(backend, run.id)).toEqual([{ + phase: 'wait', status: 'running', started_at: null, finished_at: null, message: null, + }]) + }) + + test.each([ + { name: 'TransformError', message: 'transform failed', detail: ['cover'] }, + 'legacy step failure', + '', + ])('preserves step error JSON as message: %j', async error => { + const run = await createRun() + await claimRun(run.id) + const step = await createStep(run.id, 'transform') + vi.setSystemTime(new Date('2026-06-02T05:03:00.000Z')) + await backend.failStepAttempt({ + workflowRunId: run.id, + stepAttemptId: step.id, + workerId, + error: error as unknown as RunError, + }) + + expect(await getStepAttempts(backend, run.id)).toEqual([{ + phase: 'transform', + status: 'failed', + started_at: initialTime.toISOString(), + finished_at: '2026-06-02T05:03:00.000Z', + message: JSON.stringify(error), + }]) + }) + + test('follows every SDK page in chronological order, including timestamp ties and retries', async () => { + const run = await createRun() + await claimRun(run.id) + const attempts: StepAttempt[] = [] + for (let i = 0; i < 7; i++) { + // Shared timestamps also exercise the SDK cursor ID tie-breaker. + vi.setSystemTime(new Date(initialTime.getTime() + Math.floor(i / 3) * 1000)) + const step = await createStep(run.id, `phase-${i % 3}`) + attempts.push(await backend.completeStepAttempt({ + workflowRunId: run.id, + stepAttemptId: step.id, + workerId, + output: null, + })) + } + const other = await createRun({ workflowName: 'Preview' }) + await claimRun(other.id) + await createStep(other.id, 'not-this-run') + + const original = backend.listStepAttempts.bind(backend) + const list = vi.spyOn(backend, 'listStepAttempts') + .mockImplementation(params => original({ ...params, limit: 2 })) + const rows = await getStepAttempts(backend, run.id) + const expected = attempts + .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime() || a.id.localeCompare(b.id)) + .map(attempt => ({ + phase: attempt.stepName, + status: 'completed', + started_at: attempt.startedAt!.toISOString(), + finished_at: attempt.finishedAt!.toISOString(), + message: null, + })) + + expect(rows).toHaveLength(7) + expect(rows).toEqual(expected) + expect(list).toHaveBeenCalledTimes(4) + for (let i = 1; i < list.mock.calls.length; i++) { + const previousPage = await list.mock.results[i - 1].value + expect(list.mock.calls[i][0]).toEqual({ + workflowRunId: run.id, + after: previousPage.pagination.next, + }) + } }) }) diff --git a/packages/web/jobs/src/openworkflow.ts b/packages/web/jobs/src/openworkflow.ts index 8823b306..9d019362 100644 --- a/packages/web/jobs/src/openworkflow.ts +++ b/packages/web/jobs/src/openworkflow.ts @@ -44,64 +44,69 @@ export interface StepAttemptRow { message: string | null } -/** - * Finds the latest workflow run for a given workflow name. - */ -export function getLatestWorkflowRun( - db: DatabaseSync, - workflowName: string -): WorkflowRunRow | null { - const stmt = db.prepare(` - SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs - WHERE workflow_name = ? - ORDER BY created_at DESC - LIMIT 1 - `) - const row = stmt.get(workflowName) - return row ? (row as unknown as WorkflowRunRow) : null +type WorkflowRun = NonNullable>> + +function toWorkflowRunRow(run: WorkflowRun): WorkflowRunRow { + return { + id: run.id, + workflow_name: run.workflowName, + status: run.status, + error: run.error === null ? null : JSON.stringify(run.error), + // Keep the existing non-null string DTO even for SDK runs with no input. + input: JSON.stringify(run.input), + output: run.output === null ? null : JSON.stringify(run.output), + finished_at: run.finishedAt?.toISOString() ?? null, + created_at: run.createdAt.toISOString(), + } } /** - * Finds a workflow run by the jobId nested inside its JSON input. + * Finds the latest workflow run for a given workflow name. */ -export function getWorkflowRunByJobId(db: DatabaseSync, jobId: number): WorkflowRunRow | null { - const stmt = db.prepare(` - SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs - WHERE json_extract(input, '$.jobId') = ? - LIMIT 1 - `) - const row = stmt.get(jobId) - return row ? (row as unknown as WorkflowRunRow) : null +export async function getLatestWorkflowRun( + backend: Pick, + workflowName: string +): Promise { + // The SDK lists workflow runs newest first. + const { data } = await backend.listWorkflowRuns({ workflowName, limit: 1 }) + return data[0] ? toWorkflowRunRow(data[0]) : null } /** * Finds a workflow run by its unique ID. */ -export function getWorkflowRun( - db: DatabaseSync, - id: string, -): WorkflowRunRow | null { - const stmt = db.prepare(` - SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs - WHERE id = ? - LIMIT 1 - `); - const row = stmt.get(id); - return row ? (row as unknown as WorkflowRunRow) : null; +export async function getWorkflowRun( + backend: Pick, + id: string +): Promise { + const run = await backend.getWorkflowRun({ workflowRunId: id }) + return run ? toWorkflowRunRow(run) : null } /** - * Lists all step attempts for a given workflow run ID. + * Lists all step attempts for a given workflow run ID, oldest first. */ -export function getStepAttempts(db: DatabaseSync, workflowRunId: string): StepAttemptRow[] { - const stmt = db.prepare(` - SELECT step_name as phase, status, started_at, finished_at, error as message - FROM step_attempts - WHERE workflow_run_id = ? - ORDER BY created_at ASC - `) - const rows = stmt.all(workflowRunId) - return rows as unknown as StepAttemptRow[] +export async function getStepAttempts( + backend: Pick, + workflowRunId: string +): Promise { + const rows: StepAttemptRow[] = [] + let after: string | undefined + + do { + // Forward SDK pagination preserves chronological order across pages. + const page = await backend.listStepAttempts({ workflowRunId, after }) + rows.push(...page.data.map(attempt => ({ + phase: attempt.stepName, + status: attempt.status, + started_at: attempt.startedAt?.toISOString() ?? null, + finished_at: attempt.finishedAt?.toISOString() ?? null, + message: attempt.error === null ? null : JSON.stringify(attempt.error), + }))) + after = page.pagination.next ?? undefined + } while (after) + + return rows } /** diff --git a/packages/web/worker/package.json b/packages/web/worker/package.json index 7b265205..8125d258 100644 --- a/packages/web/worker/package.json +++ b/packages/web/worker/package.json @@ -70,7 +70,7 @@ "micromark-util-symbol": "2.0.1", "micromark-util-types": "2.0.2", "mime-types": "3.0.2", - "openworkflow": "0.9.0", + "openworkflow": "0.9.2", "rehype-autolink-headings": "7.1.0", "rehype-pretty-code": "0.14.1", "rehype-raw": "7.0.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b0dec364..bbc71722 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -282,8 +282,8 @@ importers: specifier: 0.28.17 version: 0.28.17 openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.9.2 + version: 0.9.2 devDependencies: '@auth/core': specifier: 0.41.1 @@ -546,8 +546,8 @@ importers: specifier: workspace:* version: link:../../../apps/core openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.9.2 + version: 0.9.2 devDependencies: '@eslint/js': specifier: 'catalog:' @@ -816,8 +816,8 @@ importers: specifier: 3.0.2 version: 3.0.2 openworkflow: - specifier: 0.9.0 - version: 0.9.0 + specifier: 0.9.2 + version: 0.9.2 rehype-autolink-headings: specifier: 7.1.0 version: 7.1.0 @@ -3535,8 +3535,8 @@ packages: oniguruma-to-es@4.3.6: resolution: {integrity: sha512-csuQ9x3Yr0cEIs/Zgx/OEt9iBw9vqIunAPQkx19R/fiMq2oGVTgcMqO/V3Ybqefr1TBvosI6jU539ksaBULJyA==} - openworkflow@0.9.0: - resolution: {integrity: sha512-Ii6qkAh4ozbTHH5w7PdjjNYGF7u8xV2rsvTaBleo21rRlS8sEzrvDdxWsHawiY1VLfI1nfuRf8I3NDerlrugtg==} + openworkflow@0.9.2: + resolution: {integrity: sha512-NluYomuVLf+rAMJ9ASKqw8nbPAAD+C/HSTZoJacZdelLfTpndzHvPcP8SCANg9me9zek6mao0uUOGGbrWUWhGw==} engines: {node: '>=20'} peerDependencies: postgres: ^3.4.9 @@ -5939,7 +5939,7 @@ snapshots: std-env: 3.10.0 test-exclude: 7.0.2 tinyrainbow: 2.0.0 - vitest: 3.2.4(@types/debug@4.1.13)(@types/node@24.12.4)(@vitest/ui@3.2.4)(yaml@2.7.0) + vitest: 3.2.4(@types/debug@4.1.13)(@types/node@24.12.4)(@vitest/ui@3.2.4)(yaml@2.9.0) transitivePeerDependencies: - supports-color @@ -7464,7 +7464,7 @@ snapshots: regex: 6.1.0 regex-recursion: 6.0.2 - openworkflow@0.9.0: {} + openworkflow@0.9.2: {} optionator@0.9.4: dependencies: