From 3ccd9ad2986ec7aacb96abbea03e9ad9f0c6f5c3 Mon Sep 17 00:00:00 2001 From: eleith Date: Mon, 9 Nov 2026 16:37:12 -0800 Subject: [PATCH 1/2] [web] copy selected generated files --- .../worker/src/steps/cdn-copy-files.test.ts | 77 +++++++++++++++++++ .../web/worker/src/steps/cdn-copy-files.ts | 46 +++++++++++ 2 files changed, 123 insertions(+) create mode 100644 packages/web/worker/src/steps/cdn-copy-files.test.ts create mode 100644 packages/web/worker/src/steps/cdn-copy-files.ts diff --git a/packages/web/worker/src/steps/cdn-copy-files.test.ts b/packages/web/worker/src/steps/cdn-copy-files.test.ts new file mode 100644 index 00000000..edff26cf --- /dev/null +++ b/packages/web/worker/src/steps/cdn-copy-files.test.ts @@ -0,0 +1,77 @@ +import { mkdtemp, mkdir, realpath, rm, symlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { afterEach, beforeEach, expect, test, vi } from 'vitest' +import type { WorkerContext } from '../services/context.js' +import { cdnCopyFilesStep } from './cdn-copy-files.js' + +let directory: string +let assets: string +let ctx: WorkerContext + +beforeEach(async () => { + directory = await realpath(await mkdtemp(path.join(tmpdir(), 'cdn-copy-files-'))) + assets = path.join(directory, 'assets') + await mkdir(path.join(assets, 'target'), { recursive: true }) + await writeFile(path.join(assets, 'target', 'page.html'), 'page') + await writeFile(path.join(assets, 'cover.png'), 'cover') + await writeFile(path.join(assets, 'unrelated.html'), 'another piece') + ctx = { + config: { + paths: { assets }, + sync: { + config: '/app/rclone.conf', + cdn: { remote: 'cdn', path: 'published/assets/', strategy: 'sync' }, + }, + }, + logger: { info: vi.fn() }, + rclone: { copy: vi.fn().mockResolvedValue(undefined), sync: vi.fn() }, + } as unknown as WorkerContext +}) + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }) +}) + +test('copies exactly the deduplicated files, never their directory or unrelated outputs', async () => { + const result = await cdnCopyFilesStep.run( + { filePaths: ['target/page.html', 'cover.png', 'target/page.html'] }, + ctx + ) + + expect(result.status).toBe('completed') + expect( + vi.mocked(ctx.rclone.copy).mock.calls.map(([options]) => ({ + localPath: options.localPath, + remotePath: options.remotePath, + })) + ).toEqual([ + { localPath: path.join(assets, 'target', 'page.html'), remotePath: 'published/assets/target' }, + { localPath: path.join(assets, 'cover.png'), remotePath: 'published/assets' }, + ]) + expect(ctx.rclone.sync).not.toHaveBeenCalled() +}) + +test('validates the whole list before copying any file', async () => { + const outside = path.join(directory, 'outside.html') + await writeFile(outside, 'not generated output') + await symlink(outside, path.join(assets, 'escape.html')) + + await expect( + cdnCopyFilesStep.run({ filePaths: ['cover.png', 'escape.html'] }, ctx) + ).rejects.toThrow(/symbolic link/) + + expect(ctx.rclone.copy).not.toHaveBeenCalled() + expect(ctx.rclone.sync).not.toHaveBeenCalled() +}) + +test('an empty list does nothing and failed delivery remains a failure', async () => { + await cdnCopyFilesStep.run({ filePaths: [] }, ctx) + expect(ctx.rclone.copy).not.toHaveBeenCalled() + const error = new Error('CDN unavailable') + vi.mocked(ctx.rclone.copy).mockRejectedValueOnce(error) + + await expect(cdnCopyFilesStep.run({ filePaths: ['cover.png'] }, ctx)).rejects.toBe(error) + + expect(ctx.rclone.sync).not.toHaveBeenCalled() +}) diff --git a/packages/web/worker/src/steps/cdn-copy-files.ts b/packages/web/worker/src/steps/cdn-copy-files.ts new file mode 100644 index 00000000..1fe6119c --- /dev/null +++ b/packages/web/worker/src/steps/cdn-copy-files.ts @@ -0,0 +1,46 @@ +import { realpath, stat } from 'node:fs/promises' +import path from 'node:path' +import { resolveGeneratedAssetFile } from '../assets/files.js' +import { completed, skipped, type Step, type StepResult } from '../core/step.js' +import { getRcloneRemoteOptions } from '../services/rclone.js' + +export const cdnCopyFilesStep: Step<{ filePaths: string[] }, void> = { + name: 'cdn.copy-files', + async run({ filePaths }, ctx): Promise> { + const { config, logger, rclone } = ctx + const options = getRcloneRemoteOptions(config.sync, 'cdn') + + if (!options?.remote || !options.remotePath) { + logger.info('cdn.copy-files skipped: sync.cdn.remote or sync.cdn.path not configured') + return skipped('sync.cdn.remote or sync.cdn.path not configured') + } + if (!options.configPath) { + logger.info('cdn.copy-files skipped: sync.config not configured') + return skipped('sync.config not configured') + } + + const uniquePaths = [...new Set(filePaths)] + if (uniquePaths.length === 0) return completed(undefined) + + const root = await realpath(config.paths.assets) + const files: { localPath: string; remotePath: string }[] = [] + // Validate the entire manifest before transferring anything. The resolver rejects + // symlink components as well as noncanonical/escaping relative paths. + for (const filePath of uniquePaths) { + const localPath = await resolveGeneratedAssetFile(root, filePath) + if (!(await stat(localPath)).isFile()) { + throw new Error(`Generated asset is not a file: ${filePath}`) + } + files.push({ + localPath, + remotePath: path.posix.join(options.remotePath, path.posix.dirname(filePath)), + }) + } + + for (const file of files) { + await rclone.copy({ ...options, ...file }) + } + logger.info('cdn.copy-files complete', { files: files.length }) + return completed(undefined) + }, +} -- 2.54.0 From 73615eb59eeb5503bd4dbc058df87e9bd657043e Mon Sep 17 00:00:00 2001 From: eleith Date: Mon, 9 Nov 2026 16:37:33 -0800 Subject: [PATCH 2/2] [web] share filename-based publishing --- .changeset/single-piece-publication.md | 6 + .../(auth)/api/admin/publish/+server.ts | 104 ++- .../api/admin/publish/admission.test.ts | 25 +- .../(auth)/api/admin/publish/single.test.ts | 71 ++ docs/web/README.md | 8 +- packages/web/db/src/types.ts | 6 +- packages/web/jobs/src/index.ts | 1 + packages/web/jobs/src/specs.ts | 11 +- packages/web/jobs/src/types.ts | 5 +- .../worker/src/services/publication.test.ts | 35 +- packages/web/worker/src/services/publish.ts | 158 ++++ .../worker/src/steps/assets-generate.test.ts | 414 +++------ .../web/worker/src/steps/assets-generate.ts | 100 +-- .../web/worker/src/steps/publish-prepare.ts | 16 +- .../worker/src/steps/web-sync/index.test.ts | 144 ++- .../web/worker/src/steps/web-sync/index.ts | 244 +++--- .../web/worker/src/transforms/runner.test.ts | 236 +---- packages/web/worker/src/transforms/runner.ts | 25 - .../src/workflows/publish-recovery.test.ts | 715 --------------- .../web/worker/src/workflows/publish.test.ts | 817 +++++++++++------- packages/web/worker/src/workflows/publish.ts | 73 +- 21 files changed, 1305 insertions(+), 1909 deletions(-) create mode 100644 .changeset/single-piece-publication.md create mode 100644 apps/web/src/routes/(auth)/api/admin/publish/single.test.ts create mode 100644 packages/web/worker/src/services/publish.ts delete mode 100644 packages/web/worker/src/workflows/publish-recovery.test.ts diff --git a/.changeset/single-piece-publication.md b/.changeset/single-piece-publication.md new file mode 100644 index 00000000..85cc0600 --- /dev/null +++ b/.changeset/single-piece-publication.md @@ -0,0 +1,6 @@ +--- +"@luzzle/web": patch +"@luzzle/web.worker": patch +--- + +Share one filename-based publishing pipeline for individual pieces and archive selections. Read saved Markdown when the worker runs. Publish one piece through the existing endpoint without publishing or pruning unrelated pieces, copying only its generated files. 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 49876492..4e1fad2a 100644 --- a/apps/web/src/routes/(auth)/api/admin/publish/+server.ts +++ b/apps/web/src/routes/(auth)/api/admin/publish/+server.ts @@ -1,4 +1,6 @@ import type { RequestHandler } from './$types' +import { config } from '#lib/server/config.js' +import { getPieces } from '#lib/server/pieces.js' import { getOpenWorkflow, getOpenWorkflowBackend } from '#lib/server/workflow/index.js' import { findInFlightPublishRun, @@ -6,49 +8,83 @@ import { withPublishAdmission } from '#lib/server/workflow/publish.js' import { publishSpec } from '@luzzle/web.jobs/specs' -import type { PublishPayload } from '@luzzle/web.jobs' +import type { PublishPayload, PublishPiecePayload } from '@luzzle/web.jobs' + +function isCanonicalMarkdownPath(value: unknown): value is string { + if (typeof value !== 'string') return false + if (!value || /[\\\p{Control}]/u.test(value) || /^[A-Za-z]:/.test(value)) return false + if (!value.endsWith('.md')) return false + return value.split('/').every((part) => part !== '' && part !== '.' && part !== '..') +} + +interface PublishRequestBody { + filePath?: unknown + filePaths?: unknown + auditRunId?: unknown + bisync?: unknown +} + +function validatePiecePath(value: unknown): string | Response { + if (!isCanonicalMarkdownPath(value)) { + return Response.json({ message: 'Expected a canonical Markdown filePath' }, { status: 400 }) + } + const { type, format } = getPieces().parseFilename(value) + if (!type || format !== '.md' || !config.pieces.some((piece) => piece.type === type)) { + return Response.json({ message: 'Unsupported piece type' }, { status: 400 }) + } + return value +} + +async function prepareArchivePublish( + body: PublishRequestBody | null +): Promise { + const auditRunId = body?.auditRunId + const selectionProvided = Object.hasOwn(body ?? {}, 'filePaths') + let filePaths: string[] | undefined + if (selectionProvided) { + const requestedPaths = body?.filePaths + if ( + !Array.isArray(requestedPaths) || + !requestedPaths.every((path) => typeof path === 'string') + ) { + return Response.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 Response.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 + return input +} export const POST: RequestHandler = async ({ request }) => { try { - const body = await request.json().catch(() => ({})) + const body: PublishRequestBody | null = await request.json().catch(() => ({})) + let filePath: string | undefined + if (Object.hasOwn(body ?? {}, 'filePath')) { + const validatedPath = validatePiecePath(body?.filePath) + if (validatedPath instanceof Response) return validatedPath + filePath = validatedPath + } return await withPublishAdmission(async () => { const inFlight = await findInFlightPublishRun(getOpenWorkflowBackend()) if (inFlight) { return Response.json({ jobId: inFlight.id }, { status: 409 }) } - const auditRunId = body?.auditRunId - const selectionProvided = Object.hasOwn(body ?? {}, 'filePaths') - let filePaths: string[] | undefined - if (selectionProvided) { - const requestedPaths = body.filePaths - if (!Array.isArray(requestedPaths)) { - return Response.json( - { message: 'filePaths must be an array of strings' }, - { status: 400 } - ) - } - if (!requestedPaths.every((filePath: unknown) => typeof filePath === 'string')) { - return Response.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 Response.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 input: PublishPayload | PublishPiecePayload | Response = + filePath !== undefined ? { filePath } : await prepareArchivePublish(body) + if (input instanceof Response) return input const openWorkflow = getOpenWorkflow() const handle = await openWorkflow.runWorkflow(publishSpec, input) 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 ef352b93..edebfb96 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 @@ -28,7 +28,7 @@ const auditDiff = { } } -function event(body: unknown = { filePath: 'books/one.books.md' }) { +function event(body: unknown = {}) { return { request: new Request('http://localhost/api/admin/publish', { method: 'POST', @@ -37,6 +37,10 @@ function event(body: unknown = { filePath: 'books/one.books.md' }) { } as Parameters[0] & Parameters[0] & Parameters[0] } +function routeEvent(route: typeof publish | typeof audit | typeof unpublish) { + return event(route === unpublish ? { filePath: 'books/one.books.md' } : {}) +} + describe('publication admission', () => { let queue: WorkflowQueue @@ -89,7 +93,7 @@ describe('publication admission', () => { return enqueue(spec, input) }) - const first = Promise.resolve(firstRoute(event())) + const first = Promise.resolve(firstRoute(routeEvent(firstRoute))) try { await Promise.race([ started, @@ -97,7 +101,7 @@ describe('publication admission', () => { throw new Error('First request finished before enqueue') }) ]) - const other = otherRoute(event()) + const other = otherRoute(routeEvent(otherRoute)) release() const [accepted, rejected] = await Promise.all([first, other]) expect(accepted.status).toBe(200) @@ -124,7 +128,7 @@ describe('publication admission', () => { }) const request = { json: vi.fn(() => body) } as unknown as Request const slow = Promise.resolve(slowRoute({ ...event(), request })) - const other = Promise.resolve(otherRoute(event())) + const other = Promise.resolve(otherRoute(routeEvent(otherRoute))) try { await expect.poll(() => runWorkflow.mock.calls.length).toBe(1) expect((await other).status).toBe(200) @@ -198,7 +202,7 @@ describe('publication admission', () => { await claimNextRun(queue.backend) vi.setSystemTime(new Date('2026-06-20T01:00:00Z')) await createCompletedRun(queue.backend, 'Publish') - const response = await route(event()) + const response = await route(routeEvent(route)) expect(response.status).toBe(409) expect(await response.json()).toEqual({ jobId: active.id }) expect(runWorkflow).not.toHaveBeenCalled() @@ -459,6 +463,17 @@ describe('publication admission', () => { bisync: true }) await completeNextRun(queue.backend) + const malformedPublish = new Request('http://localhost/api/admin/publish', { + method: 'POST', + body: '{' + }) + expect( + (await publish({ request: malformedPublish } as Parameters[0])).status + ).toBe(200) + expect(runWorkflow).toHaveBeenLastCalledWith(expect.objectContaining({ name: 'Publish' }), { + bisync: false + }) + await completeNextRun(queue.backend) const request = new Request('http://localhost/api/admin/publish/audit', { method: 'POST', body: '{' diff --git a/apps/web/src/routes/(auth)/api/admin/publish/single.test.ts b/apps/web/src/routes/(auth)/api/admin/publish/single.test.ts new file mode 100644 index 00000000..a2dfb0ac --- /dev/null +++ b/apps/web/src/routes/(auth)/api/admin/publish/single.test.ts @@ -0,0 +1,71 @@ +import { afterEach, beforeEach, expect, test, vi } from 'vitest' +import { Pieces } from '@luzzle/core' +import { getStorage } from '#lib/server/storage.js' +import { getOpenWorkflowBackend } from '#lib/server/workflow/index.js' +import { + createWorkflowQueue, + closeWorkflowQueue, + createPendingRun, + type WorkflowQueue +} from '#lib/server/workflow/publish.fixture.js' +import { POST } from './+server.js' + +const { runWorkflow } = vi.hoisted(() => ({ runWorkflow: vi.fn() })) +vi.mock('#lib/server/config.js', () => ({ config: { pieces: [{ type: 'books' }] } })) +vi.mock('#lib/server/pieces.js', () => ({ + getPieces: () => ({ parseFilename: Pieces.prototype.parseFilename }) +})) +vi.mock('#lib/server/storage.js', () => ({ getStorage: vi.fn() })) +vi.mock('#lib/server/workflow/index.js', () => ({ + getOpenWorkflow: () => ({ runWorkflow }), + getOpenWorkflowBackend: vi.fn() +})) + +const filePath = '書 shelves/été?draft=1&tag=#%.books.md' +let queue: WorkflowQueue + +function event(body: unknown = { filePath }) { + return { + request: new Request('http://localhost/api/admin/publish', { + method: 'POST', + body: JSON.stringify(body) + }) + } as Parameters[0] +} + +beforeEach(() => { + vi.resetAllMocks() + queue = createWorkflowQueue() + vi.mocked(getOpenWorkflowBackend).mockReturnValue(queue.backend) + runWorkflow.mockImplementation(async (spec, input) => ({ + workflowRun: await createPendingRun(queue.backend, spec.name, input) + })) +}) + +afterEach(async () => { + await closeWorkflowQueue(queue) +}) + +test('queues only the canonical filename without accessing source storage', async () => { + const response = await POST(event()) + expect(response.status).toBe(200) + const { jobId } = await response.json() + expect(runWorkflow).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ name: 'Publish' }), + { filePath } + ) + expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({ + filePath + }) + expect(getStorage).not.toHaveBeenCalled() +}) + +test.each([{ filePath: null }, { filePath: `../${filePath}` }, { filePath: 'one.unsupported.md' }])( + 'rejects invalid own filePath without falling back to bulk: $filePath', + async (body) => { + expect((await POST(event({ ...body, bisync: true }))).status).toBe(400) + expect(runWorkflow).not.toHaveBeenCalled() + expect(getStorage).not.toHaveBeenCalled() + expect((await queue.backend.countWorkflowRuns()).pending).toBe(0) + } +) diff --git a/docs/web/README.md b/docs/web/README.md index e2669834..2c967b58 100644 --- a/docs/web/README.md +++ b/docs/web/README.md @@ -18,11 +18,11 @@ pieces. Save edits, then click **Saved** to start a preview of that saved file. Use `/admin/publish` to check changes and publish. Saving is not publishing. Check lists all pending changes. Choose which added or updated pieces to publish. -Publishing uses saved files, so edits saved after Check may be included. -Unselected edits stay pending. +Publishing reads saved files when the worker runs, so edits saved after Check or +while the job is waiting may be included. Unselected edits stay pending. -**Every publish also applies all pending removals**, regardless of selection. -They are listed separately; you can apply removals without selecting any pieces. +**Publishing from this screen also applies all pending removals**, regardless of +selection. They are listed separately; you can apply removals without selecting any pieces. Schema changes also apply globally. A renamed file looks like a removal and an addition: the old public version can disappear before you select its replacement. Check alone does not change the public site. diff --git a/packages/web/db/src/types.ts b/packages/web/db/src/types.ts index f3af631e..e5a9efff 100644 --- a/packages/web/db/src/types.ts +++ b/packages/web/db/src/types.ts @@ -11,11 +11,11 @@ export interface WebPieces { note?: string date_updated?: number date_added: number - date_consumed?: number + date_consumed?: number | null type: string json_metadata: string - summary?: string - keywords?: string + summary?: string | null + keywords?: string | null } export interface WebPiecesAsset { diff --git a/packages/web/jobs/src/index.ts b/packages/web/jobs/src/index.ts index 8ad37030..fc3f88f5 100644 --- a/packages/web/jobs/src/index.ts +++ b/packages/web/jobs/src/index.ts @@ -18,6 +18,7 @@ export { } from './specs.js' export type { AssetRecord, + PublishPiecePayload, PublishPayload, PublishPieceFailure, PublishResult, diff --git a/packages/web/jobs/src/specs.ts b/packages/web/jobs/src/specs.ts index 1dc938ec..52c6eba0 100644 --- a/packages/web/jobs/src/specs.ts +++ b/packages/web/jobs/src/specs.ts @@ -6,6 +6,7 @@ import type { PreviewPayload, PreviewResult, PublishPayload, + PublishPiecePayload, PublishResult, PublishAuditPayload, UnpublishPayload, @@ -36,12 +37,10 @@ export const previewSpec: WorkflowSpec = defineWo name: 'Preview', }) -export const publishSpec: WorkflowSpec = defineWorkflowSpec< - PublishPayload, - PublishResult ->({ - name: 'Publish', -}) +export const publishSpec: WorkflowSpec = + defineWorkflowSpec({ + name: 'Publish', + }) export const publishAuditSpec: WorkflowSpec = defineWorkflowSpec< PublishAuditPayload, diff --git a/packages/web/jobs/src/types.ts b/packages/web/jobs/src/types.ts index 1ff1f87f..5ff89161 100644 --- a/packages/web/jobs/src/types.ts +++ b/packages/web/jobs/src/types.ts @@ -12,10 +12,13 @@ export interface AssetRecord { export interface PublishPayload { bisync?: boolean - /** Omitted publishes all pending pieces; an empty list selects no upserts. */ filePaths?: string[] } +export interface PublishPiecePayload { + filePath: string +} + export interface PublishPieceFailure { filePath: string message: string diff --git a/packages/web/worker/src/services/publication.test.ts b/packages/web/worker/src/services/publication.test.ts index 82ad07d0..9ec8ae63 100644 --- a/packages/web/worker/src/services/publication.test.ts +++ b/packages/web/worker/src/services/publication.test.ts @@ -64,12 +64,26 @@ async function indexPiece(filePath: string, contentHash?: string, type = 'books' } } +async function syncPaths(filePaths: string[]) { + const sources = await Promise.all( + filePaths.map(async (filePath) => ({ + item: await db + .selectFrom('pieces_items') + .selectAll() + .where('file_path', '=', filePath) + .executeTakeFirstOrThrow(), + contentHash: `hash:${filePath}`, + })) + ) + return webSyncStep.run(sources, ctx) +} + async function publishMetadata( filePath: string, contentHash: string | null, lastPublishedAt: number | null = null ) { - await webSyncStep.run({ filePaths: [filePath] }, ctx) + await syncPaths([filePath]) await db .updateTable('web_pieces') .set({ content_hash: contentHash, last_published_at: lastPublishedAt }) @@ -143,7 +157,7 @@ describe('publication', () => { await indexPiece('new.books.md', 'new-hash') await publishMetadata('existing.books.md', 'old-hash', 123) - await webSyncStep.run({ filePaths: ['existing.books.md', 'new.books.md'] }, ctx) + await syncPaths(['existing.books.md', 'new.books.md']) expect(await publicationState('existing.books.md')).toEqual({ content_hash: null, @@ -155,7 +169,7 @@ describe('publication', () => { }) }) - test('prepare captures pending hashes and includes recovery paths in the report', async () => { + test('prepare selects pending filenames and includes recovery paths in the report', async () => { await indexPiece('new.books.md', 'new-hash') await indexPiece('changed.books.md', 'current-hash') await publishMetadata('changed.books.md', 'old-hash') @@ -167,10 +181,7 @@ describe('publication', () => { expect(prepared).toEqual({ status: 'completed', value: { - pieces: [ - { filePath: 'changed.books.md', contentHash: 'current-hash' }, - { filePath: 'new.books.md', contentHash: 'new-hash' }, - ], + filePaths: ['changed.books.md', 'new.books.md'], summary: { schemas: summary.schemas, pieces: { added: ['new.books.md'], updated: ['changed.books.md'], pruned: [] }, @@ -216,9 +227,9 @@ describe('publication', () => { 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' }] : []), + filePaths: [ + ...(updated.length ? ['changed.books.md'] : []), + ...(added.length ? ['new.books.md'] : []), ], summary: { schemas: summary.schemas, @@ -268,7 +279,7 @@ describe('publication', () => { test('first completion and a new captured hash record the current time atomically', async () => { await indexPiece('new.books.md', 'indexed-hash') - await webSyncStep.run({ filePaths: ['new.books.md'] }, ctx) + await syncPaths(['new.books.md']) const input = { pieces: [{ filePath: 'new.books.md', contentHash: "captured'hash" }], failedPieces: [], @@ -391,7 +402,7 @@ describe('publication', () => { test('complete and prepare accept an empty archive', async () => { expect(await publishPrepareStep.run({ summary: emptyPiecesDiff() }, ctx)).toEqual({ status: 'completed', - value: { pieces: [], summary: emptyPiecesDiff() }, + value: { filePaths: [], summary: emptyPiecesDiff() }, }) expect(await publishCompleteStep.run({ pieces: [], failedPieces: [] }, ctx)).toEqual({ status: 'completed', diff --git a/packages/web/worker/src/services/publish.ts b/packages/web/worker/src/services/publish.ts new file mode 100644 index 00000000..8847edaa --- /dev/null +++ b/packages/web/worker/src/services/publish.ts @@ -0,0 +1,158 @@ +import { createHash } from 'node:crypto' +import path from 'node:path' +import { + Pieces, + StorageFileSystem, + extractFullMarkdown, + makePieceMarkdown, + makePieceItemInsertable, + type PieceFrontmatter, + type PiecesDiff, +} from '@luzzle/core' +import type { PublishPayload, PublishPiecePayload, PublishPieceFailure } from '@luzzle/web.jobs' +import { completed, type Step, type StepContext } from '../core/step.js' +import { runProgressPhase, type DurableStepApi } from '../core/run-progress-phase.js' +import type { JobProgress } from '../core/job-progress.js' +import { archiveSyncStep } from '../steps/archive-sync.js' +import { luzzleSyncStep } from '../steps/luzzle-sync.js' +import { publishPrepareStep } from '../steps/publish-prepare.js' +import { webSyncStep, pruneWebPiecesStep, type PublicationSource } from '../steps/web-sync/index.js' +import { assetsGenerateStep } from '../steps/assets-generate.js' +import { cdnSyncStep } from '../steps/cdn-sync.js' +import { cdnCopyFilesStep } from '../steps/cdn-copy-files.js' +import type { PublicationPiece } from './publication.js' +import { emptyPiecesDiff } from '../workflows/pieces-diff.js' + +export interface PublishExecution { + step: DurableStepApi + ctx: StepContext + jobId: string + progress: JobProgress +} + +export interface PublicationReport { + summary: PiecesDiff + pieces: PublicationPiece[] + failedPieces: PublishPieceFailure[] +} + +const publicationSourcesStep: Step = { + name: 'publish.sources', + async run(filePaths, { db, config }) { + const storage = new StorageFileSystem(config.storage.root) + const pieces = new Pieces(storage) + const sources: PublicationSource[] = [] + for (const filePath of filePaths) { + if (typeof filePath !== 'string' || !filePath) throw new Error('Invalid piece filename') + if (path.posix.isAbsolute(filePath) || filePath.includes('\\') || filePath.includes('\0')) { + throw new Error('Invalid piece filename') + } + if (filePath.split('/').some((part) => !part || part === '.' || part === '..')) { + throw new Error('Invalid piece filename') + } + const { type, format } = pieces.parseFilename(filePath) + if (!type || format !== '.md' || !config.pieces.some((piece) => piece.type === type)) { + throw new Error('Expected a configured Markdown piece') + } + const source = (await storage.readFile(filePath, 'text')) as string + const piece = await pieces.getPiece(type) + const data = await extractFullMarkdown(source) + const markdown = makePieceMarkdown( + filePath, + type, + data.markdown, + data.frontmatter as PieceFrontmatter + ) + const validation = piece.validate(markdown) + if (!validation.isValid) { + throw new Error(`Invalid piece source: ${validation.errors.join('\n')}`) + } + const identity = await db + .selectFrom('pieces_items') + .select(['id', 'date_added']) + .where('file_path', '=', filePath) + .executeTakeFirst() + const item = makePieceItemInsertable(type, markdown, piece.schema) + const id = identity?.id ?? item.id + if (!id) throw new Error('Piece has no identity') + sources.push({ + item: { + id, + file_path: filePath, + type, + date_added: identity?.date_added ?? Date.now(), + date_updated: Date.now(), + note_markdown: item.note_markdown, + frontmatter_json: item.frontmatter_json, + assets_json_array: item.assets_json_array, + }, + contentHash: createHash('md5').update(source).digest('hex'), + }) + } + return completed(sources) + }, +} + +export async function publishPieces( + filePaths: string[], + execution: PublishExecution, + copyGeneratedFiles = true +): Promise { + const { step, ctx, jobId, progress } = execution + const sources = await runProgressPhase( + step, + ctx, + jobId, + progress, + publicationSourcesStep, + filePaths + ) + if (!sources) throw new Error('Publication sources could not be prepared') + const prepared = await runProgressPhase(step, ctx, jobId, progress, webSyncStep, sources) + if (!prepared) throw new Error('Publication preparation did not complete') + const generated = await runProgressPhase(step, ctx, jobId, progress, assetsGenerateStep, prepared) + if (!generated) throw new Error('Asset generation did not complete') + if (copyGeneratedFiles) { + await runProgressPhase(step, ctx, jobId, progress, cdnCopyFilesStep, { + filePaths: generated.filePaths, + }) + } + const summary = emptyPiecesDiff() + for (const piece of prepared) { + const paths = piece.wasPublished ? summary.pieces.updated : summary.pieces.added + paths.push(piece.webPiece.file_path) + } + return { + summary, + pieces: prepared.map((piece) => ({ + filePath: piece.webPiece.file_path, + contentHash: piece.contentHash, + })), + failedPieces: generated.failedPieces, + } +} + +export async function publishPiece( + input: PublishPiecePayload, + execution: PublishExecution +): Promise { + return publishPieces([input.filePath], execution) +} + +export async function publishArchive( + input: PublishPayload, + execution: PublishExecution +): Promise { + const { step, ctx, jobId, progress } = execution + if (input.bisync) await runProgressPhase(step, ctx, jobId, progress, archiveSyncStep, undefined) + const changes = await runProgressPhase(step, ctx, jobId, progress, luzzleSyncStep, undefined) + const plan = await runProgressPhase(step, ctx, jobId, progress, publishPrepareStep, { + summary: changes ?? emptyPiecesDiff(), + filePaths: input.filePaths, + }) + if (!plan) throw new Error('Publication selection did not complete') + await runProgressPhase(step, ctx, jobId, progress, pruneWebPiecesStep, undefined) + const report = await publishPieces(plan.filePaths, execution, false) + await runProgressPhase(step, ctx, jobId, progress, cdnSyncStep, undefined) + return { ...report, summary: plan.summary } +} diff --git a/packages/web/worker/src/steps/assets-generate.test.ts b/packages/web/worker/src/steps/assets-generate.test.ts index 48a2bb42..b063b7f1 100644 --- a/packages/web/worker/src/steps/assets-generate.test.ts +++ b/packages/web/worker/src/steps/assets-generate.test.ts @@ -1,325 +1,149 @@ -import { describe, test, expect, vi, beforeEach, afterEach } from 'vitest' -import type { Kysely } from 'kysely' -import { sql } from 'kysely' -import type * as luzzleCore from '@luzzle/core' -import type { LuzzleTables } from '@luzzle/core' -import { Pieces, StorageFileSystem } from '@luzzle/core' -import type { Config } from '@luzzle/web.config' -import type { WebDatabase } from '../services/db.js' +import { describe, test, expect, vi, beforeEach } from 'vitest' import type { WorkerContext } from '../services/context.js' -import { setupDatabase, teardownDatabase } from '../../test/db.js' -import { runTransformsForPiece } from '../transforms/runner.js' +import { makeLogger } from '../../test/logger.js' +import { + produceTransformsForPiece, + persistTransforms, + type ProducedTransform, +} from '../transforms/runner.js' import { cleanupAllTransforms } from '../transforms/index.js' -import { buildAssetMaps } from '../transforms/utils/assets.js' +import type { PreparedPublication } from './web-sync/index.js' import { assetsGenerateStep } from './assets-generate.js' -vi.mock('@luzzle/core', async (importOriginal) => { - const actual = await importOriginal() - return { - ...actual, - Pieces: vi.fn(), - StorageFileSystem: vi.fn(), - } -}) vi.mock('../transforms/runner.js') vi.mock('../transforms/index.js') -vi.mock('../transforms/utils/assets.js') -const mocks = { - runTransformsForPiece: vi.mocked(runTransformsForPiece), - cleanupAllTransforms: vi.mocked(cleanupAllTransforms), - buildAssetMaps: vi.mocked(buildAssetMaps), - Pieces: vi.mocked(Pieces), - StorageFileSystem: vi.mocked(StorageFileSystem), -} +const produce = vi.mocked(produceTransformsForPiece) +const persist = vi.mocked(persistTransforms) +const cleanup = vi.mocked(cleanupAllTransforms) -type FullDb = Kysely - -function makeConfig(): Config { +function prepared(name: string): PreparedPublication { return { - storage: { root: '/app/archive' }, - paths: { database: 'db.sqlite', assets: '/app/assets/pieces', config: '/app/config.yaml' }, - assets: { salt: 'salt' }, - pieces: [], - } as unknown as Config -} - -let db: FullDb -let ctx: WorkerContext - -beforeEach(async () => { - db = (await setupDatabase()).withTables() as FullDb - ctx = { - config: makeConfig(), - logger: { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - stdout: vi.fn(), - stderr: vi.fn(), + webPiece: { + id: name, + key: `${name}-key`, + title: `Captured ${name}`, + slug: name, + file_path: `books/${name}.md`, + type: 'books', + date_added: 1, + note: 'Captured note, not the current index', + json_metadata: JSON.stringify({ cover: `${name}-cover-key` }), }, + contentHash: `${name}-hash`, + assetKeyToPath: [[`${name}-cover-key`, `books/${name}/captured-cover.png`]], + wasPublished: true, + } +} + +function generated(assetPath: string): ProducedTransform[] { + return [ + { + name: 'image', + records: [ + { transformation: 'image.original', asset_path: assetPath, mime_type: 'image/png' }, + ], + }, + ] +} + +let ctx: WorkerContext +let selectFrom: ReturnType + +beforeEach(() => { + vi.resetAllMocks() + selectFrom = vi.fn(() => { + throw new Error('Publication metadata must not be reread') + }) + ctx = { + config: { + storage: { root: process.cwd() }, + paths: { assets: '/generated/assets' }, + } as WorkerContext['config'], + logger: makeLogger(), rclone: {} as WorkerContext['rclone'], - db: db as unknown as WorkerContext['db'], + db: { selectFrom } as unknown as WorkerContext['db'], } - mocks.runTransformsForPiece.mockResolvedValue(undefined) - mocks.cleanupAllTransforms.mockResolvedValue(undefined) - mocks.buildAssetMaps.mockReturnValue({ pathToKey: new Map(), keyToPath: new Map() }) - const mockPieces = {} as Pieces - const mockStorage = {} as StorageFileSystem - mocks.Pieces.mockImplementation( - class { - constructor() { - return mockPieces - } - } as typeof Pieces - ) - mocks.StorageFileSystem.mockImplementation( - class { - constructor() { - return mockStorage - } - } as typeof StorageFileSystem - ) + persist.mockResolvedValue(undefined) + cleanup.mockResolvedValue(undefined) }) -afterEach(async () => { - await teardownDatabase(db) - vi.clearAllMocks() -}) - -async function seedPiece(over: Partial> = {}) { - const row = { - id: 'item-1', - key: 'k1', - title: 'T', - slug: String(over.id ?? 'great'), - file_path: 'books/great.md', - date_added: 1, - type: 'books', - json_metadata: '{}', - ...over, - } - await db.insertInto('web_pieces').values(row).execute() - return row -} - -async function seedItem(over: Partial> = {}) { - const row = { - id: 'item-1', - file_path: 'books/great.md', - type: 'books', - date_added: 1, - note_markdown: '', - frontmatter_json: '{}', - assets_json_array: JSON.stringify(['books/great/cover.png']), - ...over, - } - await ( - db.insertInto('pieces_items') as unknown as { - values: (v: typeof row) => { execute: () => Promise } - } - ) - .values(row) - .execute() - return row -} - describe('assetsGenerateStep', () => { - test('runs transforms for each filePath given', async () => { - await seedPiece({ id: 'a', file_path: 'books/a.md' }) - await seedPiece({ id: 'b', file_path: 'books/b.md' }) - - const result = await assetsGenerateStep.run({ filePaths: ['books/a.md', 'books/b.md'] }, ctx) - - expect(result).toEqual({ status: 'completed', value: { failedPieces: [] } }) - expect(ctx.logger.warn).not.toHaveBeenCalled() - expect(mocks.runTransformsForPiece).toHaveBeenCalledTimes(2) - const paths = mocks.runTransformsForPiece.mock.calls.map((c) => c[1].file_path).sort() - expect(paths).toEqual(['books/a.md', 'books/b.md']) - }) - - test('skips web_pieces rows not in filePaths', async () => { - await seedPiece({ id: 'a', file_path: 'books/a.md' }) - await seedPiece({ id: 'b', file_path: 'books/b.md' }) - - await assetsGenerateStep.run({ filePaths: ['books/a.md'] }, ctx) - - expect(mocks.runTransformsForPiece).toHaveBeenCalledTimes(1) - expect(mocks.runTransformsForPiece.mock.calls[0][1].file_path).toBe('books/a.md') - }) - - test('passes config.paths.assets as outDir', async () => { - await seedPiece() - await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - expect(mocks.runTransformsForPiece.mock.calls[0][3]).toBe('/app/assets/pieces') - }) - - test('builds keyToPath from pieces_items.assets_json_array', async () => { - await seedPiece() - await seedItem() - - await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - - expect(mocks.buildAssetMaps).toHaveBeenCalledWith( - JSON.stringify(['books/great/cover.png']), - 'salt' + test('uses supplied publication metadata and asset map, returning only unique generated files', async () => { + const publication = prepared('captured') + const produced = generated('captured-cover.png') + produced[0].records.push( + { transformation: 'image.copy', asset_path: 'captured-cover.png', mime_type: 'image/png' }, + { + transformation: 'image.embedded', + asset_path: 'embedded.png', + mime_type: 'image/png', + is_embedded: 1, + }, + { transformation: 'image.metadata', mime_type: 'application/json', content: '{}' } ) + produce.mockResolvedValueOnce(produced) + + const result = await assetsGenerateStep.run([publication], ctx) + + expect(produce).toHaveBeenCalledExactlyOnceWith( + publication.webPiece, + ctx.config, + ctx.config.paths.assets, + expect.anything(), + {}, + new Map(publication.assetKeyToPath), + ctx.logger + ) + expect(selectFrom).not.toHaveBeenCalled() + expect(persist).toHaveBeenCalledExactlyOnceWith( + ctx.db, + publication.webPiece, + ctx.config, + produced + ) + expect(result).toEqual({ + status: 'completed', + value: { filePaths: ['captured-cover.png'], failedPieces: [] }, + }) + expect(cleanup).toHaveBeenCalledOnce() }) - test('passes empty asset map when pieces_items row is missing', async () => { - await seedPiece() - await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - expect(mocks.buildAssetMaps).toHaveBeenCalledWith(undefined, 'salt') - }) + test('continues after generation or persistence failures and excludes their files', async () => { + const generationFailure = prepared('generation-failure') + const persistenceFailure = prepared('persistence-failure') + const healthy = prepared('healthy') + produce + .mockRejectedValueOnce(new Error('generation failed')) + .mockResolvedValueOnce(generated('failed.png')) + .mockResolvedValueOnce(generated('healthy.png')) + persist.mockRejectedValueOnce(new Error('persistence failed')) - test('calls cleanupAllTransforms after iteration', async () => { - await seedPiece() - await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - }) + const result = await assetsGenerateStep.run( + [generationFailure, persistenceFailure, healthy], + ctx + ) - test('logs start and complete', async () => { - await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - expect(ctx.logger.info).toHaveBeenCalledWith('assets.generate starting', { count: 1 }) - expect(ctx.logger.info).toHaveBeenCalledWith('assets.generate complete') - }) - - test('short-circuits and skips cleanup when filePaths is empty', async () => { - await seedPiece() - const result = await assetsGenerateStep.run({ filePaths: [] }, ctx) - expect(result).toEqual({ status: 'completed', value: { failedPieces: [] } }) - expect(mocks.runTransformsForPiece).not.toHaveBeenCalled() - expect(mocks.cleanupAllTransforms).not.toHaveBeenCalled() - }) - - test('continues after the first piece rejects', async () => { - await seedPiece({ id: 'a', file_path: 'books/a.md' }) - await seedPiece({ id: 'b', file_path: 'books/b.md' }) - mocks.runTransformsForPiece.mockRejectedValueOnce(new Error('piece processing failed')) - - const result = await assetsGenerateStep.run({ filePaths: ['books/a.md', 'books/b.md'] }, ctx) - - expect(mocks.runTransformsForPiece.mock.calls.map((call) => call[1].file_path)).toEqual([ - 'books/a.md', - 'books/b.md', + expect(produce.mock.calls.map(([piece]) => piece.file_path)).toEqual([ + generationFailure.webPiece.file_path, + persistenceFailure.webPiece.file_path, + healthy.webPiece.file_path, + ]) + expect(persist.mock.calls.map(([, piece]) => piece.file_path)).toEqual([ + persistenceFailure.webPiece.file_path, + healthy.webPiece.file_path, ]) - expect(result).toEqual({ - status: 'completed', - value: { failedPieces: [{ filePath: 'books/a.md', message: 'piece processing failed' }] }, - }) - expect(ctx.logger.error).toHaveBeenCalledWith('assets.generate failed for books/a.md', { - message: 'piece processing failed', - }) - expect(ctx.logger.warn).toHaveBeenCalledWith('assets.generate complete with failures', { - failedCount: 1, - count: 2, - }) - expect(ctx.logger.info).not.toHaveBeenCalledWith('assets.generate complete') - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - expect(mocks.cleanupAllTransforms.mock.invocationCallOrder[0]).toBeGreaterThan( - mocks.runTransformsForPiece.mock.invocationCallOrder[1] - ) - }) - - test('continues after per-piece asset preparation fails', async () => { - await seedPiece({ id: 'a', file_path: 'books/a.md' }) - await seedPiece({ id: 'b', file_path: 'books/b.md' }) - mocks.buildAssetMaps.mockImplementationOnce(() => { - throw new Error('asset map failed') - }) - - const result = await assetsGenerateStep.run({ filePaths: ['books/a.md', 'books/b.md'] }, ctx) - - expect(result).toEqual({ - status: 'completed', - value: { failedPieces: [{ filePath: 'books/a.md', message: 'asset map failed' }] }, - }) - expect(mocks.buildAssetMaps).toHaveBeenCalledTimes(2) - expect(mocks.runTransformsForPiece).toHaveBeenCalledOnce() - expect(mocks.runTransformsForPiece.mock.calls[0][1].file_path).toBe('books/b.md') - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - }) - - test.each([ - { thrown: 'string failure', message: 'string failure' }, - { thrown: null, message: 'null' }, - { thrown: undefined, message: 'undefined' }, - { thrown: { reason: 'failure' }, message: '[object Object]' }, - ])('normalizes non-Error rejection $message', async ({ thrown, message }) => { - await seedPiece() - mocks.runTransformsForPiece.mockRejectedValueOnce(thrown) - - const result = await assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx) - - expect(result).toEqual({ - status: 'completed', - value: { failedPieces: [{ filePath: 'books/great.md', message }] }, - }) - expect(JSON.parse(JSON.stringify(result))).toEqual(result) - expect(ctx.logger.error).toHaveBeenCalledWith('assets.generate failed for books/great.md', { - message, - }) - expect(ctx.logger.warn).toHaveBeenCalledWith('assets.generate complete with failures', { - failedCount: 1, - count: 1, - }) - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - }) - - test('collects every failed piece', async () => { - await seedPiece({ id: 'a', file_path: 'books/a.md' }) - await seedPiece({ id: 'b', file_path: 'books/b.md' }) - mocks.runTransformsForPiece - .mockRejectedValueOnce(new Error('first failure')) - .mockRejectedValueOnce(new Error('second failure')) - - const result = await assetsGenerateStep.run({ filePaths: ['books/a.md', 'books/b.md'] }, ctx) - expect(result).toEqual({ status: 'completed', value: { + filePaths: ['healthy.png'], failedPieces: [ - { filePath: 'books/a.md', message: 'first failure' }, - { filePath: 'books/b.md', message: 'second failure' }, + { filePath: generationFailure.webPiece.file_path, message: 'generation failed' }, + { filePath: persistenceFailure.webPiece.file_path, message: 'persistence failed' }, ], }, }) - expect(ctx.logger.warn).toHaveBeenCalledWith('assets.generate complete with failures', { - failedCount: 2, - count: 2, - }) + expect(cleanup).toHaveBeenCalledOnce() + expect(cleanup.mock.invocationCallOrder[0]).toBeGreaterThan(persist.mock.invocationCallOrder[1]) }) - - test('propagates global selection failures and still cleans up', async () => { - await sql`DROP TABLE web_pieces`.execute(db) - - await expect(assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx)).rejects.toThrow( - 'no such table: web_pieces' - ) - - expect(mocks.runTransformsForPiece).not.toHaveBeenCalled() - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - expect(ctx.logger.warn).not.toHaveBeenCalled() - expect(ctx.logger.info).not.toHaveBeenCalledWith('assets.generate complete') - }) - - test.each([false, true])( - 'propagates cleanup failures (piece failed: %s)', - async (pieceFailed) => { - await seedPiece() - if (pieceFailed) mocks.runTransformsForPiece.mockRejectedValueOnce(new Error('piece failed')) - const error = new Error('cleanup failed') - mocks.cleanupAllTransforms.mockRejectedValueOnce(error) - - await expect(assetsGenerateStep.run({ filePaths: ['books/great.md'] }, ctx)).rejects.toBe( - error - ) - - expect(mocks.runTransformsForPiece).toHaveBeenCalledOnce() - expect(mocks.cleanupAllTransforms).toHaveBeenCalledOnce() - expect(ctx.logger.warn).not.toHaveBeenCalled() - expect(ctx.logger.info).not.toHaveBeenCalledWith('assets.generate complete') - } - ) }) diff --git a/packages/web/worker/src/steps/assets-generate.ts b/packages/web/worker/src/steps/assets-generate.ts index 849b4030..6d86270b 100644 --- a/packages/web/worker/src/steps/assets-generate.ts +++ b/packages/web/worker/src/steps/assets-generate.ts @@ -1,73 +1,59 @@ -import { Pieces, StorageFileSystem, type LuzzleTables } from '@luzzle/core' +import { Pieces, StorageFileSystem } from '@luzzle/core' +import type { PublishPieceFailure } from '@luzzle/web.jobs' import type { Kysely } from 'kysely' import { completed, type Step, type StepResult } from '../core/step.js' import type { WebDatabase } from '../services/db.js' -import { runTransformsForPiece } from '../transforms/runner.js' -import { buildAssetMaps } from '../transforms/utils/assets.js' +import { produceTransformsForPiece, persistTransforms } from '../transforms/runner.js' import { cleanupAllTransforms } from '../transforms/index.js' - -export interface AssetsGenerateInput { - filePaths: string[] -} +import type { PreparedPublication } from './web-sync/index.js' export interface AssetsGenerateResult { - failedPieces: Array<{ filePath: string; message: string }> + filePaths: string[] + failedPieces: PublishPieceFailure[] } -type FullDb = Kysely - -export const assetsGenerateStep: Step = { +export const assetsGenerateStep: Step = { name: 'assets.generate', - async run({ filePaths }, ctx): Promise> { + async run(prepared, ctx): Promise> { const { db, config, logger } = ctx - const fullDb = db as unknown as FullDb + const filePaths = new Set() + const failedPieces: PublishPieceFailure[] = [] - logger.info('assets.generate starting', { count: filePaths.length }) - const failedPieces: AssetsGenerateResult['failedPieces'] = [] + logger.info('assets.generate starting', { count: prepared.length }) - if (filePaths.length === 0) { - logger.info('assets.generate complete') - return completed({ failedPieces }) - } - - let pieceCount = 0 try { - const storage = new StorageFileSystem(config.storage.root) - const pieces = new Pieces(storage) - const outDir = config.paths.assets + if (prepared.length > 0) { + const pieces = new Pieces(new StorageFileSystem(config.storage.root)) - const webPieces = await fullDb - .selectFrom('web_pieces') - .selectAll() - .where('file_path', 'in', filePaths) - .execute() - pieceCount = webPieces.length + for (const { webPiece, assetKeyToPath } of prepared) { + try { + logger.info(`assets.generate running transforms: ${webPiece.file_path}`) + const produced = await produceTransformsForPiece( + webPiece, + config, + config.paths.assets, + pieces, + {}, + new Map(assetKeyToPath), + logger + ) + await persistTransforms( + db as unknown as Kysely, + webPiece, + config, + produced + ) - for (const webPiece of webPieces) { - try { - const item = await fullDb - .selectFrom('pieces_items') - .select('assets_json_array') - .where('file_path', '=', webPiece.file_path) - .executeTakeFirst() - - const { keyToPath } = buildAssetMaps(item?.assets_json_array, config.assets.salt) - - logger.info(`assets.generate running transforms: ${webPiece.file_path}`) - await runTransformsForPiece( - fullDb as unknown as Kysely, - webPiece, - config, - outDir, - pieces, - {}, - keyToPath, - logger - ) - } catch (error) { - const message = error instanceof Error ? error.message : String(error) - failedPieces.push({ filePath: webPiece.file_path, message }) - logger.error(`assets.generate failed for ${webPiece.file_path}`, { message }) + for (const { records } of produced) { + for (const record of records) { + if (record.asset_path && !record.is_embedded) filePaths.add(record.asset_path) + } + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + failedPieces.push({ filePath: webPiece.file_path, message }) + logger.error(`assets.generate failed for ${webPiece.file_path}`, { message }) + } } } } finally { @@ -77,11 +63,11 @@ export const assetsGenerateStep: Step if (failedPieces.length > 0) { logger.warn('assets.generate complete with failures', { failedCount: failedPieces.length, - count: pieceCount, + count: prepared.length, }) } else { logger.info('assets.generate complete') } - return completed({ failedPieces }) + return completed({ filePaths: [...filePaths], failedPieces }) }, } diff --git a/packages/web/worker/src/steps/publish-prepare.ts b/packages/web/worker/src/steps/publish-prepare.ts index 948ef18d..19b66991 100644 --- a/packages/web/worker/src/steps/publish-prepare.ts +++ b/packages/web/worker/src/steps/publish-prepare.ts @@ -1,10 +1,10 @@ import type { PiecesDiff } from '@luzzle/core' import { completed, type Step } from '../core/step.js' -import { getPendingPublication, type PublicationPiece } from '../services/publication.js' +import { getPendingPublication } from '../services/publication.js' import { includePendingPieces } from '../workflows/pieces-diff.js' export interface PublishPlan { - pieces: PublicationPiece[] + filePaths: string[] summary: PiecesDiff } @@ -17,12 +17,12 @@ export const publishPrepareStep: Step = { name: 'publish.prepare', async run({ summary: indexSummary, filePaths }, { db, config, logger }) { const pending = await getPendingPublication(db, config) - let pieces = pending.pieces + let selectedPaths = pending.pieces.map((piece) => piece.filePath) if (filePaths !== undefined) { - const selectedPaths = new Set(filePaths) - pieces = pending.pieces.filter((piece) => selectedPaths.has(piece.filePath)) + const requestedPaths = new Set(filePaths) + selectedPaths = selectedPaths.filter((filePath) => requestedPaths.has(filePath)) } - const targetPaths = new Set(pieces.map((piece) => piece.filePath)) + const targetPaths = new Set(selectedPaths) // Core sync is global; only authoritative pending targets count as upserts. const globalChanges: PiecesDiff = { @@ -36,7 +36,7 @@ export const publishPrepareStep: Step = { } const publicationSummary = includePendingPieces(globalChanges, selectedChanges) - logger.info('publish.prepare complete', { count: pieces.length }) - return completed({ pieces, summary: publicationSummary }) + logger.info('publish.prepare complete', { count: selectedPaths.length }) + return completed({ filePaths: selectedPaths, summary: publicationSummary }) }, } diff --git a/packages/web/worker/src/steps/web-sync/index.test.ts b/packages/web/worker/src/steps/web-sync/index.test.ts index 22dbd815..248b6ad4 100644 --- a/packages/web/worker/src/steps/web-sync/index.test.ts +++ b/packages/web/worker/src/steps/web-sync/index.test.ts @@ -63,18 +63,46 @@ async function seedItem(over: Partial> = {}) { assets_json_array: '[]', ...over, } - await (db.insertInto('pieces_items') as unknown as { - values: (v: typeof row) => { execute: () => Promise } - }) + await ( + db.insertInto('pieces_items') as unknown as { + values: (v: typeof row) => { execute: () => Promise } + } + ) .values(row) .execute() return row } +async function syncPaths(filePaths: string[]) { + const sources = await Promise.all( + filePaths.map(async (filePath) => ({ + item: await db + .selectFrom('pieces_items') + .selectAll() + .where('file_path', '=', filePath) + .executeTakeFirstOrThrow(), + contentHash: `hash:${filePath}`, + })) + ) + return webSyncStep.run(sources, ctx) +} + describe('webSyncStep', () => { - test('upserts a web_pieces row for each filePath given', async () => { + test('upserts supplied sources and returns prepared metadata with the captured hash', async () => { await seedItem() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) + const result = await syncPaths(['books/great.md']) + + expect(result).toMatchObject({ + status: 'completed', + value: [ + { + webPiece: { id: 'item-1', title: 'Great Book', slug: 'great', type: 'books' }, + contentHash: 'hash:books/great.md', + assetKeyToPath: [], + wasPublished: false, + }, + ], + }) const rows = await db.selectFrom('web_pieces').selectAll().execute() expect(rows).toHaveLength(1) @@ -84,92 +112,49 @@ describe('webSyncStep', () => { expect(rows[0].type).toBe('books') }) - test('inserts web_pieces_tags from frontmatter', async () => { + test('keeps the public slug while preparing and persisting changed content and configuration', async () => { await seedItem() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) - - const tags = await db.selectFrom('web_pieces_tags').selectAll().orderBy('tag').execute() - expect(tags.map((t) => t.tag)).toEqual(['classic', 'fiction']) - }) - - test('keeps slug stable across re-runs', async () => { - await seedItem() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) - const first = await db.selectFrom('web_pieces').select('slug').executeTakeFirst() + await syncPaths(['books/great.md']) + const first = await db.selectFrom('web_pieces').select(['slug', 'key']).executeTakeFirst() + ctx.config.assets.salt = 'new-salt' await db .updateTable('pieces_items') .set({ frontmatter_json: JSON.stringify({ title: 'Renamed', tags: [] }) }) .where('id', '=', 'item-1') .execute() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) + const result = await syncPaths(['books/great.md']) const after = await db.selectFrom('web_pieces').selectAll().executeTakeFirst() expect(after!.slug).toBe(first!.slug) expect(after!.title).toBe('Renamed') - }) - - test('replaces tags on re-sync', async () => { - await seedItem() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) - - await db - .updateTable('pieces_items') - .set({ frontmatter_json: JSON.stringify({ title: 'Great Book', tags: ['updated'] }) }) - .where('id', '=', 'item-1') - .execute() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) - - const tags = await db.selectFrom('web_pieces_tags').selectAll().execute() - expect(tags.map((t) => t.tag)).toEqual(['updated']) + expect(after!.key).not.toBe(first!.key) + expect(result).toMatchObject({ + status: 'completed', + value: [ + { webPiece: { slug: first!.slug, key: after!.key, title: 'Renamed' }, wasPublished: true }, + ], + }) }) test('generates unique slugs when filenames collide within type', async () => { await seedItem({ id: 'a', file_path: 'books/great.md' }) await seedItem({ id: 'b', file_path: 'archive/great.md' }) - await webSyncStep.run({ filePaths: ['books/great.md', 'archive/great.md'] }, ctx) + const result = await syncPaths(['books/great.md', 'archive/great.md']) const rows = await db.selectFrom('web_pieces').select(['id', 'slug']).orderBy('id').execute() expect(rows.map((r) => r.slug).sort()).toEqual(['great', 'great--1']) + expect(result).toMatchObject({ + status: 'completed', + value: [ + { webPiece: { id: 'a', slug: 'great' }, contentHash: 'hash:books/great.md' }, + { webPiece: { id: 'b', slug: 'great--1' }, contentHash: 'hash:archive/great.md' }, + ], + }) }) - test('prunes web_pieces rows whose file_path is no longer in pieces_items', async () => { - await seedItem() - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) - - await db.deleteFrom('pieces_items').where('id', '=', 'item-1').execute() - await webSyncStep.run({ filePaths: [] }, ctx) - - const rows = await db.selectFrom('web_pieces').selectAll().execute() - expect(rows).toHaveLength(0) - }) - - test('does not touch web_pieces rows whose file_path is not in filePaths', async () => { - await seedItem({ id: 'a', file_path: 'books/keep.md' }) - await seedItem({ id: 'b', file_path: 'books/touch.md' }) - await webSyncStep.run({ filePaths: ['books/keep.md', 'books/touch.md'] }, ctx) - - const originalTitle = ( - await db.selectFrom('web_pieces').select('title').where('id', '=', 'a').executeTakeFirst() - )?.title - - await db - .updateTable('pieces_items') - .set({ frontmatter_json: JSON.stringify({ title: 'Mutated', tags: [] }) }) - .where('id', '=', 'a') - .execute() - await webSyncStep.run({ filePaths: ['books/touch.md'] }, ctx) - - const after = await db - .selectFrom('web_pieces') - .select('title') - .where('id', '=', 'a') - .executeTakeFirst() - expect(after?.title).toBe(originalTitle) - }) - - test('sanitizes asset paths in json_metadata to keys', async () => { + test('sanitizes metadata assets and returns the matching key-to-path map for transforms', async () => { await seedItem({ frontmatter_json: JSON.stringify({ title: 'X', @@ -179,25 +164,28 @@ describe('webSyncStep', () => { assets_json_array: JSON.stringify(['books/great/cover.png']), }) - await webSyncStep.run({ filePaths: ['books/great.md'] }, ctx) + const result = await syncPaths(['books/great.md']) const row = await db.selectFrom('web_pieces').select('json_metadata').executeTakeFirst() const parsed = JSON.parse(row!.json_metadata) expect(parsed.cover).not.toBe('books/great/cover.png') expect(parsed.cover).toMatch(/^[a-f0-9]{64}$/) + expect(result).toMatchObject({ + status: 'completed', + value: [ + { + webPiece: { json_metadata: row!.json_metadata }, + assetKeyToPath: [[parsed.cover, 'books/great/cover.png']], + }, + ], + }) }) - test('skips item types not in config.pieces', async () => { + test('rejects supplied source types not in config.pieces', async () => { await seedItem({ id: 'x', type: 'unknown', file_path: 'unknown/x.md' }) - await webSyncStep.run({ filePaths: ['unknown/x.md'] }, ctx) + await expect(syncPaths(['unknown/x.md'])).rejects.toThrow('Piece type is not configured') const rows = await db.selectFrom('web_pieces').selectAll().execute() expect(rows).toHaveLength(0) }) - - test('logs sync start and completion', async () => { - await webSyncStep.run({ filePaths: [] }, ctx) - expect(ctx.logger.info).toHaveBeenCalledWith('web.sync starting', { count: 0 }) - expect(ctx.logger.info).toHaveBeenCalledWith('web.sync complete') - }) }) diff --git a/packages/web/worker/src/steps/web-sync/index.ts b/packages/web/worker/src/steps/web-sync/index.ts index 1e199235..3e1e6b3d 100644 --- a/packages/web/worker/src/steps/web-sync/index.ts +++ b/packages/web/worker/src/steps/web-sync/index.ts @@ -1,140 +1,128 @@ -import { - type LuzzleSelectable, - type LuzzleTables, -} from '@luzzle/core' -import type { Config } from '@luzzle/web.config' -import type { Kysely } from 'kysely' -import type { Logger } from '../../services/logger.js' -import type { WebDatabase, WebPieceTags } from '../../services/db.js' +import type { LuzzleSelectable } from '@luzzle/core' +import { completed, type Step } from '../../core/step.js' +import type { WebPieces } from '../../services/db.js' import { buildAssetMaps } from '../../transforms/utils/assets.js' -import { completed, type Step, type StepResult } from '../../core/step.js' -import { - buildWebPiece, - generateUniqueSlug, - sanitizeMetadata, - slugify, -} from './helpers.js' +import { buildWebPiece, generateUniqueSlug, sanitizeMetadata, slugify } from './helpers.js' -type PiecesItem = LuzzleSelectable<'pieces_items'> -type FullDb = Kysely - -export interface WebSyncInput { - filePaths: string[] +export interface PublicationSource { + item: LuzzleSelectable<'pieces_items'> + contentHash: string } -export const webSyncStep: Step = { +export interface PreparedPublication { + webPiece: WebPieces + contentHash: string + assetKeyToPath: Array<[string, string]> + wasPublished: boolean +} + +export const webSyncStep: Step = { name: 'web.sync', - async run({ filePaths }, ctx): Promise> { - const { db, config, logger } = ctx - const fullDb = db as unknown as FullDb - - logger.info('web.sync starting', { count: filePaths.length }) - - const targets = new Set(filePaths) - const livePaths = new Set() - - for (const pieceConfig of config.pieces) { - const items = await fullDb - .selectFrom('pieces_items') - .selectAll() - .where('type', '=', pieceConfig.type) - .execute() - - const existing = await fullDb - .selectFrom('web_pieces') - .select(['slug', 'file_path']) - .where('type', '=', pieceConfig.type) - .execute() - const usedSlugs = new Set(existing.map((p) => p.slug)) - const slugByFilePath = new Map( - existing.map((p) => [p.file_path, p.slug]) + async run(sources, { db, config, logger }) { + const existing = await db + .selectFrom('web_pieces') + .select(['id', 'slug', 'type', 'file_path', 'date_added']) + .execute() + const publicationByPath = new Map(existing.map((piece) => [piece.file_path, piece])) + const usedSlugs = new Map>() + for (const piece of existing) { + if (!usedSlugs.has(piece.type)) usedSlugs.set(piece.type, new Set()) + usedSlugs.get(piece.type)!.add(piece.slug) + } + const prepared: PreparedPublication[] = [] + for (const { item, contentHash } of sources) { + const pieceConfig = config.pieces.find((piece) => piece.type === item.type) + if (!pieceConfig) throw new Error(`Piece type is not configured: ${item.type}`) + const previous = publicationByPath.get(item.file_path) + if (previous && previous.type !== item.type) { + throw new Error('Unpublish the previous piece type before publishing this source') + } + if (!usedSlugs.has(item.type)) usedSlugs.set(item.type, new Set()) + const slug = + previous?.slug ?? + generateUniqueSlug( + usedSlugs.get(item.type)!, + item.file_path.split('/').pop()!.replace(/\.md$/, '').split('.')[0] + ) + const frontmatter = JSON.parse(item.frontmatter_json) + const { pathToKey, keyToPath } = buildAssetMaps(item.assets_json_array, config.assets.salt) + const webPiece = buildWebPiece( + { + ...item, + id: previous?.id ?? item.id, + date_added: previous?.date_added ?? item.date_added, + frontmatter_json: sanitizeMetadata(item.frontmatter_json, pathToKey), + }, + pieceConfig, + slug, + config.assets.salt, + frontmatter ) - - for (const item of items) { - livePaths.add(item.file_path) - if (!targets.has(item.file_path)) continue - await syncOne(fullDb, item, pieceConfig, config, usedSlugs, slugByFilePath, logger) + await db + .insertInto('web_pieces') + .values({ ...webPiece, content_hash: null }) + .onConflict((conflict) => + conflict.column('id').doUpdateSet({ + content_hash: null, + key: webPiece.key, + title: webPiece.title, + summary: webPiece.summary ?? null, + note: webPiece.note, + keywords: webPiece.keywords ?? null, + json_metadata: webPiece.json_metadata, + date_updated: webPiece.date_updated, + date_consumed: webPiece.date_consumed ?? null, + }) + ) + .execute() + await db.deleteFrom('web_pieces_tags').where('piece_id', '=', webPiece.id).execute() + const keywords: string[] = webPiece.keywords ? JSON.parse(webPiece.keywords) : [] + if (keywords.length) { + await db + .insertInto('web_pieces_tags') + .values( + keywords.map((tag) => ({ + piece_slug: webPiece.slug, + piece_type: webPiece.type, + piece_id: webPiece.id, + tag: tag.trim(), + slug: slugify(tag.trim()), + })) + ) + .execute() } + prepared.push({ + webPiece, + contentHash, + assetKeyToPath: [...keyToPath], + wasPublished: Boolean(previous), + }) + logger.info(`web piece synced: ${webPiece.file_path}`) } - - const stored = await fullDb.selectFrom('web_pieces').select('file_path').execute() - for (const row of stored) { - if (!livePaths.has(row.file_path)) { - await fullDb.deleteFrom('web_pieces').where('file_path', '=', row.file_path).execute() - logger.info(`web piece pruned: ${row.file_path}`) - } - } - - logger.info('web.sync complete') - return completed(undefined) + return completed(prepared) }, } -async function syncOne( - db: FullDb, - item: PiecesItem, - pieceConfig: Config['pieces'][number], - config: Config, - usedSlugs: Set, - slugByFilePath: Map, - logger: Logger -): Promise { - let slug: string - const existingSlug = slugByFilePath.get(item.file_path) - if (existingSlug) { - slug = existingSlug - } else { - const filename = item.file_path.split('/').pop()!.replace(/\.md$/, '').split('.')[0] - slug = generateUniqueSlug(usedSlugs, filename) - slugByFilePath.set(item.file_path, slug) - } - - const frontmatter = JSON.parse(item.frontmatter_json) - - const { pathToKey } = buildAssetMaps(item.assets_json_array, config.assets.salt) - const sanitizedItem: PiecesItem = { - ...item, - frontmatter_json: sanitizeMetadata(item.frontmatter_json, pathToKey), - } - - const webPiece = buildWebPiece( - sanitizedItem, - pieceConfig, - slug, - config.assets.salt, - frontmatter, - ) - - await db - .insertInto('web_pieces') - .values(webPiece) - .onConflict((oc) => - oc.column('id').doUpdateSet({ - content_hash: null, - title: webPiece.title, - summary: webPiece.summary, - note: webPiece.note, - keywords: webPiece.keywords, - json_metadata: webPiece.json_metadata, - date_updated: webPiece.date_updated, - date_consumed: webPiece.date_consumed, - }) - ) - .execute() - - await db.deleteFrom('web_pieces_tags').where('piece_id', '=', item.id).execute() - - const keywords: string[] = webPiece.keywords ? JSON.parse(webPiece.keywords) : [] - if (keywords.length > 0) { - const tags: WebPieceTags[] = keywords.map((tag) => ({ - piece_slug: slug, - piece_type: item.type, - piece_id: item.id, - tag: tag.trim(), - slug: slugify(tag.trim()), - })) - await db.insertInto('web_pieces_tags').values(tags).execute() - } - - logger.info(`web piece synced: ${item.file_path}`) +export const pruneWebPiecesStep: Step = { + name: 'web.prune', + async run(_input, { db, config, logger }) { + const live = await db + .selectFrom('pieces_items') + .select('file_path') + .where( + 'type', + 'in', + config.pieces.map((piece) => piece.type) + ) + .execute() + const livePaths = new Set(live.map((piece) => piece.file_path)) + const publications = await db.selectFrom('web_pieces').select('file_path').execute() + for (const piece of publications) { + if (!livePaths.has(piece.file_path)) { + await db.deleteFrom('web_pieces').where('file_path', '=', piece.file_path).execute() + logger.info(`web piece pruned: ${piece.file_path}`) + } + } + return completed(undefined) + }, } diff --git a/packages/web/worker/src/transforms/runner.test.ts b/packages/web/worker/src/transforms/runner.test.ts index fc8b8bf5..b08a478a 100644 --- a/packages/web/worker/src/transforms/runner.test.ts +++ b/packages/web/worker/src/transforms/runner.test.ts @@ -1,5 +1,5 @@ import { describe, test, expect, vi, afterEach, beforeEach } from 'vitest' -import { runTransformsForPiece, produceTransformsForPiece, persistTransforms } from './runner.js' +import { produceTransformsForPiece, persistTransforms } from './runner.js' import type { Config } from '@luzzle/web.config' import type { WebPieces } from '../services/db.js' import { setupDatabase, teardownDatabase } from '../../test/db.js' @@ -69,201 +69,47 @@ afterEach(async () => { vi.clearAllMocks() }) -describe('transforms/runner', () => { - test('runs all transforms and writes to DB by default', async () => { - transforms.attachment.run.mockResolvedValueOnce([ - { - transformation: 'attachment.original', - asset_path: 'books/key/file.pdf', - mime_type: 'application/pdf', - }, - ]) - - await runTransformsForPiece(db, webPiece, config, '/out', {} as Pieces, {}, new Map(), logger) - - const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() - expect(assets).toHaveLength(1) - expect(assets[0]).toMatchObject({ +describe('transforms/produceTransformsForPiece', () => { + test('produces records and shares prior assets and public URLs with later transforms', async () => { + const record = { transformation: 'attachment.original', asset_path: 'books/key/file.pdf', mime_type: 'application/pdf', - piece_file_path: 'book.md', - piece_key: 'key123', - asset_key: 'asset-key', - }) + } + transforms.attachment.run.mockResolvedValueOnce([record]) + transforms.palette.run.mockImplementationOnce( + async ({ priorAssets, buildAssetUrl }: TransformInput) => { + expect(priorAssets).toEqual([ + expect.objectContaining({ ...record, asset_key: 'asset-key' }), + ]) + expect(buildAssetUrl?.(record.asset_path)).toBe( + 'https://cdn.test/garden/pieces/assets/books/key/file.pdf' + ) + return [] + } + ) + + const produced = await produceTransformsForPiece( + webPiece, + config, + '/out', + {} as Pieces, + {}, + new Map(), + logger + ) + + expect(produced).toEqual([ + { name: 'attachment', records: [record] }, + { name: 'palette', records: [] }, + { name: 'opengraph', records: [] }, + ]) expect(transforms.attachment.run).toHaveBeenCalledOnce() expect(transforms.palette.run).toHaveBeenCalledOnce() expect(transforms.opengraph.run).toHaveBeenCalledOnce() - const { buildAssetUrl } = transforms.attachment.run.mock.calls[0][0] as TransformInput - expect(buildAssetUrl?.('books/key/file.pdf')).toBe( - 'https://cdn.test/garden/pieces/assets/books/key/file.pdf' - ) }) - test('filters to only the specified transform when typeFilter is given', async () => { - await runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { typeFilter: 'palette' }, - new Map(), - logger - ) - - expect(transforms.palette.run).toHaveBeenCalledOnce() - expect(transforms.attachment.run).not.toHaveBeenCalled() - expect(transforms.opengraph.run).not.toHaveBeenCalled() - }) - - test('skips DB delete and insert when dryRun is true', async () => { - transforms.attachment.run.mockResolvedValueOnce([ - { - transformation: 'attachment.original', - asset_path: 'books/key/file.pdf', - mime_type: 'application/pdf', - }, - ]) - - await runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { dryRun: true }, - new Map(), - logger - ) - - expect(transforms.attachment.run).toHaveBeenCalledOnce() - const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() - expect(assets).toHaveLength(0) - }) - - test('does not insert when transform returns empty records', async () => { - await runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { typeFilter: 'palette' }, - new Map(), - logger - ) - - const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() - expect(assets).toHaveLength(0) - }) - - test('logs generated asset path', async () => { - transforms.palette.run.mockResolvedValueOnce([ - { - transformation: 'palette', - asset_path: null as unknown as string, - mime_type: 'application/json', - content: '{}', - is_embedded: 1 as const, - }, - ]) - - await runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { typeFilter: 'palette' }, - new Map(), - logger - ) - - expect(logger.info).toHaveBeenCalledWith( - 'transform.palette generated content of application/json' - ) - }) - - test('logs and propagates a transform failure instead of producing empty records', async () => { - transforms.attachment.run.mockRejectedValueOnce(new Error('boom')) - - await expect( - runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { typeFilter: 'attachment' }, - new Map(), - logger - ) - ).rejects.toThrow('boom') - - expect(logger.error).toHaveBeenCalledWith( - 'transform.attachment error for book.md', - expect.objectContaining({ error: 'boom' }) - ) - }) - - test('retains existing asset rows when a later transform fails after earlier production succeeds', async () => { - const oldRecords = [ - { - piece_file_path: webPiece.file_path, - piece_key: webPiece.key, - asset_key: 'old-attachment', - transformation: 'attachment', - asset_path: 'old.pdf', - mime_type: 'application/pdf', - }, - { - piece_file_path: webPiece.file_path, - piece_key: webPiece.key, - asset_key: 'old-palette', - transformation: 'palette', - content: 'old palette', - mime_type: 'application/json', - }, - ] - await db.insertInto('web_pieces_assets').values(oldRecords).execute() - transforms.attachment.run.mockResolvedValueOnce([ - { transformation: 'attachment', asset_path: 'new.pdf', mime_type: 'application/pdf' }, - ]) - transforms.palette.run.mockRejectedValueOnce(new Error('palette failed')) - - await expect( - runTransformsForPiece(db, webPiece, config, '/out', {} as Pieces, {}, new Map(), logger) - ).rejects.toThrow('palette failed') - const assets = await db - .selectFrom('web_pieces_assets') - .selectAll() - .orderBy('asset_key') - .execute() - expect(assets).toMatchObject(oldRecords) - expect(transforms.opengraph.run).not.toHaveBeenCalled() - }) - - test('does nothing when typeFilter matches no transform', async () => { - await runTransformsForPiece( - db, - webPiece, - config, - '/out', - {} as Pieces, - { typeFilter: 'nonexistent' }, - new Map(), - logger - ) - - const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() - expect(assets).toHaveLength(0) - expect(transforms.attachment.run).not.toHaveBeenCalled() - }) -}) - -describe('transforms/produceTransformsForPiece', () => { - test('returns produced records without touching the DB', async () => { + test('returns records for only the selected transform', async () => { const record = { transformation: 'palette', asset_path: null as unknown as string, @@ -284,8 +130,8 @@ describe('transforms/produceTransformsForPiece', () => { ) expect(produced).toEqual([{ name: 'palette', records: [record] }]) - const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() - expect(assets).toHaveLength(0) + expect(transforms.attachment.run).not.toHaveBeenCalled() + expect(transforms.opengraph.run).not.toHaveBeenCalled() }) test('rejects failed production without returning a successful result or running later transforms', async () => { @@ -339,7 +185,15 @@ describe('transforms/persistTransforms', () => { const assets = await db.selectFrom('web_pieces_assets').selectAll().execute() expect(assets).toHaveLength(1) - expect(assets[0]).toMatchObject({ content: 'fresh', asset_key: 'asset-key' }) + expect(assets[0]).toMatchObject({ + content: 'fresh', + asset_key: 'asset-key', + transformation: 'palette', + mime_type: 'application/json', + is_embedded: 1, + piece_file_path: webPiece.file_path, + piece_key: webPiece.key, + }) }) test('still clears prior records after successful production of an empty result', async () => { diff --git a/packages/web/worker/src/transforms/runner.ts b/packages/web/worker/src/transforms/runner.ts index ca35d327..47ab4cc5 100644 --- a/packages/web/worker/src/transforms/runner.ts +++ b/packages/web/worker/src/transforms/runner.ts @@ -115,28 +115,3 @@ export async function persistTransforms( .execute() } } - -export async function runTransformsForPiece( - db: Kysely, - webPiece: WebPieces, - config: Config, - outDir: string, - pieces: Pieces, - options: { typeFilter?: string; dryRun?: boolean }, - assetKeyToPath: Map, - logger: Logger -): Promise { - const produced = await produceTransformsForPiece( - webPiece, - config, - outDir, - pieces, - { typeFilter: options.typeFilter }, - assetKeyToPath, - logger - ) - - if (!options.dryRun) { - await persistTransforms(db, webPiece, config, produced) - } -} diff --git a/packages/web/worker/src/workflows/publish-recovery.test.ts b/packages/web/worker/src/workflows/publish-recovery.test.ts deleted file mode 100644 index 082d4082..00000000 --- a/packages/web/worker/src/workflows/publish-recovery.test.ts +++ /dev/null @@ -1,715 +0,0 @@ -import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' -import { mkdir, mkdtemp, rm, utimes, writeFile } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import path from 'node:path' -import { migrate, type PiecesDiff } from '@luzzle/core' -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 { 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, 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' -import { emptyPiecesDiff } from './pieces-diff.js' - -type PublishWorkflow = Workflow['fn'] -type WorkflowStep = Parameters[0]['step'] -type AssetsRunner = ( - input: AssetsGenerateInput, - ctx: WorkerContext -) => Promise> - -const mocks = vi.hoisted(() => ({ - implementWorkflow: vi.fn<(spec: typeof publishSpec, fn: PublishWorkflow) => void>(), - assets: vi.fn(), - cdn: vi.fn<() => Promise>>(), - cache: vi.fn<() => Promise>>(), -})) - -vi.mock('@luzzle/web.jobs', () => ({ - getOpenWorkflow: () => ({ implementWorkflow: mocks.implementWorkflow }), -})) -vi.mock('../steps/assets-generate.js', () => ({ - assetsGenerateStep: { name: 'assets.generate', run: mocks.assets }, -})) -vi.mock('../steps/cdn-sync.js', () => ({ - cdnSyncStep: { name: 'cdn.sync', run: mocks.cdn }, -})) -vi.mock('../steps/cache-purge.js', () => ({ - cachePurgeStep: { name: 'cache.purge', run: mocks.cache }, -})) - -const healthy = 'healthy.books.md' -const broken = 'broken.books.md' -const batch = [broken, healthy] -const completionTime = 1_900_000_000_000 - -function valueOf(result: StepResult): T { - expect(result.status).toBe('completed') - if (result.status !== 'completed') throw new Error('Expected a completed step') - return result.value -} - -describe('publish hash recovery with a real archive and SQLite', () => { - let directory: string - let ctx: WorkerContext - let workflow: PublishWorkflow - - beforeEach(async () => { - vi.resetAllMocks() - vi.spyOn(Date, 'now').mockReturnValue(completionTime) - mocks.assets.mockResolvedValue(completed({ failedPieces: [] })) - mocks.cdn.mockResolvedValue(completed(undefined)) - mocks.cache.mockResolvedValue(completed(undefined)) - - directory = await mkdtemp(path.join(tmpdir(), 'luzzle-publish-recovery-')) - const archive = path.join(directory, 'archive') - await mkdir(path.join(archive, '.luzzle', 'schemas'), { recursive: true }) - await writeFile( - path.join(archive, '.luzzle', 'schemas', 'books.json'), - JSON.stringify({ - type: 'object', - title: 'books', - properties: { title: { type: 'string' } }, - required: ['title'], - additionalProperties: true, - }) - ) - const config = validateConfig( - { - storage: { root: archive }, - paths: { - config: path.join(directory, 'config.yaml'), - database: 'index.sqlite', - assets: path.join(directory, 'assets'), - }, - assets: { salt: 'recovery-test' }, - pieces: [{ type: 'books', fields: { title: 'title', date_consumed: 'date_consumed' } }], - }, - {} - ) - const db = createAppDb(path.join(directory, 'index.sqlite')) - expect((await migrate(db)).error).toBeUndefined() - expect((await runWebMigrations(db)).error).toBeUndefined() - const logger = { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - stdout: vi.fn(), - stderr: vi.fn(), - } - ctx = { db, config, logger, rclone: new RcloneClient(logger) } - setWorkerContext(ctx) - registerPublishWorkflow() - expect(mocks.implementWorkflow).toHaveBeenCalledOnce() - const [spec, callback] = mocks.implementWorkflow.mock.calls[0]! - expect(spec).toBe(publishSpec) - workflow = callback - }) - - afterEach(async () => { - vi.restoreAllMocks() - await ctx?.db.destroy() - await rm(directory, { recursive: true, force: true }) - }) - - async function writePiece(filePath: string, title: string) { - const target = path.join(ctx.config.storage.root, filePath) - await writeFile(target, `---\ntitle: ${title}\n---\n\nMarkdown for ${title}.\n`) - await utimes(target, new Date(), new Date(Date.now() + 10_000)) - } - - // Each invocation is a fresh run with no saved checkpoints. - // The optional boundary models losing a phase result, not an SDK process crash. - async function publish( - jobId: string, - afterPhase?: (name: string, value: unknown) => Promise - ) { - const phaseResults = new Map() - const step: WorkflowStep = { - run: async ( - { name }: { name: string }, - fn: () => Promise | Output | undefined - ): Promise => { - const value = await fn() - await afterPhase?.(name, value) - phaseResults.set(name, value) - return value as Output - }, - } as WorkflowStep - const result = await workflow({ - input: {}, - step, - version: null, - run: { - id: jobId, - workflowName: publishSpec.name, - createdAt: new Date(), - startedAt: null, - }, - }) - 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') - .select('content_hash') - .where('file_path', '=', filePath) - .executeTakeFirstOrThrow() - const published = await ctx.db - .selectFrom('web_pieces') - .select(['content_hash', 'title', 'last_published_at']) - .where('file_path', '=', filePath) - .executeTakeFirstOrThrow() - return { - indexed: indexed.content_hash, - published: published.content_hash, - title: published.title, - lastPublishedAt: published.last_published_at, - } - } - - async function audit(): Promise { - 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') - const original = await hashes(healthy) - await writePiece(healthy, 'Changed') - await writePiece(broken, 'New') - - const interruption = new Error('lost sync result before checkpoint') - await expect( - publish('interrupted', async (name, value) => { - if (name !== 'luzzle.sync') return - expect(value).toMatchObject({ pieces: { added: [broken], updated: [healthy] } }) - throw interruption - }) - ).rejects.toBe(interruption) - expect(await hashes(healthy)).toMatchObject({ - published: original.published, - title: 'Original', - }) - expect(await ctx.db.selectFrom('web_pieces').select('file_path').execute()).toEqual([ - { file_path: healthy }, - ]) - expect((await audit()).pieces).toEqual({ added: [broken], updated: [healthy], pruned: [] }) - - mocks.assets.mockClear() - const recovered = await publish('fresh-recovery') - expect(recovered.phaseResults.get('luzzle.sync')).toEqual(emptyPiecesDiff()) - expect([...recovered.phaseResults.keys()]).toEqual([ - 'luzzle.sync', - 'publish.prepare', - 'web.sync', - 'assets.generate', - 'cdn.sync', - 'cache.purge', - 'publish.complete', - ]) - expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: batch }, ctx) - expect(recovered.result).toEqual({ - schemas: emptyPiecesDiff().schemas, - pieces: { added: [broken], updated: [healthy], pruned: [] }, - failedPieces: [], - }) - for (const filePath of batch) { - const state = await hashes(filePath) - expect(state.published).toBe(state.indexed) - } - expect(await audit()).toEqual(emptyPiecesDiff()) - }) - - 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({ 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) - - await publish('baseline') - const original = await hashes(healthy) - await writePiece(healthy, 'Metadata changed') - const diff = valueOf(await luzzleSyncStep.run(undefined, 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') - expect(state.indexed).not.toBe(original.indexed) - expect(state.published).toBeNull() - expect(original.lastPublishedAt).toBe(completionTime) - expect(state.lastPublishedAt).toBe(original.lastPublishedAt) - expect((await getPendingPublication(ctx.db, ctx.config)).pieces).toEqual(plan.pieces) - expect((await audit()).pieces).toEqual({ added: [], updated: [healthy], pruned: [] }) - }) - - test('partial transform failure completes healthy siblings and a fresh run targets only the failure', async () => { - await writePiece(healthy, 'Original healthy') - await writePiece(broken, 'Original broken') - await publish('baseline') - const oldBroken = await hashes(broken) - vi.mocked(Date.now).mockReturnValue(completionTime + 1000) - await writePiece(healthy, 'Changed healthy') - await writePiece(broken, 'Changed broken') - const failedPieces = [{ filePath: broken, message: 'markdown transform failed' }] - mocks.assets.mockResolvedValueOnce(completed({ failedPieces })) - - const partial = await publish('partial') - expect(partial.result.failedPieces).toEqual(failedPieces) - const healthyState = await hashes(healthy) - expect(healthyState.published).toBe(healthyState.indexed) - expect(healthyState.lastPublishedAt).toBe(completionTime + 1000) - const brokenState = await hashes(broken) - expect(brokenState.title).toBe('Changed broken') - expect(oldBroken.published).toBe(oldBroken.indexed) - expect(brokenState.published).toBeNull() - expect(brokenState.published).not.toBe(brokenState.indexed) - expect(oldBroken.lastPublishedAt).toBe(completionTime) - expect(brokenState.lastPublishedAt).toBe(oldBroken.lastPublishedAt) - expect((await audit()).pieces).toEqual({ added: [], updated: [broken], pruned: [] }) - - mocks.assets.mockClear() - vi.mocked(Date.now).mockReturnValue(completionTime + 2000) - const retry = await publish('fresh-retry') - expect(retry.phaseResults.get('luzzle.sync')).toEqual(emptyPiecesDiff()) - expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [broken] }, ctx) - expect(retry.result.pieces).toEqual({ added: [], updated: [broken], pruned: [] }) - expect(retry.result.failedPieces).toEqual([]) - expect(await hashes(broken)).toMatchObject({ - published: brokenState.indexed, - lastPublishedAt: completionTime + 2000, - }) - expect((await hashes(healthy)).lastPublishedAt).toBe(healthyState.lastPublishedAt) - expect(await audit()).toEqual(emptyPiecesDiff()) - }) - - test.each(['cdn.sync', 'cache.purge'])( - '%s failure leaves the entire batch pending for a fresh publish', - async (phase) => { - await writePiece(healthy, 'Original healthy') - await writePiece(broken, 'Original broken') - await publish('baseline') - const originalHealthy = await hashes(healthy) - const originalBroken = await hashes(broken) - vi.mocked(Date.now).mockReturnValue(completionTime + 1000) - await writePiece(healthy, 'Changed healthy') - await writePiece(broken, 'Changed broken') - const error = new Error(`${phase} unavailable`) - if (phase === 'cdn.sync') mocks.cdn.mockRejectedValueOnce(error) - if (phase === 'cache.purge') mocks.cache.mockRejectedValueOnce(error) - - await expect(publish('global-failure')).rejects.toBe(error) - expect(originalHealthy.published).toBe(originalHealthy.indexed) - expect(originalBroken.published).toBe(originalBroken.indexed) - expect(await hashes(healthy)).toMatchObject({ - published: null, - lastPublishedAt: originalHealthy.lastPublishedAt, - }) - expect(await hashes(broken)).toMatchObject({ - published: null, - lastPublishedAt: originalBroken.lastPublishedAt, - }) - const completion = await ctx.db - .selectFrom('job_progress') - .selectAll() - .where('job_id', '=', 'global-failure') - .where('phase', '=', 'publish.complete') - .execute() - expect(completion).toEqual([]) - expect((await audit()).pieces).toEqual({ added: [], updated: batch, pruned: [] }) - - mocks.assets.mockClear() - vi.mocked(Date.now).mockReturnValue(completionTime + 2000) - const retry = await publish('fresh-global-retry') - expect(retry.phaseResults.get('luzzle.sync')).toEqual(emptyPiecesDiff()) - expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: batch }, ctx) - expect(retry.result.pieces).toEqual({ added: [], updated: batch, pruned: [] }) - for (const filePath of batch) { - const state = await hashes(filePath) - expect(state.published).toBe(state.indexed) - expect(state.lastPublishedAt).toBe(completionTime + 2000) - } - expect(await audit()).toEqual(emptyPiecesDiff()) - } - ) - - test('completion records the captured revision, leaving a newer indexed revision pending', async () => { - await writePiece(healthy, 'Revision one') - await publish('baseline') - await writePiece(healthy, 'Revision two') - let processedHash: string | undefined - mocks.assets.mockImplementationOnce(async ({ filePaths }) => { - expect(filePaths).toEqual([healthy]) - processedHash = (await hashes(healthy)).indexed - await writePiece(healthy, 'Revision three') - expect(valueOf(await luzzleSyncStep.run(undefined, ctx)).pieces.updated).toEqual([healthy]) - return completed({ failedPieces: [] }) - }) - - const completedRun = await publish('revision-two') - expect(completedRun.phaseResults.get('publish.prepare')).toMatchObject({ - pieces: [{ filePath: healthy, contentHash: processedHash }], - }) - const state = await hashes(healthy) - expect(processedHash).toBeTypeOf('string') - expect(state.published).toBe(processedHash) - expect(state.indexed).not.toBe(processedHash) - expect(state.title).toBe('Revision two') - expect((await audit()).pieces).toEqual({ added: [], updated: [healthy], pruned: [] }) - - mocks.assets.mockClear() - const retry = await publish('fresh-revision-three') - expect(retry.phaseResults.get('luzzle.sync')).toEqual(emptyPiecesDiff()) - expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx) - expect(await hashes(healthy)).toEqual({ - indexed: state.indexed, - published: state.indexed, - title: 'Revision three', - lastPublishedAt: completionTime, - }) - expect(await audit()).toEqual(emptyPiecesDiff()) - }) - - test.each([false, true])( - 'reverted Markdown stays pending after a failed publication (already indexed: %s)', - async (alreadyIndexed) => { - await writePiece(healthy, 'Revision A') - await publish('baseline-A') - const original = await hashes(healthy) - await writePiece(healthy, 'Revision B') - mocks.assets.mockResolvedValueOnce( - completed({ - failedPieces: [{ filePath: healthy, message: 'image failed' }], - }) - ) - await publish('failed-B') - expect(await hashes(healthy)).toMatchObject({ title: 'Revision B', published: null }) - - await writePiece(healthy, 'Revision A') - if (alreadyIndexed) await luzzleSyncStep.run(undefined, ctx) - expect((await audit()).pieces.updated).toEqual([healthy]) - - mocks.assets.mockClear() - await publish('restore-A') - expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx) - expect(await hashes(healthy)).toEqual({ - indexed: original.indexed, - published: original.indexed, - title: 'Revision A', - lastPublishedAt: completionTime, - }) - expect(await audit()).toEqual(emptyPiecesDiff()) - } - ) - - test('a read-only audit unions unsynced source changes with pending indexed publication', async () => { - await writePiece(healthy, 'Already indexed') - valueOf(await luzzleSyncStep.run(undefined, ctx)) - await writePiece(broken, 'Not yet indexed') - const before = await ctx.db.selectFrom('pieces_cache').selectAll().execute() - const diff = await audit() - expect([...diff.pieces.added].sort()).toEqual(batch) - expect(diff.pieces.updated).toEqual([]) - expect(await ctx.db.selectFrom('pieces_cache').selectAll().execute()).toEqual(before) - expect(await ctx.db.selectFrom('web_pieces').selectAll().execute()).toEqual([]) - }) -}) diff --git a/packages/web/worker/src/workflows/publish.test.ts b/packages/web/worker/src/workflows/publish.test.ts index 0a2ffc11..276f999d 100644 --- a/packages/web/worker/src/workflows/publish.test.ts +++ b/packages/web/worker/src/workflows/publish.test.ts @@ -1,347 +1,574 @@ -import { describe, test, expect, vi, beforeEach } from 'vitest' -import type { Workflow } from 'openworkflow' -import type { PiecesDiff } from '@luzzle/core' -import type { PublishPayload, PublishResult } from '@luzzle/web.jobs' +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from 'node:fs/promises' +import { createHash } from 'node:crypto' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { asCoreDatabase, migrate } from '@luzzle/core' +import { validateConfig } from '@luzzle/web.config' +import { createAppDb, runWebMigrations } from '@luzzle/web.db' +import type { PublishPayload, PublishPiecePayload, PublishResult } from '@luzzle/web.jobs' import { publishSpec } from '@luzzle/web.jobs/specs' -import type { StepResult } from '../core/step.js' -import type { WorkerContext } from '../services/context.js' +import type { Workflow } from 'openworkflow' +import { completed } from '../core/step.js' +import { RcloneClient } from '../services/rclone.js' +import { getPendingPublication } from '../services/publication.js' +import { setWorkerContext, type WorkerContext } from '../services/context.js' +import { archiveSyncStep } from '../steps/archive-sync.js' +import { luzzleSyncStep } from '../steps/luzzle-sync.js' +import type * as TransformRunner from '../transforms/runner.js' import { registerPublishWorkflow } from './publish.js' -import { emptyPiecesDiff } from './pieces-diff.js' -type PublishWorkflow = Workflow['fn'] +type PublishInput = PublishPayload | PublishPiecePayload +type PublishWorkflow = Workflow['fn'] type WorkflowStep = Parameters[0]['step'] -const mocks = vi.hoisted(() => { - const jobStep = (name: string) => ({ - name, - run: vi.fn<(input: unknown) => Promise>>(), - }) - return { - implementWorkflow: vi.fn<(spec: typeof publishSpec, fn: PublishWorkflow) => void>(), - logger: { info: vi.fn(), warn: vi.fn() }, - progress: { start: vi.fn(), complete: vi.fn(), skip: vi.fn(), fail: vi.fn() }, - archiveSync: jobStep('archive.sync'), - luzzleSync: jobStep('luzzle.sync'), - webSync: jobStep('web.sync'), - assetsGenerate: jobStep('assets.generate'), - cdnSync: jobStep('cdn.sync'), - cachePurge: jobStep('cache.purge'), - publishPrepare: jobStep('publish.prepare'), - publishComplete: jobStep('publish.complete'), - } -}) +const mocks = vi.hoisted(() => ({ + implementWorkflow: vi.fn<(spec: typeof publishSpec, fn: PublishWorkflow) => void>(), + produce: vi.fn(), + copy: vi.fn(), + cdn: vi.fn(), + cache: vi.fn(), +})) vi.mock('@luzzle/web.jobs', () => ({ getOpenWorkflow: () => ({ implementWorkflow: mocks.implementWorkflow }), })) -vi.mock('../services/context.js', () => ({ - getWorkerContext: () => ({ logger: mocks.logger, db: {} }), - createPhaseContext: (ctx: WorkerContext) => ctx, +vi.mock('../transforms/runner.js', async (importOriginal) => ({ + ...(await importOriginal()), + produceTransformsForPiece: mocks.produce, })) -vi.mock('../core/job-progress.js', () => ({ - JobProgress: class { - constructor() { - return mocks.progress - } - }, +vi.mock('../steps/cdn-copy-files.js', () => ({ + cdnCopyFilesStep: { name: 'cdn.copy-files', run: mocks.copy }, +})) +vi.mock('../steps/cdn-sync.js', () => ({ + cdnSyncStep: { name: 'cdn.sync', run: mocks.cdn }, +})) +vi.mock('../steps/cache-purge.js', () => ({ + cachePurgeStep: { name: 'cache.purge', run: mocks.cache }, })) -vi.mock('../steps/archive-sync.js', () => ({ archiveSyncStep: mocks.archiveSync })) -vi.mock('../steps/luzzle-sync.js', () => ({ luzzleSyncStep: mocks.luzzleSync })) -vi.mock('../steps/web-sync/index.js', () => ({ webSyncStep: mocks.webSync })) -vi.mock('../steps/assets-generate.js', () => ({ assetsGenerateStep: mocks.assetsGenerate })) -vi.mock('../steps/cdn-sync.js', () => ({ cdnSyncStep: mocks.cdnSync })) -vi.mock('../steps/cache-purge.js', () => ({ cachePurgeStep: mocks.cachePurge })) -vi.mock('../steps/publish-prepare.js', () => ({ publishPrepareStep: mocks.publishPrepare })) -vi.mock('../steps/publish-complete.js', () => ({ publishCompleteStep: mocks.publishComplete })) -const summary: PiecesDiff = { - schemas: { added: ['books'], updated: [], pruned: ['old'] }, - pieces: { - added: ['healthy.books.md'], - updated: ['broken.books.md'], - pruned: ['removed.books.md'], - }, -} +const target = 'target.books.md' +const pending = 'pending.books.md' +const removed = 'removed.books.md' +const publishedAt = 1_900_000_000_000 +const hash = (rawContent: string) => createHash('md5').update(rawContent).digest('hex') -// Exercise the app's registered callback and real runProgressPhase helper, without -// the SDK's private registry or any filesystem/transform/CDN side effects. -async function runPublish(input: PublishPayload = {}, checkpoints = new Map()) { - registerPublishWorkflow() - expect(mocks.implementWorkflow).toHaveBeenCalledOnce() - const [spec, callback] = mocks.implementWorkflow.mock.calls[0]! - expect(spec).toBe(publishSpec) +// Real archive, SQLite, source loading, upserts, transform persistence and completion. +// Only expensive rendering and external delivery are replaced. +describe('registered publication workflow', () => { + let directory: string + let ctx: WorkerContext + let workflow: PublishWorkflow - const run = vi.fn( - async ( - { name }: { name: string }, - fn: () => Promise | Output | undefined - ): Promise => { - if (checkpoints.has(name)) return checkpoints.get(name) as Output - return (await fn()) as Output - } - ) - const result = await callback({ - input, - step: { run } as unknown as WorkflowStep, - version: null, - run: { - id: 'publish-1', - workflowName: publishSpec.name, - createdAt: new Date(), - startedAt: null, - }, - }) - return { result, phases: run.mock.calls.map(([config]) => config.name) } -} - -function expectHealthyPhasesComplete() { - expect(mocks.cdnSync.run).toHaveBeenCalledOnce() - expect(mocks.cachePurge.run).toHaveBeenCalledOnce() - expect(mocks.progress.complete).toHaveBeenCalledWith('publish-1', 'cdn.sync') - expect(mocks.progress.complete).toHaveBeenCalledWith('publish-1', 'cache.purge') - expect(mocks.progress.fail).not.toHaveBeenCalled() -} - -function expectNormalCompletion() { - expect(mocks.logger.info).toHaveBeenCalledWith('openworkflow publish complete', { - jobId: 'publish-1', - }) - expect(mocks.logger.warn).not.toHaveBeenCalled() -} - -describe('workflows/publish', () => { - beforeEach(() => { + beforeEach(async () => { vi.resetAllMocks() - for (const jobStep of [ - mocks.archiveSync, - mocks.luzzleSync, - mocks.webSync, - mocks.assetsGenerate, - mocks.cdnSync, - mocks.cachePurge, - mocks.publishPrepare, - mocks.publishComplete, - ]) { - jobStep.run.mockResolvedValue({ status: 'completed', value: undefined }) + vi.spyOn(Date, 'now').mockReturnValue(publishedAt) + for (const delivery of [mocks.copy, mocks.cdn, mocks.cache]) { + delivery.mockResolvedValue(completed(undefined)) } - mocks.luzzleSync.run.mockResolvedValue({ status: 'completed', value: summary }) - mocks.assetsGenerate.run.mockResolvedValue({ status: 'completed', value: { failedPieces: [] } }) - mocks.publishPrepare.run.mockResolvedValue({ - status: 'completed', - value: { - pieces: [ - { filePath: 'healthy.books.md', contentHash: 'healthy-hash' }, - { filePath: 'broken.books.md', contentHash: 'broken-hash' }, - ], - summary, - }, + mocks.produce.mockImplementation(async (piece, _config, outDir) => { + const assetPath = `${piece.key}/note.html` + await mkdir(path.join(outDir, piece.key), { recursive: true }) + await writeFile(path.join(outDir, assetPath), piece.note ?? '') + return [ + { + name: 'markdown', + records: [ + { transformation: 'markdown.file', asset_path: assetPath, mime_type: 'text/html' }, + { + transformation: 'markdown', + mime_type: 'text/html', + is_embedded: 1, + content: piece.note, + }, + ], + }, + ] }) - }) - - test('reports failed pieces in JSON while publishing healthy siblings and retaining the source diff', async () => { - const failedPieces = [{ filePath: 'broken.books.md', message: 'markdown transform failed' }] - mocks.assetsGenerate.run.mockResolvedValue({ status: 'completed', value: { failedPieces } }) - - const { result, phases } = await runPublish() - - expect(phases).toEqual([ - 'luzzle.sync', - 'publish.prepare', - 'web.sync', - 'assets.generate', - 'cdn.sync', - 'cache.purge', - 'publish.complete', - ]) - expect(mocks.webSync.run).toHaveBeenCalledWith( - { filePaths: ['healthy.books.md', 'broken.books.md'] }, - expect.anything() + directory = await mkdtemp(path.join(tmpdir(), 'luzzle-publish-')) + const archive = path.join(directory, 'archive') + await mkdir(path.join(archive, '.luzzle', 'schemas'), { recursive: true }) + await writeFile( + path.join(archive, '.luzzle', 'schemas', 'books.json'), + JSON.stringify({ + type: 'object', + title: 'books', + properties: { + title: { type: 'string' }, + tags: { type: 'array', items: { type: 'string' } }, + }, + required: ['title'], + additionalProperties: true, + }) ) - expect(mocks.assetsGenerate.run).toHaveBeenCalledWith( - { filePaths: ['healthy.books.md', 'broken.books.md'] }, - expect.anything() - ) - expect(mocks.publishComplete.run).toHaveBeenCalledExactlyOnceWith( + const config = validateConfig( { + storage: { root: archive }, + paths: { + config: path.join(directory, 'config.yaml'), + database: 'index.sqlite', + assets: path.join(directory, 'assets'), + }, + assets: { salt: 'publish-test' }, pieces: [ - { filePath: 'healthy.books.md', contentHash: 'healthy-hash' }, - { filePath: 'broken.books.md', contentHash: 'broken-hash' }, + { + type: 'books', + fields: { title: 'title', date_consumed: 'date_consumed', tags: 'tags' }, + }, ], - failedPieces, }, - expect.anything() + {} ) - expect(JSON.parse(JSON.stringify(result))).toEqual({ ...summary, failedPieces }) - expect(result.pieces.updated).toContain('broken.books.md') - expect(summary).not.toHaveProperty('failedPieces') - expect(mocks.progress.complete).toHaveBeenCalledWith('publish-1', 'assets.generate') - expectHealthyPhasesComplete() - expect(mocks.logger.warn).toHaveBeenCalledWith('openworkflow publish complete with failures', { - jobId: 'publish-1', - failedCount: 1, + const db = createAppDb(path.join(directory, 'index.sqlite')) + expect((await migrate(asCoreDatabase(db))).error).toBeUndefined() + expect((await runWebMigrations(db)).error).toBeUndefined() + const logger = { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + stdout: vi.fn(), + stderr: vi.fn(), + } + ctx = { db, config, logger, rclone: new RcloneClient(logger) } + setWorkerContext(ctx) + registerPublishWorkflow() + expect(mocks.implementWorkflow).toHaveBeenCalledOnce() + expect(mocks.implementWorkflow.mock.calls[0]![0]).toBe(publishSpec) + workflow = mocks.implementWorkflow.mock.calls[0]![1] + }) + + afterEach(async () => { + vi.restoreAllMocks() + await ctx?.db.destroy() + await rm(directory, { recursive: true, force: true }) + }) + + async function writePiece(filePath: string, title: string) { + const rawContent = `---\ntitle: ${title}\ntags: [${title}]\n---\n\nNote for ${title}.\n` + const destination = path.join(ctx.config.storage.root, filePath) + await writeFile(destination, rawContent) + // Avoid relying on filesystem timestamp resolution for the real indexer. + await utimes(destination, new Date(), new Date(Date.now() + 10_000)) + return { rawContent } + } + + async function publish(input: PublishInput = {}, jobId = 'publish-test') { + const step = { + run: async ( + _config: { name: string }, + fn: () => Promise | Output | undefined + ) => (await fn()) as Output, + } as WorkflowStep + return workflow({ + input, + step, + version: null, + run: { id: jobId, workflowName: publishSpec.name, createdAt: new Date(), startedAt: null }, }) - expect(mocks.logger.info).not.toHaveBeenCalledWith( - 'openworkflow publish complete', - expect.anything() - ) - }) + } - 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: [] }) + async function publicRows(filePath: string) { + const piece = await ctx.db + .selectFrom('web_pieces') + .selectAll() + .where('file_path', '=', filePath) + .executeTakeFirstOrThrow() + return { + piece, + assets: await ctx.db + .selectFrom('web_pieces_assets') + .selectAll() + .where('piece_file_path', '=', filePath) + .orderBy('transformation') + .execute(), + tags: await ctx.db + .selectFrom('web_pieces_tags') + .selectAll() + .where('piece_id', '=', piece.id) + .orderBy('tag') + .execute(), } - ) + } - 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 }, - }, + async function coreRows() { + return { + items: await ctx.db.selectFrom('pieces_items').selectAll().orderBy('file_path').execute(), + cache: await ctx.db.selectFrom('pieces_cache').selectAll().orderBy('file_path').execute(), } - 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() + function clearEffects() { + for (const effect of [mocks.produce, mocks.copy, mocks.cdn, mocks.cache]) effect.mockClear() + } + + test('a queued filename publishes saved B without touching pending C or removed public D', async () => { + for (const filePath of [target, pending, removed]) await writePiece(filePath, 'Original') + await publish() + const original = (await publicRows(target)).piece + await writePiece(target, 'Queued A') + const queued: PublishPiecePayload = { filePath: target } + const saved = await writePiece(target, 'Saved B') + await writePiece(pending, 'Pending C') + await rm(path.join(ctx.config.storage.root, removed)) + await luzzleSyncStep.run(undefined, ctx) + const coreBefore = await coreRows() + expect(coreBefore.cache.find((row) => row.file_path === target)?.content_hash).toBe( + hash(saved.rawContent) ) - expect(mocks.assetsGenerate.run).toHaveBeenCalledExactlyOnceWith( - { filePaths: ['healthy.books.md'] }, - expect.anything() + const pendingBefore = await publicRows(pending) + const removedBefore = await publicRows(removed) + clearEffects() + const archive = vi.spyOn(archiveSyncStep, 'run') + const index = vi.spyOn(luzzleSyncStep, 'run') + + const result = await publish(queued) + + expect(result).toMatchObject({ + pieces: { added: [], updated: [target], pruned: [] }, + failedPieces: [], + }) + const published = await publicRows(target) + expect(published.piece).toMatchObject({ + id: original.id, + slug: original.slug, + date_added: original.date_added, + title: 'Saved B', + note: 'Note for Saved B.', + content_hash: hash(saved.rawContent), + }) + expect(JSON.parse(published.piece.json_metadata).title).toBe('Saved B') + expect(published.tags.map(({ tag }) => tag)).toEqual(['Saved B']) + expect( + published.assets.find(({ transformation }) => transformation === 'markdown') + ).toMatchObject({ content: 'Note for Saved B.', is_embedded: 1 }) + const manifest = [`${published.piece.key}/note.html`] + expect(mocks.produce).toHaveBeenCalledOnce() + expect(mocks.produce.mock.calls[0]![0]).toMatchObject({ + title: 'Saved B', + note: 'Note for Saved B.', + }) + expect(mocks.copy).toHaveBeenCalledExactlyOnceWith({ filePaths: manifest }, ctx) + expect(await readFile(path.join(ctx.config.paths.assets, manifest[0]!), 'utf8')).toBe( + 'Note for Saved B.' ) - expect(mocks.publishComplete.run).toHaveBeenCalledExactlyOnceWith( - { pieces: plan.pieces, failedPieces: [] }, - expect.anything() + expect(mocks.cache).toHaveBeenCalledOnce() + expect(mocks.cdn).not.toHaveBeenCalled() + expect(archive).not.toHaveBeenCalled() + expect(index).not.toHaveBeenCalled() + expect(await coreRows()).toEqual(coreBefore) + expect(await readFile(path.join(ctx.config.storage.root, target), 'utf8')).toBe( + saved.rawContent ) - expect(result).toEqual({ ...plan.summary, failedPieces: [] }) - expectHealthyPhasesComplete() + expect(await publicRows(pending)).toEqual(pendingBefore) + expect(await publicRows(removed)).toEqual(removedBefore) + expect((await getPendingPublication(ctx.db, ctx.config)).diff).toEqual({ + added: [], + updated: [pending], + pruned: [removed], + }) }) - test.each(['cdn.sync', 'cache.purge'])( - 'keeps a global %s rejection as a real publish failure', - async (phase) => { - const error = new Error(`${phase} failed`) - const jobStep = phase === 'cdn.sync' ? mocks.cdnSync : mocks.cachePurge - jobStep.run.mockRejectedValueOnce(error) - await expect(runPublish()).rejects.toBe(error) - expect(mocks.progress.fail).toHaveBeenCalledWith('publish-1', phase, error) - if (phase === 'cdn.sync') expect(mocks.cachePurge.run).not.toHaveBeenCalled() - expect(mocks.logger.info).not.toHaveBeenCalledWith( - 'openworkflow publish complete', - expect.anything() - ) - expect(mocks.logger.warn).not.toHaveBeenCalled() - expect(mocks.publishComplete.run).not.toHaveBeenCalled() - } - ) + test('archive selection uses the shared renderer/upsert, prunes globally and delivers the whole CDN once', async () => { + for (const filePath of [target, pending, removed]) await writePiece(filePath, 'Original') + await publish() + const original = (await publicRows(target)).piece + const pendingBefore = await publicRows(pending) + const selected = await writePiece(target, 'Selected revision') + await writePiece(pending, 'Unselected revision') + await writePiece('new.books.md', 'New unselected') + await rm(path.join(ctx.config.storage.root, removed)) + clearEffects() - test('returns successful reports and keeps the normal completion log', async () => { - const { result } = await runPublish() - expect(result).toEqual({ ...summary, failedPieces: [] }) - expect(mocks.publishComplete.run).toHaveBeenCalledExactlyOnceWith( - { - pieces: [ - { filePath: 'healthy.books.md', contentHash: 'healthy-hash' }, - { filePath: 'broken.books.md', contentHash: 'broken-hash' }, - ], - failedPieces: [], - }, - expect.anything() - ) - expect(mocks.archiveSync.run).not.toHaveBeenCalled() - expectHealthyPhasesComplete() - expectNormalCompletion() + const result = await publish({ filePaths: [target, target, removed, 'missing.books.md'] }) + + expect(result.pieces).toEqual({ added: [], updated: [target], pruned: [removed] }) + expect(result.failedPieces).toEqual([]) + const published = await publicRows(target) + expect(published.piece).toMatchObject({ + id: original.id, + slug: original.slug, + title: 'Selected revision', + content_hash: hash(selected.rawContent), + last_published_at: publishedAt, + }) + expect(published.tags.map(({ tag }) => tag)).toEqual(['Selected revision']) + expect( + published.assets.find(({ transformation }) => transformation === 'markdown') + ).toMatchObject({ content: 'Note for Selected revision.' }) + expect(mocks.produce).toHaveBeenCalledOnce() + expect(mocks.produce.mock.calls[0]![0].title).toBe('Selected revision') + expect(mocks.cdn).toHaveBeenCalledExactlyOnceWith(undefined, ctx) + expect(mocks.copy).not.toHaveBeenCalled() + expect(await publicRows(pending)).toEqual(pendingBefore) + expect( + await ctx.db + .selectFrom('web_pieces') + .select('file_path') + .where('file_path', 'in', [removed, 'new.books.md']) + .execute() + ).toEqual([]) + expect( + await ctx.db + .selectFrom('web_pieces_assets') + .selectAll() + .where('piece_file_path', '=', removed) + .execute() + ).toEqual([]) }) - test('still completes after configured delivery steps are skipped', async () => { - mocks.cdnSync.run.mockResolvedValue({ status: 'skipped', message: 'no remote' }) - mocks.cachePurge.run.mockResolvedValue({ status: 'skipped', message: 'no proxy' }) + test('explicit empty archive selection still prunes and delivers, without publishing pending rows', async () => { + await writePiece(target, 'Original') + await writePiece(removed, 'Removed') + await publish() + const before = await publicRows(target) + await writePiece(target, 'Pending') + await rm(path.join(ctx.config.storage.root, removed)) + clearEffects() - const { result, phases } = await runPublish() + const result = await publish({ filePaths: [] }) - expect(phases.slice(-3)).toEqual(['cdn.sync', 'cache.purge', 'publish.complete']) - expect(mocks.progress.skip).toHaveBeenCalledWith('publish-1', 'cdn.sync', 'no remote') - expect(mocks.progress.skip).toHaveBeenCalledWith('publish-1', 'cache.purge', 'no proxy') - expect(mocks.publishComplete.run).toHaveBeenCalledOnce() - expect(result).toEqual({ ...summary, failedPieces: [] }) - expectNormalCompletion() + expect(result.pieces).toEqual({ added: [], updated: [], pruned: [removed] }) + expect(await publicRows(target)).toEqual(before) + expect( + await ctx.db.selectFrom('web_pieces').selectAll().where('file_path', '=', removed).execute() + ).toEqual([]) + expect(mocks.produce).not.toHaveBeenCalled() + expect(mocks.cdn).toHaveBeenCalledOnce() + expect(mocks.copy).not.toHaveBeenCalled() }) - test('runs optional archive bisync before the remaining publish phases', async () => { - const { result, phases } = await runPublish({ bisync: true }) - expect(phases).toEqual([ - 'archive.sync', - 'luzzle.sync', - 'publish.prepare', - 'web.sync', - 'assets.generate', - 'cdn.sync', - 'cache.purge', - 'publish.complete', + test('a sibling render failure is reported and unstamped while healthy siblings complete; a fresh run retries only the failure', async () => { + await writePiece(target, 'Healthy original') + await writePiece(pending, 'Broken original') + await publish() + const oldFailed = (await publicRows(pending)).piece + const healthy = await writePiece(target, 'Healthy revision') + const broken = await writePiece(pending, 'Broken revision') + const render = mocks.produce.getMockImplementation()! + mocks.produce.mockImplementation(async (...args) => { + if (args[0].file_path === pending) throw new Error('markdown transform failed') + return render(...args) + }) + vi.mocked(Date.now).mockReturnValue(publishedAt + 1000) + + const result = await publish({ filePaths: [target, pending] }) + + expect(JSON.parse(JSON.stringify(result)).failedPieces).toEqual([ + { filePath: pending, message: 'markdown transform failed' }, ]) - expect(result).toEqual({ ...summary, failedPieces: [] }) - expectHealthyPhasesComplete() - expectNormalCompletion() - }) - - test('returns an empty diff and report when an older luzzle checkpoint has no pending work', async () => { - mocks.publishPrepare.run.mockResolvedValue({ - status: 'completed', - value: { pieces: [], summary: emptyPiecesDiff() }, + expect((await publicRows(target)).piece).toMatchObject({ + content_hash: hash(healthy.rawContent), + last_published_at: publishedAt + 1000, }) - const { result } = await runPublish({}, new Map([['luzzle.sync', undefined]])) - expect(result).toEqual({ ...emptyPiecesDiff(), failedPieces: [] }) - expect(mocks.luzzleSync.run).not.toHaveBeenCalled() - expect(mocks.webSync.run).toHaveBeenCalledWith({ filePaths: [] }, expect.anything()) - expect(mocks.assetsGenerate.run).toHaveBeenCalledWith({ filePaths: [] }, expect.anything()) - expectHealthyPhasesComplete() - expectNormalCompletion() + expect((await publicRows(pending)).piece).toMatchObject({ + title: 'Broken revision', + content_hash: null, + last_published_at: oldFailed.last_published_at, + }) + mocks.produce.mockImplementation(render) + clearEffects() + + const retry = await publish() + + expect(retry.pieces).toEqual({ added: [], updated: [pending], pruned: [] }) + expect(retry.failedPieces).toEqual([]) + expect(mocks.produce).toHaveBeenCalledOnce() + expect(mocks.produce.mock.calls[0]![0].file_path).toBe(pending) + expect((await publicRows(pending)).piece.content_hash).toBe(hash(broken.rawContent)) }) - test.each([undefined, null, {}, { failedPieces: [] }])( - 'accepts an empty/older assets checkpoint (%j)', - async (checkpoint) => { - const { result } = await runPublish({}, new Map([['assets.generate', checkpoint]])) - expect(result).toEqual({ ...summary, failedPieces: [] }) - expect(mocks.assetsGenerate.run).not.toHaveBeenCalled() - expectHealthyPhasesComplete() - expectNormalCompletion() + test('archive CDN failure leaves publication pending until a fresh publish rediscovers it', async () => { + await writePiece(target, 'Original') + await publish() + const revision = await writePiece(target, 'Retry revision') + clearEffects() + const error = new Error('CDN unavailable') + mocks.cdn.mockRejectedValueOnce(error) + vi.mocked(Date.now).mockReturnValue(publishedAt + 1000) + + await expect(publish({}, 'failed-archive')).rejects.toBe(error) + + expect(mocks.cdn).toHaveBeenCalledExactlyOnceWith(undefined, ctx) + expect(mocks.cache).not.toHaveBeenCalled() + expect((await publicRows(target)).piece).toMatchObject({ + title: 'Retry revision', + content_hash: null, + last_published_at: publishedAt, + }) + expect( + await ctx.db + .selectFrom('job_progress') + .selectAll() + .where('job_id', '=', 'failed-archive') + .where('phase', '=', 'publish.complete') + .execute() + ).toEqual([]) + clearEffects() + vi.mocked(Date.now).mockReturnValue(publishedAt + 2000) + + const retry = await publish({}, 'fresh-archive') + + expect(retry.pieces).toEqual({ added: [], updated: [target], pruned: [] }) + expect(retry.failedPieces).toEqual([]) + expect(mocks.produce).toHaveBeenCalledOnce() + expect(mocks.produce.mock.calls[0]![0].file_path).toBe(target) + expect(mocks.cdn).toHaveBeenCalledExactlyOnceWith(undefined, ctx) + expect(mocks.cache).toHaveBeenCalledOnce() + expect((await publicRows(target)).piece).toMatchObject({ + content_hash: hash(revision.rawContent), + last_published_at: publishedAt + 2000, + }) + }) + + test.each(['delivery', 'cache purge'] as const)( + '%s failure prevents completion until explicit direct retry', + async (failure) => { + await writePiece(target, 'Original') + await publish({ filePath: target }) + const saved = await writePiece(target, 'Retry revision') + clearEffects() + const error = new Error(`${failure} unavailable`) + const failingStep = failure === 'delivery' ? mocks.copy : mocks.cache + failingStep.mockRejectedValueOnce(error) + vi.mocked(Date.now).mockReturnValue(publishedAt + 1000) + + await expect(publish({ filePath: target }, 'failed')).rejects.toBe(error) + + expect((await publicRows(target)).piece).toMatchObject({ + title: 'Retry revision', + content_hash: null, + last_published_at: publishedAt, + }) + expect( + await ctx.db + .selectFrom('job_progress') + .selectAll() + .where('job_id', '=', 'failed') + .where('phase', '=', 'publish.complete') + .execute() + ).toEqual([]) + if (failure === 'delivery') expect(mocks.cache).not.toHaveBeenCalled() + vi.mocked(Date.now).mockReturnValue(publishedAt + 2000) + + const retry = await publish({ filePath: target }, 'explicit-retry') + + expect(retry.failedPieces).toEqual([]) + expect((await publicRows(target)).piece).toMatchObject({ + content_hash: hash(saved.rawContent), + last_published_at: publishedAt + 2000, + }) } ) - test('preserves failed piece reports from a durable assets checkpoint', async () => { - const failedPieces = [{ filePath: 'broken.books.md', message: 'cached failure' }] - const { result } = await runPublish({}, new Map([['assets.generate', { failedPieces }]])) - expect(result).toEqual({ ...summary, failedPieces }) - expect(mocks.publishComplete.run).toHaveBeenCalledWith( - expect.objectContaining({ failedPieces }), - expect.anything() + test('removing optional source fields clears previously published values and tags', async () => { + await writePiece(target, 'Original') + await publish({ filePath: target }) + await ctx.db + .updateTable('web_pieces') + .set({ + summary: 'Previous summary', + keywords: '["previous"]', + date_consumed: 123, + }) + .where('file_path', '=', target) + .execute() + const rawContent = '---\ntitle: Cleared metadata\n---\n\nNew note.\n' + + await writeFile(path.join(ctx.config.storage.root, target), rawContent) + await publish({ filePath: target }) + + const published = await publicRows(target) + expect(published.piece).toMatchObject({ + title: 'Cleared metadata', + summary: null, + keywords: null, + date_consumed: null, + content_hash: hash(rawContent), + }) + expect(published.tags).toEqual([]) + }) + + test('an unindexed direct publication keeps its public identity through a later archive publish', async () => { + await writePiece(target, 'Unindexed source') + const result = await publish({ filePath: target }) + expect(result.pieces.added).toEqual([target]) + expect(await coreRows()).toEqual({ items: [], cache: [] }) + const original = (await publicRows(target)).piece + const saved = await writePiece(target, 'Later indexed save') + + await publish() + + const indexed = await ctx.db + .selectFrom('pieces_items') + .select('id') + .where('file_path', '=', target) + .executeTakeFirstOrThrow() + expect(indexed.id).not.toBe(original.id) + const published = await publicRows(target) + expect(published.piece).toMatchObject({ + id: original.id, + slug: original.slug, + date_added: original.date_added, + title: 'Later indexed save', + content_hash: hash(saved.rawContent), + }) + expect(await ctx.db.selectFrom('web_pieces').selectAll().execute()).toEqual([published.piece]) + expect(published.tags).toEqual([ + expect.objectContaining({ + piece_id: original.id, + piece_slug: original.slug, + tag: 'Later indexed save', + }), + ]) + expect(await ctx.db.selectFrom('web_pieces_tags').selectAll().execute()).toEqual(published.tags) + }) + + test('invalid saved Markdown is rejected before public writes', async () => { + await writePiece(target, 'Original') + await publish({ filePath: target }) + const before = await publicRows(target) + const coreBefore = await coreRows() + clearEffects() + + await writeFile( + path.join(ctx.config.storage.root, target), + '---\ntags: [Invalid]\n---\n\nMissing required title.\n' ) - expect(mocks.assetsGenerate.run).not.toHaveBeenCalled() - expectHealthyPhasesComplete() - expect(mocks.logger.warn).toHaveBeenCalledWith('openworkflow publish complete with failures', { - jobId: 'publish-1', - failedCount: 1, + + await expect(publish({ filePath: target })).rejects.toThrow(/title/) + + expect(await publicRows(target)).toEqual(before) + expect(await coreRows()).toEqual(coreBefore) + expect(mocks.produce).not.toHaveBeenCalled() + expect(mocks.copy).not.toHaveBeenCalled() + expect(mocks.cache).not.toHaveBeenCalled() + }) + + test('a save during rendering stays pending after the source read by the worker completes', async () => { + const sourceB = await writePiece(target, 'Source B') + await luzzleSyncStep.run(undefined, ctx) + let savedC: Awaited> | undefined + const render = mocks.produce.getMockImplementation()! + mocks.produce.mockImplementationOnce(async (...args) => { + savedC = await writePiece(target, 'Saved C') + await luzzleSyncStep.run(undefined, ctx) + return render(...args) + }) + + await publish({ filePath: target }) + + expect((await publicRows(target)).piece).toMatchObject({ + title: 'Source B', + content_hash: hash(sourceB.rawContent), + last_published_at: publishedAt, + }) + expect((await coreRows()).cache.find((row) => row.file_path === target)?.content_hash).toBe( + hash(savedC!.rawContent) + ) + expect(await getPendingPublication(ctx.db, ctx.config)).toEqual({ + pieces: [{ filePath: target, contentHash: hash(savedC!.rawContent) }], + diff: { added: [], updated: [target], pruned: [] }, }) }) }) diff --git a/packages/web/worker/src/workflows/publish.ts b/packages/web/worker/src/workflows/publish.ts index 8696e341..a655a5d1 100644 --- a/packages/web/worker/src/workflows/publish.ts +++ b/packages/web/worker/src/workflows/publish.ts @@ -1,67 +1,36 @@ import { publishSpec } from '@luzzle/web.jobs/specs' import { getOpenWorkflow } from '@luzzle/web.jobs' import { getWorkerContext } from '../services/context.js' +import { publishPiece, publishArchive, type PublicationReport } from '../services/publish.js' import { JobProgress } from '../core/job-progress.js' import { runProgressPhase } from '../core/run-progress-phase.js' -import { archiveSyncStep } from '../steps/archive-sync.js' -import { luzzleSyncStep } from '../steps/luzzle-sync.js' -import { webSyncStep } from '../steps/web-sync/index.js' -import { assetsGenerateStep } from '../steps/assets-generate.js' -import { cdnSyncStep } from '../steps/cdn-sync.js' import { cachePurgeStep } from '../steps/cache-purge.js' -import { publishPrepareStep } from '../steps/publish-prepare.js' import { publishCompleteStep } from '../steps/publish-complete.js' -import { emptyPiecesDiff } from './pieces-diff.js' export function registerPublishWorkflow(): void { - const openWorkflow = getOpenWorkflow() - - openWorkflow.implementWorkflow(publishSpec, async ({ input, step, run }) => { + getOpenWorkflow().implementWorkflow(publishSpec, async ({ input, step, run }) => { const ctx = getWorkerContext() - const { logger, db } = ctx - const jobId = run.id - const progress = new JobProgress(db) - - logger.info('openworkflow publish starting', { jobId, bisync: input.bisync ?? false }) - - if (input.bisync) { - await runProgressPhase(step, ctx, jobId, progress, archiveSyncStep, undefined) - } - - const summary = await runProgressPhase(step, ctx, jobId, progress, luzzleSyncStep, undefined) - const plan = await runProgressPhase( - step, - ctx, - jobId, - progress, - publishPrepareStep, - { 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) - - await runProgressPhase(step, ctx, jobId, progress, webSyncStep, { filePaths: changedPaths }) - const assetsReport = await runProgressPhase(step, ctx, jobId, progress, assetsGenerateStep, { - filePaths: changedPaths, - }) - await runProgressPhase(step, ctx, jobId, progress, cdnSyncStep, undefined) - await runProgressPhase(step, ctx, jobId, progress, cachePurgeStep, undefined) - - const failedPieces = assetsReport?.failedPieces ?? [] - await runProgressPhase(step, ctx, jobId, progress, publishCompleteStep, { - pieces: plan.pieces, - failedPieces, - }) - - if (failedPieces.length > 0) { - logger.warn('openworkflow publish complete with failures', { - jobId, - failedCount: failedPieces.length, - }) + const progress = new JobProgress(ctx.db) + const execution = { step, ctx, jobId, progress } + ctx.logger.info('openworkflow publish starting', { jobId }) + let report: PublicationReport + if ('filePath' in input) { + report = await publishPiece(input, execution) } else { - logger.info('openworkflow publish complete', { jobId }) + report = await publishArchive(input, execution) } - return { ...plan.summary, failedPieces } + await runProgressPhase(step, ctx, jobId, progress, cachePurgeStep, undefined) + await runProgressPhase(step, ctx, jobId, progress, publishCompleteStep, { + pieces: report.pieces, + failedPieces: report.failedPieces, + }) + if (report.failedPieces.length) { + ctx.logger.warn('openworkflow publish complete with failures', { + jobId, + failedCount: report.failedPieces.length, + }) + } else ctx.logger.info('openworkflow publish complete', { jobId }) + return { ...report.summary, failedPieces: report.failedPieces } }) } -- 2.54.0