From 4755c18ef5e023aff682b4cafa4fa284dd942f97 Mon Sep 17 00:00:00 2001 From: "Claude Opus 4.8" Date: Fri, 10 Jul 2026 04:06:27 +0200 Subject: [PATCH] Add conversion job queue and pandoc sidecar integration (#62) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Import/export conversions run asynchronously against an internal pandoc-server sidecar with limits and graceful failure (ADR 0009). This is the plumbing; the import (#63) and export (#65) features enqueue jobs onto it. Sidecar & config: - pandoc/core:3.6 in HTTP server mode added to the Compose stack, internal network only, with a wget healthcheck on /version; the api depends on it healthy and reaches it via the new PANDOC_URL env (default http://pandoc:3030). - readyz gains a warning-level `converter` check: an unreachable sidecar degrades import/export but never flips the instance to unready (new `warn` status on ReadinessCheck). Conversion flow (apps/api/src/import-export/): - ConversionJob table (per-request work queue, distinct from the name-keyed maintenance Job table): owner, formats, input/result bytes, status, attempts, lockedAt. Migration + owner cascade. - PandocConverter (abstract) + PandocServerConverter: POST / with {text,from,to,standalone}; binary input formats (docx/odt/…) are base64-encoded in `text`; 60 s AbortController timeout; input/output size caps. Failures map to distinct localized codes — converter_unavailable / converter_timeout (retryable) and conversion_failed (final). - ConversionWorker: claims one job at a time with `FOR UPDATE SKIP LOCKED` (safe against overlapping sweeps and a second process), recovers a stale RUNNING lock, retries transient failures up to 3 attempts then fails. A 2 s sweep plus wake-on-enqueue means a queued job survives an API restart. - ConversionJobService.enqueue (size-limited) + owner-scoped GET /jobs/:id (poll) and GET /jobs/:id/result (stream the output); a foreign/unknown id is 404. ConversionJobView in @dorfteich/shared. Tests: - conversion-job.e2e.db.test.ts (fake converter injected via a new createTestApp override hook): enqueue→convert→poll→result; foreign/unknown job 404; a persisted PENDING job picked up by a fresh app's worker (restart survival); sidecar-down fails after 3 retries while the API stays healthy. - pandoc.converter.test.ts: success, non-200→conversion_failed, refused→ converter_unavailable, and a delay-injecting server→converter_timeout. - Verified locally against a real pandoc/core:3.6 container: markdown→html, markdown→docx (valid PK/OOXML bytes), and a docx→markdown round-trip. Local: typecheck, lint, i18n:check, build all green; api 193 tests (9 new), shared 121, web 50. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EwZ4jR4KFAPvpjWevfUGX1 --- .../migration.sql | 30 +++ apps/api/prisma/schema.prisma | 53 ++++- apps/api/src/app.module.ts | 2 + apps/api/src/health/readiness.service.ts | 38 ++- .../conversion-job.e2e.db.test.ts | 223 ++++++++++++++++++ .../import-export/conversion-job.service.ts | 110 +++++++++ .../conversion-worker.service.ts | 152 ++++++++++++ .../src/import-export/import-export.module.ts | 24 ++ apps/api/src/import-export/jobs.controller.ts | 34 +++ .../import-export/pandoc.converter.test.ts | 96 ++++++++ .../api/src/import-export/pandoc.converter.ts | 146 ++++++++++++ apps/api/src/testing/test-app.ts | 12 +- deploy/compose/docker-compose.yml | 19 ++ packages/shared/i18n/de/errors.json | 3 + packages/shared/i18n/en/errors.json | 3 + packages/shared/src/conversion.ts | 19 ++ packages/shared/src/env.ts | 7 + packages/shared/src/index.ts | 1 + 18 files changed, 958 insertions(+), 14 deletions(-) create mode 100644 apps/api/prisma/migrations/20260710035316_conversion_jobs/migration.sql create mode 100644 apps/api/src/import-export/conversion-job.e2e.db.test.ts create mode 100644 apps/api/src/import-export/conversion-job.service.ts create mode 100644 apps/api/src/import-export/conversion-worker.service.ts create mode 100644 apps/api/src/import-export/import-export.module.ts create mode 100644 apps/api/src/import-export/jobs.controller.ts create mode 100644 apps/api/src/import-export/pandoc.converter.test.ts create mode 100644 apps/api/src/import-export/pandoc.converter.ts create mode 100644 packages/shared/src/conversion.ts diff --git a/apps/api/prisma/migrations/20260710035316_conversion_jobs/migration.sql b/apps/api/prisma/migrations/20260710035316_conversion_jobs/migration.sql new file mode 100644 index 0000000..14eee4c --- /dev/null +++ b/apps/api/prisma/migrations/20260710035316_conversion_jobs/migration.sql @@ -0,0 +1,30 @@ +-- CreateEnum +CREATE TYPE "ConversionJobStatus" AS ENUM ('PENDING', 'RUNNING', 'SUCCEEDED', 'FAILED'); + +-- CreateTable +CREATE TABLE "conversion_jobs" ( + "id" TEXT NOT NULL, + "owner_id" TEXT NOT NULL, + "kind" TEXT NOT NULL, + "source_format" TEXT NOT NULL, + "target_format" TEXT NOT NULL, + "standalone" BOOLEAN NOT NULL DEFAULT true, + "input" BYTEA NOT NULL, + "status" "ConversionJobStatus" NOT NULL DEFAULT 'PENDING', + "attempts" INTEGER NOT NULL DEFAULT 0, + "result" BYTEA, + "result_mime_type" TEXT, + "error_code" TEXT, + "locked_at" TIMESTAMP(3), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "conversion_jobs_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "conversion_jobs_status_created_at_idx" ON "conversion_jobs"("status", "created_at"); + +-- AddForeignKey +ALTER TABLE "conversion_jobs" ADD CONSTRAINT "conversion_jobs_owner_id_fkey" FOREIGN KEY ("owner_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; + diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 6951326..a54cb94 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -42,12 +42,13 @@ model User { createdAt DateTime @default(now()) @map("created_at") lastLoginAt DateTime? @map("last_login_at") - identities UserIdentity[] - sessions Session[] - authTokens AuthToken[] - ponds Pond[] - pages Page[] - attachments Attachment[] + identities UserIdentity[] + sessions Session[] + authTokens AuthToken[] + ponds Pond[] + pages Page[] + attachments Attachment[] + conversionJobs ConversionJob[] @@map("users") } @@ -515,3 +516,43 @@ model Job { @@map("jobs") } + +enum ConversionJobStatus { + PENDING + RUNNING + SUCCEEDED + FAILED +} + +/// One import/export conversion (ADR 0009, issue #62). Unlike the name-keyed +/// maintenance `Job` table, this is a per-request work queue: a row is +/// enqueued PENDING, a worker claims it (`FOR UPDATE SKIP LOCKED`, `lockedAt` +/// recovers a crashed run), calls the pandoc sidecar with a timeout, and +/// stores the output bytes or an `errorCode`. `input`/`result` are the raw +/// document bytes — kept small by the request size limit and pruned by a +/// later maintenance job (they are transient, not the durable copy an +/// Attachment is). The polling endpoint `GET /jobs/:id` is owner-scoped. +model ConversionJob { + id String @id @default(uuid()) + ownerId String @map("owner_id") + /// Free-form label for the higher-level operation (e.g. 'export_docx', + /// 'import_docx') that later stories (#63/#65) set; #62 uses it only for logs. + kind String + sourceFormat String @map("source_format") + targetFormat String @map("target_format") + standalone Boolean @default(true) + input Bytes + status ConversionJobStatus @default(PENDING) + attempts Int @default(0) + result Bytes? + resultMimeType String? @map("result_mime_type") + errorCode String? @map("error_code") + lockedAt DateTime? @map("locked_at") + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @updatedAt @map("updated_at") + + owner User @relation(fields: [ownerId], references: [id], onDelete: Cascade) + + @@index([status, createdAt]) + @@map("conversion_jobs") +} diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 7a24d9c..2c0cf63 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -11,6 +11,7 @@ import { ConfigModule } from './config/config.module'; import { FilesModule } from './files/files.module'; import { GrantsModule } from './grants/grants.module'; import { HealthModule } from './health/health.module'; +import { ImportExportModule } from './import-export/import-export.module'; import { LabelsModule } from './labels/labels.module'; import { LinksModule } from './links/links.module'; import { MailModule } from './mail/mail.module'; @@ -48,6 +49,7 @@ import { VersionsModule } from './versions/versions.module'; GrantsModule, MembersModule, PublicModule, + ImportExportModule, AuthModule, AdminModule, LoggerModule.forRootAsync({ diff --git a/apps/api/src/health/readiness.service.ts b/apps/api/src/health/readiness.service.ts index 3053977..43136ac 100644 --- a/apps/api/src/health/readiness.service.ts +++ b/apps/api/src/health/readiness.service.ts @@ -1,10 +1,14 @@ import { Injectable } from '@nestjs/common'; +import { AppConfig } from '../config/app-config.service'; import { PrismaService } from '../prisma/prisma.service'; export interface ReadinessCheck { + /** `warn` reports a degraded-but-serving dependency: the instance still + * works, only some feature is unavailable. It never flips overall + * readiness to `unready` (that is reserved for `failed`). */ name: string; - status: 'ok' | 'failed'; + status: 'ok' | 'warn' | 'failed'; detail?: string; } @@ -13,27 +17,53 @@ export interface ReadinessReport { checks: ReadinessCheck[]; } +/** Converter reachability is a warning, not a failure — the pandoc probe is + * given a short budget so readyz stays fast even when the sidecar is down. */ +const CONVERTER_PROBE_TIMEOUT_MS = 2000; + @Injectable() export class ReadinessService { - constructor(private readonly prisma: PrismaService) {} + constructor( + private readonly prisma: PrismaService, + private readonly config: AppConfig, + ) {} /** * Readiness = the api can do real work: database reachable and all * migrations applied. Further checks (converters, backup freshness) are * added by later stories (issues #62, #85) — each as one more entry in - * the checks array, never as a separate endpoint. + * the checks array, never as a separate endpoint. Converter reachability + * is warning-level: import/export degrades, but the instance stays ready. */ async report(): Promise { const checks: ReadinessCheck[] = [ await this.databaseReachable(), await this.migrationsApplied(), + await this.converterReachable(), ]; return { - status: checks.every((c) => c.status === 'ok') ? 'ok' : 'unready', + status: checks.some((c) => c.status === 'failed') ? 'unready' : 'ok', checks, }; } + private async converterReachable(): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), CONVERTER_PROBE_TIMEOUT_MS); + try { + const response = await fetch(`${this.config.env.PANDOC_URL}/version`, { + signal: controller.signal, + }); + return response.ok + ? { name: 'converter', status: 'ok' } + : { name: 'converter', status: 'warn', detail: `pandoc returned ${response.status}` }; + } catch (error) { + return { name: 'converter', status: 'warn', detail: shortMessage(error) }; + } finally { + clearTimeout(timer); + } + } + private async databaseReachable(): Promise { try { await this.prisma.$queryRaw`SELECT 1`; diff --git a/apps/api/src/import-export/conversion-job.e2e.db.test.ts b/apps/api/src/import-export/conversion-job.e2e.db.test.ts new file mode 100644 index 0000000..c829278 --- /dev/null +++ b/apps/api/src/import-export/conversion-job.e2e.db.test.ts @@ -0,0 +1,223 @@ +import { INestApplication } from '@nestjs/common'; +import { PrismaClient } from '@prisma/client'; +import request from 'supertest'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { AuthTokensService } from '../auth/auth-tokens.service'; +import { createTestApp, sessionCookieOf } from '../testing/test-app'; +import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db'; +import { UsersService } from '../users/users.service'; + +import { ConversionJobService } from './conversion-job.service'; +import { ConversionWorker } from './conversion-worker.service'; +import { + ConversionError, + ConversionRequest, + ConversionResult, + PandocConverter, +} from './pandoc.converter'; + +/** + * Conversion job queue (issue #62): the enqueue → worker → poll flow, driven + * by an injected fake converter so no live pandoc sidecar is needed. The real + * PandocServerConverter's transport (timeout, error mapping) is covered by + * pandoc.converter.test.ts. + */ + +/** A converter whose behaviour each test sets: succeed with fixed bytes, or + * throw a chosen ConversionError (to exercise retries and failure). */ +class FakeConverter extends PandocConverter { + behaviour: (request: ConversionRequest) => ConversionResult = () => ({ + output: Buffer.from('CONVERTED'), + mimeType: 'application/octet-stream', + }); + calls = 0; + + convert(request: ConversionRequest): Promise { + this.calls += 1; + return Promise.resolve(this.behaviour(request)); + } + reachable(): Promise { + return Promise.resolve(true); + } +} + +describe.skipIf(!hasTestDb)('conversion job queue (e2e, issue #62)', () => { + let app: INestApplication; + let prisma: PrismaClient; + let jobs: ConversionJobService; + let worker: ConversionWorker; + let fake: FakeConverter; + const suffix = uniqueSuffix(); + const password = 'konvertiere meine dokumente 1'; + + const owner = { username: `carla-convert-${suffix}`, displayName: `Carla Convert ${suffix}` }; + const other = { username: `oscar-other-${suffix}`, displayName: `Oscar Other ${suffix}` }; + let ownerId: string; + let ownerCookie: string; + let otherCookie: string; + + const api = () => request(app.getHttpServer()); + + async function loginOf(username: string): Promise { + const res = await api() + .post('/api/v1/auth/login') + .send({ usernameOrEmail: username, password }) + .expect(200); + return sessionCookieOf(res); + } + + beforeAll(async () => { + prisma = createTestPrisma(); + await prisma.rateLimit.deleteMany({}); + fake = new FakeConverter(); + app = await createTestApp((builder) => + builder.overrideProvider(PandocConverter).useValue(fake), + ); + jobs = app.get(ConversionJobService); + worker = app.get(ConversionWorker); + + const users = app.get(UsersService); + const tokens = app.get(AuthTokensService); + const ownerUser = await users.createUser({ + username: owner.username, + email: `${owner.username}@example.org`, + displayName: owner.displayName, + password, + locale: 'en', + }); + ownerId = ownerUser.id; + const verify = await tokens.issue(ownerUser.id, 'EMAIL_VERIFICATION', 600); + await api().post('/api/v1/auth/verify-email').send({ token: verify }).expect(204); + ownerCookie = await loginOf(owner.username); + + const otherUser = await users.createUser({ + username: other.username, + email: `${other.username}@example.org`, + displayName: other.displayName, + password, + locale: 'en', + }); + await users.markEmailVerified(otherUser.id); + otherCookie = await loginOf(other.username); + }); + + afterAll(async () => { + await prisma.conversionJob.deleteMany({ where: { owner: { username: { contains: suffix } } } }); + // Verifying the owner's e-mail created a personal pond (+ owner-admin + // grant); clear those before the users they reference. + const where = { pond: { owner: { username: { contains: suffix } } } }; + await prisma.roleGrant.deleteMany({ where }); + await prisma.pond.deleteMany({ where: { owner: { username: { contains: suffix } } } }); + await prisma.user.deleteMany({ where: { username: { contains: suffix } } }); + await prisma.$disconnect(); + await app.close(); + }); + + it('enqueues, the worker converts, and the owner polls status and result', async () => { + fake.behaviour = () => ({ output: Buffer.from('DOCX-BYTES'), mimeType: 'application/x-test' }); + const job = await jobs.enqueue({ + ownerId, + kind: 'export_test', + from: 'markdown', + to: 'docx', + input: Buffer.from('# Hi'), + }); + + // Pending until the worker runs (wake() is a no-op under test). + const pending = await api() + .get(`/api/v1/jobs/${job.id}`) + .set('Cookie', ownerCookie) + .expect(200); + expect(pending.body.status).toBe('pending'); + + await worker.drain(); + + const done = await api().get(`/api/v1/jobs/${job.id}`).set('Cookie', ownerCookie).expect(200); + expect(done.body.status).toBe('succeeded'); + expect(done.body.errorCode).toBeNull(); + + const result = await api() + .get(`/api/v1/jobs/${job.id}/result`) + .set('Cookie', ownerCookie) + .expect(200); + expect(result.headers['content-type']).toContain('application/x-test'); + expect(result.headers['content-disposition']).toContain('attachment'); + expect(result.text).toBe('DOCX-BYTES'); + }); + + it('hides a foreign or unknown job (404, not 403)', async () => { + fake.behaviour = () => ({ output: Buffer.from('x'), mimeType: 'text/plain' }); + const job = await jobs.enqueue({ + ownerId, + kind: 'export_test', + from: 'markdown', + to: 'html', + input: Buffer.from('hi'), + }); + await worker.drain(); + + await api().get(`/api/v1/jobs/${job.id}`).set('Cookie', otherCookie).expect(404); + await api().get(`/api/v1/jobs/${job.id}/result`).set('Cookie', otherCookie).expect(404); + await api().get(`/api/v1/jobs/${crypto.randomUUID()}`).set('Cookie', ownerCookie).expect(404); + }); + + it('a queued job survives an API restart and completes', async () => { + // Persist a PENDING job directly (as if enqueued just before a crash), + // with no worker having touched it. + const persisted = await prisma.conversionJob.create({ + data: { + ownerId, + kind: 'export_test', + sourceFormat: 'markdown', + targetFormat: 'docx', + input: new Uint8Array(Buffer.from('# survives restart')), + }, + }); + expect(persisted.status).toBe('PENDING'); + + // A brand-new application = a fresh worker with empty memory, exactly like + // a real process restart. It must pick the persisted job up from the DB. + const restartFake = new FakeConverter(); + restartFake.behaviour = () => ({ + output: Buffer.from('AFTER-RESTART'), + mimeType: 'text/plain', + }); + const restarted = await createTestApp((builder) => + builder.overrideProvider(PandocConverter).useValue(restartFake), + ); + try { + await restarted.get(ConversionWorker).drain(); + const row = await prisma.conversionJob.findUniqueOrThrow({ where: { id: persisted.id } }); + expect(row.status).toBe('SUCCEEDED'); + expect(Buffer.from(row.result!).toString()).toBe('AFTER-RESTART'); + } finally { + await restarted.close(); + } + }); + + it('sidecar down → job fails after retries; the API stays healthy', async () => { + fake.calls = 0; + fake.behaviour = () => { + throw new ConversionError('converter_unavailable', true, 'sidecar down'); + }; + const job = await jobs.enqueue({ + ownerId, + kind: 'export_test', + from: 'markdown', + to: 'docx', + input: Buffer.from('# retry me'), + }); + + await worker.drain(); + + const row = await prisma.conversionJob.findUniqueOrThrow({ where: { id: job.id } }); + expect(row.status).toBe('FAILED'); + expect(row.errorCode).toBe('converter_unavailable'); + expect(row.attempts).toBe(3); // MAX_ATTEMPTS — retried, then given up + expect(fake.calls).toBe(3); + + // The API is unharmed by the failed conversion. + await api().get('/api/v1/healthz').expect(200); + }); +}); diff --git a/apps/api/src/import-export/conversion-job.service.ts b/apps/api/src/import-export/conversion-job.service.ts new file mode 100644 index 0000000..4667f25 --- /dev/null +++ b/apps/api/src/import-export/conversion-job.service.ts @@ -0,0 +1,110 @@ +import { Injectable, NotFoundException, PayloadTooLargeException } from '@nestjs/common'; +import { ConversionJob, ConversionJobStatus as PrismaStatus } from '@prisma/client'; +import { ConversionJobStatus, ConversionJobView } from '@dorfteich/shared'; +import { PinoLogger } from 'nestjs-pino'; + +import { PrismaService } from '../prisma/prisma.service'; + +import { ConversionWorker } from './conversion-worker.service'; +import { MAX_CONVERSION_INPUT_BYTES } from './pandoc.converter'; + +export interface EnqueueConversion { + ownerId: string; + kind: string; + from: string; + to: string; + input: Buffer; + standalone?: boolean; +} + +export interface ConversionResultPayload { + bytes: Buffer; + mimeType: string; +} + +const STATUS_VIEW: Record = { + PENDING: 'pending', + RUNNING: 'running', + SUCCEEDED: 'succeeded', + FAILED: 'failed', +}; + +/** + * Enqueues import/export conversions and answers owner-scoped polling + * (ADR 0009, issue #62). Enqueue persists the job (so it survives a restart) + * and wakes the worker for interactive latency; the worker + * ({@link ConversionWorker}) does the sidecar call out of band. + */ +@Injectable() +export class ConversionJobService { + constructor( + private readonly prisma: PrismaService, + private readonly worker: ConversionWorker, + private readonly logger: PinoLogger, + ) { + this.logger.setContext(ConversionJobService.name); + } + + async enqueue(request: EnqueueConversion): Promise { + if (request.input.byteLength > MAX_CONVERSION_INPUT_BYTES) { + throw new PayloadTooLargeException({ + code: 'file_too_large', + details: { limitBytes: MAX_CONVERSION_INPUT_BYTES }, + }); + } + const job = await this.prisma.conversionJob.create({ + data: { + ownerId: request.ownerId, + kind: request.kind, + sourceFormat: request.from, + targetFormat: request.to, + standalone: request.standalone ?? true, + // Prisma's Bytes maps to Uint8Array; a Node Buffer's + // backing store is ArrayBufferLike, so copy into a plain Uint8Array. + input: new Uint8Array(request.input), + }, + }); + this.logger.info( + { jobId: job.id, kind: job.kind, ownerId: request.ownerId }, + 'audit: conversion enqueued', + ); + this.worker.wake(); + return job; + } + + /** The job if it belongs to `userId`, else 404 — existence stays hidden from + * anyone but its owner (permissions.md §non-page objects). */ + async getForOwner(id: string, userId: string): Promise { + return this.viewOf(await this.ownedJob(id, userId)); + } + + /** The finished output bytes, or 404 while the job is not yet succeeded (so a + * caller cannot distinguish "still running" from "never existed"). */ + async resultForOwner(id: string, userId: string): Promise { + const job = await this.ownedJob(id, userId); + if (job.status !== 'SUCCEEDED' || !job.result) throw new NotFoundException(); + return { + bytes: Buffer.from(job.result), + mimeType: job.resultMimeType ?? 'application/octet-stream', + }; + } + + private async ownedJob(id: string, userId: string): Promise { + const job = await this.prisma.conversionJob.findFirst({ where: { id, ownerId: userId } }); + if (!job) throw new NotFoundException(); + return job; + } + + viewOf(job: ConversionJob): ConversionJobView { + return { + id: job.id, + status: STATUS_VIEW[job.status], + kind: job.kind, + sourceFormat: job.sourceFormat, + targetFormat: job.targetFormat, + errorCode: job.errorCode, + createdAt: job.createdAt.toISOString(), + updatedAt: job.updatedAt.toISOString(), + }; + } +} diff --git a/apps/api/src/import-export/conversion-worker.service.ts b/apps/api/src/import-export/conversion-worker.service.ts new file mode 100644 index 0000000..aaa9639 --- /dev/null +++ b/apps/api/src/import-export/conversion-worker.service.ts @@ -0,0 +1,152 @@ +import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { ConversionJob } from '@prisma/client'; +import { PinoLogger } from 'nestjs-pino'; + +import { AppConfig } from '../config/app-config.service'; +import { PrismaService } from '../prisma/prisma.service'; + +import { ConversionError, PandocConverter } from './pandoc.converter'; + +/** How often the worker sweeps for pending jobs on its own — the safety net + * that makes a queued conversion survive an API restart even if no new + * enqueue wakes it. Enqueues also wake it immediately (interactive latency). */ +const SWEEP_MS = 2000; +/** A job left RUNNING past this (crashed worker, never released) is treated as + * available again — comfortably longer than the 60 s conversion timeout. */ +const STALE_LOCK_MS = 5 * 60_000; +/** Transient failures (sidecar down / timed out) are retried up to this many + * total attempts before the job is marked failed. */ +const MAX_ATTEMPTS = 3; + +/** + * Drains the conversion job queue (ADR 0009, issue #62). Claims one PENDING + * job at a time with `FOR UPDATE SKIP LOCKED` (safe against a second worker + * and against its own overlapping sweeps), calls the pandoc sidecar with a + * timeout, and stores the output or a localizable error code. Transient + * failures are retried; a genuine conversion failure is final. Durability and + * restart-survival live in the row, not memory: a PENDING row is picked up by + * the next sweep, a crashed RUNNING row recovered once its lock goes stale. + */ +@Injectable() +export class ConversionWorker implements OnModuleInit, OnModuleDestroy { + private timer: NodeJS.Timeout | undefined; + private draining = false; + private wakeAgain = false; + + constructor( + private readonly prisma: PrismaService, + private readonly converter: PandocConverter, + private readonly config: AppConfig, + private readonly logger: PinoLogger, + ) { + this.logger.setContext(ConversionWorker.name); + } + + onModuleInit(): void { + if (this.config.env.NODE_ENV === 'test') return; // tests drive drain() directly + this.timer = setInterval(() => void this.drain(), SWEEP_MS); + this.timer.unref(); + } + + onModuleDestroy(): void { + if (this.timer) clearInterval(this.timer); + } + + /** Nudge the worker after an enqueue without blocking the caller. Coalesces + * concurrent wakes: a drain already in progress is asked to run once more. + * No-op under test, where tests drive {@link drain} deterministically. */ + wake(): void { + if (this.config.env.NODE_ENV === 'test') return; + void this.drain(); + } + + /** Process every claimable job, then stop. Re-entrant-safe: a second call + * while draining just flags one more pass instead of running in parallel. */ + async drain(): Promise { + if (this.draining) { + this.wakeAgain = true; + return; + } + this.draining = true; + try { + do { + this.wakeAgain = false; + for (let job = await this.claimNext(); job; job = await this.claimNext()) { + await this.process(job); + } + } while (this.wakeAgain); + } finally { + this.draining = false; + } + } + + /** Atomically claim the oldest available job (PENDING, or a RUNNING one + * whose lock has gone stale), or return null when the queue is drained. */ + private async claimNext(): Promise { + const staleBefore = new Date(Date.now() - STALE_LOCK_MS); + const claimed = await this.prisma.$queryRaw<{ id: string }[]>` + UPDATE conversion_jobs + SET status = 'RUNNING', locked_at = now(), attempts = attempts + 1 + WHERE id = ( + SELECT id FROM conversion_jobs + WHERE status = 'PENDING' + OR (status = 'RUNNING' AND locked_at < ${staleBefore}) + ORDER BY created_at + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING id`; + const id = claimed[0]?.id; + if (!id) return null; + return this.prisma.conversionJob.findUnique({ where: { id } }); + } + + private async process(job: ConversionJob): Promise { + try { + const result = await this.converter.convert({ + from: job.sourceFormat, + to: job.targetFormat, + input: Buffer.from(job.input), + standalone: job.standalone, + }); + await this.prisma.conversionJob.update({ + where: { id: job.id }, + data: { + status: 'SUCCEEDED', + // Prisma Bytes = Uint8Array; copy the Buffer in (#40). + result: new Uint8Array(result.output), + resultMimeType: result.mimeType, + errorCode: null, + }, + }); + this.logger.info( + { jobId: job.id, from: job.sourceFormat, to: job.targetFormat }, + 'audit: conversion succeeded', + ); + } catch (error) { + await this.recordFailure(job, error); + } + } + + private async recordFailure(job: ConversionJob, error: unknown): Promise { + const code = error instanceof ConversionError ? error.code : 'conversion_failed'; + const retryable = error instanceof ConversionError ? error.retryable : false; + // `attempts` was already incremented by the claim, so it reflects this try. + if (retryable && job.attempts < MAX_ATTEMPTS) { + await this.prisma.conversionJob.update({ + where: { id: job.id }, + data: { status: 'PENDING', lockedAt: null, errorCode: code }, + }); + this.logger.warn( + { jobId: job.id, code, attempt: job.attempts }, + 'conversion attempt failed, will retry', + ); + return; + } + await this.prisma.conversionJob.update({ + where: { id: job.id }, + data: { status: 'FAILED', errorCode: code }, + }); + this.logger.error({ jobId: job.id, code, attempts: job.attempts }, 'conversion failed'); + } +} diff --git a/apps/api/src/import-export/import-export.module.ts b/apps/api/src/import-export/import-export.module.ts new file mode 100644 index 0000000..7bf16f3 --- /dev/null +++ b/apps/api/src/import-export/import-export.module.ts @@ -0,0 +1,24 @@ +import { Module } from '@nestjs/common'; + +import { ConversionJobService } from './conversion-job.service'; +import { ConversionWorker } from './conversion-worker.service'; +import { JobsController } from './jobs.controller'; +import { PandocConverter, PandocServerConverter } from './pandoc.converter'; + +/** + * Import/export orchestration (ADR 0009, issue #62): the conversion job queue, + * its worker, and the pandoc-server client. Later stories (#63 import, #65 + * export) add the feature endpoints that enqueue jobs here. + */ +@Module({ + controllers: [JobsController], + providers: [ + ConversionJobService, + ConversionWorker, + // Bind the abstract converter to the HTTP implementation; tests override + // this provider with a fake so the queue mechanics need no live sidecar. + { provide: PandocConverter, useClass: PandocServerConverter }, + ], + exports: [ConversionJobService, PandocConverter], +}) +export class ImportExportModule {} diff --git a/apps/api/src/import-export/jobs.controller.ts b/apps/api/src/import-export/jobs.controller.ts new file mode 100644 index 0000000..c8db3ff --- /dev/null +++ b/apps/api/src/import-export/jobs.controller.ts @@ -0,0 +1,34 @@ +import { Controller, Get, Param, Req, Res, StreamableFile } from '@nestjs/common'; +import { ConversionJobView } from '@dorfteich/shared'; +import type { Response } from 'express'; + +import { AuthedRequest } from '../auth/auth.guard'; +import { AuthenticatedOnly } from '../permissions/permission.decorators'; + +import { ConversionJobService } from './conversion-job.service'; + +/** Conversion job polling (ADR 0009, issue #62). Both routes are owner-scoped + * inside the service (a foreign or unknown id is 404), so the authenticated + * marker is the only guard needed. */ +@Controller('jobs') +export class JobsController { + constructor(private readonly jobs: ConversionJobService) {} + + @Get(':id') + @AuthenticatedOnly() + get(@Param('id') id: string, @Req() request: AuthedRequest): Promise { + return this.jobs.getForOwner(id, request.user!.id); + } + + @Get(':id/result') + @AuthenticatedOnly() + async result( + @Param('id') id: string, + @Req() request: AuthedRequest, + @Res({ passthrough: true }) response: Response, + ): Promise { + const { bytes, mimeType } = await this.jobs.resultForOwner(id, request.user!.id); + response.set('X-Content-Type-Options', 'nosniff'); + return new StreamableFile(bytes, { type: mimeType, disposition: 'attachment' }); + } +} diff --git a/apps/api/src/import-export/pandoc.converter.test.ts b/apps/api/src/import-export/pandoc.converter.test.ts new file mode 100644 index 0000000..1df8172 --- /dev/null +++ b/apps/api/src/import-export/pandoc.converter.test.ts @@ -0,0 +1,96 @@ +import { createServer, Server } from 'node:http'; +import type { RequestListener } from 'node:http'; +import type { AddressInfo } from 'node:net'; + +import { afterEach, describe, expect, it } from 'vitest'; + +import type { AppConfig } from '../config/app-config.service'; + +import { ConversionError, PandocServerConverter } from './pandoc.converter'; + +/** A PandocServerConverter pointed at a test URL, with the 60 s timeout + * shortened so the timeout path is exercisable in milliseconds. */ +class TestConverter extends PandocServerConverter { + constructor(url: string, timeoutMs: number) { + super({ env: { PANDOC_URL: url } } as unknown as AppConfig); + this.timeoutMs = timeoutMs; + } +} + +describe('PandocServerConverter (transport)', () => { + let server: Server | undefined; + + afterEach(() => { + server?.close(); + server = undefined; + }); + + async function serve(handler: RequestListener): Promise { + server = createServer(handler); + await new Promise((resolve) => server!.listen(0, '127.0.0.1', resolve)); + const { port } = server!.address() as AddressInfo; + return `http://127.0.0.1:${port}`; + } + + it('returns the converted bytes and the format MIME type on success', async () => { + const url = await serve((_req, res) => { + res.writeHead(200); + res.end('# converted'); + }); + const result = await new TestConverter(url, 5000).convert({ + from: 'docx', + to: 'markdown', + input: Buffer.from('binary-docx'), + }); + expect(result.output.toString()).toBe('# converted'); + expect(result.mimeType).toBe('text/markdown'); + }); + + it('maps a non-200 pandoc response to a non-retryable conversion_failed', async () => { + const url = await serve((_req, res) => { + res.writeHead(500); + res.end('Unknown input format nonsense'); + }); + await expect( + new TestConverter(url, 5000).convert({ from: 'x', to: 'html', input: Buffer.from('a') }), + ).rejects.toMatchObject({ code: 'conversion_failed', retryable: false }); + }); + + it('maps an unreachable sidecar to a retryable converter_unavailable', async () => { + // Nothing listening on this port. + await expect( + new TestConverter('http://127.0.0.1:1', 5000).convert({ + from: 'markdown', + to: 'html', + input: Buffer.from('a'), + }), + ).rejects.toMatchObject({ code: 'converter_unavailable', retryable: true }); + }); + + it('times out a slow conversion and fails it (delay-injecting fake)', async () => { + const url = await serve((_req, res) => { + // Never responds within the (shortened) timeout. + setTimeout(() => { + res.writeHead(200); + res.end('too late'); + }, 2000).unref(); + }); + await expect( + new TestConverter(url, 100).convert({ + from: 'markdown', + to: 'html', + input: Buffer.from('a'), + }), + ).rejects.toMatchObject({ code: 'converter_timeout', retryable: true }); + }); + + it('is a ConversionError so callers can branch on the code', async () => { + await expect( + new TestConverter('http://127.0.0.1:1', 5000).convert({ + from: 'markdown', + to: 'html', + input: Buffer.from('a'), + }), + ).rejects.toBeInstanceOf(ConversionError); + }); +}); diff --git a/apps/api/src/import-export/pandoc.converter.ts b/apps/api/src/import-export/pandoc.converter.ts new file mode 100644 index 0000000..1ee23d9 --- /dev/null +++ b/apps/api/src/import-export/pandoc.converter.ts @@ -0,0 +1,146 @@ +import { Injectable } from '@nestjs/common'; + +import { AppConfig } from '../config/app-config.service'; + +/** A single pandoc conversion request (ADR 0009). `input` is the raw source + * bytes; the converter base64-encodes it for pandoc when `from` is a binary + * format (docx/odt/…) and passes it as text otherwise. */ +export interface ConversionRequest { + from: string; + to: string; + input: Buffer; + standalone?: boolean; +} + +export interface ConversionResult { + output: Buffer; + mimeType: string; +} + +export type ConversionErrorCode = + 'converter_unavailable' | 'converter_timeout' | 'conversion_failed'; + +/** A conversion failure with a stable, localizable code. `retryable` marks + * the transient causes (sidecar down / timed out) the worker retries before + * giving up; a genuine `conversion_failed` (bad/unsupported content) is not + * retried. */ +export class ConversionError extends Error { + constructor( + readonly code: ConversionErrorCode, + readonly retryable: boolean, + message?: string, + ) { + super(message ?? code); + this.name = 'ConversionError'; + } +} + +/** Server-side conversion limits (ADR 0009). Input is checked before the + * sidecar call; output is capped while reading the response so a runaway + * conversion can't exhaust memory. */ +export const MAX_CONVERSION_INPUT_BYTES = 25 * 1024 * 1024; +export const MAX_CONVERSION_OUTPUT_BYTES = 50 * 1024 * 1024; +/** Per ADR 0009 / issue #62: a single conversion may run for at most 60 s. */ +export const CONVERSION_TIMEOUT_MS = 60_000; + +/** Formats pandoc reads as binary — their bytes are base64-encoded in the + * request `text` field; text formats are sent verbatim. */ +const BINARY_INPUT_FORMATS = new Set(['docx', 'odt', 'epub', 'pptx']); + +/** Served content type per pandoc output format. */ +const OUTPUT_MIME_TYPES: Readonly> = { + html: 'text/html', + markdown: 'text/markdown', + gfm: 'text/markdown', + commonmark: 'text/markdown', + plain: 'text/plain', + docx: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', + odt: 'application/vnd.oasis.opendocument.text', +}; + +/** + * Typed client for the pandoc-server sidecar (ADR 0009). Abstract so the job + * worker and its tests depend on the contract, not the HTTP transport — the + * queue-mechanics tests inject a fake, this real implementation is exercised + * against a running container in the integration test. + */ +export abstract class PandocConverter { + abstract convert(request: ConversionRequest): Promise; + abstract reachable(): Promise; +} + +@Injectable() +export class PandocServerConverter extends PandocConverter { + /** Overridable so the timeout path is testable without a 60 s wait. */ + protected timeoutMs = CONVERSION_TIMEOUT_MS; + + constructor(private readonly config: AppConfig) { + super(); + } + + private get baseUrl(): string { + return this.config.env.PANDOC_URL; + } + + async reachable(): Promise { + try { + const response = await fetch(`${this.baseUrl}/version`); + return response.ok; + } catch { + return false; + } + } + + async convert(request: ConversionRequest): Promise { + if (request.input.byteLength > MAX_CONVERSION_INPUT_BYTES) { + throw new ConversionError('conversion_failed', false, 'input exceeds size limit'); + } + const text = BINARY_INPUT_FORMATS.has(request.from) + ? request.input.toString('base64') + : request.input.toString('utf8'); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), this.timeoutMs); + let response: Response; + try { + response = await fetch(`${this.baseUrl}/`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + text, + from: request.from, + to: request.to, + standalone: request.standalone ?? true, + }), + signal: controller.signal, + }); + } catch (error) { + // AbortController.abort() surfaces as an AbortError → the run timed out; + // anything else means the sidecar could not be reached. + if (error instanceof Error && error.name === 'AbortError') { + throw new ConversionError('converter_timeout', true, 'pandoc timed out'); + } + throw new ConversionError('converter_unavailable', true, shortMessage(error)); + } finally { + clearTimeout(timer); + } + + if (!response.ok) { + // Non-200 = pandoc rejected the content (unknown format, malformed + // document). Not retryable — the same input fails the same way. + const detail = (await response.text().catch(() => '')).slice(0, 300); + throw new ConversionError('conversion_failed', false, detail || `pandoc ${response.status}`); + } + + const output = Buffer.from(await response.arrayBuffer()); + if (output.byteLength > MAX_CONVERSION_OUTPUT_BYTES) { + throw new ConversionError('conversion_failed', false, 'output exceeds size limit'); + } + return { output, mimeType: OUTPUT_MIME_TYPES[request.to] ?? 'application/octet-stream' }; + } +} + +function shortMessage(error: unknown): string { + const message = error instanceof Error ? error.message : String(error); + return message.split('\n')[0]?.slice(0, 200) ?? 'unknown error'; +} diff --git a/apps/api/src/testing/test-app.ts b/apps/api/src/testing/test-app.ts index a9854e1..c7b0509 100644 --- a/apps/api/src/testing/test-app.ts +++ b/apps/api/src/testing/test-app.ts @@ -3,7 +3,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { INestApplication } from '@nestjs/common'; -import { Test } from '@nestjs/testing'; +import { Test, TestingModuleBuilder } from '@nestjs/testing'; import type { NestExpressApplication } from '@nestjs/platform-express'; import cookieParser from 'cookie-parser'; @@ -12,9 +12,12 @@ import { AppModule } from '../app.module'; /** * Boots the full application for e2e tests, mirroring main.ts middleware. * Requires TEST_DATABASE_URL; DATABASE_URL is pointed at it so the app - * under test uses the test database. + * under test uses the test database. `customize` can override providers + * (e.g. inject a fake converter for the conversion queue tests, issue #62). */ -export async function createTestApp(): Promise { +export async function createTestApp( + customize?: (builder: TestingModuleBuilder) => TestingModuleBuilder, +): Promise { process.env.NODE_ENV = 'test'; if (process.env.TEST_DATABASE_URL) { process.env.DATABASE_URL = process.env.TEST_DATABASE_URL; @@ -24,7 +27,8 @@ export async function createTestApp(): Promise { // repository or collide with each other. process.env.UPLOADS_DIR ??= mkdtempSync(join(tmpdir(), 'dorfteich-uploads-')); - const moduleRef = await Test.createTestingModule({ imports: [AppModule] }).compile(); + const base = Test.createTestingModule({ imports: [AppModule] }); + const moduleRef = await (customize ? customize(base) : base).compile(); const app = moduleRef.createNestApplication(); app.use(cookieParser()); // Mirrors main.ts: base64 Yjs page state needs more than Express's 100kb default. diff --git a/deploy/compose/docker-compose.yml b/deploy/compose/docker-compose.yml index b4bb35b..2982b1f 100644 --- a/deploy/compose/docker-compose.yml +++ b/deploy/compose/docker-compose.yml @@ -60,6 +60,8 @@ services: SMTP_FROM: ${SMTP_FROM:-Dorfteich } # Matches the `uploads` volume mount below (ADR 0011). UPLOADS_DIR: /data/uploads + # Internal pandoc-server sidecar for import/export (ADR 0009, issue #62). + PANDOC_URL: http://pandoc:3030 ports: - '127.0.0.1:${API_PORT:-8101}:3000' networks: [frontend, internal] @@ -68,6 +70,8 @@ services: depends_on: db: condition: service_healthy + pandoc: + condition: service_healthy <<: *logging collab: @@ -117,6 +121,21 @@ services: retries: 12 <<: *logging + # Import/export converter (ADR 0009, issue #62): pandoc in HTTP server mode + # on the internal network only — never exposed. Pinned image; the api reaches + # it at http://pandoc:3030. `wget` ships in the (busybox-based) image. + pandoc: + image: pandoc/core:3.6 + command: ['server'] + restart: unless-stopped + networks: [internal] + healthcheck: + test: ['CMD-SHELL', 'wget -q -O /dev/null http://127.0.0.1:3030/version || exit 1'] + interval: 30s + timeout: 5s + retries: 3 + <<: *logging + networks: frontend: internal: diff --git a/packages/shared/i18n/de/errors.json b/packages/shared/i18n/de/errors.json index 7dafd1b..1e99c26 100644 --- a/packages/shared/i18n/de/errors.json +++ b/packages/shared/i18n/de/errors.json @@ -31,6 +31,9 @@ "unsupported_file_type": "Dieser Dateityp wird nicht unterstützt.", "upload_type_not_allowed": "Dieser Dateityp ist auf dieser Instanz nicht erlaubt.", "file_too_large": "Die Datei ist zu groß (Limit: {{limitBytes}} Bytes).", + "converter_unavailable": "Der Dokument-Konverter ist derzeit nicht verfügbar. Bitte versuche es später erneut.", + "converter_timeout": "Die Konvertierung hat zu lange gedauert und wurde abgebrochen.", + "conversion_failed": "Dieses Dokument konnte nicht konvertiert werden.", "network": "Der Server war nicht erreichbar.", "grant_exists": "Diese Berechtigung existiert bereits.", "grant_pond_admin_scope": "Eine Teich-Admin-Berechtigung muss für den ganzen Teich und eine bestimmte Person gelten.", diff --git a/packages/shared/i18n/en/errors.json b/packages/shared/i18n/en/errors.json index b2ce64c..6e1258b 100644 --- a/packages/shared/i18n/en/errors.json +++ b/packages/shared/i18n/en/errors.json @@ -31,6 +31,9 @@ "unsupported_file_type": "This file type is not supported.", "upload_type_not_allowed": "This file type is not allowed on this instance.", "file_too_large": "The file is too large (limit: {{limitBytes}} bytes).", + "converter_unavailable": "The document converter is currently unavailable. Please try again later.", + "converter_timeout": "The conversion took too long and was cancelled.", + "conversion_failed": "This document could not be converted.", "network": "The server could not be reached.", "grant_exists": "This grant already exists.", "grant_pond_admin_scope": "A Pond Admin grant must apply to the whole pond and a specific user.", diff --git a/packages/shared/src/conversion.ts b/packages/shared/src/conversion.ts new file mode 100644 index 0000000..c42b9ea --- /dev/null +++ b/packages/shared/src/conversion.ts @@ -0,0 +1,19 @@ +/** + * Import/export conversion job types shared between api and web (ADR 0009, + * issue #62). A conversion runs asynchronously against the pandoc sidecar; + * the client enqueues it and polls `GET /jobs/:id` for this view. + */ +export type ConversionJobStatus = 'pending' | 'running' | 'succeeded' | 'failed'; + +export interface ConversionJobView { + id: string; + status: ConversionJobStatus; + kind: string; + sourceFormat: string; + targetFormat: string; + /** Set only when `status` is `failed` — a code from the errors namespace + * (`converter_unavailable` | `converter_timeout` | `conversion_failed`). */ + errorCode: string | null; + createdAt: string; + updatedAt: string; +} diff --git a/packages/shared/src/env.ts b/packages/shared/src/env.ts index 2d61ea0..da8c80c 100644 --- a/packages/shared/src/env.ts +++ b/packages/shared/src/env.ts @@ -62,6 +62,13 @@ export const apiEnvSchema = z.object({ * dev/test runs. */ UPLOADS_DIR: z.string().min(1).default('./data/uploads'), + /** + * Base URL of the internal pandoc-server sidecar (ADR 0009, issue #62). + * The default matches the compose service name; native dev/test runs point + * it at a locally running container or leave it unreachable (the converter + * readiness check is warning-level, so an unset sidecar never fails readyz). + */ + PANDOC_URL: z.string().url().default('http://pandoc:3030'), }); export type ApiEnv = z.infer; diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 759c71c..9465e00 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -4,6 +4,7 @@ export * from './auth'; export * from './collab-token'; export * from './editor-schema'; export * from './env'; +export * from './conversion'; export * from './files'; export * from './health'; export * from './i18n-tools';