[web] share publishing for pieces and archives #201

Merged
eleith merged 2 commits from web/single-piece-publishing into main 2026-11-10 00:42:24 +00:00
23 changed files with 1428 additions and 1909 deletions

No files matched your search

+6
View File
@@ -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.
@@ -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<PublishPayload | Response> {
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)
@@ -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<typeof publish>[0] & Parameters<typeof audit>[0] & Parameters<typeof unpublish>[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<typeof publish>[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: '{'
@@ -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<typeof POST>[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)
}
)
+4 -4
View File
@@ -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.
+3 -3
View File
@@ -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 {
+1
View File
@@ -18,6 +18,7 @@ export {
} from './specs.js'
export type {
AssetRecord,
PublishPiecePayload,
PublishPayload,
PublishPieceFailure,
PublishResult,
+5 -6
View File
@@ -6,6 +6,7 @@ import type {
PreviewPayload,
PreviewResult,
PublishPayload,
PublishPiecePayload,
PublishResult,
PublishAuditPayload,
UnpublishPayload,
@@ -36,12 +37,10 @@ export const previewSpec: WorkflowSpec<PreviewPayload, PreviewResult> = defineWo
name: 'Preview',
})
export const publishSpec: WorkflowSpec<PublishPayload, PublishResult> = defineWorkflowSpec<
PublishPayload,
PublishResult
>({
name: 'Publish',
})
export const publishSpec: WorkflowSpec<PublishPayload | PublishPiecePayload, PublishResult> =
defineWorkflowSpec<PublishPayload | PublishPiecePayload, PublishResult>({
name: 'Publish',
})
export const publishAuditSpec: WorkflowSpec<PublishAuditPayload, PiecesDiff> = defineWorkflowSpec<
PublishAuditPayload,
+4 -1
View File
@@ -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
@@ -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',
+158
View File
@@ -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<string[], PublicationSource[]> = {
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<PublicationReport> {
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<PublicationReport> {
return publishPieces([input.filePath], execution)
}
export async function publishArchive(
input: PublishPayload,
execution: PublishExecution
): Promise<PublicationReport> {
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 }
}
@@ -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<typeof luzzleCore>()
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<WebDatabase & LuzzleTables>
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<LuzzleTables>() 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<typeof vi.fn>
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<Record<string, unknown>> = {}) {
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<Record<string, unknown>> = {}) {
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<void> }
}
)
.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')
}
)
})
@@ -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<WebDatabase & LuzzleTables>
export const assetsGenerateStep: Step<AssetsGenerateInput, AssetsGenerateResult> = {
export const assetsGenerateStep: Step<PreparedPublication[], AssetsGenerateResult> = {
name: 'assets.generate',
async run({ filePaths }, ctx): Promise<StepResult<AssetsGenerateResult>> {
async run(prepared, ctx): Promise<StepResult<AssetsGenerateResult>> {
const { db, config, logger } = ctx
const fullDb = db as unknown as FullDb
const filePaths = new Set<string>()
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<WebDatabase>,
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<WebDatabase>,
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<AssetsGenerateInput, AssetsGenerateResult>
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 })
},
}
@@ -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()
})
@@ -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<StepResult<void>> {
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)
},
}
@@ -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<PublishPrepareInput, PublishPlan> = {
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<PublishPrepareInput, PublishPlan> = {
}
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 })
},
}
@@ -63,18 +63,46 @@ async function seedItem(over: Partial<Record<string, unknown>> = {}) {
assets_json_array: '[]',
...over,
}
await (db.insertInto('pieces_items') as unknown as {
values: (v: typeof row) => { execute: () => Promise<void> }
})
await (
db.insertInto('pieces_items') as unknown as {
values: (v: typeof row) => { execute: () => Promise<void> }
}
)
.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')
})
})
+116 -128
View File
@@ -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<WebDatabase & LuzzleTables>
export interface WebSyncInput {
filePaths: string[]
export interface PublicationSource {
item: LuzzleSelectable<'pieces_items'>
contentHash: string
}
export const webSyncStep: Step<WebSyncInput, void> = {
export interface PreparedPublication {
webPiece: WebPieces
contentHash: string
assetKeyToPath: Array<[string, string]>
wasPublished: boolean
}
export const webSyncStep: Step<PublicationSource[], PreparedPublication[]> = {
name: 'web.sync',
async run({ filePaths }, ctx): Promise<StepResult<void>> {
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<string>()
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<string>(existing.map((p) => p.slug))
const slugByFilePath = new Map<string, string>(
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<string, Set<string>>()
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<string>,
slugByFilePath: Map<string, string>,
logger: Logger
): Promise<void> {
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<void, void> = {
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)
},
}
+45 -191
View File
@@ -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 () => {
@@ -115,28 +115,3 @@ export async function persistTransforms(
.execute()
}
}
export async function runTransformsForPiece(
db: Kysely<WebDatabase>,
webPiece: WebPieces,
config: Config,
outDir: string,
pieces: Pieces,
options: { typeFilter?: string; dryRun?: boolean },
assetKeyToPath: Map<string, string>,
logger: Logger
): Promise<void> {
const produced = await produceTransformsForPiece(
webPiece,
config,
outDir,
pieces,
{ typeFilter: options.typeFilter },
assetKeyToPath,
logger
)
if (!options.dryRun) {
await persistTransforms(db, webPiece, config, produced)
}
}
@@ -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<PublishPayload, PublishResult, PublishPayload>['fn']
type WorkflowStep = Parameters<PublishWorkflow>[0]['step']
type AssetsRunner = (
input: AssetsGenerateInput,
ctx: WorkerContext
) => Promise<StepResult<AssetsGenerateResult>>
const mocks = vi.hoisted(() => ({
implementWorkflow: vi.fn<(spec: typeof publishSpec, fn: PublishWorkflow) => void>(),
assets: vi.fn<AssetsRunner>(),
cdn: vi.fn<() => Promise<StepResult<void>>>(),
cache: vi.fn<() => Promise<StepResult<void>>>(),
}))
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<T>(result: StepResult<T>): 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<void>
) {
const phaseResults = new Map<string, unknown>()
const step: WorkflowStep = {
run: async <Output>(
{ name }: { name: string },
fn: () => Promise<Output | undefined> | Output | undefined
): Promise<Output> => {
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<OpenWorkflow['newWorker']>,
backend: BackendSqlite
) => Promise<void>
) {
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<OpenWorkflow['newWorker']>,
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<PiecesDiff> {
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([])
})
})
+522 -295
View File
@@ -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<PublishPayload, PublishResult, PublishPayload>['fn']
type PublishInput = PublishPayload | PublishPiecePayload
type PublishWorkflow = Workflow<PublishInput, PublishResult, PublishInput>['fn']
type WorkflowStep = Parameters<PublishWorkflow>[0]['step']
const mocks = vi.hoisted(() => {
const jobStep = (name: string) => ({
name,
run: vi.fn<(input: unknown) => Promise<StepResult<unknown>>>(),
})
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<typeof TransformRunner.produceTransformsForPiece>(),
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<typeof TransformRunner>()),
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<string, unknown>()) {
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 <Output>(
{ name }: { name: string },
fn: () => Promise<Output | undefined> | Output | undefined
): Promise<Output> => {
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 <Output>(
_config: { name: string },
fn: () => Promise<Output | undefined> | 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<string, unknown>([['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<ReturnType<typeof writePiece>> | 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: [] },
})
})
})
+21 -52
View File
@@ -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 }
})
}