.changeset/unpublish-worker-operation.md

@@ -0,0 +1,9 @@
+---
+"@luzzle/web.jobs": patch
+"@luzzle/web.worker": patch
+---
+
+Add a one-attempt Unpublish worker operation that withdraws the public record and
+removes its recorded, unshared generated files. Preserve archive originals and
+report cleanup failures without automatic recovery. Reuse shared rclone remote
+configuration and generated-file validation helpers.

packages/web/jobs/src/index.ts

@@ -13,6 +13,7 @@ export {
 	previewSpec,
 	publishSpec,
 	publishAuditSpec,
+	unpublishSpec,
 	testConnectivitySpec,
 } from './specs.js'
 export type {
@@ -21,6 +22,8 @@ export type {
 	PublishPieceFailure,
 	PublishResult,
 	PublishAuditPayload,
+	UnpublishPayload,
+	UnpublishResult,
 	PreviewPayload,
 	PreviewAsset,
 	PreviewResult,

packages/web/jobs/src/specs.ts

@@ -8,6 +8,8 @@ import type {
 	PublishPayload,
 	PublishResult,
 	PublishAuditPayload,
+	UnpublishPayload,
+	UnpublishResult,
 	TestConnectivityPayload,
 	TestConnectivityResult,
 } from './types.js'
@@ -39,6 +41,10 @@ export const publishAuditSpec = defineWorkflowSpec<PublishAuditPayload, PiecesDi
 	name: 'PublishAudit',
 })
 
+export const unpublishSpec = defineWorkflowSpec<UnpublishPayload, UnpublishResult>({
+	name: 'Unpublish',
+})
+
 export const testConnectivitySpec = defineWorkflowSpec<
 	TestConnectivityPayload,
 	TestConnectivityResult

packages/web/jobs/src/types.ts

@@ -29,6 +29,12 @@ export interface PublishAuditPayload {
 	bisync?: boolean
 }
 
+export interface UnpublishPayload {
+	filePath: string
+}
+
+export type UnpublishResult = 'ok'
+
 export interface PreviewPayload {
 	filePath: string
 }

packages/web/worker/src/assets/files.test.ts

@@ -0,0 +1,54 @@
+import { promises as fs } from 'node:fs'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { afterEach, beforeEach, expect, test } from 'vitest'
+import { resolveGeneratedAssetFile } from './files.js'
+
+let directory: string
+let root: string
+
+beforeEach(async () => {
+	directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-asset-files-'))
+	root = path.join(directory, 'assets')
+	await fs.mkdir(root)
+})
+
+afterEach(async () => {
+	await fs.rm(directory, { recursive: true, force: true })
+})
+
+test('resolves a generated file under its configured root without changing it', async () => {
+	await fs.mkdir(path.join(root, 'books'))
+	const file = path.join(root, 'books/cover.png')
+	await fs.writeFile(file, 'generated')
+	expect(await resolveGeneratedAssetFile(root, 'books/cover.png')).toBe(file)
+	expect(await fs.readFile(file, 'utf8')).toBe('generated')
+})
+
+test('allows a missing local file so callers can still handle its remote copy', async () => {
+	expect(await resolveGeneratedAssetFile(root, 'books/missing.png')).toBe(
+		path.join(root, 'books/missing.png')
+	)
+})
+
+test.each(['../outside.png', '/outside.png'])(
+	'rejects unsafe generated path %j',
+	async (assetPath) => {
+		await expect(resolveGeneratedAssetFile(root, assetPath)).rejects.toThrow('asset path')
+	}
+)
+
+test('rejects symbolic links without following or changing their targets', async () => {
+	const outside = path.join(directory, 'outside')
+	await fs.mkdir(outside)
+	await fs.writeFile(path.join(outside, 'keep.png'), 'keep')
+	await fs.symlink(outside, path.join(root, 'linked'))
+	await expect(resolveGeneratedAssetFile(root, 'linked/keep.png')).rejects.toThrow('symbolic link')
+	expect(await fs.readFile(path.join(outside, 'keep.png'), 'utf8')).toBe('keep')
+})
+
+test('rejects directories where an asset file is expected', async () => {
+	await fs.mkdir(path.join(root, 'directory'))
+	await expect(resolveGeneratedAssetFile(root, 'directory')).rejects.toThrow('not a file')
+	expect(await fs.readdir(path.join(root, 'directory'))).toEqual([])
+})

