starr-philproperties/apps/api/services/assetTranscode.service.js

// services/assetTranscode.service.js
//
// Orchestrates the background remux job for a single video asset: mints a
// presigned read URL, hands it to ffmpeg.service, uploads the result back to
// storage (streamed, not buffered), then swaps the asset row over to the new
// object. Called two ways:
//   1. Fire-and-forget from assets.controller.js#finalizeAssetFromStorage,
//      right after a .mov/.mkv upload is finalized.
//   2. cron/jobs/retry_stuck_transcodes.cron.js — safety net for jobs that
//      never got picked up (server restarted mid-remux) or are still marked
//      "pending" (the fire-and-forget call in 1. never actually started,
//      e.g. this process crashed between the DB commit and the call).
//
// The original .mov/.mkv object is intentionally left in storage on success
// — this only swaps which object the asset *plays from* (storage_key), it
// doesn't delete anything. Reclaiming that storage is a separate decision.

const fs = require("fs");

const Asset = require("../models/assets/assets.mdl");
const s3 = require("../services/s3.service");
const ffmpegSvc = require("../services/ffmpeg.service");

// ── One retry-worth of guardrails ──────────────────────────────────────────
// Only ever the fire-and-forget call or the retry cron should be racing to
// pick up a given asset — this claim step (pending/failed -> processing)
// makes double-processing harmless even if both fire close together.
async function claimForProcessing(assetId) {
  const [count] = await Asset.update(
    { transcode_status: "processing", transcode_error: null },
    { where: { asset_id: assetId, transcode_status: ["pending", "failed"] } },
  );
  return count > 0;
}

async function transcodeAsset(asset) {
  if (asset.storage_provider !== "s3" || !ffmpegSvc.needsRemux(asset.extension)) return;

  const claimed = await claimForProcessing(asset.asset_id);
  if (!claimed) return; // already being processed, or already done

  let outputPath = null;
  try {
    const inputUrl = await s3.getSignedDownloadUrl(asset.storage_key);
    outputPath = await ffmpegSvc.remuxToFaststartMp4(inputUrl);

    const { size: file_size } = await fs.promises.stat(outputPath);
    const readStream = fs.createReadStream(outputPath);

    const originalname = `${(asset.original_name || asset.uuid || "video").replace(/\.[^.]+$/, "")}.mp4`;
    const { url: file_url, uuid: storage_key } = await s3.uploadStream({
      stream: readStream,
      originalname,
      mimetype: "video/mp4",
      ownerType: "video",
    });

    await asset.update({
      storage_key,
      file_url,
      file_size,
      mime_type: "video/mp4",
      extension: "mp4",
      transcode_status: "done",
      transcode_error: null,
    });

    console.log(`[ASSET][TRANSCODE] Remuxed asset ${asset.asset_id} (${asset.original_name}) to faststart mp4.`);

  } catch (err) {
    console.error(`[ASSET][TRANSCODE] Remux failed for asset ${asset.asset_id}:`, err.message);
    await asset.update({
      transcode_status: "failed",
      transcode_error: String(err.message || err).slice(0, 2000),
    }).catch(() => {});

  } finally {
    if (outputPath) fs.promises.unlink(outputPath).catch(() => {});
  }
}

module.exports = { transcodeAsset };