Build a File Format Converter API

// src/converter/engine.ts — File format conversion with queue processing and webhooks
import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes } from "node:crypto";
import { execSync } from "node:child_process";
import { readFile, writeFile, unlink } from "node:fs/promises";
import { join } from "node:path";

const redis = new Redis(process.env.REDIS_URL!);

interface ConversionJob {
  id: string;
  inputFormat: string;
  outputFormat: string;
  inputPath: string;
  outputPath: string | null;
  status: "queued" | "processing" | "completed" | "failed";
  progress: number;
  error: string | null;
  webhookUrl: string | null;
  metadata: { originalName: string; inputSize: number; outputSize?: number };
  createdAt: string;
  completedAt: string | null;
}

const SUPPORTED_CONVERSIONS: Record<string, string[]> = {
  // Documents
  docx: ["pdf", "html", "txt", "md"],
  xlsx: ["csv", "pdf", "json"],
  csv: ["json", "xlsx"],
  html: ["pdf", "png", "md"],
  md: ["html", "pdf"],
  // Images
  png: ["jpg", "webp", "avif", "pdf", "svg"],
  jpg: ["png", "webp", "avif", "pdf"],
  heic: ["jpg", "png", "webp"],
  svg: ["png", "jpg", "pdf"],
  webp: ["png", "jpg"],
  // Audio
  mp3: ["wav", "ogg", "aac"],
  wav: ["mp3", "ogg", "aac"],
  ogg: ["mp3", "wav"],
  // Video
  mp4: ["webm", "gif", "mp3"],
  mov: ["mp4", "webm", "gif"],
  webm: ["mp4", "gif"],
};

// Submit conversion job
export async function submitJob(params: {
  inputBuffer: Buffer;
  inputFormat: string;
  outputFormat: string;
  originalName: string;
  webhookUrl?: string;
}): Promise<ConversionJob> {
  // Validate conversion is supported
  const supported = SUPPORTED_CONVERSIONS[params.inputFormat];
  if (!supported || !supported.includes(params.outputFormat)) {
    throw new Error(`Conversion from ${params.inputFormat} to ${params.outputFormat} not supported`);
  }

  const id = `conv-${randomBytes(8).toString("hex")}`;
  const inputPath = join("/tmp/conversions", `${id}.${params.inputFormat}`);
  await writeFile(inputPath, params.inputBuffer);

  const job: ConversionJob = {
    id,
    inputFormat: params.inputFormat,
    outputFormat: params.outputFormat,
    inputPath,
    outputPath: null,
    status: "queued",
    progress: 0,
    error: null,
    webhookUrl: params.webhookUrl || null,
    metadata: { originalName: params.originalName, inputSize: params.inputBuffer.length },
    createdAt: new Date().toISOString(),
    completedAt: null,
  };

  await pool.query(
    `INSERT INTO conversion_jobs (id, input_format, output_format, input_path, status, webhook_url, metadata, created_at)
     VALUES ($1, $2, $3, $4, 'queued', $5, $6, NOW())`,
    [id, params.inputFormat, params.outputFormat, inputPath, params.webhookUrl, JSON.stringify(job.metadata)]
  );

  // Queue for processing
  await redis.rpush("conversion:queue", id);

  return job;
}

// Process conversion (called by worker)
export async function processJob(jobId: string): Promise<void> {
  await pool.query("UPDATE conversion_jobs SET status = 'processing' WHERE id = $1", [jobId]);

  const { rows: [job] } = await pool.query("SELECT * FROM conversion_jobs WHERE id = $1", [jobId]);
  if (!job) return;

  const outputPath = join("/tmp/conversions", `${jobId}.${job.output_format}`);

  try {
    // Route to appropriate converter
    const category = getCategory(job.input_format);
    switch (category) {
      case "document":
        await convertDocument(job.input_path, outputPath, job.input_format, job.output_format);
        break;
      case "image":
        await convertImage(job.input_path, outputPath, job.output_format);
        break;
      case "audio":
      case "video":
        await convertMedia(job.input_path, outputPath, job.output_format);
        break;
    }

    const outputBuffer = await readFile(outputPath);
    const metadata = JSON.parse(job.metadata);
    metadata.outputSize = outputBuffer.length;

    await pool.query(
      "UPDATE conversion_jobs SET status = 'completed', output_path = $2, metadata = $3, completed_at = NOW() WHERE id = $1",
      [jobId, outputPath, JSON.stringify(metadata)]
    );

    // Send webhook
    if (job.webhook_url) {
      await fetch(job.webhook_url, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ jobId, status: "completed", outputSize: metadata.outputSize }),
      }).catch(() => {});
    }
  } catch (error: any) {
    await pool.query(
      "UPDATE conversion_jobs SET status = 'failed', error = $2 WHERE id = $1",
      [jobId, error.message]
    );

    if (job.webhook_url) {
      await fetch(job.webhook_url, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ jobId, status: "failed", error: error.message }),
      }).catch(() => {});
    }
  } finally {
    // Cleanup input file
    await unlink(job.input_path).catch(() => {});
  }
}

