dorfteich/apps/api/src/import-export/conversion-worker.service.ts

import { readFile } from 'node:fs/promises';
import { join } from 'node:path';

import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { ModuleRef } from '@nestjs/core';
import { ConversionJob } from '@prisma/client';
import { PinoLogger } from 'nestjs-pino';

import { AppConfig } from '../config/app-config.service';
import { PrismaService } from '../prisma/prisma.service';

import {
  DATA_EXPORT_KIND,
  DATA_EXPORT_PROCESSOR,
  DATA_EXPORT_TTL_MS,
  DataExportProcessor,
} from './data-export.constants';
import { GotenbergRenderer, RenderError } from './gotenberg.renderer';
import { IMPORT_PROCESSOR, ImportProcessor, isImportKind } from './import.constants';
import {
  ConversionError,
  ConversionResult,
  conversionInputOf,
  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 renderer: GotenbergRenderer,
    private readonly config: AppConfig,
    private readonly logger: PinoLogger,
    // Resolved lazily to break the construction cycle (the import service
    // enqueues via the job service, which wakes this worker).
    private readonly moduleRef: ModuleRef,
  ) {
    this.logger.setContext(ConversionWorker.name);
  }

  private async convert(job: ConversionJob): Promise<{ bytes: Buffer; mimeType: string }> {
    const result: ConversionResult = await this.converter.convert({
      from: job.sourceFormat,
      to: job.targetFormat,
      input: Buffer.from(conversionInputOf(job)),
      standalone: job.standalone,
      referenceDoc: await this.classifiedReferenceDoc(job),
    });
    return { bytes: result.output, mimeType: result.mimeType };
  }

  /**
   * The classified reference document for a marked docx/odt export (issue
   * #209, ADR 0022): pandoc copies its header/footer — which carry the
   * VS-NfD marking — into the output, so the marking repeats on every page
   * in Word/LibreOffice and is not deletable body text. Only present when
   * the enqueue put a `marking` into the job options; the binaries ship in
   * `apps/api/assets/` (see `scripts/gen-classified-reference-docs.mjs`).
   */
  private async classifiedReferenceDoc(
    job: ConversionJob,
  ): Promise<{ name: string; bytes: Buffer } | undefined> {
    const marked = Boolean((job.options as { marking?: string } | null)?.marking);
    if (!marked || (job.targetFormat !== 'docx' && job.targetFormat !== 'odt')) return undefined;
    const name = `reference-vs-nfd.${job.targetFormat}`;
    const cached = this.referenceDocs.get(name);
    if (cached) return { name, bytes: cached };
    const bytes = await readFile(join(__dirname, '../../assets', name));
    this.referenceDocs.set(name, bytes);
    return { name, bytes };
  }

  private readonly referenceDocs = new Map<string, Buffer>();

  onModuleInit(): void {
    if (this.config.env.NODE_ENV === 'test') return; // tests drive drain() directly
    this.timer = setInterval(() => this.drainSafely(), SWEEP_MS);
    this.timer.unref();
  }

  /** A sweep hitting a transient database failure (outage, or the backup
   * sidecar terminating connections mid-restore, #103) must degrade and
   * retry on the next tick — an unhandled rejection here killed the whole
   * api once. */
  private drainSafely(): void {
    this.drain().catch((error: unknown) => {
      this.logger.warn({ err: error }, 'conversion sweep failed; retrying on the next tick');
    });
  }

  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;
    this.drainSafely();
  }

  /** 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<void> {
    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<ConversionJob | null> {
    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<void> {
    try {
      if (isImportKind(job.kind)) {
        // Import runs a multi-step pipeline and records its own success (the
        // created page id) on the job (#63). Resolved lazily via a token so the
        // worker's file never imports the import service's (avoids a cycle).
        await this.moduleRef.get<ImportProcessor>(IMPORT_PROCESSOR, { strict: false }).run(job);
        return;
      }
      // A data export gathers the account's own data into a ZIP (#68) whose
      // download link expires; a PDF export renders HTML through Gotenberg
      // (#67); everything else is a pandoc byte→byte conversion (#62/#65).
      let output: { bytes: Buffer; mimeType: string };
      let expiresAt: Date | null = null;
      if (job.kind === DATA_EXPORT_KIND) {
        // Resolved lazily via a token so this file never imports the export
        // service's (avoids a construction cycle), mirroring the import path.
        output = await this.moduleRef
          .get<DataExportProcessor>(DATA_EXPORT_PROCESSOR, { strict: false })
          .build(job);
        expiresAt = new Date(Date.now() + DATA_EXPORT_TTL_MS);
      } else if (job.targetFormat === 'pdf') {
        // A classified page's export carries its marking as a job option
        // (issue #208) — Gotenberg repeats it in header/footer of every page.
        const marking = (job.options as { marking?: string } | null)?.marking ?? null;
        output = {
          bytes: await this.renderer.renderHtmlToPdf(
            Buffer.from(conversionInputOf(job)).toString('utf8'),
            { marking },
          ),
          mimeType: 'application/pdf',
        };
      } else {
        output = await this.convert(job);
      }
      await this.prisma.conversionJob.update({
        where: { id: job.id },
        data: {
          status: 'SUCCEEDED',
          // Prisma Bytes = Uint8Array<ArrayBuffer>; copy the Buffer in (#40).
          result: new Uint8Array(output.bytes),
          resultMimeType: output.mimeType,
          errorCode: null,
          expiresAt,
        },
      });
      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<void> {
    const typed = error instanceof ConversionError || error instanceof RenderError ? error : null;
    const code = typed?.code ?? 'conversion_failed';
    const retryable = typed?.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');
  }
}