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

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<PrismaStatus, ConversionJobStatus> = {
  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<ConversionJob> {
    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<ArrayBuffer>; 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<ConversionJobView> {
    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<ConversionResultPayload> {
    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<ConversionJob> {
    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(),
    };
  }
}