packages/web/worker/src/assets/files.ts

@@ -0,0 +1,45 @@
+import { promises as fs } from 'node:fs'
+import type { Stats } from 'node:fs'
+import path from 'node:path'
+
+export function isMissingFile(error: unknown): boolean {
+	if (!(error instanceof Error)) return false
+	return 'code' in error && error.code === 'ENOENT'
+}
+
+export async function resolveGeneratedAssetFile(root: string, assetPath: string): Promise<string> {
+	if (path.posix.isAbsolute(assetPath) || path.win32.isAbsolute(assetPath)) {
+		throw new Error(`Generated asset path must be relative: ${assetPath}`)
+	}
+	if (assetPath.includes('\\') || assetPath.includes('\0')) {
+		throw new Error(`Invalid generated asset path: ${assetPath}`)
+	}
+	const segments = assetPath.split('/')
+	if (segments.some((segment) => segment === '' || segment === '.' || segment === '..')) {
+		throw new Error(`Invalid generated asset path: ${assetPath}`)
+	}
+
+	const target = path.resolve(root, assetPath)
+	let current = path.resolve(root)
+	for (const segment of segments) {
+		current = path.join(current, segment)
+		let entry: Stats
+		try {
+			entry = await fs.lstat(current)
+		} catch (error) {
+			if (isMissingFile(error)) break
+			throw error
+		}
+		if (entry.isSymbolicLink()) {
+			throw new Error(`Generated asset path contains a symbolic link: ${assetPath}`)
+		}
+		if (current === target) {
+			if (!entry.isFile()) throw new Error(`Generated asset is not a file: ${assetPath}`)
+			continue
+		}
+		if (!entry.isDirectory()) {
+			throw new Error(`Generated asset parent is not a directory: ${assetPath}`)
+		}
+	}
+	return target
+}

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

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import { describe, it, expect, vi, beforeEach } from 'vitest'
-import { RcloneClient } from './rclone.js'
+import { getRcloneRemoteOptions, RcloneClient } from './rclone.js'
+import type { Config } from '@luzzle/web.config'
 import type { Logger } from './logger.js'
 import { EventEmitter } from 'events'
 
@@ -32,6 +33,47 @@ async function mockSpawnSuccess() {
 	return spawnMock
 }
 
