feat(growing-files): Phase 1 - live HLS preview during recording

While a recorder is running, the capture container tees an HLS stream into /live/<assetId>/ alongside the ProRes master upload. The asset row is pre-created at recorder start with status='live' so the clip appears in the library immediately. /api/v1/assets/:id/stream returns the HLS playlist URL until recording stops, then proxy. * docker-compose: shared wild-dragon-live mount on api/capture/web-ui * migration 001-add-live-status: idempotent ALTER TYPE for asset_status * mam-api: runMigrations() on boot; recorders.js pre-creates live asset + passes ASSET_ID; assets.js POST upserts on existing live row instead of inserting a duplicate, and stream route returns HLS for live assets * capture: parallel HLS ffmpeg into /live/<assetId>/; ASSET_ID env * web-ui: nginx serves /live/, preview.js loads hls.js, LIVE badge added
2026-05-18 07:29:50 -04:00 · 2026-05-18 07:29:50 -04:00 · 7d76f9c549
commit 7d76f9c549
parent 6a8e4ac250
10 changed files with 272 additions and 51 deletions
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -34,6 +34,7 @@ services:
      - "${PORT_MAM_API:-7432}:3000"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
+      - /mnt/NVME/MAM/wild-dragon-live:/live
    environment:
      DATABASE_URL: ${DATABASE_URL}
      REDIS_URL: ${REDIS_URL}
