dragonflight/services/capture/src/index.js

import express from 'express';
import cors from 'cors';
import dotenv from 'dotenv';
import captureRoutes from './routes/capture.js';
import captureManager from './capture-manager.js';

dotenv.config();

const app = express();
const PORT = process.env.PORT || 3001;
const MAM_API_URL = process.env.MAM_API_URL || 'http://mam-api:3000';
const MAM_API_TOKEN = process.env.MAM_API_TOKEN || '';

app.use(cors());
app.use(express.json());

app.get('/health', (req, res) => {
  res.json({ status: 'ok' });
});

app.use('/capture', captureRoutes);

const server = app.listen(PORT, () => {
  console.log(`Wild Dragon Capture Service listening on port ${PORT}`);
  const _srcType = process.env.SOURCE_TYPE;
  const _standby = process.env.STANDBY === '1';

  if (_standby) {
    // Standby mode — sidecar pre-spawned at recorder create time.
    // Don't auto-start a recording session; wait for POST /capture/start.
    // Still start idle preview so the live signal thumbnail is visible.
    console.log('[bootstrap] standby mode — waiting for /capture/start HTTP call');
    if (process.env.RECORDER_ID && (_srcType === 'deltacast' || _srcType === 'sdi' || _srcType === 'blackmagic')) {
      setTimeout(() => captureManager.startIdlePreview(), 3000);
    }
  } else {
    // Legacy mode — env vars carry the session params, start immediately.
    bootstrapAutoStart();
    // Auto-start idle signal preview for deltacast/sdi sidecars.
    // 3s delay lets the deltacast bridge FIFOs come up first.
    if (process.env.RECORDER_ID && (_srcType === 'deltacast' || _srcType === 'sdi')) {
      setTimeout(() => captureManager.startIdlePreview(), 3000);
    }
  }
});

// Mapped from the env vars routes/recorders.js writes into the container.
// Empty strings collapse to undefined so capture-manager's defaults win.
function envOpt(name) {
  const v = process.env[name];
  return v === undefined || v === '' ? undefined : v;
}
function envInt(name) {
  const v = envOpt(name);
  if (v === undefined) return undefined;
  const n = parseInt(v, 10);
  return Number.isFinite(n) ? n : undefined;
}
function envBool(name) {
  const v = envOpt(name);
  if (v === undefined) return undefined;
  return v === 'true' || v === '1' || v === 'yes';
}

async function bootstrapAutoStart() {
  const recorderId = process.env.RECORDER_ID;
  const sourceType = process.env.SOURCE_TYPE;
  if (!recorderId || !sourceType) {
    console.log('[bootstrap] no RECORDER_ID/SOURCE_TYPE - on-demand sidecar');
    return;
  }

  const projectId = process.env.PROJECT_ID;
  const clipName  = process.env.CLIP_NAME;
  if (!projectId || !clipName) {
    console.error('[bootstrap] missing PROJECT_ID or CLIP_NAME - cannot start');
    return;
  }

  const listen = process.env.LISTEN === '1' || process.env.LISTEN === 'true';
  const listenPort = envInt('LISTEN_PORT');
  const streamKey = envOpt('STREAM_KEY');
  const sourceUrl = envOpt('SOURCE_URL');
  const device    = envInt('DEVICE_INDEX');
  // SOURCE_CONFIG is the recorder's source_config JSON (set by recorders.js).
  // For deltacast it carries the capture channel (`port`) and optional `board`.
  let sourceConfig = {};
  try { sourceConfig = JSON.parse(process.env.SOURCE_CONFIG || '{}') || {}; }
  catch (e) { console.warn('[bootstrap] bad SOURCE_CONFIG JSON:', e.message); }
  const port  = Number.isInteger(sourceConfig.port)  ? sourceConfig.port  : undefined;
  const board = Number.isInteger(sourceConfig.board) ? sourceConfig.board : undefined;

  console.log(`[bootstrap] starting ${sourceType} ingest (listen=${listen} port=${listenPort || 'n/a'})...`);
  try {
    const session = await captureManager.start({
      assetId: envOpt('ASSET_ID') || null,
      projectId,
      binId: envOpt('BIN_ID') || null,
      clipName,
      device,
      port,
      board,
      sourceType,
      sourceUrl,
      listen,
      listenPort,
      streamKey,

      // Recording codec — recorders.js passes these straight through
      videoCodec:    envOpt('RECORDING_CODEC')           || 'prores_hq',
      videoBitrate:  envOpt('RECORDING_VIDEO_BITRATE'),
      framerate:     envOpt('RECORDING_FRAMERATE'),
      audioCodec:    envOpt('RECORDING_AUDIO_CODEC')     || 'pcm_s24le',
      audioBitrate:  envOpt('RECORDING_AUDIO_BITRATE'),
      audioChannels: envInt('RECORDING_AUDIO_CHANNELS')  ?? 2,
      container:     envOpt('RECORDING_CONTAINER')       || 'mov',

      proxyEnabled:       envBool('PROXY_ENABLED') ?? true,
      proxyVideoCodec:    envOpt('PROXY_CODEC')          || 'h264',
      proxyVideoBitrate:  envOpt('PROXY_VIDEO_BITRATE')  || '8M',
      proxyFramerate:     envOpt('PROXY_FRAMERATE'),
      proxyAudioCodec:    envOpt('PROXY_AUDIO_CODEC')    || 'aac',
      proxyAudioBitrate:  envOpt('PROXY_AUDIO_BITRATE')  || '192k',
      proxyAudioChannels: envInt('PROXY_AUDIO_CHANNELS') ?? 2,
      proxyContainer:     envOpt('PROXY_CONTAINER')      || 'mp4',
    });
    console.log(`[bootstrap] session ${session.sessionId} started for clip ${clipName}`);
  } catch (err) {
    console.error('[bootstrap] failed to start capture:', err);
  }
}

