diff --git a/.changeset/selective-publishing-backend.md b/.changeset/selective-publishing-backend.md new file mode 100644 index 00000000..957bd856 --- /dev/null +++ b/.changeset/selective-publishing-backend.md @@ -0,0 +1,9 @@ +--- +"@luzzle/web": minor +"@luzzle/web.jobs": minor +"@luzzle/web.worker": minor +--- + +Accept audited selections of pending additions and updates. Keep unselected work +pending, apply global removals and schema changes, and preserve selected plans +and hashes across worker retries. Omitted selections retain publish-all behavior. diff --git a/apps/web/src/lib/server/workflow/publish.fixture.ts b/apps/web/src/lib/server/workflow/publish.fixture.ts index 31760248..f125a971 100644 --- a/apps/web/src/lib/server/workflow/publish.fixture.ts +++ b/apps/web/src/lib/server/workflow/publish.fixture.ts @@ -20,7 +20,7 @@ export async function closeWorkflowQueue(queue: WorkflowQueue): Promise { export function createPendingRun( backend: BackendSqlite, workflowName: string, - input: Record = {} + input: Parameters[0]['input'] = {} ) { return backend.createWorkflowRun({ workflowName, @@ -45,18 +45,25 @@ export async function claimNextRun(backend: BackendSqlite) { return run } -export async function completeNextRun(backend: BackendSqlite) { +export async function completeNextRun( + backend: BackendSqlite, + output: Parameters[0]['output'] = null +) { const run = await claimNextRun(backend) return backend.completeWorkflowRun({ workflowRunId: run.id, workerId: 'fixture-worker', - output: null + output }) } -export async function createCompletedRun(backend: BackendSqlite, workflowName: string) { +export async function createCompletedRun( + backend: BackendSqlite, + workflowName: string, + output: Parameters[0]['output'] = null +) { const run = await createPendingRun(backend, workflowName) - const completed = await completeNextRun(backend) + const completed = await completeNextRun(backend, output) 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 9ea69992..aa8f111b 100644 --- a/apps/web/src/lib/server/workflow/publish.test.ts +++ b/apps/web/src/lib/server/workflow/publish.test.ts @@ -232,6 +232,71 @@ describe('validateAuditForPublish', () => { expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true }) }) + test.each([ + { filePaths: [] }, + { filePaths: ['added.book.md', 'updated.book.md', 'added.book.md'] } + ])( + 'validates and deduplicates explicit selection $filePaths against audit upserts', + async ({ filePaths }) => { + const run = makeRun({ + output: JSON.stringify({ + schemas: { added: ['book'], updated: [], pruned: [] }, + pieces: { + added: ['added.book.md'], + updated: ['updated.book.md'], + pruned: ['removed.book.md'] + } + }) + }) + mocks.getWorkflowRun.mockResolvedValue(run) + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => + name === 'PublishAudit' ? run : null + ) + expect(await validateAuditForPublish(backend, run.id, filePaths)).toEqual({ + ok: true, + filePaths: [...new Set(filePaths)] + }) + } + ) + + test.each([null, 'not JSON', '"ok"', '{}', '{"pieces":{"added":[]}}'])( + 'requires a valid audit report even for an empty selection: %s', + async (output) => { + const run = makeRun({ output }) + mocks.getWorkflowRun.mockResolvedValue(run) + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => + name === 'PublishAudit' ? run : null + ) + expect(await validateAuditForPublish(backend, run.id, [])).toEqual({ + ok: false, + reason: 'audit has no valid changes report; re-check before publishing' + }) + // Omitted selection keeps accepting legacy completed audits. + expect(await validateAuditForPublish(backend, run.id)).toEqual({ ok: true }) + } + ) + + test.each(['removed.book.md', 'book', './added.book.md', 'unknown.book.md'])( + 'rejects a path outside exact audit additions/updates: %s', + async (filePath) => { + const run = makeRun({ + output: JSON.stringify({ + schemas: { added: ['book'], updated: [], pruned: [] }, + pieces: { added: ['added.book.md'], updated: [], pruned: ['removed.book.md'] } + }) + }) + mocks.getWorkflowRun.mockResolvedValue(run) + mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) => + name === 'PublishAudit' ? run : null + ) + expect(await validateAuditForPublish(backend, run.id, [filePath])).toEqual({ + ok: false, + reason: 'selected file paths must be pending additions or updates from this audit', + status: 400 + }) + } + ) + 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' }) diff --git a/apps/web/src/lib/server/workflow/publish.ts b/apps/web/src/lib/server/workflow/publish.ts index 55b6fd29..7be56401 100644 --- a/apps/web/src/lib/server/workflow/publish.ts +++ b/apps/web/src/lib/server/workflow/publish.ts @@ -30,11 +30,14 @@ export async function findInFlightPublishRun( return null } -export type AuditGuard = { ok: true } | { ok: false; reason: string } +export type AuditGuard = + | { ok: true; filePaths?: string[] } + | { ok: false; reason: string; status?: 400 } export async function validateAuditForPublish( backend: Pick, - auditRunId: unknown + auditRunId: unknown, + filePaths?: string[] ): Promise { if (typeof auditRunId !== 'string' || auditRunId.length === 0) { return { ok: false, reason: 'no audit run provided; check for changes before publishing' } @@ -62,6 +65,22 @@ export async function validateAuditForPublish( } } + if (filePaths !== undefined) { + const diff = parsePiecesDiff(run.output) + if (!diff) { + return { ok: false, reason: 'audit has no valid changes report; re-check before publishing' } + } + const pending = new Set([...diff.pieces.added, ...diff.pieces.updated]) + if (filePaths.some((filePath) => !pending.has(filePath))) { + return { + ok: false, + reason: 'selected file paths must be pending additions or updates from this audit', + status: 400 + } + } + return { ok: true, filePaths: [...new Set(filePaths)] } + } + return { ok: true } } diff --git a/apps/web/src/routes/(auth)/api/admin/publish/+server.ts b/apps/web/src/routes/(auth)/api/admin/publish/+server.ts index ef16f76d..059241d1 100644 --- a/apps/web/src/routes/(auth)/api/admin/publish/+server.ts +++ b/apps/web/src/routes/(auth)/api/admin/publish/+server.ts @@ -7,6 +7,7 @@ import { withPublishAdmission } from '$lib/server/workflow/publish.js' import { publishSpec } from '@luzzle/web.jobs/specs' +import type { PublishPayload } from '@luzzle/web.jobs' export const POST: RequestHandler = async ({ request }) => { try { @@ -18,19 +19,34 @@ export const POST: RequestHandler = async ({ request }) => { } 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 selectionProvided = Object.hasOwn(body ?? {}, 'filePaths') + let filePaths: string[] | undefined + if (selectionProvided) { + const requestedPaths = body.filePaths + if (!Array.isArray(requestedPaths)) { + return json({ message: 'filePaths must be an array of strings' }, { status: 400 }) } + if (!requestedPaths.every((filePath: unknown) => typeof filePath === 'string')) { + return json({ message: 'filePaths must be an array of strings' }, { status: 400 }) + } + filePaths = requestedPaths } + if (auditRunId || selectionProvided) { + const guard = await validateAuditForPublish(getOpenWorkflowBackend(), auditRunId, filePaths) + if (!guard.ok) { + return json({ message: guard.reason }, { status: guard.status ?? 412 }) + } + filePaths = guard.filePaths + } + + let bisync = body?.bisync === true + if (auditRunId) bisync = false + const input: PublishPayload = { bisync } + if (filePaths !== undefined) input.filePaths = filePaths + const openWorkflow = getOpenWorkflow() - const handle = await openWorkflow.runWorkflow(publishSpec, { - bisync: auditRunId ? false : bisync - }) + const handle = await openWorkflow.runWorkflow(publishSpec, input) return json({ jobId: handle.workflowRun.id }) }) } catch (error) { diff --git a/apps/web/src/routes/(auth)/api/admin/publish/admission.test.ts b/apps/web/src/routes/(auth)/api/admin/publish/admission.test.ts index e84a8be0..c9f70b35 100644 --- a/apps/web/src/routes/(auth)/api/admin/publish/admission.test.ts +++ b/apps/web/src/routes/(auth)/api/admin/publish/admission.test.ts @@ -18,6 +18,15 @@ vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflowBackend: vi.fn() })) +const auditDiff = { + schemas: { added: ['book'], updated: [], pruned: [] }, + pieces: { + added: ['added.book.md', '書 shelves/été?draft=1&tag=#%.book.md'], + updated: ['updated.book.md'], + pruned: ['removed.book.md'] + } +} + function event(body: unknown = {}) { return { request: new Request('http://localhost/api/admin/publish', { @@ -30,7 +39,7 @@ function event(body: unknown = {}) { describe('publish/audit admission', () => { let queue: WorkflowQueue - async function enqueue(spec: { name: string }, input: { bisync: boolean }) { + async function enqueue(spec: { name: string }, input: { bisync: boolean; filePaths?: string[] }) { vi.setSystemTime(new Date('2026-06-21T00:00:00Z')) const workflowRun = await createPendingRun(queue.backend, spec.name, input) return { workflowRun } @@ -220,6 +229,201 @@ describe('publish/audit admission', () => { expect(runWorkflow).toHaveBeenCalledTimes(1) }) + test.each([false, true])( + 'persists omitted selection as publish-all with bisync=%s', + async (bisync) => { + const response = await publish(event({ bisync })) + expect(response.status).toBe(200) + const { jobId } = await response.json() + const run = await queue.backend.getWorkflowRun({ workflowRunId: jobId }) + expect(run?.input).toEqual({ bisync }) + expect(run?.input).not.toHaveProperty('filePaths') + } + ) + + test('keeps selection omitted for a checked publish-all with a legacy audit report', async () => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', 'ok') + const response = await publish(event({ auditRunId: auditRun.id, bisync: true })) + expect(response.status).toBe(200) + const { jobId } = await response.json() + expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({ + bisync: false + }) + }) + + test.each([ + { label: 'pending upserts', diff: auditDiff }, + { + label: 'removals only', + diff: { + schemas: { added: [], updated: [], pruned: [] }, + pieces: { added: [], updated: [], pruned: ['removed.book.md'] } + } + }, + { + label: 'schemas only', + diff: { + schemas: { added: [], updated: ['book'], pruned: [] }, + pieces: { added: [], updated: [], pruned: [] } + } + }, + { + label: 'no changes', + diff: { + schemas: { added: [], updated: [], pruned: [] }, + pieces: { added: [], updated: [], pruned: [] } + } + } + ])('preserves explicit [] with $label for global processing', async ({ diff }) => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', diff) + const response = await publish(event({ auditRunId: auditRun.id, bisync: true, filePaths: [] })) + expect(response.status).toBe(200) + const { jobId } = await response.json() + expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({ + bisync: false, + filePaths: [] + }) + expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), { + bisync: false, + filePaths: [] + }) + }) + + test('persists deduplicated exact added/updated paths without decoding special characters', async () => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const filePaths = [auditDiff.pieces.added[1], 'updated.book.md', 'added.book.md'] + const response = await publish( + event({ auditRunId: auditRun.id, filePaths: [...filePaths, filePaths[0]] }) + ) + expect(response.status).toBe(200) + const { jobId } = await response.json() + expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({ + bisync: false, + filePaths + }) + }) + + test.each([ + { filePaths: null }, + { filePaths: 'added.book.md' }, + { filePaths: {} }, + { filePaths: true }, + { filePaths: 1 }, + { filePaths: [null] }, + { filePaths: [1] }, + { filePaths: [['added.book.md']] }, + { filePaths: ['added.book.md', {}] } + ])('rejects malformed selection $filePaths before enqueue', async ({ filePaths }) => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const response = await publish(event({ auditRunId: auditRun.id, filePaths })) + expect(response.status).toBe(400) + expect(await response.json()).toEqual({ message: 'filePaths must be an array of strings' }) + expect(runWorkflow).not.toHaveBeenCalled() + expect((await queue.backend.countWorkflowRuns()).pending).toBe(0) + }) + + test.each([ + '', + 'unknown.book.md', + 'removed.book.md', + 'book', + '/tmp/added.book.md', + '../added.book.md', + './added.book.md', + 'folder/../added.book.md', + encodeURIComponent(auditDiff.pieces.added[1]) + ])('rejects paths outside exact audit upsert membership: %s', async (filePath) => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const response = await publish( + event({ auditRunId: auditRun.id, filePaths: ['added.book.md', filePath] }) + ) + expect(response.status).toBe(400) + expect(await response.json()).toEqual({ + message: 'selected file paths must be pending additions or updates from this audit' + }) + expect(runWorkflow).not.toHaveBeenCalled() + }) + + test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])( + 'requires an audit for explicit selection $filePaths', + async ({ filePaths }) => { + const response = await publish(event({ filePaths, bisync: true })) + expect(response.status).toBe(412) + expect(await response.json()).toEqual({ + message: 'no audit run provided; check for changes before publishing' + }) + expect(runWorkflow).not.toHaveBeenCalled() + } + ) + + test.each([null, '', 42, 'missing-run'])( + 'rejects invalid audit identity for explicit []: %s', + async (auditRunId) => { + expect((await publish(event({ auditRunId, filePaths: [] }))).status).toBe(412) + expect(runWorkflow).not.toHaveBeenCalled() + } + ) + + const invalidAuditOutputs: Parameters[2][] = [ + null, + 'ok', + {}, + { pieces: auditDiff.pieces }, + { ...auditDiff, pieces: { ...auditDiff.pieces, updated: [42] } } + ] + + test.each(invalidAuditOutputs)( + 'rejects absent/legacy/malformed completed audit output for []: %j', + async (output) => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', output) + const response = await publish(event({ auditRunId: auditRun.id, filePaths: [] })) + expect(response.status).toBe(412) + expect(await response.json()).toEqual({ + message: 'audit has no valid changes report; re-check before publishing' + }) + expect(runWorkflow).not.toHaveBeenCalled() + } + ) + + test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])( + 'rejects stale audit selection $filePaths and releases admission', + async ({ filePaths }) => { + const oldAudit = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) + const latestAudit = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const rejected = await publish(event({ auditRunId: oldAudit.id, filePaths })) + expect(rejected.status).toBe(412) + expect(runWorkflow).not.toHaveBeenCalled() + expect((await publish(event({ auditRunId: latestAudit.id, filePaths }))).status).toBe(200) + } + ) + + test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])( + 'consumes the audit after selected publish $filePaths completes', + async ({ filePaths }) => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const response = await publish(event({ auditRunId: auditRun.id, filePaths })) + expect(response.status).toBe(200) + await completeNextRun(queue.backend) + const rejected = await publish(event({ auditRunId: auditRun.id, filePaths })) + 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 active-run conflict precedence for an explicit selection', async () => { + const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff) + const accepted = await publish(event({ auditRunId: auditRun.id, filePaths: [] })) + const rejected = await publish(event({ auditRunId: auditRun.id, filePaths: ['unknown'] })) + expect(accepted.status).toBe(200) + expect(rejected.status).toBe(409) + expect(await rejected.json()).toEqual(await accepted.json()) + 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) diff --git a/packages/web/jobs/src/types.ts b/packages/web/jobs/src/types.ts index 4b3e81d1..1d0b776e 100644 --- a/packages/web/jobs/src/types.ts +++ b/packages/web/jobs/src/types.ts @@ -12,6 +12,8 @@ export interface AssetRecord { export interface PublishPayload { bisync?: boolean + /** Omitted publishes all pending pieces; an empty list selects no upserts. */ + filePaths?: string[] } export interface PublishPieceFailure { diff --git a/packages/web/worker/src/services/publication.test.ts b/packages/web/worker/src/services/publication.test.ts index 9d1d392f..82ad07d0 100644 --- a/packages/web/worker/src/services/publication.test.ts +++ b/packages/web/worker/src/services/publication.test.ts @@ -162,7 +162,7 @@ describe('publication', () => { const summary = emptyPiecesDiff() summary.pieces.added.push('new.books.md') - const prepared = await publishPrepareStep.run(summary, ctx) + const prepared = await publishPrepareStep.run({ summary }, ctx) expect(prepared).toEqual({ status: 'completed', @@ -180,6 +180,57 @@ describe('publication', () => { expect(summary.pieces.updated).toEqual([]) }) + test.each([ + { filePaths: undefined, added: ['new.books.md'], updated: ['changed.books.md'] }, + { filePaths: [], added: [], updated: [] }, + { + filePaths: ['changed.books.md', 'changed.books.md', 'unchanged.books.md', 'missing.books.md'], + added: [], + updated: ['changed.books.md'], + }, + ])('prepare intersects fresh pending work with selection $filePaths, retaining global work', async ({ + filePaths, + added, + updated, + }) => { + await indexPiece('new.books.md', 'new-hash') + await indexPiece('changed.books.md', 'current-hash') + await indexPiece('unchanged.books.md', 'same-hash') + await indexPiece('gone.books.md', 'gone-hash') + await publishMetadata('changed.books.md', 'old-hash', 123) + await publishMetadata('unchanged.books.md', 'same-hash', 456) + await publishMetadata('gone.books.md', 'gone-hash', 789) + await db.deleteFrom('pieces_items').where('file_path', '=', 'gone.books.md').execute() + const summary = { + schemas: { added: ['books'], updated: ['films'], pruned: ['old'] }, + pieces: { + added: ['changed.books.md', 'missing.books.md'], + updated: ['unchanged.books.md', 'new.books.md'], + pruned: ['already-pruned.books.md', 'gone.books.md'], + }, + } + const before = await db.selectFrom('web_pieces').selectAll().orderBy('file_path').execute() + + const result = await publishPrepareStep.run({ summary, filePaths }, ctx) + + expect(result.status).toBe('completed') + if (result.status !== 'completed') throw new Error('Expected a completed plan') + expect(result.value).toEqual({ + pieces: [ + ...(updated.length ? [{ filePath: 'changed.books.md', contentHash: 'current-hash' }] : []), + ...(added.length ? [{ filePath: 'new.books.md', contentHash: 'new-hash' }] : []), + ], + summary: { + schemas: summary.schemas, + pieces: { added, updated, pruned: ['already-pruned.books.md', 'gone.books.md'] }, + }, + }) + expect(await db.selectFrom('web_pieces').selectAll().orderBy('file_path').execute()).toEqual( + before + ) + expect(summary.pieces.added).toEqual(['changed.books.md', 'missing.books.md']) + }) + test('complete advances healthy pieces only, using their captured hashes', async () => { await indexPiece('healthy.books.md', 'even-newer-hash') await indexPiece('failed.books.md', 'failed-hash') @@ -338,7 +389,7 @@ describe('publication', () => { }) test('complete and prepare accept an empty archive', async () => { - expect(await publishPrepareStep.run(emptyPiecesDiff(), ctx)).toEqual({ + expect(await publishPrepareStep.run({ summary: emptyPiecesDiff() }, ctx)).toEqual({ status: 'completed', value: { pieces: [], summary: emptyPiecesDiff() }, }) diff --git a/packages/web/worker/src/steps/publish-prepare.ts b/packages/web/worker/src/steps/publish-prepare.ts index 74034d44..948ef18d 100644 --- a/packages/web/worker/src/steps/publish-prepare.ts +++ b/packages/web/worker/src/steps/publish-prepare.ts @@ -8,14 +8,35 @@ export interface PublishPlan { summary: PiecesDiff } -export const publishPrepareStep: Step = { +export interface PublishPrepareInput { + summary: PiecesDiff + filePaths?: string[] +} + +export const publishPrepareStep: Step = { name: 'publish.prepare', - async run(summary, { db, config, logger }) { + async run({ summary: indexSummary, filePaths }, { db, config, logger }) { const pending = await getPendingPublication(db, config) - logger.info('publish.prepare complete', { count: pending.pieces.length }) - return completed({ - pieces: pending.pieces, - summary: includePendingPieces(summary, pending.diff), - }) + let pieces = pending.pieces + if (filePaths !== undefined) { + const selectedPaths = new Set(filePaths) + pieces = pending.pieces.filter((piece) => selectedPaths.has(piece.filePath)) + } + const targetPaths = new Set(pieces.map((piece) => piece.filePath)) + + // Core sync is global; only authoritative pending targets count as upserts. + const globalChanges: PiecesDiff = { + schemas: indexSummary.schemas, + pieces: { added: [], updated: [], pruned: indexSummary.pieces.pruned }, + } + const selectedChanges: PiecesDiff['pieces'] = { + added: pending.diff.added.filter((filePath) => targetPaths.has(filePath)), + updated: pending.diff.updated.filter((filePath) => targetPaths.has(filePath)), + pruned: pending.diff.pruned, + } + const publicationSummary = includePendingPieces(globalChanges, selectedChanges) + + logger.info('publish.prepare complete', { count: pieces.length }) + return completed({ pieces, summary: publicationSummary }) }, } diff --git a/packages/web/worker/src/workflows/publish-recovery.test.ts b/packages/web/worker/src/workflows/publish-recovery.test.ts index 63729307..082d4082 100644 --- a/packages/web/worker/src/workflows/publish-recovery.test.ts +++ b/packages/web/worker/src/workflows/publish-recovery.test.ts @@ -7,14 +7,16 @@ import { validateConfig } from '@luzzle/web.config' import { createAppDb, runWebMigrations } from '@luzzle/web.db' import type { PublishPayload, PublishResult } from '@luzzle/web.jobs' import { publishSpec } from '@luzzle/web.jobs/specs' -import type { Workflow } from 'openworkflow' +import { OpenWorkflow, type Workflow } from 'openworkflow' +import { BackendSqlite } from 'openworkflow/sqlite' import { completed, type StepResult } from '../core/step.js' import { RcloneClient } from '../services/rclone.js' +import { createOpenWorkflowLogger } from '../services/logger.js' import { setWorkerContext, type WorkerContext } from '../services/context.js' import { getPendingPublication } from '../services/publication.js' import { luzzleSyncStep } from '../steps/luzzle-sync.js' import { luzzleAuditStep } from '../steps/luzzle-audit.js' -import { publishPrepareStep } from '../steps/publish-prepare.js' +import { publishPrepareStep, type PublishPlan } from '../steps/publish-prepare.js' import { webSyncStep } from '../steps/web-sync/index.js' import type { AssetsGenerateInput, AssetsGenerateResult } from '../steps/assets-generate.js' import { registerPublishWorkflow } from './publish.js' @@ -160,6 +162,77 @@ describe('publish hash recovery with a real archive and SQLite', () => { return { result, phaseResults } } + // Real SDK queue/checkpoint storage; only external delivery and transforms are stubbed. + async function withQueue( + run: ( + client: OpenWorkflow, + worker: ReturnType, + backend: BackendSqlite + ) => Promise + ) { + const backend = BackendSqlite.connect(path.join(directory, 'queue.sqlite')) + const client = new OpenWorkflow({ backend, logger: createOpenWorkflowLogger(ctx.logger) }) + client.implementWorkflow(publishSpec, workflow) + const worker = client.newWorker({ concurrency: 1 }) + try { + await run(client, worker, backend) + } finally { + await worker.stop() + await backend.stop() + } + } + + async function finishQueued( + worker: ReturnType, + backend: BackendSqlite, + id: string + ) { + await vi.waitFor( + async () => { + await worker.tick() + expect((await backend.getWorkflowRun({ workflowRunId: id }))?.status).toBe('completed') + }, + { timeout: 5000 } + ) + const run = await backend.getWorkflowRun({ workflowRunId: id }) + return run!.output as unknown as PublishResult + } + + async function publicRows(filePath: string) { + return { + pieces: await ctx.db + .selectFrom('web_pieces') + .selectAll() + .where('file_path', '=', filePath) + .execute(), + assets: await ctx.db + .selectFrom('web_pieces_assets') + .selectAll() + .where('piece_file_path', '=', filePath) + .execute(), + } + } + + async function seedAsset(filePath: string) { + const piece = await ctx.db + .selectFrom('web_pieces') + .select('key') + .where('file_path', '=', filePath) + .executeTakeFirstOrThrow() + await ctx.db + .insertInto('web_pieces_assets') + .values({ + piece_file_path: filePath, + piece_key: piece.key, + piece_asset_path: '.assets/cover.jpg', + asset_key: `${piece.key}-cover`, + transformation: 'image', + asset_path: `${piece.key}/cover.webp`, + mime_type: 'image/webp', + }) + .execute() + } + async function hashes(filePath: string) { const indexed = await ctx.db .selectFrom('pieces_cache') @@ -183,6 +256,219 @@ describe('publish hash recovery with a real archive and SQLite', () => { return valueOf(await luzzleAuditStep.run(undefined, ctx)) } + test.each(['added', 'updated'] as const)( + 'queued selection publishes only the pending %s target, preserving unselected rows and pruning globally', + async (action) => { + const removed = 'removed.books.md' + const unchanged = 'unchanged.books.md' + if (action === 'updated') await writePiece(healthy, 'Original selected') + await writePiece(broken, 'Original unselected') + await writePiece(removed, 'Removed globally') + await writePiece(unchanged, 'Unchanged') + await withQueue(async (client, worker, backend) => { + const baseline = await client.runWorkflow(publishSpec, {}) + await finishQueued(worker, backend, baseline.workflowRun.id) + await seedAsset(broken) + await seedAsset(removed) + const unselectedBefore = await publicRows(broken) + const unchangedBefore = await publicRows(unchanged) + await writePiece(healthy, 'Selected current') + await writePiece(broken, 'Unselected current') + await writePiece('new-unselected.books.md', 'New unselected') + await rm(path.join(ctx.config.storage.root, removed)) + mocks.assets.mockClear() + mocks.cdn.mockClear() + mocks.cache.mockClear() + + const selected = await client.runWorkflow(publishSpec, { + filePaths: [healthy, healthy, unchanged, removed, 'missing.books.md'], + }) + const result = await finishQueued(worker, backend, selected.workflowRun.id) + + expect(result).toEqual({ + schemas: emptyPiecesDiff().schemas, + pieces: { + added: action === 'added' ? [healthy] : [], + updated: action === 'updated' ? [healthy] : [], + pruned: [removed], + }, + failedPieces: [], + }) + expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx) + expect(mocks.cdn).toHaveBeenCalledOnce() + expect(mocks.cache).toHaveBeenCalledOnce() + expect(await publicRows(broken)).toEqual(unselectedBefore) + expect(await publicRows(unchanged)).toEqual(unchangedBefore) + expect(await publicRows(removed)).toEqual({ pieces: [], assets: [] }) + expect(await publicRows('new-unselected.books.md')).toEqual({ pieces: [], assets: [] }) + const state = await hashes(healthy) + expect(state.published).toBe(state.indexed) + expect(state.title).toBe('Selected current') + expect((await audit()).pieces).toEqual({ + added: ['new-unselected.books.md'], + updated: [broken], + pruned: [], + }) + }) + } + ) + + test.each([false, true])( + 'queued explicit empty selection keeps schema housekeeping and global removals (%s)', + async (removePiece) => { + const removed = 'removed.books.md' + for (const filePath of [healthy, broken, removed]) { + await writePiece(filePath, `Original ${filePath}`) + } + await publish('baseline-empty-selection') + await seedAsset(broken) + await seedAsset(removed) + const healthyBefore = await publicRows(healthy) + const brokenBefore = await publicRows(broken) + const removedBefore = await publicRows(removed) + await writePiece(healthy, 'Pending selected none') + await writePiece(broken, 'Pending unselected') + if (removePiece) await rm(path.join(ctx.config.storage.root, removed)) + await writeFile( + path.join(ctx.config.storage.root, '.luzzle', 'schemas', 'books.json'), + JSON.stringify({ + type: 'object', + title: 'books', + description: 'Updated globally', + properties: { title: { type: 'string' } }, + required: ['title'], + additionalProperties: true, + }) + ) + mocks.assets.mockClear() + mocks.cdn.mockClear() + mocks.cache.mockClear() + + await withQueue(async (client, worker, backend) => { + const empty = await client.runWorkflow(publishSpec, { filePaths: [] }) + const result = await finishQueued(worker, backend, empty.workflowRun.id) + expect(result).toEqual({ + schemas: { added: [], updated: ['books'], pruned: [] }, + pieces: { added: [], updated: [], pruned: removePiece ? [removed] : [] }, + failedPieces: [], + }) + expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [] }, ctx) + expect(mocks.cdn).toHaveBeenCalledOnce() + expect(mocks.cache).toHaveBeenCalledOnce() + expect(await publicRows(healthy)).toEqual(healthyBefore) + expect(await publicRows(broken)).toEqual(brokenBefore) + expect(await publicRows(removed)).toEqual( + removePiece ? { pieces: [], assets: [] } : removedBefore + ) + expect((await audit()).pieces).toEqual({ added: [], updated: batch, pruned: [] }) + }) + } + ) + + test('queued interrupted delivery replays the captured selected plan without widening membership or hashes', async () => { + const late = 'late.books.md' + await writePiece(healthy, 'Original selected') + await writePiece(broken, 'Original unselected') + await publish('baseline-interruption') + await seedAsset(broken) + const unselectedBefore = await publicRows(broken) + await writePiece(healthy, 'Selected revision two') + await writePiece(broken, 'Unselected revision two') + mocks.assets.mockClear() + mocks.cache.mockRejectedValueOnce(new Error('interrupted delivery')) + + await withQueue(async (client, worker, backend) => { + const handle = await client.runWorkflow(publishSpec, { filePaths: [healthy, late] }) + const id = handle.workflowRun.id + await worker.tick() + await vi.waitFor( + async () => { + const run = await backend.getWorkflowRun({ workflowRunId: id }) + expect(run?.status).toBe('pending') + expect(run?.attempts).toBe(1) + }, + { timeout: 5000 } + ) + const attempts = (await backend.listStepAttempts({ workflowRunId: id })).data + const plan = attempts.find((attempt) => attempt.stepName === 'publish.prepare')! + .output as unknown as PublishPlan + expect(plan.pieces.map((piece) => piece.filePath)).toEqual([healthy]) + const capturedHash = plan.pieces[0]!.contentHash + await writePiece(healthy, 'Selected revision three') + await writePiece(late, 'Selected but not pending when prepared') + await luzzleSyncStep.run(undefined, ctx) + + const result = await finishQueued(worker, backend, id) + + expect(result).toEqual({ ...plan.summary, failedPieces: [] }) + expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx) + expect(await publicRows(broken)).toEqual(unselectedBefore) + expect(await publicRows(late)).toEqual({ pieces: [], assets: [] }) + const state = await hashes(healthy) + expect(state.published).toBe(capturedHash) + expect(state.indexed).not.toBe(capturedHash) + expect(state.title).toBe('Selected revision two') + const retried = (await backend.listStepAttempts({ workflowRunId: id })).data + expect(retried.filter((attempt) => attempt.stepName === 'publish.prepare')).toHaveLength(1) + expect(retried.find((attempt) => attempt.stepName === 'publish.prepare')!.output).toEqual(plan) + expect(retried.filter((attempt) => attempt.stepName === 'cache.purge')).toHaveLength(2) + expect((await audit()).pieces).toEqual({ added: [late], updated: batch, pruned: [] }) + }) + }) + + test('queued partial-failure rerun retains selected hashes and leaves unselected pending work untouched', async () => { + const unselected = 'unselected.books.md' + const late = 'late.books.md' + for (const filePath of [...batch, unselected]) await writePiece(filePath, `Original ${filePath}`) + await publish('baseline-partial-selection') + await seedAsset(unselected) + const unselectedBefore = await publicRows(unselected) + for (const filePath of [...batch, unselected]) await writePiece(filePath, `Changed ${filePath}`) + const failedPieces = [{ filePath: broken, message: 'selected transform failed' }] + mocks.assets.mockClear() + mocks.assets.mockResolvedValueOnce(completed({ failedPieces })) + + await withQueue(async (client, worker, backend) => { + const partial = await client.runWorkflow(publishSpec, { filePaths: [...batch, late] }) + const partialResult = await finishQueued(worker, backend, partial.workflowRun.id) + expect(partialResult.failedPieces).toEqual(failedPieces) + expect((await hashes(healthy)).published).toBe((await hashes(healthy)).indexed) + expect((await hashes(broken)).published).toBeNull() + expect(await publicRows(unselected)).toEqual(unselectedBefore) + const attempts = ( + await backend.listStepAttempts({ workflowRunId: partial.workflowRun.id }) + ).data + const plan = attempts.find((attempt) => attempt.stepName === 'publish.prepare')! + .output as unknown as PublishPlan + expect(plan.pieces.map((piece) => piece.filePath)).toEqual(batch) + await writePiece(healthy, 'Newer than captured selected hash') + await writePiece(late, 'Late selected addition') + await luzzleSyncStep.run(undefined, ctx) + mocks.assets.mockClear() + + const retry = await client.rerunWorkflowRun(partial.workflowRun.id, { + fromStep: 'assets.generate', + }) + const result = await finishQueued(worker, backend, retry.id) + + expect(result).toEqual({ ...plan.summary, failedPieces: [] }) + expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: batch }, ctx) + for (const piece of plan.pieces) { + expect((await hashes(piece.filePath)).published).toBe(piece.contentHash) + } + expect((await hashes(healthy)).indexed).not.toBe((await hashes(healthy)).published) + expect(await publicRows(unselected)).toEqual(unselectedBefore) + expect(await publicRows(late)).toEqual({ pieces: [], assets: [] }) + const retried = (await backend.listStepAttempts({ workflowRunId: retry.id })).data + expect(retried.find((attempt) => attempt.stepName === 'publish.prepare')!.output).toEqual(plan) + expect((await audit()).pieces).toEqual({ + added: [late], + updated: [healthy, unselected], + pruned: [], + }) + }) + }) + test('a fresh publish rediscovers new and updated pieces after the sync result is discarded', async () => { await writePiece(healthy, 'Original') await publish('baseline') @@ -235,7 +521,7 @@ describe('publish hash recovery with a real archive and SQLite', () => { test('metadata sync alone never records publication success for either new or existing rows', async () => { await writePiece(healthy, 'Original') const initialDiff = valueOf(await luzzleSyncStep.run(undefined, ctx)) - const initialPlan = valueOf(await publishPrepareStep.run(initialDiff, ctx)) + const initialPlan = valueOf(await publishPrepareStep.run({ summary: initialDiff }, ctx)) await webSyncStep.run({ filePaths: initialPlan.pieces.map((piece) => piece.filePath) }, ctx) expect(await hashes(healthy)).toMatchObject({ published: null, lastPublishedAt: null }) expect((await getPendingPublication(ctx.db, ctx.config)).pieces).toEqual(initialPlan.pieces) @@ -244,7 +530,7 @@ describe('publish hash recovery with a real archive and SQLite', () => { const original = await hashes(healthy) await writePiece(healthy, 'Metadata changed') const diff = valueOf(await luzzleSyncStep.run(undefined, ctx)) - const plan = valueOf(await publishPrepareStep.run(diff, ctx)) + const plan = valueOf(await publishPrepareStep.run({ summary: diff }, ctx)) await webSyncStep.run({ filePaths: plan.pieces.map((piece) => piece.filePath) }, ctx) const state = await hashes(healthy) expect(state.title).toBe('Metadata changed') diff --git a/packages/web/worker/src/workflows/publish.test.ts b/packages/web/worker/src/workflows/publish.test.ts index 6faebf12..0a2ffc11 100644 --- a/packages/web/worker/src/workflows/publish.test.ts +++ b/packages/web/worker/src/workflows/publish.test.ts @@ -186,6 +186,57 @@ describe('workflows/publish', () => { ) }) + test.each([ + { filePaths: undefined }, + { filePaths: [] }, + { filePaths: ['healthy.books.md'] }, + ])( + 'passes selection $filePaths to prepare without narrowing global sync or delivery', + async ({ filePaths }) => { + const { result } = await runPublish({ filePaths }) + + expect(mocks.luzzleSync.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything()) + expect(mocks.publishPrepare.run).toHaveBeenCalledExactlyOnceWith( + { summary, filePaths }, + expect.anything() + ) + expect(mocks.cdnSync.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything()) + expect(mocks.cachePurge.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything()) + expect(result).toEqual({ ...summary, failedPieces: [] }) + } + ) + + test('uses only the saved selected plan and hashes when retrying, not the input or global diff', async () => { + const plan = { + pieces: [{ filePath: 'healthy.books.md', contentHash: 'captured-hash' }], + summary: { + schemas: summary.schemas, + pieces: { added: ['healthy.books.md'], updated: [], pruned: summary.pieces.pruned }, + }, + } + const { result } = await runPublish( + { filePaths: ['healthy.books.md', 'broken.books.md'] }, + new Map([['luzzle.sync', summary], ['publish.prepare', plan]]) + ) + + expect(mocks.luzzleSync.run).not.toHaveBeenCalled() + expect(mocks.publishPrepare.run).not.toHaveBeenCalled() + expect(mocks.webSync.run).toHaveBeenCalledExactlyOnceWith( + { filePaths: ['healthy.books.md'] }, + expect.anything() + ) + expect(mocks.assetsGenerate.run).toHaveBeenCalledExactlyOnceWith( + { filePaths: ['healthy.books.md'] }, + expect.anything() + ) + expect(mocks.publishComplete.run).toHaveBeenCalledExactlyOnceWith( + { pieces: plan.pieces, failedPieces: [] }, + expect.anything() + ) + expect(result).toEqual({ ...plan.summary, failedPieces: [] }) + expectHealthyPhasesComplete() + }) + test.each(['cdn.sync', 'cache.purge'])( 'keeps a global %s rejection as a real publish failure', async (phase) => { diff --git a/packages/web/worker/src/workflows/publish.ts b/packages/web/worker/src/workflows/publish.ts index ad9fbd1f..8696e341 100644 --- a/packages/web/worker/src/workflows/publish.ts +++ b/packages/web/worker/src/workflows/publish.ts @@ -36,7 +36,7 @@ export function registerPublishWorkflow(): void { jobId, progress, publishPrepareStep, - summary ?? emptyPiecesDiff() + { summary: summary ?? emptyPiecesDiff(), filePaths: input.filePaths } ) if (!plan) throw new Error('publish.prepare did not return a plan') const changedPaths = plan.pieces.map((piece) => piece.filePath)