@ -64,6 +65,8 @@ services:
      S3_SECRET_KEY: ${S3_SECRET_KEY}
      S3_REGION: ${S3_REGION:-us-east-1}
      MAM_API_URL: ${MAM_API_URL:-http://mam-api:3000}
+    volumes:
+      - /mnt/NVME/MAM/wild-dragon-live:/live
    networks:
      - wild-dragon

@ -87,6 +90,8 @@ services:
    build: ./services/web-ui
    ports:
      - "${PORT_WEB_UI:-7434}:80"
+    volumes:
+      - /mnt/NVME/MAM/wild-dragon-live:/live
    networks:
      - wild-dragon

--- a/services/capture/src/capture-manager.js
+++ b/services/capture/src/capture-manager.js
@ -72,6 +72,7 @@ class CaptureManager {
   * @returns {Object} Session info
   */
  async start({
+    assetId,
    projectId,
    binId,
    clipName,
@ -82,6 +83,7 @@ class CaptureManager {
    listenPort,
    streamKey,
  }) {
+    this._assetIdForHls = assetId || null;
    if (this.state.recording) {
      throw new Error('Capture already in progress');
    }
@ -138,6 +140,34 @@ class CaptureManager {

    const processes = { hires: hiresProcess };
    const uploads = { hires: hiresUpload };
+    let hlsProcess = null;
+    let hlsDir = null;
+    if (isNetwork && this._assetIdForHls) {
+      try {
+        const fs = await import('node:fs');
+        hlsDir = '/live/' + this._assetIdForHls;
+        fs.mkdirSync(hlsDir, { recursive: true });
+        const hlsArgs = [
+          ...inputArgs,
+          '-map', '0:v:0?', '-map', '0:a:0?',
+          '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency',
+          '-pix_fmt', 'yuv420p', '-b:v', '2M', '-g', '60', '-sc_threshold', '0',
+          '-c:a', 'aac', '-b:a', '128k', '-ar', '44100',
+          '-f', 'hls', '-hls_time', '2', '-hls_list_size', '15',
+          '-hls_flags', 'delete_segments+append_list+omit_endlist',
+          '-hls_segment_filename', hlsDir + '/seg-%05d.ts',
+          hlsDir + '/index.m3u8',
+        ];
+        hlsProcess = spawn('ffmpeg', hlsArgs, { stdio: ['ignore', 'pipe', 'pipe'] });
+        hlsProcess.stderr.on('data', (d) => { console.error('[HLS] ' + d); });
+        hlsProcess.on('exit', (c) => console.log('[HLS] exited ' + c));
+        processes.hls = hlsProcess;
+        console.log('[HLS] tee started -> ' + hlsDir);
+      } catch (err) {
+        console.error('[HLS] tee failed:', err.message);
+      }
+    }
+

    hiresProcess.stderr.on('data', (data) => {
      const text = data.toString();
@ -223,9 +253,8 @@ class CaptureManager {
    if (processes.hires) {
      processes.hires.kill('SIGINT');
    }
-    if (processes.proxy) {
-      processes.proxy.kill('SIGINT');
-    }
+    if (processes.proxy) processes.proxy.kill('SIGINT');
+    if (processes.hls) { try { processes.hls.kill('SIGINT'); } catch (_) {} }

    try {
      // Wait for all in-flight S3 uploads to complete
--- a/services/capture/src/index.js
+++ b/services/capture/src/index.js
@ -24,3 +24,103 @@ app.use('/capture', captureRoutes);
 app.listen(PORT, () => {
  console.log(`Wild Dragon Capture Service listening on port ${PORT}`);
 });
+
+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 = process.env.LISTEN_PORT
+    ? parseInt(process.env.LISTEN_PORT, 10)
+    : undefined;
+  const streamKey = process.env.STREAM_KEY || undefined;
+  const sourceUrl = process.env.SOURCE_URL || undefined;
+
+  if (sourceType === 'sdi') {
+    console.warn('[bootstrap] SDI auto-start not supported');
+    return;
+  }
+
+  console.log(`[bootstrap] starting ${sourceType} ingest (listen=${listen} port=${listenPort || 'n/a'})...`);
+  try {
+    const session = await captureManager.start({
+      assetId: process.env.ASSET_ID || null,
+      projectId,
+      binId: process.env.BIN_ID || null,
+      clipName,
+      sourceType,
+      sourceUrl,
+      listen,
+      listenPort,
+      streamKey,
+    });
+    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`);
+
+      try {
+        const res = await fetch(`${MAM_API_URL}/api/v1/assets`, {
+          method: 'POST',
+          headers: { 'Content-Type': 'application/json' },
+          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'));
--- a/services/mam-api/src/db/migrations/001-add-live-status.sql
+++ b/services/mam-api/src/db/migrations/001-add-live-status.sql
@ -0,0 +1,7 @@
+-- 2026-05: add 'live' to asset_status for growing-file ingest
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_enum WHERE enumlabel = 'live' AND enumtypid = 'asset_status'::regtype) THEN
+    ALTER TYPE asset_status ADD VALUE 'live' BEFORE 'ingesting';
+  END IF;
+END $$;
--- a/services/mam-api/src/index.js
+++ b/services/mam-api/src/index.js
@ -68,6 +68,27 @@ app.use('/api/v1/ampp',      amppRouter);
 app.use(errorHandler);

 // ── Start ────────────────────────────────────────────────────────────────────
+import { readdirSync, readFileSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { dirname, join } from 'node:path';
+
+const __dirnameMig = dirname(fileURLToPath(import.meta.url));
+async function runMigrations() {
+  const dir = join(__dirnameMig, 'db', 'migrations');
+  let files = [];
+  try { files = readdirSync(dir).filter(f => f.endsWith('.sql')).sort(); } catch { return; }
+  for (const f of files) {
+    const sql = readFileSync(join(dir, f), 'utf8');
+    try {
+      await pool.query(sql);
+      console.log('[migration] applied ' + f);
+    } catch (err) {
+      console.error('[migration] failed ' + f, err.message);
+    }
+  }
+}
+await runMigrations();
+
 app.listen(PORT, () => {
  const authMode = process.env.AUTH_ENABLED === 'true' ? 'ENABLED' : 'DISABLED (set AUTH_ENABLED=true for production)';
  console.log(`MAM API listening on port ${PORT}`);
--- a/services/mam-api/src/routes/assets.js
+++ b/services/mam-api/src/routes/assets.js
@ -112,38 +112,69 @@ router.post('/', async (req, res, next) => {
      return res.status(400).json({ error: 'projectId and clipName are required' });
    }

-    const id = uuidv4();
-    const thumbnailKey = `thumbnails/${id}.jpg`;
    const durationMs = duration ? Math.round(duration * 1000) : null;

-    const result = await pool.query(
-      `INSERT INTO assets (
-        id, project_id, bin_id,
-        filename, display_name,
-        status, media_type,
-        original_s3_key, proxy_s3_key,
-        duration_ms,
-        created_at, updated_at
-      )
-      VALUES (
-        $1, $2, $3,
-        $4, $4,
-        'processing', 'video',
-        $5, $6,
-        $7,
-        COALESCE($8::timestamptz, NOW()), NOW()
-      )
-      RETURNING *`,
-      [
-        id, projectId, binId || null,
-        clipName,
-        hiresKey || null, proxyKey || null,
-        durationMs,
-        capturedAt || null,
-      ]
+    // Phase 1 growing-files: an asset row may already exist in status='live'
+    // (pre-created at recorder start so the library shows the recording while
+    // it is happening). If so we UPDATE that row instead of inserting a new
+    // one -- otherwise we would have two rows per recording.
+    const existing = await pool.query(
+      `SELECT * FROM assets
+         WHERE project_id = $1 AND display_name = $2 AND status = 'live'
+         ORDER BY created_at DESC LIMIT 1`,
+      [projectId, clipName]
    );

-    const asset = result.rows[0];
+    let id;
+    let asset;
+    if (existing.rows.length > 0) {
+      id = existing.rows[0].id;
+      const upd = await pool.query(
+        `UPDATE assets
+           SET status          = 'processing',
+               original_s3_key = COALESCE($2, original_s3_key),
+               proxy_s3_key    = COALESCE($3, proxy_s3_key),
+               duration_ms     = COALESCE($4, duration_ms),
+               bin_id          = COALESCE(bin_id, $5),
+               updated_at      = NOW()
+         WHERE id = $1
+         RETURNING *`,
+        [id, hiresKey || null, proxyKey || null, durationMs, binId || null]
+      );
+      asset = upd.rows[0];
+    } else {
+      id = uuidv4();
+      const ins = await pool.query(
+        `INSERT INTO assets (
+          id, project_id, bin_id,
+          filename, display_name,
+          status, media_type,
+          original_s3_key, proxy_s3_key,
+          duration_ms,
+          created_at, updated_at
+        )
+        VALUES (
+          $1, $2, $3,
+          $4, $4,
+          'processing', 'video',
+          $5, $6,
+          $7,
+          COALESCE($8::timestamptz, NOW()), NOW()
+        )
+        RETURNING *`,
+        [
+          id, projectId, binId || null,
+          clipName,
+          hiresKey || null, proxyKey || null,
+          durationMs,
+          capturedAt || null,
+        ]
+      );
+      asset = ins.rows[0];
+    }
+
+    const thumbnailKey = `thumbnails/${id}.jpg`;
+

    // Dispatch thumbnail job — proxy already in S3 from capture
    if (proxyKey) {
@ -161,7 +192,7 @@ router.post('/', async (req, res, next) => {
      asset.status = 'ready';
    }

-    res.status(201).json(asset);
+    res.status(existing.rows.length > 0 ? 200 : 201).json(asset);
  } catch (err) {
    next(err);
  }
@ -339,28 +370,20 @@ router.delete('/:id', async (req, res, next) => {
 router.get('/:id/stream', async (req, res, next) => {
  try {
    const { id } = req.params;
-    const result = await pool.query(
-      'SELECT proxy_s3_key FROM assets WHERE id = $1',
-      [id]
-    );
-
-    if (result.rows.length === 0) {
-      return res.status(404).json({ error: 'Asset not found' });
+    const r = await pool.query('SELECT * FROM assets WHERE id = $1', [id]);
+    if (r.rows.length === 0) return res.status(404).json({ error: 'Asset not found' });
+    const a = r.rows[0];
+    if (a.status === 'live') {
+      return res.json({ url: `/live/${a.id}/index.m3u8`, type: 'hls', live: true });
    }
-
-    const { proxy_s3_key } = result.rows[0];
-
-    if (!proxy_s3_key) {
-      return res.status(400).json({ error: 'No proxy available for this asset' });
-    }
-
-    const url = await getSignedUrlForObject(proxy_s3_key);
+    const key = a.proxy_s3_key || a.original_s3_key;
+    if (!key) return res.status(404).json({ error: 'No stream available yet' });
+    const url = await getSignedUrlForObject(key);
    res.json({ url });
  } catch (err) {
    next(err);
  }
 });
-
 // GET /:id/thumbnail - Signed URL for thumbnail image
 router.get('/:id/thumbnail', async (req, res, next) => {
  try {
--- a/services/mam-api/src/routes/recorders.js
+++ b/services/mam-api/src/routes/recorders.js
@ -197,6 +197,22 @@ router.post('/:id/start', async (req, res, next) => {
    // Generate clip name with timestamp
    const clipName = generateClipName(recorder.name);

+    // live-asset: create the asset row right now (status='live') so the library
+    // shows the recording while it is happening. The capture container will
+    // tee an HLS stream into /live/<assetId>/.
+    const assetIdLive = (await import('uuid')).v4();
+    try {
+      await pool.query(
+        `INSERT INTO assets (
+          id, project_id, bin_id, filename, display_name, status, media_type,
+          original_s3_key, created_at, updated_at
+        ) VALUES ($1, $2, NULL, $3, $3, 'live', 'video', $4, NOW(), NOW())`,
+        [assetIdLive, recorder.project_id, clipName, `projects/${recorder.project_id}/masters/${clipName}.mov`]
+      );
+    } catch (e) {
+      console.warn('[recorders] could not pre-create live asset:', e.message);
+    }
+
    // Determine source config and whether this is a listener-mode recorder
    const sourceConfig = recorder.source_config || {};
    const isListener = sourceConfig.mode === 'listener';
@ -224,6 +240,7 @@ router.post('/:id/start', async (req, res, next) => {
      `PROXY_RESOLUTION=${recorder.proxy_resolution}`,
      `PROJECT_ID=${recorder.project_id}`,
      `CLIP_NAME=${clipName}`,
+      `ASSET_ID=${assetIdLive}`,
    ];

    // Add source-specific env vars for SRT/RTMP
@ -250,6 +267,7 @@ router.post('/:id/start', async (req, res, next) => {
        Privileged: true,
        NetworkMode: dockerNetwork,
        PortBindings: Object.keys(portBindings).length > 0 ? portBindings : undefined,
+        Binds: ['/mnt/NVME/MAM/wild-dragon-live:/live'],
      },
      NetworkingConfig: {
        EndpointsConfig: {
--- a/services/web-ui/nginx.conf
+++ b/services/web-ui/nginx.conf
@ -38,6 +38,14 @@ server {
        add_header Cache-Control "no-cache, no-store, must-revalidate";
    }

+    # Live HLS — served from /live (bind-mounted shared volume), low cache so playlist refreshes
+    location /live/ {
+        alias /live/;
+        types { application/vnd.apple.mpegurl m3u8; video/mp2t ts; }
+        add_header Cache-Control "no-cache";
+        add_header Access-Control-Allow-Origin *;
+    }
+
    # API proxy - forward to mam-api service
    location /api/ {
        set $api_upstream http://mam-api:3000;
--- a/services/web-ui/public/index.html
+++ b/services/web-ui/public/index.html
@ -302,6 +302,8 @@
    .first-splash-dot{width:8px;height:8px;background:oklch(55% 0.20 266);border-radius:50%;animation:fsPulse 1.4s ease-in-out infinite}
    @keyframes fsPulse{0%,100%{opacity:.35;transform:scale(.9)}50%{opacity:1;transform:scale(1.1)}}
    .first-splash-title{font-size:13px;color:var(--text-secondary);letter-spacing:.04em}
+    .badge-live { background: oklch(64% 0.22 25 / 0.18); color: oklch(70% 0.22 25); border: 1px solid oklch(64% 0.22 25 / 0.4); animation: liveBlink 1.4s ease-in-out infinite; }
+    @keyframes liveBlink { 0%,100% { opacity: 0.7 } 50% { opacity: 1 } }
  </style>
 </head>
 <body>
@ -477,7 +479,7 @@

 <script src="js/api.js?v=5"></script>
 <script src="js/topbar-strip.js?v=1"></script>
-<script src="js/preview.js?v=3"></script>
+<script src="js/preview.js?v=4"></script>
 <script src="js/selection.js?v=1"></script>
 <script>
  const state = {
@ -703,7 +705,7 @@
  }

  function statusBadgeClass(s) {
-    const map = { ingesting:'badge-ingesting', processing:'badge-processing', ready:'badge-ready', error:'badge-error', archived:'badge-archived' };
+    const map = { live:'badge-live', ingesting:'badge-ingesting', processing:'badge-processing', ready:'badge-ready', error:'badge-error', archived:'badge-archived' };
    return map[s] || 'badge-idle';
  }

--- a/services/web-ui/public/js/preview.js
+++ b/services/web-ui/public/js/preview.js
@ -79,7 +79,15 @@
        if (!r.ok) throw new Error(r.statusText);
        const { url } = await r.json();
        const tag = mt === 'audio' ? 'audio' : 'video';
-        stage.innerHTML = `<${tag} controls autoplay playsinline src="${esc(url)}"></${tag}>`;
+        if (url && url.endsWith('.m3u8')) {
+          stage.innerHTML = `<${tag} id="prevPlayer" controls autoplay playsinline></${tag}>`;
+          const v = stage.querySelector('#prevPlayer');
+          if (v.canPlayType('application/vnd.apple.mpegurl')) { v.src = url; }
+          else if (window.Hls && window.Hls.isSupported()) { const h = new window.Hls(); h.loadSource(url); h.attachMedia(v); }
+          else { await new Promise(r => { const sc = document.createElement('script'); sc.src = 'https://cdn.jsdelivr.net/npm/hls.js@1.5.0/dist/hls.min.js'; sc.onload = r; document.head.appendChild(sc); }); const h = new window.Hls({ liveSyncDuration: 6 }); h.loadSource(url); h.attachMedia(v); }
+        } else {
+          stage.innerHTML = `<${tag} controls autoplay playsinline src="${esc(url)}"></${tag}>`;
+        }
      } catch (err) {
        stage.innerHTML = `<div class="preview-empty">Stream URL failed: ${esc(err.message)}</div>`;
      }