+describe('getRcloneRemoteOptions', () => {
+	function syncConfig(): Config['sync'] {
+		return {
+			config: '/config/rclone.conf',
+			archive: { remote: 'archive', path: 'sources/', flags: ['--no-traverse'] },
+			cdn: { remote: 'cdn', path: 'published/', flags: ['--dry-run'], strategy: 'copy' },
+		}
+	}
+
+	it.each(['archive', 'cdn'] as const)('maps the configured %s target', (target) => {
+		const sync = syncConfig()
+		expect(getRcloneRemoteOptions(sync, target)).toEqual({
+			remote: sync[target].remote,
+			remotePath: sync[target].path,
+			configPath: sync.config,
+			flags: sync[target].flags,
+		})
+	})
+
+	it('returns no target when both the name and path are empty', () => {
+		const sync = syncConfig()
+		sync.cdn.remote = ''
+		sync.cdn.path = ''
+		expect(getRcloneRemoteOptions(sync, 'cdn')).toBeNull()
+	})
+
+	it.each([
+		{ remote: 'cdn', path: '' },
+		{ remote: '', path: 'published/' },
+	])('leaves incomplete target policy to callers: %j', (target) => {
+		const sync = syncConfig()
+		Object.assign(sync.cdn, target)
+		expect(getRcloneRemoteOptions(sync, 'cdn')).toEqual({
+			remote: target.remote,
+			remotePath: target.path,
+			configPath: sync.config,
+			flags: sync.cdn.flags,
+		})
+	})
+})
+
 describe('RcloneClient flags passthrough', () => {
 	beforeEach(() => vi.clearAllMocks())
 

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

@@ -1,5 +1,6 @@
 import { spawn } from 'child_process'
 import type { Logger } from './logger.js'
+import type { Config } from '@luzzle/web.config'
 
 export interface RcloneBisyncOptions {
 	localPath: string
@@ -43,6 +44,17 @@ export interface RcloneCheckOptions {
 
 export type RcloneCheckResult = { ok: true } | { ok: false; reason: string }
 
+export function getRcloneRemoteOptions(sync: Config['sync'], target: 'archive' | 'cdn') {
+	const destination = sync[target]
+	if (!destination?.remote && !destination?.path) return null
+	return {
+		remote: destination?.remote ?? '',
+		remotePath: destination?.path ?? '',
+		configPath: sync.config,
+		flags: destination?.flags,
+	}
+}
+
 export class RcloneClient {
 	private readonly logger: Logger
 

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

@@ -0,0 +1,274 @@
+import { promises as fs } from 'node:fs'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { sql } from 'kysely'
+import { afterEach, beforeEach, expect, test, vi } from 'vitest'
+import type { WorkerContext } from './context.js'
+import type { AppDatabase } from './db.js'
+import { setupDatabase, teardownDatabase } from '../../test/db.js'
+import { makeLogger } from '../../test/logger.js'
+import { cachePurgeStep } from '../steps/cache-purge.js'
+import { unpublishPiece } from './unpublish.js'
+
+const filePath = 'books/withdraw.books.md'
+const otherPiece = 'books/keep.books.md'
+let directory: string
+let ctx: WorkerContext
+let deleteFile: ReturnType<typeof vi.fn>
+
+beforeEach(async () => {
+	directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-unpublish-'))
+	const assets = path.join(directory, 'assets')
+	const cache = path.join(directory, 'cache')
+	const archive = path.join(directory, 'archive')
+	for (const folder of [assets, cache, archive]) await fs.mkdir(folder)
+	deleteFile = vi.fn().mockResolvedValue(undefined)
+	ctx = {
+		db: (await setupDatabase()).withTables<AppDatabase>(),
+		config: {
+			paths: { assets, cache },
+			storage: { root: archive },
+			sync: {
+				config: '/config/rclone.conf',
+				cdn: { remote: 'cdn', path: 'published/', strategy: 'copy', flags: ['--no-traverse'] },
+			},
+		} as WorkerContext['config'],
+		logger: makeLogger(),
+		rclone: { deleteFile, sync: vi.fn(), copy: vi.fn() } as unknown as WorkerContext['rclone'],
+	}
+	await seedPiece(filePath, 'Withdrawn')
+})
+
+afterEach(async () => {
+	vi.restoreAllMocks()
+	await teardownDatabase(ctx.db)
+	await fs.rm(directory, { recursive: true, force: true })
+})
+
+async function seedPiece(file: string, title: string) {
+	await ctx.db
+		.insertInto('pieces_items')
+		.values({
+			id: file,
+			file_path: file,
+			type: 'books',
+			date_added: 1,
+			frontmatter_json: '{}',
+			note_markdown: 'source note',
+		})
+		.execute()
+	await ctx.db
+		.insertInto('pieces_cache')
+		.values({ id: file, file_path: file, content_hash: 'source-hash' })
+		.execute()
+	await ctx.db
+		.insertInto('web_pieces')
+		.values({
+			id: file,
+			file_path: file,
+			key: file,
+			slug: file,
+			type: 'books',
+			title,
+			date_added: 1,
+			json_metadata: '{}',
+		})
+		.execute()
+}
+
+async function recordAsset(
+	file: string,
+	assetPath: string | null,
+	transformation = 'image.original',
+	isEmbedded: 0 | 1 = 0
+) {
+	await ctx.db
+		.insertInto('web_pieces_assets')
+		.values({
+			piece_file_path: file,
+			piece_key: file,
+			piece_asset_path: assetPath,
+			asset_key: 'asset',
+			asset_path: assetPath,
+			transformation,
+			mime_type: 'image/png',
+			is_embedded: isEmbedded,
+		})
+		.execute()
+}
+
+async function writeAsset(assetPath: string) {
+	const target = path.join(ctx.config.paths.assets, assetPath)
+	await fs.mkdir(path.dirname(target), { recursive: true })
+	await fs.writeFile(target, 'generated')
+	return target
+}
+
+function published() {
+	return ctx.db
+		.selectFrom('web_pieces')
+		.select('file_path')
+		.where('file_path', '=', filePath)
+		.executeTakeFirst()
+}
+
+test('withdraws only the target public piece and its recorded unshared files', async () => {
+	await seedPiece(otherPiece, 'Retained')
+	const cover = await writeAsset('books/target/cover.png')
+	const shared = await writeAsset('books/shared.png')
+	const unrecorded = await writeAsset('books/target/unrecorded.png')
+	const inline = await writeAsset('books/target/inline.png')
+	await recordAsset(filePath, 'books/target/cover.png')
+	await recordAsset(filePath, 'books/target/cover.png', 'image.s.png')
+	await recordAsset(filePath, 'books/shared.png')
+	await recordAsset(otherPiece, 'books/shared.png')
+	await recordAsset(filePath, null, 'markdown')
+	await recordAsset(filePath, 'books/target/inline.png', 'inline', 1)
+	await ctx.db
+		.insertInto('web_pieces_tags')
+		.values({
+			piece_id: filePath,
+			piece_slug: filePath,
+			piece_type: 'books',
+			tag: 'test',
+			slug: 'test',
+		})
+		.execute()
+	const source = path.join(ctx.config.storage.root, 'source.md')
+	const original = path.join(ctx.config.storage.root, 'original.png')
+	await fs.writeFile(source, 'original Markdown')
+	await fs.writeFile(original, 'original attachment')
+	await fs.writeFile(path.join(ctx.config.paths.cache, 'page'), 'cached page')
+	const searchBefore = await sql`
+		SELECT file_path FROM web_pieces_fts5 WHERE web_pieces_fts5 MATCH 'Withdrawn'
+	`.execute(ctx.db)
+	expect(searchBefore.rows).toHaveLength(1)
+
+	await unpublishPiece(filePath, ctx)
+
+	expect(await published()).toBeUndefined()
+	const searchAfter = await sql`
+		SELECT file_path FROM web_pieces_fts5 WHERE web_pieces_fts5 MATCH 'Withdrawn'
+	`.execute(ctx.db)
+	expect(searchAfter.rows).toHaveLength(0)
+	expect(await ctx.db.selectFrom('web_pieces_tags').selectAll().execute()).toEqual([])
+	expect(await ctx.db.selectFrom('web_pieces_assets').selectAll().execute()).toMatchObject([
+		{ piece_file_path: otherPiece, asset_path: 'books/shared.png' },
+	])
+	expect(await ctx.db.selectFrom('web_pieces').select('file_path').execute()).toEqual([
+		{ file_path: otherPiece },
+	])
+	expect(await ctx.db.selectFrom('pieces_items').selectAll().execute()).toHaveLength(2)
+	expect(await ctx.db.selectFrom('pieces_cache').selectAll().execute()).toHaveLength(2)
+	await expect(fs.readFile(cover)).rejects.toMatchObject({ code: 'ENOENT' })
+	for (const retained of [shared, unrecorded, inline]) {
+		expect(await fs.readFile(retained, 'utf8')).toBe('generated')
+	}
+	expect(await fs.readFile(source, 'utf8')).toBe('original Markdown')
+	expect(await fs.readFile(original, 'utf8')).toBe('original attachment')
+	expect(await fs.readdir(ctx.config.paths.cache)).toEqual([])
+	expect(deleteFile).toHaveBeenCalledExactlyOnceWith({
+		remote: 'cdn',
+		remotePath: 'published/books/target/cover.png',
+		configPath: '/config/rclone.conf',
+		flags: ['--no-traverse'],
+	})
+	expect(ctx.rclone.sync).not.toHaveBeenCalled()
+	expect(ctx.rclone.copy).not.toHaveBeenCalled()
+})
+
+test('attempts remote deletion when the local file is already absent', async () => {
+	await recordAsset(filePath, 'books/target/missing.png')
+	await unpublishPiece(filePath, ctx)
+	expect(deleteFile).toHaveBeenCalledExactlyOnceWith(
+		expect.objectContaining({ remotePath: 'published/books/target/missing.png' })
+	)
+})
+
+test('works without a remote and without generated files', async () => {
+	ctx.config.sync.cdn.remote = ''
+	ctx.config.sync.cdn.path = ''
+	await unpublishPiece(filePath, ctx)
+	expect(await published()).toBeUndefined()
+	expect(deleteFile).not.toHaveBeenCalled()
+})
+
+test('deletes local files when no remote is configured', async () => {
+	ctx.config.sync.cdn.remote = ''
+	ctx.config.sync.cdn.path = ''
+	const target = await writeAsset('books/target/local.png')
+	await recordAsset(filePath, 'books/target/local.png')
+	await unpublishPiece(filePath, ctx)
+	await expect(fs.readFile(target)).rejects.toMatchObject({ code: 'ENOENT' })
+	expect(deleteFile).not.toHaveBeenCalled()
+})
+
+test('reports cleanup failures but attempts remaining local and remote files', async () => {
+	const first = await writeAsset('books/target/a.png')
+	const second = await writeAsset('books/target/b.png')
+	await recordAsset(filePath, 'books/target/a.png')
+	await recordAsset(filePath, 'books/target/b.png')
+	vi.spyOn(cachePurgeStep, 'run').mockRejectedValueOnce(new Error('cache unavailable'))
+	const unlink = fs.unlink
+	vi.spyOn(fs, 'unlink').mockImplementation(async (target) => {
+		if (target === first) throw new Error('local deletion refused')
+		return unlink(target)
+	})
+	deleteFile.mockRejectedValueOnce(new Error('remote deletion refused'))
+
+	await expect(unpublishPiece(filePath, ctx)).rejects.toThrow('Unpublish cleanup incomplete')
+
+	expect(await published()).toBeUndefined()
+	expect(await fs.readFile(first, 'utf8')).toBe('generated')
+	await expect(fs.readFile(second)).rejects.toMatchObject({ code: 'ENOENT' })
+	expect(deleteFile).toHaveBeenCalledTimes(2)
+	expect(deleteFile).toHaveBeenNthCalledWith(
+		2,
+		expect.objectContaining({ remotePath: 'published/books/target/b.png' })
+	)
+	for (const error of ['cache unavailable', 'local deletion refused', 'remote deletion refused']) {
+		expect(ctx.logger.error).toHaveBeenCalledWith(
+			'unpublish cleanup failed',
+			expect.objectContaining({ filePath, error })
+		)
+	}
+})
+
+test('validates recorded paths before withdrawing the public record', async () => {
+	await recordAsset(filePath, '../outside.png')
+	await expect(unpublishPiece(filePath, ctx)).rejects.toThrow('asset path')
+	expect(await published()).toBeDefined()
+	expect(deleteFile).not.toHaveBeenCalled()
+})
+
+test('does not withdraw the record when remote configuration is incomplete', async () => {
+	await recordAsset(filePath, 'books/target/missing.png')
+	ctx.config.sync.cdn.remote = ''
+	await expect(unpublishPiece(filePath, ctx)).rejects.toThrow('without a remote name')
+	expect(await published()).toBeDefined()
+	expect(deleteFile).not.toHaveBeenCalled()
+})
+
+test('stops before cleanup if public-record deletion fails', async () => {
+	const target = await writeAsset('books/target/cover.png')
+	await recordAsset(filePath, 'books/target/cover.png')
+	vi.spyOn(ctx.db, 'deleteFrom').mockImplementationOnce(() => {
+		throw new Error('database refused deletion')
+	})
+	const purge = vi.spyOn(cachePurgeStep, 'run')
+	await expect(unpublishPiece(filePath, ctx)).rejects.toThrow('database refused deletion')
+	expect(await published()).toBeDefined()
+	expect(await fs.readFile(target, 'utf8')).toBe('generated')
+	expect(purge).not.toHaveBeenCalled()
+	expect(deleteFile).not.toHaveBeenCalled()
+})
+
+test('reports an absent public record without claiming earlier cleanup succeeded', async () => {
+	const purge = vi.spyOn(cachePurgeStep, 'run')
+	await expect(unpublishPiece('books/missing.books.md', ctx)).rejects.toThrow(
+		'No published piece found'
+	)
+	expect(await published()).toBeDefined()
+	expect(purge).not.toHaveBeenCalled()
+	expect(deleteFile).not.toHaveBeenCalled()
+})

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

@@ -0,0 +1,98 @@
+import { promises as fs } from 'node:fs'
+import path from 'node:path'
+import type { WorkerContext } from './context.js'
+import { getRcloneRemoteOptions } from './rclone.js'
+import { isMissingFile, resolveGeneratedAssetFile } from '../assets/files.js'
+import { cachePurgeStep } from '../steps/cache-purge.js'
+
+async function findAssetDeletionCandidates(
+	filePath: string,
+	db: WorkerContext['db']
+): Promise<string[]> {
+	const assets = await db
+		.selectFrom('web_pieces_assets')
+		.select(['asset_path', 'is_embedded'])
+		.where('piece_file_path', '=', filePath)
+		.orderBy('asset_path')
+		.execute()
+	const paths = new Set<string>()
+	for (const asset of assets) {
+		if (asset.is_embedded) continue
+		if (!asset.asset_path) continue
+		paths.add(asset.asset_path)
+	}
+	if (paths.size === 0) return []
+
+	const shared = await db
+		.selectFrom('web_pieces_assets as asset')
+		.innerJoin('web_pieces as piece', 'piece.file_path', 'asset.piece_file_path')
+		.select('asset.asset_path')
+		.where('piece.file_path', '!=', filePath)
+		.where('asset.asset_path', 'in', [...paths])
+		.execute()
+	const sharedPaths = new Set(shared.map((asset) => asset.asset_path))
+	return [...paths].filter((assetPath) => !sharedPaths.has(assetPath))
+}
+
+export async function unpublishPiece(filePath: string, ctx: WorkerContext): Promise<void> {
+	if (typeof filePath !== 'string' || filePath.length === 0) {
+		throw new Error('Unpublish requires a filePath')
+	}
+	const { db, config, logger, rclone } = ctx
+	const piece = await db
+		.selectFrom('web_pieces')
+		.select('id')
+		.where('file_path', '=', filePath)
+		.executeTakeFirst()
+	if (!piece) throw new Error(`No published piece found: ${filePath}`)
+
+	const assetPaths = await findAssetDeletionCandidates(filePath, db)
+	const files: { assetPath: string; localPath: string }[] = []
+	for (const assetPath of assetPaths) {
+		const localPath = await resolveGeneratedAssetFile(config.paths.assets, assetPath)
+		files.push({ assetPath, localPath })
+	}
+	const remote = getRcloneRemoteOptions(config.sync, 'cdn')
+	if (files.length > 0 && remote) {
+		if (!remote.remote) throw new Error('Remote asset path is configured without a remote name')
+		if (!remote.remotePath) throw new Error('Remote asset name is configured without a path')
+		if (!remote.configPath) throw new Error('Remote asset deletion requires rclone configuration')
+	}
+
+	logger.info('unpublish targets selected', { filePath, assetPaths })
+	await db.deleteFrom('web_pieces').where('id', '=', piece.id).execute()
+	logger.info('unpublish public record removed', { filePath })
+
+	const failures: string[] = []
+	function recordFailure(target: string, error: unknown) {
+		let message = String(error)
+		if (error instanceof Error) message = error.message
+		failures.push(`${target}: ${message}`)
+		logger.error('unpublish cleanup failed', { filePath, target, error: message })
+	}
+
+	try {
+		await cachePurgeStep.run(undefined, ctx)
+	} catch (error) {
+		recordFailure(`cache ${config.paths.cache}`, error)
+	}
+
+	for (const file of files) {
+		try {
+			await fs.unlink(file.localPath)
+		} catch (error) {
+			if (!isMissingFile(error)) recordFailure(`local ${file.localPath}`, error)
+		}
+		if (!remote) continue
+		const remotePath = path.posix.join(remote.remotePath, file.assetPath)
+		try {
+			await rclone.deleteFile({ ...remote, remotePath })
+		} catch (error) {
+			recordFailure(`remote ${remote.remote}:${remotePath}`, error)
+		}
+	}
+
+	if (failures.length > 0) {
+		throw new Error(`Unpublish cleanup incomplete for ${filePath}:\n${failures.join('\n')}`)
+	}
+}

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

@@ -1,26 +1,26 @@
 import path from 'node:path'
 import { mkdir, readdir } from 'node:fs/promises'
 import { completed, skipped, type Step, type StepResult } from '../core/step.js'
+import { getRcloneRemoteOptions } from '../services/rclone.js'
 
 export const archiveSyncStep: Step<void, void> = {
 	name: 'archive.sync',
 	async run(_input, ctx): Promise<StepResult<void>> {
 		const { config, logger, rclone } = ctx
 
-		const remote = config.sync.archive?.remote
-		const remotePath = config.sync.archive?.path
-		const configPath = config.sync.config
+		const options = getRcloneRemoteOptions(config.sync, 'archive')
 
-		if (!remote || !remotePath) {
+		if (!options?.remote || !options.remotePath) {
 			logger.info('archive.sync skipped: sync.archive.remote or sync.archive.path not configured')
 			return skipped('sync.archive.remote or sync.archive.path not configured')
 		}
 
-		if (!configPath) {
+		if (!options.configPath) {
 			logger.info('archive.sync skipped: sync.config not configured')
 			return skipped('sync.config not configured')
 		}
 
+		const { remote, remotePath } = options
 		const localPath = config.storage.root
 		const workdir = path.resolve(localPath, '..', 'rclone', 'bisync')
 
@@ -30,15 +30,7 @@ export const archiveSyncStep: Step<void, void> = {
 
 		logger.info('archive.sync starting bisync', { localPath, remote, remotePath, resync })
 
-		await rclone.bisync({
-			localPath,
-			remote,
-			remotePath,
-			configPath,
-			workdir,
-			resync,
-			flags: config.sync.archive?.flags,
-		})
+		await rclone.bisync({ ...options, localPath, workdir, resync })
 
 		logger.info('archive.sync complete')
 		return completed(undefined)

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

@@ -1,45 +1,33 @@
 import { completed, skipped, type Step, type StepResult } from '../core/step.js'
+import { getRcloneRemoteOptions } from '../services/rclone.js'
 
 export const cdnSyncStep: Step<void, void> = {
 	name: 'cdn.sync',
 	async run(_input, ctx): Promise<StepResult<void>> {
 		const { config, logger, rclone } = ctx
 
-		const remote = config.sync.cdn?.remote
-		const remotePath = config.sync.cdn?.path
-		const configPath = config.sync.config
+		const options = getRcloneRemoteOptions(config.sync, 'cdn')
 
-		if (!remote || !remotePath) {
+		if (!options?.remote || !options.remotePath) {
 			logger.info('cdn.sync skipped: sync.cdn.remote or sync.cdn.path not configured')
 			return skipped('sync.cdn.remote or sync.cdn.path not configured')
 		}
 
-		if (!configPath) {
+		if (!options.configPath) {
 			logger.info('cdn.sync skipped: sync.config not configured')
 			return skipped('sync.config not configured')
 		}
 
+		const { remote, remotePath } = options
 		const localPath = config.paths.assets
 		const strategy = config.sync.cdn?.strategy ?? 'sync'
 
 		logger.info(`cdn.sync starting ${strategy}`, { localPath, remote, remotePath, strategy })
 
 		if (strategy === 'copy') {
-			await rclone.copy({
-				localPath,
-				remote,
-				remotePath,
-				configPath,
-				flags: config.sync.cdn?.flags,
-			})
+			await rclone.copy({ ...options, localPath })
 		} else {
-			await rclone.sync({
-				localPath,
-				remote,
-				remotePath,
-				configPath,
-				flags: config.sync.cdn?.flags,
-			})
+			await rclone.sync({ ...options, localPath })
 		}
 
 		logger.info('cdn.sync complete')

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

@@ -3,11 +3,13 @@ import { registerPreviewWorkflow } from './preview.js'
 import { registerPublishWorkflow } from './publish.js'
 import { registerPublishAuditWorkflow } from './publishAudit.js'
 import { registerTestConnectivityWorkflow } from './testConnectivity.js'
+import { registerUnpublishWorkflow } from './unpublish.js'
 
 export function registerWorkflows(): void {
 	registerJobProgressPurgeWorkflow()
 	registerPreviewWorkflow()
 	registerPublishWorkflow()
 	registerPublishAuditWorkflow()
+	registerUnpublishWorkflow()
 	registerTestConnectivityWorkflow()
 }

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

@@ -2,6 +2,7 @@ import { testConnectivitySpec } from '@luzzle/web.jobs/specs'
 import { getOpenWorkflow } from '@luzzle/web.jobs'
 import type { TestConnectivityResult } from '@luzzle/web.jobs'
 import { getWorkerContext } from '../services/context.js'
+import { getRcloneRemoteOptions } from '../services/rclone.js'
 
 export function registerTestConnectivityWorkflow(): void {
 	const openWorkflow = getOpenWorkflow()
@@ -10,19 +11,18 @@ export function registerTestConnectivityWorkflow(): void {
 		testConnectivitySpec,
 		async ({ input }): Promise<TestConnectivityResult> => {
 			const { config, rclone, logger } = getWorkerContext()
-			const sync = input.target === 'archive' ? config.sync.archive : config.sync.cdn
-			const configPath = config.sync.config
+			const options = getRcloneRemoteOptions(config.sync, input.target)
 
-			if (!sync?.remote || !sync?.path || !configPath) {
+			if (!options?.remote || !options.remotePath || !options.configPath) {
 				return { ok: false, reason: `sync.${input.target} is not fully configured` }
 			}
 
-			logger.info(`connectivity check starting for ${input.target}`, { remote: sync.remote })
+			logger.info(`connectivity check starting for ${input.target}`, { remote: options.remote })
 
 			return rclone.checkConnectivity({
-				remote: sync.remote,
-				remotePath: sync.path,
-				configPath,
+				remote: options.remote,
+				remotePath: options.remotePath,
+				configPath: options.configPath,
 			})
 		}
 	)

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

@@ -0,0 +1,62 @@
+import { beforeEach, expect, test, vi } from 'vitest'
+import type { UnpublishPayload, UnpublishResult } from '@luzzle/web.jobs'
+import { unpublishSpec } from '@luzzle/web.jobs/specs'
+import { getWorkerContext } from '../services/context.js'
+import type { WorkerContext } from '../services/context.js'
+import { unpublishPiece } from '../services/unpublish.js'
+import { makeLogger } from '../../test/logger.js'
+import { registerUnpublishWorkflow } from './unpublish.js'
+
+const { implementWorkflow } = vi.hoisted(() => ({ implementWorkflow: vi.fn() }))
+vi.mock('@luzzle/web.jobs', () => ({
+	getOpenWorkflow: () => ({ implementWorkflow }),
+}))
+vi.mock('../services/context.js', () => ({ getWorkerContext: vi.fn() }))
+vi.mock('../services/unpublish.js', () => ({ unpublishPiece: vi.fn() }))
+
+type Handler = (args: {
+	input: UnpublishPayload
+	run: { id: string }
+}) => Promise<UnpublishResult>
+
+const filePath = 'books/withdraw.books.md'
+let ctx: WorkerContext
+
+beforeEach(() => {
+	vi.clearAllMocks()
+	ctx = { logger: makeLogger() } as WorkerContext
+	vi.mocked(getWorkerContext).mockReturnValue(ctx)
+	vi.mocked(unpublishPiece).mockResolvedValue(undefined)
+})
+
+function register(): Handler {
+	registerUnpublishWorkflow()
+	return implementWorkflow.mock.calls[0][1] as Handler
+}
+
+test('registers the declared Unpublish job without executing it', () => {
+	register()
+	expect(implementWorkflow).toHaveBeenCalledExactlyOnceWith(unpublishSpec, expect.any(Function))
+	expect(unpublishPiece).not.toHaveBeenCalled()
+})
+
+test('runs the worker operation once and reports completion', async () => {
+	const run = register()
+	expect(await run({ input: { filePath }, run: { id: 'job-1' } })).toBe('ok')
+	expect(unpublishPiece).toHaveBeenCalledExactlyOnceWith(filePath, ctx)
+	expect(ctx.logger.info).toHaveBeenCalledWith('unpublish complete', { jobId: 'job-1', filePath })
+})
+
+test('reports failure without retrying or claiming completion', async () => {
+	const error = new Error('remote deletion refused')
+	vi.mocked(unpublishPiece).mockRejectedValueOnce(error)
+	const run = register()
+	await expect(run({ input: { filePath }, run: { id: 'job-1' } })).rejects.toBe(error)
+	expect(unpublishPiece).toHaveBeenCalledTimes(1)
+	expect(ctx.logger.error).toHaveBeenCalledWith('unpublish failed', {
+		jobId: 'job-1',
+		filePath,
+		error: 'remote deletion refused',
+	})
+	expect(ctx.logger.info).not.toHaveBeenCalledWith('unpublish complete', expect.anything())
+})

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

@@ -0,0 +1,26 @@
+import { unpublishSpec } from '@luzzle/web.jobs/specs'
+import { getOpenWorkflow } from '@luzzle/web.jobs'
+import type { UnpublishResult } from '@luzzle/web.jobs'
+import { getWorkerContext } from '../services/context.js'
+import { unpublishPiece } from '../services/unpublish.js'
+
+export function registerUnpublishWorkflow(): void {
+	getOpenWorkflow().implementWorkflow(
+		unpublishSpec,
+		async ({ input, run }): Promise<UnpublishResult> => {
+			const ctx = getWorkerContext()
+			const fields = { jobId: run.id, filePath: input.filePath }
+			ctx.logger.info('unpublish starting', fields)
+			try {
+				await unpublishPiece(input.filePath, ctx)
+			} catch (error) {
+				let detail = String(error)
+				if (error instanceof Error) detail = error.message
+				ctx.logger.error('unpublish failed', { ...fields, error: detail })
+				throw error
+			}
+			ctx.logger.info('unpublish complete', fields)
+			return 'ok'
+		}
+	)
+}