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; /** Set for import jobs (#63): the pond the document is imported into and the * original upload file name (a title fallback). */ pondId?: string; sourceName?: string; } 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, pondId: request.pondId ?? null, sourceName: request.sourceName ?? null, 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(); // A data-export link expires (#68): past its window the result is treated as // gone (a scheduled purge deletes the bytes), so the download 404s. if (job.expiresAt && job.expiresAt.getTime() < Date.now()) 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, // Set once an import job succeeds (#63) so the client can open the page. resultPageId: job.resultPageId, // Set for a data-export job (#68): when its download link stops working. expiresAt: job.expiresAt?.toISOString() ?? null, createdAt: job.createdAt.toISOString(), updatedAt: job.updatedAt.toISOString(), }; } }