.changeset/publish-recovery-hashes.md

@@ -0,0 +1,6 @@
+---
+"@luzzle/web": patch
+"@luzzle/web.worker": patch
+---
+
+Track successfully published Markdown content hashes so Check and Publish can find unfinished pieces after indexing has already advanced. Record success only after asset, CDN, and cache phases finish, leaving failed pieces pending for a later publish. Initialize existing web rows from their indexed hashes during migration. Deploy after existing publish jobs have finished; older interrupted jobs must not resume across this upgrade.

packages/web/db/package.json

@@ -8,7 +8,8 @@
 		"prebuild": "node scripts/generate-migrations.js",
 		"build": "tsc -p tsconfig.build.json",
 		"pretest": "node scripts/generate-migrations.js",
-		"test": "vitest run --silent --coverage"
+		"test": "vitest run --silent --coverage",
+		"schema:generate": "node scripts/generate-test-schema.js"
 	},
 	"engines": {
 		"node": ">=24"

packages/web/db/scripts/generate-test-schema.js

@@ -0,0 +1,42 @@
+import { writeFile } from 'node:fs/promises'
+import { getDatabaseClient, migrate, sql } from '@luzzle/core'
+import { runWebMigrations } from '@luzzle/web.db'
+
+const db = getDatabaseClient(':memory:')
+
+try {
+	const coreResult = await migrate(db)
+	if (coreResult.error) throw coreResult.error
+
+	const webResult = await runWebMigrations(db)
+	if (webResult.error) throw webResult.error
+
+	const schema = await sql`
+		SELECT sql FROM sqlite_schema
+		WHERE sql IS NOT NULL
+			AND name NOT LIKE 'sqlite_%'
+			AND name NOT LIKE 'kysely_%'
+			AND name NOT IN (
+				SELECT name FROM pragma_table_list WHERE type = 'shadow'
+			)
+		ORDER BY CASE type
+			WHEN 'table' THEN 0
+			WHEN 'index' THEN 1
+			ELSE 2
+		END, name
+	`.execute(db)
+
+	const statements = schema.rows.map(row => `${row.sql};`)
+	const snapshot = [
+		'-- Generated by pnpm --filter @luzzle/web.db schema:generate. Do not edit.',
+		'-- Statement boundaries let the test helper load complete triggers without parsing SQL.',
+		'',
+		statements.join('\n\n-- statement-breakpoint\n\n'),
+		'',
+	].join('\n')
+
+	await writeFile(new URL('../../worker/test/db.sql', import.meta.url), snapshot)
+	console.log(`Generated worker/test/db.sql (${statements.length} schema objects)`)
+} finally {
+	await db.destroy()
+}

packages/web/db/src/migrations/2026-09-15T00:00:00Z-add-web-pieces-content-hash.ts

@@ -0,0 +1,29 @@
+import type { Kysely } from "kysely";
+import { sql } from "kysely";
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export async function up(db: Kysely<any>): Promise<void> {
+	await db.schema
+		.alterTable("web_pieces")
+		.addColumn("content_hash", "text")
+		.execute();
+
+	const tables = await db.introspection.getTables();
+	const hasCache = tables.some((table) => table.name === "pieces_cache");
+	if (!hasCache) return;
+
+	// Trust the initial indexed state for existing web rows. Missing cache entries stay null.
+	await sql`
+		UPDATE web_pieces
+		SET content_hash = (
+			SELECT pieces_cache.content_hash
+			FROM pieces_cache
+			WHERE pieces_cache.file_path = web_pieces.file_path
+		)
+	`.execute(db);
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export async function down(db: Kysely<any>): Promise<void> {
+	await db.schema.alterTable("web_pieces").dropColumn("content_hash").execute();
+}

packages/web/db/src/types.ts

@@ -6,6 +6,7 @@ export interface WebPieces {
 	title: string
 	slug: string
 	file_path: string
+	content_hash?: string | null
 	note?: string
 	date_updated?: number
 	date_added: number
@@ -71,7 +72,7 @@ export type PublicWebPieceAsset = Pick<
 	'asset_key' | 'transformation' | 'asset_path' | 'mime_type' | 'is_embedded' | 'content'
 >
 
-export type PublicWebPiece = Omit<WebPieces, 'file_path' | 'json_metadata'> & {
+export type PublicWebPiece = Omit<WebPieces, 'file_path' | 'json_metadata' | 'content_hash'> & {
 	metadata: PieceFrontmatter
 	assets: PublicWebPieceAsset[]
 }

packages/web/db/vite.config.ts

@@ -6,6 +6,7 @@ export default defineConfig({
 		coverage: {
 			exclude: [
 				'eslint.config.js',
+				'scripts/generate-test-schema.js',
 				'dist/**',
 				'node_modules/**',
 				'src/migrations/**',

packages/web/worker/src/services/publication.test.ts

@@ -0,0 +1,200 @@
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+import type { Config } from '@luzzle/web.config'
+import type { Kysely } from 'kysely'
+import type { AppDatabase } from './db.js'
+import type { WorkerContext } from './context.js'
+import { setupDatabase, teardownDatabase } from '../../test/db.js'
+import { getPendingPublication } from './publication.js'
+import { publishPrepareStep } from '../steps/publish-prepare.js'
+import { publishCompleteStep } from '../steps/publish-complete.js'
+import { webSyncStep } from '../steps/web-sync/index.js'
+import { emptyPiecesDiff } from '../workflows/pieces-diff.js'
+
+let db: Kysely<AppDatabase>
+let ctx: WorkerContext
+
+beforeEach(async () => {
+	db = (await setupDatabase()).withTables<AppDatabase>()
+	ctx = {
+		db,
+		config: {
+			assets: { salt: 'test' },
+			pieces: [{ type: 'books', fields: { title: 'title', date_consumed: 'date_read' } }],
+		} as Config,
+		logger: {
+			debug: vi.fn(),
+			info: vi.fn(),
+			warn: vi.fn(),
+			error: vi.fn(),
+			stdout: vi.fn(),
+			stderr: vi.fn(),
+		},
+		rclone: {} as WorkerContext['rclone'],
+	}
+})
+
+afterEach(async () => {
+	await teardownDatabase(db)
+})
+
+async function indexPiece(filePath: string, contentHash?: string, type = 'books') {
+	await db
+		.insertInto('pieces_items')
+		.values({
+			id: filePath,
+			file_path: filePath,
+			type,
+			date_added: 1,
+			frontmatter_json: '{"title":"Test piece"}',
+			note_markdown: 'Test note',
+		})
+		.execute()
+	if (contentHash !== undefined) {
+		await db
+			.insertInto('pieces_cache')
+			.values({
+				id: filePath,
+				file_path: filePath,
+				content_hash: contentHash,
+			})
+			.execute()
+	}
+}
+
+async function publishMetadata(filePath: string, contentHash: string | null) {
+	await webSyncStep.run({ filePaths: [filePath] }, ctx)
+	await db
+		.updateTable('web_pieces')
+		.set({ content_hash: contentHash })
+		.where('file_path', '=', filePath)
+		.execute()
+}
+
+async function publishedHash(filePath: string) {
+	const piece = await db
+		.selectFrom('web_pieces')
+		.select('content_hash')
+		.where('file_path', '=', filePath)
+		.executeTakeFirstOrThrow()
+	return piece.content_hash
+}
+
+describe('publication', () => {
+	test('finds missing, stale and never-completed web rows, excluding published and unconfigured pieces', async () => {
+		await indexPiece('new.books.md', 'new-hash')
+		await indexPiece('changed.books.md', 'current-hash')
+		await indexPiece('unfinished.books.md', 'unfinished-hash')
+		await indexPiece('healthy.books.md', 'healthy-hash')
+		await indexPiece('unconfigured.films.md', 'film-hash', 'films')
+		await publishMetadata('changed.books.md', 'old-hash')
+		await publishMetadata('unfinished.books.md', null)
+		await publishMetadata('healthy.books.md', 'healthy-hash')
+
+		const pending = await getPendingPublication(db, ctx.config)
+
+		expect(pending).toEqual({
+			pieces: [
+				{ filePath: 'changed.books.md', contentHash: 'current-hash' },
+				{ filePath: 'new.books.md', contentHash: 'new-hash' },
+				{ filePath: 'unfinished.books.md', contentHash: 'unfinished-hash' },
+			],
+			diff: {
+				added: ['new.books.md'],
+				updated: ['changed.books.md', 'unfinished.books.md'],
+				pruned: [],
+			},
+		})
+	})
+
+	test('reports deleted web rows even after the core index has consumed the deletion', async () => {
+		await indexPiece('gone.books.md', 'gone-hash')
+		await publishMetadata('gone.books.md', 'gone-hash')
+		await db.deleteFrom('pieces_items').where('file_path', '=', 'gone.books.md').execute()
+
+		expect(await getPendingPublication(db, ctx.config)).toEqual({
+			pieces: [],
+			diff: { added: [], updated: [], pruned: ['gone.books.md'] },
+		})
+	})
+
+	test('does not label a missing core hash successfully published or prune its live row', async () => {
+		await indexPiece('uncached.books.md')
+		await publishMetadata('uncached.books.md', null)
+
+		expect(await getPendingPublication(db, ctx.config)).toEqual({
+			pieces: [],
+			diff: { added: [], updated: [], pruned: [] },
+		})
+		expect(await publishedHash('uncached.books.md')).toBeNull()
+	})
+
+	test('metadata refresh invalidates success before output changes; new metadata starts with no success hash', async () => {
+		await indexPiece('existing.books.md', 'new-hash')
+		await indexPiece('new.books.md', 'new-hash')
+		await publishMetadata('existing.books.md', 'old-hash')
+
+		await webSyncStep.run({ filePaths: ['existing.books.md', 'new.books.md'] }, ctx)
+
+		expect(await publishedHash('existing.books.md')).toBeNull()
+		expect(await publishedHash('new.books.md')).toBeNull()
+	})
+
+	test('prepare captures pending hashes 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')
+		const summary = emptyPiecesDiff()
+		summary.pieces.added.push('new.books.md')
+
+		const prepared = await publishPrepareStep.run(summary, ctx)
+
+		expect(prepared).toEqual({
+			status: 'completed',
+			value: {
+				pieces: [
+					{ filePath: 'changed.books.md', contentHash: 'current-hash' },
+					{ filePath: 'new.books.md', contentHash: 'new-hash' },
+				],
+				summary: {
+					schemas: summary.schemas,
+					pieces: { added: ['new.books.md'], updated: ['changed.books.md'], pruned: [] },
+				},
+			},
+		})
+		expect(summary.pieces.updated).toEqual([])
+	})
+
+	test('complete advances healthy pieces only, using their captured hashes', async () => {
+		await indexPiece('healthy.books.md', 'even-newer-hash')
+		await indexPiece('failed.books.md', 'failed-hash')
+		await publishMetadata('healthy.books.md', 'old-healthy-hash')
+		await publishMetadata('failed.books.md', 'old-failed-hash')
+
+		await publishCompleteStep.run(
+			{
+				pieces: [
+					{ filePath: 'healthy.books.md', contentHash: 'processed-hash' },
+					{ filePath: 'failed.books.md', contentHash: 'failed-hash' },
+				],
+				failedPieces: [{ filePath: 'failed.books.md', message: 'image failed' }],
+			},
+			ctx
+		)
+
+		expect(await publishedHash('healthy.books.md')).toBe('processed-hash')
+		expect(await publishedHash('failed.books.md')).toBe('old-failed-hash')
+		const pending = await getPendingPublication(db, ctx.config)
+		expect(pending.diff.updated).toEqual(['failed.books.md', 'healthy.books.md'])
+	})
+
+	test('complete and prepare accept an empty archive', async () => {
+		expect(await publishPrepareStep.run(emptyPiecesDiff(), ctx)).toEqual({
+			status: 'completed',
+			value: { pieces: [], summary: emptyPiecesDiff() },
+		})
+		expect(await publishCompleteStep.run({ pieces: [], failedPieces: [] }, ctx)).toEqual({
+			status: 'completed',
+			value: undefined,
+		})
+	})
+})

packages/web/worker/src/services/publication.ts

@@ -0,0 +1,63 @@
+import type { PiecesDiff } from '@luzzle/core'
+import type { Config } from '@luzzle/web.config'
+import type { Kysely } from 'kysely'
+import type { AppDatabase } from './db.js'
+
+export interface PublicationPiece {
+	filePath: string
+	contentHash: string
+}
+
+export interface PendingPublication {
+	pieces: PublicationPiece[]
+	diff: PiecesDiff['pieces']
+}
+
+export async function getPendingPublication(
+	db: Kysely<AppDatabase>,
+	config: Config
+): Promise<PendingPublication> {
+	const pending: PendingPublication = {
+		pieces: [],
+		diff: { added: [], updated: [], pruned: [] },
+	}
+	const livePaths = new Set<string>()
+
+	for (const pieceConfig of config.pieces) {
+		const items = await db
+			.selectFrom('pieces_items')
+			.leftJoin('pieces_cache', 'pieces_cache.file_path', 'pieces_items.file_path')
+			.leftJoin('web_pieces', 'web_pieces.file_path', 'pieces_items.file_path')
+			.select([
+				'pieces_items.file_path as filePath',
+				'pieces_cache.content_hash as indexedHash',
+				'web_pieces.content_hash as publishedHash',
+				'web_pieces.id as publishedId',
+			])
+			.where('pieces_items.type', '=', pieceConfig.type)
+			.orderBy('pieces_items.file_path')
+			.execute()
+
+		for (const item of items) {
+			livePaths.add(item.filePath)
+			if (item.indexedHash === null) continue
+			if (item.indexedHash === item.publishedHash) continue
+
+			pending.pieces.push({ filePath: item.filePath, contentHash: item.indexedHash })
+			if (item.publishedId === null) {
+				pending.diff.added.push(item.filePath)
+			} else {
+				pending.diff.updated.push(item.filePath)
+			}
+		}
+	}
+
+	const published = await db.selectFrom('web_pieces').select('file_path').execute()
+	for (const piece of published) {
+		if (!livePaths.has(piece.file_path)) {
+			pending.diff.pruned.push(piece.file_path)
+		}
+	}
+
+	return pending
+}

packages/web/worker/src/steps/assets-generate.test.ts

@@ -79,7 +79,7 @@ async function seedPiece(over: Partial<Record<string, unknown>> = {}) {
 		id: 'item-1',
 		key: 'k1',
 		title: 'T',
-		slug: 'great',
+		slug: String(over.id ?? 'great'),
 		file_path: 'books/great.md',
 		date_added: 1,
 		type: 'books',

packages/web/worker/src/steps/luzzle-audit.test.ts

@@ -3,6 +3,7 @@ import { luzzleAuditStep } from './luzzle-audit.js'
 import { Pieces, StorageFileSystem, getDatabaseClient } from '@luzzle/core'
 import type { WorkerContext } from '../services/context.js'
 import type { Config } from '@luzzle/web.config'
+import { getPendingPublication } from '../services/publication.js'
 
 vi.mock('@luzzle/core', () => ({
 	Pieces: vi.fn(),
@@ -10,6 +11,8 @@ vi.mock('@luzzle/core', () => ({
 	getDatabaseClient: vi.fn(),
 }))
 
+vi.mock('../services/publication.js', () => ({ getPendingPublication: vi.fn() }))
+
 vi.mock('../services/db.js', () => ({
 	resolveDbPath: vi.fn(() => '/app/data/db.sqlite'),
 }))
@@ -42,6 +45,10 @@ function makeCtx(): WorkerContext {
 describe('luzzleAuditStep', () => {
 	beforeEach(() => {
 		vi.clearAllMocks()
+		vi.mocked(getPendingPublication).mockResolvedValue({
+			pieces: [],
+			diff: { added: [], updated: [], pruned: [] },
+		})
 	})
 
 	test('runs pieces.diff and returns the structured diff', async () => {
@@ -66,6 +73,34 @@ describe('luzzleAuditStep', () => {
 		}
 	})
 
+	test('includes indexed but unpublished changes without duplicating source changes', async () => {
+		const ctx = makeCtx()
+		const sourceDiff = {
+			schemas: { added: [], updated: [], pruned: [] },
+			pieces: { added: ['new.md'], updated: ['changed.md'], pruned: [] },
+		}
+		mocks.StorageFileSystem.mockReturnValue({} as never)
+		mocks.getDatabaseClient.mockReturnValue({} as never)
+		mocks.Pieces.mockReturnValue({ diff: vi.fn().mockResolvedValue(sourceDiff) } as never)
+		vi.mocked(getPendingPublication).mockResolvedValue({
+			pieces: [],
+			diff: { added: ['new.md'], updated: ['changed.md', 'failed.md'], pruned: ['gone.md'] },
+		})
+
+		const result = await luzzleAuditStep.run(undefined, ctx)
+
+		expect(getPendingPublication).toHaveBeenCalledWith(ctx.db, ctx.config)
+		expect(result).toEqual({
+			status: 'completed',
+			value: {
+				schemas: sourceDiff.schemas,
+				pieces: { added: ['new.md'], updated: ['changed.md', 'failed.md'], pruned: ['gone.md'] },
+			},
+			message: '4 pending change(s)',
+		})
+		expect(sourceDiff.pieces.updated).toEqual(['changed.md'])
+	})
+
 	test('reports zero pending changes for an empty diff', async () => {
 		const ctx = makeCtx()
 		const diff = {

packages/web/worker/src/steps/luzzle-audit.ts

@@ -1,6 +1,8 @@
 import { Pieces, StorageFileSystem, getDatabaseClient, type PiecesDiff } from '@luzzle/core'
 import { completed, type Step, type StepResult } from '../core/step.js'
 import { resolveDbPath } from '../services/db.js'
+import { getPendingPublication } from '../services/publication.js'
+import { includePendingPieces } from '../workflows/pieces-diff.js'
 
 export const luzzleAuditStep: Step<void, PiecesDiff> = {
 	name: 'luzzle.audit',
@@ -13,7 +15,9 @@ export const luzzleAuditStep: Step<void, PiecesDiff> = {
 		const db = getDatabaseClient(dbPath)
 
 		const pieces = new Pieces(storage)
-		const diff = await pieces.diff(db)
+		const sourceDiff = await pieces.diff(db)
+		const pending = await getPendingPublication(ctx.db, config)
+		const diff = includePendingPieces(sourceDiff, pending.diff)
 
 		const changedCount =
 			diff.pieces.added.length +

packages/web/worker/src/steps/publish-complete.ts

@@ -0,0 +1,30 @@
+import type { PublishPieceFailure } from '@luzzle/web.jobs'
+import { completed, type Step } from '../core/step.js'
+import type { PublicationPiece } from '../services/publication.js'
+
+export interface PublishCompleteInput {
+	pieces: PublicationPiece[]
+	failedPieces: PublishPieceFailure[]
+}
+
+export const publishCompleteStep: Step<PublishCompleteInput, void> = {
+	name: 'publish.complete',
+	async run({ pieces, failedPieces }, { db, logger }) {
+		const failedPaths = new Set(failedPieces.map((piece) => piece.filePath))
+		let count = 0
+
+		for (const piece of pieces) {
+			if (failedPaths.has(piece.filePath)) continue
+
+			await db
+				.updateTable('web_pieces')
+				.set({ content_hash: piece.contentHash })
+				.where('file_path', '=', piece.filePath)
+				.execute()
+			count += 1
+		}
+
+		logger.info('publish.complete recorded published hashes', { count })
+		return completed(undefined)
+	},
+}

packages/web/worker/src/steps/publish-prepare.ts

@@ -0,0 +1,21 @@
+import type { PiecesDiff } from '@luzzle/core'
+import { completed, type Step } from '../core/step.js'
+import { getPendingPublication, type PublicationPiece } from '../services/publication.js'
+import { includePendingPieces } from '../workflows/pieces-diff.js'
+
+export interface PublishPlan {
+	pieces: PublicationPiece[]
+	summary: PiecesDiff
+}
+
+export const publishPrepareStep: Step<PiecesDiff, PublishPlan> = {
+	name: 'publish.prepare',
+	async run(summary, { db, config, logger }) {
+		const pending = await getPendingPublication(db, config)
+		logger.info('publish.prepare complete', { count: pending.pieces.length })
+		return completed({
+			pieces: pending.pieces,
+			summary: includePendingPieces(summary, pending.diff),
+		})
+	},
+}

packages/web/worker/src/steps/web-sync/index.ts

@@ -110,6 +110,7 @@ async function syncOne(
 		.values(webPiece)
 		.onConflict((oc) =>
 			oc.column('id').doUpdateSet({
+				content_hash: null,
 				title: webPiece.title,
 				summary: webPiece.summary,
 				note: webPiece.note,

packages/web/worker/src/workflows/pieces-diff.ts

@@ -1,5 +1,24 @@
 import type { PiecesDiff } from '@luzzle/core'
 
+export function includePendingPieces(diff: PiecesDiff, pending: PiecesDiff['pieces']): PiecesDiff {
+	const pieces = {
+		added: [...diff.pieces.added],
+		updated: [...diff.pieces.updated],
+		pruned: [...diff.pieces.pruned],
+	}
+	const reportedPaths = new Set([...pieces.added, ...pieces.updated, ...pieces.pruned])
+
+	for (const action of ['added', 'updated', 'pruned'] as const) {
+		for (const filePath of pending[action]) {
+			if (reportedPaths.has(filePath)) continue
+			pieces[action].push(filePath)
+			reportedPaths.add(filePath)
+		}
+	}
+
+	return { schemas: diff.schemas, pieces }
+}
+
 export function emptyPiecesDiff(): PiecesDiff {
 	return {
 		schemas: { added: [], updated: [], pruned: [] },

packages/web/worker/src/workflows/publish-recovery.test.ts

@@ -0,0 +1,403 @@
+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 type { Workflow } from 'openworkflow'
+import { completed, type StepResult } from '../core/step.js'
+import { RcloneClient } from '../services/rclone.js'
+import { setWorkerContext, type WorkerContext } from '../services/context.js'
+import { getPendingPublication } from '../services/publication.js'
+import { luzzleSyncStep } from '../steps/luzzle-sync.js'
+import { luzzleAuditStep } from '../steps/luzzle-audit.js'
+import { publishPrepareStep } from '../steps/publish-prepare.js'
+import { 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]
+
+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()
+		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 () => {
+		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 }
+	}
+
+	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'])
+			.where('file_path', '=', filePath)
+			.executeTakeFirstOrThrow()
+		return {
+			indexed: indexed.content_hash,
+			published: published.content_hash,
+			title: published.title,
+		}
+	}
+
+	async function audit(): Promise<PiecesDiff> {
+		return valueOf(await luzzleAuditStep.run(undefined, ctx))
+	}
+
+	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(initialDiff, ctx))
+		await webSyncStep.run({ filePaths: initialPlan.pieces.map((piece) => piece.filePath) }, ctx)
+		expect((await hashes(healthy)).published).toBeNull()
+		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(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((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)
+		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)
+		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((await audit()).pieces).toEqual({ added: [], updated: [broken], pruned: [] })
+
+		mocks.assets.mockClear()
+		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)).published).toBe(brokenState.indexed)
+		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)
+			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)).published).toBeNull()
+			expect((await hashes(broken)).published).toBeNull()
+			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()
+			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(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',
+		})
+		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',
+			})
+			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([])
+	})
+})

packages/web/worker/src/workflows/publish.test.ts

@@ -26,6 +26,8 @@ const mocks = vi.hoisted(() => {
 		assetsGenerate: jobStep('assets.generate'),
 		cdnSync: jobStep('cdn.sync'),
 		cachePurge: jobStep('cache.purge'),
+		publishPrepare: jobStep('publish.prepare'),
+		publishComplete: jobStep('publish.complete'),
 	}
 })
 
@@ -49,6 +51,8 @@ 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'] },
@@ -115,11 +119,23 @@ describe('workflows/publish', () => {
 			mocks.assetsGenerate,
 			mocks.cdnSync,
 			mocks.cachePurge,
+			mocks.publishPrepare,
+			mocks.publishComplete,
 		]) {
 			jobStep.run.mockResolvedValue({ status: 'completed', value: 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,
+			},
+		})
 	})
 
 	test('reports failed pieces in JSON while publishing healthy siblings and retaining the source diff', async () => {
@@ -130,10 +146,12 @@ describe('workflows/publish', () => {
 
 		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'] },
@@ -167,8 +185,12 @@ describe('workflows/publish', () => {
 			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.info).not.toHaveBeenCalledWith(
+				'openworkflow publish complete',
+				expect.anything()
+			)
 			expect(mocks.logger.warn).not.toHaveBeenCalled()
+			expect(mocks.publishComplete.run).not.toHaveBeenCalled()
 		}
 	)
 
@@ -185,17 +207,23 @@ describe('workflows/publish', () => {
 		expect(phases).toEqual([
 			'archive.sync',
 			'luzzle.sync',
+			'publish.prepare',
 			'web.sync',
 			'assets.generate',
 			'cdn.sync',
 			'cache.purge',
+			'publish.complete',
 		])
 		expect(result).toEqual({ ...summary, failedPieces: [] })
 		expectHealthyPhasesComplete()
 		expectNormalCompletion()
 	})
 
-	test('returns an empty diff and report for an empty/older luzzle checkpoint', async () => {
+	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() },
+		})
 		const { result } = await runPublish({}, new Map([['luzzle.sync', undefined]]))
 		expect(result).toEqual({ ...emptyPiecesDiff(), failedPieces: [] })
 		expect(mocks.luzzleSync.run).not.toHaveBeenCalled()

packages/web/worker/src/workflows/publish.ts

@@ -9,6 +9,8 @@ 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 {
@@ -28,7 +30,16 @@ export function registerPublishWorkflow(): void {
 		}
 
 		const summary = await runProgressPhase(step, ctx, jobId, progress, luzzleSyncStep, undefined)
-		const changedPaths = [...(summary?.pieces.added ?? []), ...(summary?.pieces.updated ?? [])]
+		const plan = await runProgressPhase(
+			step,
+			ctx,
+			jobId,
+			progress,
+			publishPrepareStep,
+			summary ?? emptyPiecesDiff()
+		)
+		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, {
@@ -38,6 +49,11 @@ export function registerPublishWorkflow(): void {
 		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,
@@ -46,6 +62,6 @@ export function registerPublishWorkflow(): void {
 		} else {
 			logger.info('openworkflow publish complete', { jobId })
 		}
-		return { ...(summary ?? emptyPiecesDiff()), failedPieces }
+		return { ...plan.summary, failedPieces }
 	})
 }

packages/web/worker/test/db.sql

@@ -0,0 +1,89 @@
+-- Generated by pnpm --filter @luzzle/web.db schema:generate. Do not edit.
+-- Statement boundaries let the test helper load complete triggers without parsing SQL.
+
+CREATE TABLE "job_progress" ("job_id" text not null, "phase" text not null, "status" text not null, "started_at" integer not null, "finished_at" integer, "message" text, constraint "job_progress_pk" primary key ("job_id", "phase"));
+
+-- statement-breakpoint
+
+CREATE TABLE "job_progress_logs" ("job_id" text not null, "phase" text not null, "line_number" integer not null, "ts" integer not null, "level" text not null, "message" text not null, constraint "job_progress_logs_pk" primary key ("job_id", "phase", "line_number"));
+
+-- statement-breakpoint
+
+CREATE TABLE "pieces_cache" ("id" text not null primary key, "file_path" text not null, "content_hash" text, "date_added" datetime default ((julianday('now') - 2440587.5)*86400000) not null, "date_updated" datetime, constraint "unique-files" unique ("file_path"));
+
+-- statement-breakpoint
+
+CREATE TABLE "pieces_items" ("id" text not null primary key, "date_added" datetime default ((julianday('now') - 2440587.5)*86400000) not null, "date_updated" datetime, "file_path" text not null, "type" text not null, "note_markdown" text, "frontmatter_json" text, "assets_json_array" text, constraint "unique-files" unique ("file_path"));
+
+-- statement-breakpoint
+
+CREATE TABLE "pieces_manager" ("id" text not null primary key, "date_added" datetime default ((julianday('now') - 2440587.5)*86400000) not null, "date_updated" datetime, "name" text not null unique, "schema" text not null);
+
+-- statement-breakpoint
+
+CREATE TABLE "web_pieces" ("id" text not null primary key, "key" text not null, "slug" text not null, "type" text not null, "file_path" text not null, "title" text, "summary" text, "note" text, "keywords" text, "json_metadata" text, "date_added" datetime, "date_updated" datetime, "date_consumed" datetime, "content_hash" text, constraint "slug-type" unique ("slug", "type"));
+
+-- statement-breakpoint
+
+CREATE TABLE "web_pieces_assets" (
+		piece_file_path TEXT NOT NULL,
+		piece_key TEXT NOT NULL,
+		piece_asset_path TEXT DEFAULT NULL,
+		transformation TEXT NOT NULL,
+		asset_path TEXT,
+		piece_field_path TEXT,
+		asset_key TEXT NOT NULL DEFAULT '',
+		mime_type TEXT NOT NULL,
+		is_embedded BOOLEAN DEFAULT 0,
+		content TEXT,
+		PRIMARY KEY (piece_file_path, transformation, piece_asset_path)
+	);
+
+-- statement-breakpoint
+
+CREATE VIRTUAL TABLE "web_pieces_fts5" USING fts5(id UNINDEXED, key UNINDEXED, slug, type UNINDEXED, title, summary, note, keywords, json_metadata, date_added UNINDEXED, date_updated UNINDEXED, date_consumed UNINDEXED, file_path UNINDEXED, tokenize = 'porter ascii', prefix='3 4 5', content = 'web_pieces', content_rowid="rowid");
+
+-- statement-breakpoint
+
+CREATE TABLE "web_pieces_tags" ("piece_type" text not null, "piece_slug" text not null, "piece_id" text not null, "tag" text not null, "slug" text not null);
+
+-- statement-breakpoint
+
+CREATE INDEX web_pieces_date_added_index ON web_pieces (date_added DESC);
+
+-- statement-breakpoint
+
+CREATE INDEX web_pieces_date_consumed_date_added_index ON web_pieces (date_consumed DESC, date_added DESC);
+
+-- statement-breakpoint
+
+CREATE INDEX web_pieces_type_date_consumed_date_added_index ON web_pieces (type, date_consumed DESC, date_added DESC);
+
+-- statement-breakpoint
+
+CREATE TRIGGER web_pieces_after_delete AFTER DELETE ON web_pieces
+	BEGIN
+		INSERT INTO web_pieces_fts5(web_pieces_fts5, rowid, key, slug, title, summary, note, keywords, json_metadata)
+		VALUES('delete', old.rowid, old.key, old.slug, old.title, old.summary, old.note, old.keywords, old.json_metadata);
+
+		DELETE FROM web_pieces_tags WHERE piece_id = old.id;
+		DELETE FROM web_pieces_assets WHERE piece_file_path = old.file_path;
+	END;
+
+-- statement-breakpoint
+
+CREATE TRIGGER web_pieces_after_insert AFTER INSERT ON web_pieces		BEGIN
+		INSERT INTO web_pieces_fts5(rowid, key, slug, type, title, summary, note, keywords, json_metadata, date_added, date_updated, date_consumed)
+		VALUES(new.rowid, new.key, new.slug, new.type, new.title, new.summary, new.note, new.keywords, new.json_metadata, new.date_added, new.date_updated, new.date_consumed);
+	END;
+
+-- statement-breakpoint
+
+CREATE TRIGGER web_pieces_after_update AFTER UPDATE ON web_pieces
+	BEGIN
+		INSERT INTO web_pieces_fts5(web_pieces_fts5, rowid, key, slug, title, summary, note, keywords, json_metadata)
+		VALUES('delete', old.rowid, old.key, old.slug, old.title, old.summary, old.note, old.keywords, old.json_metadata);
+
+		INSERT INTO web_pieces_fts5(rowid, key, slug, title, summary, note, keywords, json_metadata)
+		VALUES (new.rowid, new.key, new.slug, new.title, new.summary, new.note, new.keywords, new.json_metadata);
+	END;

packages/web/worker/test/db.ts

@@ -1,5 +1,7 @@
+import { readFileSync } from 'node:fs'
 import { getDatabaseClient } from '@luzzle/core'
-import { Kysely, sql } from 'kysely'
+import { sql } from 'kysely'
+import type { Kysely } from 'kysely'
 import type { WebDatabase } from '../src/services/db.js'
 
 let cachedDb: ReturnType<typeof getDatabaseClient> | null = null
@@ -7,86 +9,10 @@ let cachedDb: ReturnType<typeof getDatabaseClient> | null = null
 export async function setupDatabase(): Promise<Kysely<WebDatabase>> {
 	if (!cachedDb) {
 		cachedDb = getDatabaseClient(':memory:')
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS pieces_items (
-				id TEXT PRIMARY KEY,
-				file_path TEXT NOT NULL,
-				type TEXT NOT NULL,
-				date_added INTEGER NOT NULL,
-				date_updated INTEGER,
-				note_markdown TEXT,
-				frontmatter_json TEXT NOT NULL DEFAULT '{}',
-				assets_json_array TEXT
-			)
-		`.execute(cachedDb)
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS web_pieces (
-				id TEXT PRIMARY KEY,
-				key TEXT NOT NULL,
-				title TEXT NOT NULL,
-				slug TEXT NOT NULL,
-				file_path TEXT NOT NULL,
-				note TEXT,
-				date_updated INTEGER,
-				date_added INTEGER NOT NULL,
-				date_consumed INTEGER,
-				type TEXT NOT NULL,
-				json_metadata TEXT NOT NULL DEFAULT '{}',
-				summary TEXT,
-				keywords TEXT
-			)
-		`.execute(cachedDb)
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS web_pieces_tags (
-				piece_slug TEXT NOT NULL,
-				piece_type TEXT NOT NULL,
-				piece_id TEXT NOT NULL,
-				tag TEXT NOT NULL,
-				slug TEXT NOT NULL
-			)
-		`.execute(cachedDb)
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS web_pieces_assets (
-				piece_file_path TEXT NOT NULL,
-				piece_key TEXT NOT NULL,
-				piece_asset_path TEXT,
-				piece_field_path TEXT,
-				asset_key TEXT NOT NULL,
-				transformation TEXT NOT NULL,
-				asset_path TEXT,
-				mime_type TEXT NOT NULL,
-				is_embedded INTEGER DEFAULT 0,
-				content TEXT
-			)
-		`.execute(cachedDb)
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS job_progress (
-				job_id TEXT NOT NULL,
-				phase TEXT NOT NULL,
-				status TEXT NOT NULL,
-				started_at INTEGER NOT NULL,
-				finished_at INTEGER,
-				message TEXT,
-				PRIMARY KEY (job_id, phase)
-			)
-		`.execute(cachedDb)
-
-		await sql`
-			CREATE TABLE IF NOT EXISTS job_progress_logs (
-				job_id TEXT NOT NULL,
-				phase TEXT NOT NULL,
-				line_number INTEGER NOT NULL,
-				ts INTEGER NOT NULL,
-				level TEXT NOT NULL,
-				message TEXT NOT NULL,
-				PRIMARY KEY (job_id, phase, line_number)
-			)
-		`.execute(cachedDb)
+		const schema = readFileSync(new URL('./db.sql', import.meta.url), 'utf8')
+		for (const statement of schema.split('-- statement-breakpoint')) {
+			await sql.raw(statement).execute(cachedDb)
+		}
 	}
 
 	await sql`BEGIN`.execute(cachedDb)