diff --git a/.changeset/unpublish-worker-operation.md b/.changeset/unpublish-worker-operation.md new file mode 100644 index 00000000..6a789d6f --- /dev/null +++ b/.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. diff --git a/packages/web/jobs/src/index.ts b/packages/web/jobs/src/index.ts index c2e56967..8ad37030 100644 --- a/packages/web/jobs/src/index.ts +++ b/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, diff --git a/packages/web/jobs/src/specs.ts b/packages/web/jobs/src/specs.ts index cfddab1b..70875fcb 100644 --- a/packages/web/jobs/src/specs.ts +++ b/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({ + name: 'Unpublish', +}) + export const testConnectivitySpec = defineWorkflowSpec< TestConnectivityPayload, TestConnectivityResult diff --git a/packages/web/jobs/src/types.ts b/packages/web/jobs/src/types.ts index 1d0b776e..1ff1f87f 100644 --- a/packages/web/jobs/src/types.ts +++ b/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 } diff --git a/packages/web/worker/src/assets/files.test.ts b/packages/web/worker/src/assets/files.test.ts new file mode 100644 index 00000000..f98dd0f7 --- /dev/null +++ b/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([]) +}) diff --git a/packages/web/worker/src/assets/files.ts b/packages/web/worker/src/assets/files.ts new file mode 100644 index 00000000..dd76afc7 --- /dev/null +++ b/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 { + 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 +} diff --git a/packages/web/worker/src/services/rclone.test.ts b/packages/web/worker/src/services/rclone.test.ts index b98c6844..759667f5 100644 --- a/packages/web/worker/src/services/rclone.test.ts +++ b/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()) diff --git a/packages/web/worker/src/services/rclone.ts b/packages/web/worker/src/services/rclone.ts index 00c5cc7f..ce3b1d2d 100644 --- a/packages/web/worker/src/services/rclone.ts +++ b/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 diff --git a/packages/web/worker/src/services/unpublish.test.ts b/packages/web/worker/src/services/unpublish.test.ts new file mode 100644 index 00000000..f08d8383 --- /dev/null +++ b/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 + +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(), + 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() +}) diff --git a/packages/web/worker/src/services/unpublish.ts b/packages/web/worker/src/services/unpublish.ts new file mode 100644 index 00000000..d12cf3be --- /dev/null +++ b/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 { + 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() + 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 { + 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')}`) + } +} diff --git a/packages/web/worker/src/steps/archive-sync.ts b/packages/web/worker/src/steps/archive-sync.ts index 1e955c54..68a739bf 100644 --- a/packages/web/worker/src/steps/archive-sync.ts +++ b/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 = { name: 'archive.sync', async run(_input, ctx): Promise> { 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 = { 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) diff --git a/packages/web/worker/src/steps/cdn-sync.ts b/packages/web/worker/src/steps/cdn-sync.ts index ef7b44b8..012bfba3 100644 --- a/packages/web/worker/src/steps/cdn-sync.ts +++ b/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 = { name: 'cdn.sync', async run(_input, ctx): Promise> { 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') diff --git a/packages/web/worker/src/workflows/index.ts b/packages/web/worker/src/workflows/index.ts index e7fa1a09..908182d8 100644 --- a/packages/web/worker/src/workflows/index.ts +++ b/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() } diff --git a/packages/web/worker/src/workflows/testConnectivity.ts b/packages/web/worker/src/workflows/testConnectivity.ts index 9810ecd8..ec90bf93 100644 --- a/packages/web/worker/src/workflows/testConnectivity.ts +++ b/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 => { 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, }) } ) diff --git a/packages/web/worker/src/workflows/unpublish.test.ts b/packages/web/worker/src/workflows/unpublish.test.ts new file mode 100644 index 00000000..fe5c23c1 --- /dev/null +++ b/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 + +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()) +}) diff --git a/packages/web/worker/src/workflows/unpublish.ts b/packages/web/worker/src/workflows/unpublish.ts new file mode 100644 index 00000000..0396704c --- /dev/null +++ b/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 => { + 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' + } + ) +}