let shuttingDown = false;
async function gracefulShutdown(signal) {
  if (shuttingDown) return;
  shuttingDown = true;
  console.log(`[shutdown] ${signal} received`);

  const status = captureManager.getStatus();

  if (status.recording) {
    console.log(`[shutdown] stopping active session ${status.sessionId}...`);
    try {
      const completed = await captureManager.stop(status.sessionId);
      console.log(`[shutdown] session ${completed.sessionId} finalised; duration=${completed.duration}s frames=${completed.framesReceived}`);

      const liveAssetId = process.env.ASSET_ID || null;

      // No frames received → the source never connected (bad SRT URL, dead
      // SDI signal, RTMP stream key mismatch, etc.). The S3 upload at this
      // point is 0 bytes and would just clog the proxy queue with "moov
      // atom not found" failures. Mark the pre-created live asset as
      // 'error' and skip the POST /assets registration entirely.
      if (completed.empty) {
        console.warn('[shutdown] no frames received — marking asset as error and skipping registration');
        if (liveAssetId) {
          try {
            await fetch(`${MAM_API_URL}/api/v1/assets/${liveAssetId}/mark-empty`, {
              method: 'POST',
              headers: { 'Content-Type': 'application/json', ...(MAM_API_TOKEN ? { 'Authorization': `Bearer ${MAM_API_TOKEN}` } : {}) },
            });
          } catch (e) {
            console.error('[shutdown] failed to flag empty asset:', e.message);
          }
        }
      } else if (completed.growingPath) {
        // Growing-files recorder: the master lives on the SMB share. Mark the asset
        // as pending_migration so the UI shows it is on SMB and provides a manual
        // right-click option to promote it to S3.
        console.log(`[shutdown] growing capture finalized on share (${completed.growingPath}); flagging pending_migration`);
        try {
          const res = await fetch(`${MAM_API_URL}/api/v1/assets/${liveAssetId}/pending-migration`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json', ...(MAM_API_TOKEN ? { 'Authorization': `Bearer ${MAM_API_TOKEN}` } : {}) },
            body: JSON.stringify({ duration: completed.duration }),
          });
          if (!res.ok) {
            console.warn(`[shutdown] mam-api pending-migration returned ${res.status}: ${await res.text()}`);
          } else {
            console.log('[shutdown] live asset flagged pending_migration with mam-api');
          }
        } catch (mamErr) {
          console.error('[shutdown] failed to flag pending_migration:', mamErr.message);
        }
      } else if (liveAssetId) {
        // Finalise the pre-created live asset by id (avoids POST / 409 collision).
        try {
          const res = await fetch(`${MAM_API_URL}/api/v1/assets/${liveAssetId}/finalize`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json', ...(MAM_API_TOKEN ? { 'Authorization': `Bearer ${MAM_API_TOKEN}` } : {}) },
            body: JSON.stringify({ hiresKey: completed.hiresKey, proxyKey: completed.proxyKey, duration: completed.duration }),
          });
          if (!res.ok) {
            console.warn(`[shutdown] mam-api finalize returned ${res.status}: ${await res.text()}`);
          } else {
            console.log('[shutdown] live asset finalised with mam-api');
          }
        } catch (mamErr) {
          console.error('[shutdown] failed to finalise asset:', mamErr.message);
        }
      } else {
        try {
          const res = await fetch(`${MAM_API_URL}/api/v1/assets`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json', ...(MAM_API_TOKEN ? { 'Authorization': `Bearer ${MAM_API_TOKEN}` } : {}) },
            body: JSON.stringify({
              projectId:  completed.projectId,
              binId:      completed.binId,
              clipName:   completed.clipName,
              sourceType: completed.sourceType,
              hiresKey:   completed.hiresKey,
              proxyKey:   completed.proxyKey,
              needsProxy: completed.proxyKey === null,
              duration:   completed.duration,
              capturedAt: completed.startedAt,
            }),
          });
          if (!res.ok) {
            console.warn(`[shutdown] mam-api /assets returned ${res.status}: ${await res.text()}`);
          } else {
            console.log('[shutdown] asset registered with mam-api');
          }
        } catch (mamErr) {
          console.error('[shutdown] failed to register asset:', mamErr.message);
        }
      }
    } catch (err) {
      console.error('[shutdown] error during stop:', err);
    }
  }

  server.close(() => {
    console.log('[shutdown] http server closed - exiting');
    process.exit(0);
  });

  setTimeout(() => process.exit(0), 5000).unref();
}

process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT',  () => gracefulShutdown('SIGINT'));