async function convertDocument(input: string, output: string, fromFormat: string, toFormat: string): Promise<void> {
  // Use LibreOffice for document conversions
  if (["docx", "xlsx"].includes(fromFormat) && toFormat === "pdf") {
    execSync(`libreoffice --headless --convert-to pdf --outdir /tmp/conversions ${input}`, { timeout: 120000 });
    return;
  }
  if (fromFormat === "csv" && toFormat === "json") {
    const content = await readFile(input, "utf-8");
    const lines = content.split("\n").filter((l) => l.trim());
    const headers = lines[0].split(",").map((h) => h.trim().replace(/^"|"$/g, ""));
    const rows = lines.slice(1).map((line) => {
      const values = line.split(",").map((v) => v.trim().replace(/^"|"$/g, ""));
      return Object.fromEntries(headers.map((h, i) => [h, values[i]]));
    });
    await writeFile(output, JSON.stringify(rows, null, 2));
    return;
  }
  throw new Error(`Document conversion ${fromFormat} → ${toFormat} not implemented`);
}

async function convertImage(input: string, output: string, toFormat: string): Promise<void> {
  // Use sharp for image conversions
  const sharp = require("sharp");
  let pipeline = sharp(input);
  switch (toFormat) {
    case "jpg": pipeline = pipeline.jpeg({ quality: 85 }); break;
    case "png": pipeline = pipeline.png(); break;
    case "webp": pipeline = pipeline.webp({ quality: 85 }); break;
    case "avif": pipeline = pipeline.avif({ quality: 65 }); break;
  }
  await pipeline.toFile(output);
}

async function convertMedia(input: string, output: string, toFormat: string): Promise<void> {
  // Use ffmpeg for audio/video
  execSync(`ffmpeg -i ${input} -y ${output}`, { timeout: 600000 });
}

function getCategory(format: string): string {
  if (["docx", "xlsx", "csv", "html", "md", "pdf"].includes(format)) return "document";
  if (["png", "jpg", "jpeg", "heic", "svg", "webp", "avif"].includes(format)) return "image";
  if (["mp3", "wav", "ogg", "aac"].includes(format)) return "audio";
  if (["mp4", "mov", "webm", "avi"].includes(format)) return "video";
  return "unknown";
}

// Auto-detect format from buffer
export function detectFormat(buffer: Buffer, fileName?: string): string {
  // Magic bytes detection
  const hex = buffer.slice(0, 8).toString("hex");
  if (hex.startsWith("89504e47")) return "png";
  if (hex.startsWith("ffd8ff")) return "jpg";
  if (hex.startsWith("25504446")) return "pdf";
  if (hex.startsWith("504b0304")) {
    // ZIP-based: could be docx, xlsx, etc.
    if (fileName?.endsWith(".docx")) return "docx";
    if (fileName?.endsWith(".xlsx")) return "xlsx";
    return "zip";
  }
  if (hex.startsWith("1a45dfa3")) return "webm";
  if (hex.startsWith("4f676753")) return "ogg";
  // Fallback to extension
  if (fileName) return fileName.split(".").pop()?.toLowerCase() || "unknown";
  return "unknown";
}

// Get job status
export async function getJobStatus(jobId: string): Promise<ConversionJob | null> {
  const { rows: [row] } = await pool.query("SELECT * FROM conversion_jobs WHERE id = $1", [jobId]);
  return row ? { ...row, metadata: JSON.parse(row.metadata) } : null;
}