From bdbf3c01387614048b25a04205bdab5b97408077 Mon Sep 17 00:00:00 2001 From: legop3 Date: Tue, 28 Apr 2026 01:52:25 -0400 Subject: [PATCH] reapaltle --- server/index.js | 2 +- server/src/services/healthService.js | 45 +- server/src/services/llmCommentaryService.js | 4 +- server/src/services/replayBuildService.js | 216 +----- server/src/services/replayEngineV2.js | 718 ++++++++++++++++++++ server/src/services/replaySegmentManager.js | 325 --------- 6 files changed, 726 insertions(+), 584 deletions(-) create mode 100644 server/src/services/replayEngineV2.js delete mode 100644 server/src/services/replaySegmentManager.js diff --git a/server/index.js b/server/index.js index 57ab12e6..bec6976e 100644 --- a/server/index.js +++ b/server/index.js @@ -39,6 +39,6 @@ require('./src/services/buttonBoxService'); require('./src/services/sessionService'); require('./src/services/batteryManager'); require('./src/services/replaySocketService'); -require('./src/services/replaySegmentManager'); +require('./src/services/replayEngineV2'); require('./src/services/discordBotService'); require('./src/services/httpServer'); diff --git a/server/src/services/healthService.js b/server/src/services/healthService.js index a6c4d07c..599bae6c 100644 --- a/server/src/services/healthService.js +++ b/server/src/services/healthService.js @@ -3,8 +3,7 @@ const path = require('path'); const roverManager = require('./roverManager'); const { getRoomCameras } = require('./roomCameraService'); const { getRoomCameraState } = require('./roomCameraSnapshotService'); -const { getReplaySources } = require('./replaySourceService'); -const { replaySegmentsDir, segmentSeconds, bufferSeconds } = require('./replaySegmentManager'); +const { getReplayHealthSnapshot } = require('./replayEngineV2'); const ROVER_SNAPSHOT_DIR = process.env.ROVER_SNAPSHOT_DIR || '/var/lib/rover-snapshots'; const HEALTH_INTERVAL_MS = 5000; @@ -17,44 +16,8 @@ let latest = { snapshots: { rovers: [], rooms: [] }, }; -async function collectReplayHealth(now) { - const neededCount = Math.max(1, Math.ceil(20000 / (segmentSeconds * 1000))); - const sources = getReplaySources(); - const list = []; - let readyCount = 0; - for (const source of sources) { - const key = `${source.type}__${source.id}`; - const dir = path.join(replaySegmentsDir, key); - let lastSegmentAt = null; - let recentCount = 0; - try { - const entries = await fsp.readdir(dir, { withFileTypes: true }); - for (const entry of entries) { - if (!entry.isFile() || !entry.name.endsWith('.mp4')) continue; - const stat = await fsp.stat(path.join(dir, entry.name)); - if (stat.mtimeMs > (lastSegmentAt || 0)) { - lastSegmentAt = stat.mtimeMs; - } - if (now - stat.mtimeMs <= bufferSeconds * 1000) { - recentCount += 1; - } - } - } catch { - // directory missing or unreadable - } - const ready = recentCount >= neededCount; - if (ready) readyCount += 1; - list.push({ - type: source.type, - id: source.id, - label: source.label || `${source.type}:${source.id}`, - recentCount, - neededCount, - lastSegmentAt, - ready, - }); - } - return { sources: list, readyCount, totalCount: list.length }; +function collectReplayHealth() { + return getReplayHealthSnapshot(); } async function collectSnapshotHealth(now) { @@ -109,7 +72,7 @@ async function collectSnapshotHealth(now) { async function refreshHealth() { const now = Date.now(); - const replay = await collectReplayHealth(now); + const replay = collectReplayHealth(now); const snapshots = await collectSnapshotHealth(now); latest = { updatedAt: now, replay, snapshots }; } diff --git a/server/src/services/llmCommentaryService.js b/server/src/services/llmCommentaryService.js index 6973d54d..7cb0677e 100644 --- a/server/src/services/llmCommentaryService.js +++ b/server/src/services/llmCommentaryService.js @@ -22,8 +22,8 @@ const ACTIVITY_WINDOW_MS = 60000; const ACTIVITY_BUCKET_MS = 1000; const ACTIVITY_SCORE_WINDOW_MS = 30000; const SELF_TALK_WINDOW_MS = 30 * 60 * 1000; -const MAX_CONTEXT_EVENTS = 15; -const MAX_RUN_HISTORY = 30; +const MAX_CONTEXT_EVENTS = 8; +const MAX_RUN_HISTORY = 100; const MAX_ROVER_EVENTS = 400; const POST_COOLDOWN_MS = 10000; diff --git a/server/src/services/replayBuildService.js b/server/src/services/replayBuildService.js index 8dbe9d7c..3a2eade3 100644 --- a/server/src/services/replayBuildService.js +++ b/server/src/services/replayBuildService.js @@ -1,218 +1,4 @@ -const { execFile } = require('child_process'); -const fsp = require('fs/promises'); -const os = require('os'); -const path = require('path'); -const { promisify } = require('util'); -const logger = require('../globals/logger').child('replayBuild'); -const { replaySegmentsDir, segmentSeconds } = require('./replaySegmentManager'); - -const execFileAsync = promisify(execFile); - -const REPLAY_DURATION_MS = 20000; -const REPLAY_FPS = 15; -const MAX_REPLAY_WIDTH = 1280; -const MAX_REPLAY_HEIGHT = 720; -const REPLAY_MAX_BYTES = Math.floor(9.5 * 1024 * 1024); - -function buildGridLayout(count) { - const cols = Math.ceil(Math.sqrt(count)); - const rows = Math.ceil(count / cols); - return { cols, rows }; -} - -function clampEven(value) { - return Math.max(2, Math.floor(value / 2) * 2); -} - -function buildScalePadFilter(tileWidth, tileHeight) { - return `scale=${tileWidth}:${tileHeight}:force_original_aspect_ratio=decrease:flags=lanczos,pad=${tileWidth}:${tileHeight}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1`; -} - -async function probeMaxFrameSize(paths) { - let maxWidth = 0; - let maxHeight = 0; - for (const filePath of paths) { - try { - const { stdout } = await execFileAsync('ffprobe', [ - '-v', - 'error', - '-select_streams', - 'v:0', - '-show_entries', - 'stream=width,height', - '-of', - 'csv=p=0', - filePath, - ]); - const [widthRaw, heightRaw] = stdout.trim().split(','); - const width = Number(widthRaw); - const height = Number(heightRaw); - if (Number.isFinite(width) && Number.isFinite(height)) { - maxWidth = Math.max(maxWidth, width); - maxHeight = Math.max(maxHeight, height); - } - } catch (err) { - logger.warn('Failed to probe replay clip size', err.message); - } - } - return { maxWidth, maxHeight }; -} - -async function listLatestSegments(sourceKey, neededCount) { - const dir = path.join(replaySegmentsDir, sourceKey); - const entries = await fsp.readdir(dir, { withFileTypes: true }); - const files = []; - const now = Date.now(); - for (const entry of entries) { - if (!entry.isFile() || !entry.name.endsWith('.mp4')) continue; - const filePath = path.join(dir, entry.name); - const stat = await fsp.stat(filePath); - if (stat.size < 16 * 1024) { - continue; - } - if (now - stat.mtimeMs < segmentSeconds * 1000) { - continue; - } - files.push({ filePath, mtimeMs: stat.mtimeMs }); - } - files.sort((a, b) => a.mtimeMs - b.mtimeMs); - return files.slice(-neededCount).map((file) => file.filePath); -} - -async function buildReplayVideo({ sources = [] } = {}) { - if (!sources.length) { - throw new Error('No replay sources selected'); - } - const segmentCount = Math.max(1, Math.ceil(REPLAY_DURATION_MS / (segmentSeconds * 1000))); - const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'rover-replay-')); - try { - const clipPaths = []; - const usedSources = []; - const missingSources = []; - for (let i = 0; i < sources.length; i += 1) { - const source = sources[i]; - const key = `${source.type}__${source.id}`; - let segmentPaths; - try { - segmentPaths = await listLatestSegments(key, segmentCount); - } catch (err) { - missingSources.push({ ...source, reason: err.message || 'missing segments' }); - continue; - } - if (segmentPaths.length < segmentCount) { - missingSources.push({ - ...source, - reason: `only ${segmentPaths.length}/${segmentCount} segments available`, - }); - continue; - } - const concatPath = path.join(tmpDir, `concat-${clipPaths.length}.txt`); - const concatBody = segmentPaths.map((file) => `file '${file}'`).join('\n'); - await fsp.writeFile(concatPath, concatBody); - const clipPath = path.join(tmpDir, `clip-${clipPaths.length}.mp4`); - await execFileAsync('ffmpeg', [ - '-hide_banner', - '-loglevel', - 'error', - '-f', - 'concat', - '-safe', - '0', - '-i', - concatPath, - '-c', - 'copy', - clipPath, - ]); - clipPaths.push(clipPath); - usedSources.push(source); - } - if (!clipPaths.length) { - throw new Error('No replay segments available for selected sources'); - } - if (missingSources.length) { - logger.warn('Replay sources missing segments', { - missing: missingSources.map((source) => `${source.type}:${source.id}`), - }); - } - - const { maxWidth, maxHeight } = await probeMaxFrameSize(clipPaths); - const layout = buildGridLayout(clipPaths.length); - let tileWidth = maxWidth || 640; - let tileHeight = maxHeight || 360; - let outputWidth = tileWidth * layout.cols; - let outputHeight = tileHeight * layout.rows; - if (outputWidth > MAX_REPLAY_WIDTH || outputHeight > MAX_REPLAY_HEIGHT) { - const scale = Math.min(MAX_REPLAY_WIDTH / outputWidth, MAX_REPLAY_HEIGHT / outputHeight); - tileWidth *= scale; - tileHeight *= scale; - outputWidth = tileWidth * layout.cols; - outputHeight = tileHeight * layout.rows; - } - tileWidth = clampEven(tileWidth); - tileHeight = clampEven(tileHeight); - outputWidth = clampEven(tileWidth * layout.cols); - outputHeight = clampEven(tileHeight * layout.rows); - - const inputArgs = []; - const filterParts = []; - const layoutParts = []; - for (let i = 0; i < clipPaths.length; i += 1) { - inputArgs.push('-i', clipPaths[i]); - filterParts.push(`[${i}:v]${buildScalePadFilter(tileWidth, tileHeight)}[v${i}]`); - const x = (i % layout.cols) * tileWidth; - const y = Math.floor(i / layout.cols) * tileHeight; - layoutParts.push(`${x}_${y}`); - } - if (clipPaths.length === 1) { - filterParts.push('[v0]null[v]'); - } else { - filterParts.push( - `${clipPaths.map((_, i) => `[v${i}]`).join('')}` + - `xstack=inputs=${clipPaths.length}:layout=${layoutParts.join('|')}:fill=black[v]`, - ); - } - - const durationSec = Math.max(1, REPLAY_DURATION_MS / 1000); - const targetBitrateKbps = Math.max(300, Math.floor((REPLAY_MAX_BYTES * 8) / durationSec / 1000)); - const maxrateKbps = Math.floor(targetBitrateKbps * 1.1); - const bufsizeKbps = Math.floor(targetBitrateKbps * 2); - - const outPath = path.join(tmpDir, 'replay.mp4'); - await execFileAsync('ffmpeg', [ - '-y', - '-hide_banner', - '-loglevel', - 'error', - ...inputArgs, - '-filter_complex', - filterParts.join(';'), - '-map', - '[v]', - '-r', - String(REPLAY_FPS), - '-c:v', - 'libx264', - '-b:v', - `${targetBitrateKbps}k`, - '-maxrate', - `${maxrateKbps}k`, - '-bufsize', - `${bufsizeKbps}k`, - '-pix_fmt', - 'yuv420p', - outPath, - ]); - const buffer = await fsp.readFile(outPath); - return { buffer, usedSources, missingSources }; - } finally { - try { - await fsp.rm(tmpDir, { recursive: true, force: true }); - } catch (err) { - logger.warn('Failed to cleanup replay temp dir', err.message); - } - } -} +const { buildReplayVideo } = require('./replayEngineV2'); module.exports = { buildReplayVideo, diff --git a/server/src/services/replayEngineV2.js b/server/src/services/replayEngineV2.js new file mode 100644 index 00000000..41f282b6 --- /dev/null +++ b/server/src/services/replayEngineV2.js @@ -0,0 +1,718 @@ +const { spawn, execFile } = require('child_process'); +const fsp = require('fs/promises'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { promisify } = require('util'); +const EventEmitter = require('events'); +const logger = require('../globals/logger').child('replayEngineV2'); +const roverManager = require('./roverManager'); +const { getRoomCameras, roomCameraEvents } = require('./roomCameraService'); + +const execFileAsync = promisify(execFile); + +const FFMPEG_BIN = process.env.FFMPEG_BIN || 'ffmpeg'; +const SEGMENT_ROOT = process.env.REPLAY_SEGMENT_DIR || '/var/lib/replay-segments-v2'; +const SEGMENT_SECONDS = Math.max(1, Number.parseInt(process.env.REPLAY_SEGMENT_SECONDS || '1', 10)); +const BUFFER_SECONDS = Math.max(20, Number.parseInt(process.env.REPLAY_BUFFER_SECONDS || '45', 10)); +const CLEANUP_INTERVAL_MS = 10_000; +const BUILD_DURATION_MS = Math.max(5000, Number.parseInt(process.env.REPLAY_DURATION_MS || '20000', 10)); +const BUILD_GUARD_MS = Math.max(200, Number.parseInt(process.env.REPLAY_GUARD_MS || '1200', 10)); +const TARGET_FPS = Math.max(10, Number.parseInt(process.env.REPLAY_TARGET_FPS || '30', 10)); +const MAX_WIDTH = Math.max(320, Number.parseInt(process.env.REPLAY_MAX_WIDTH || '1280', 10)); +const MAX_HEIGHT = Math.max(180, Number.parseInt(process.env.REPLAY_MAX_HEIGHT || '720', 10)); +const MAX_BYTES = Math.floor(Number.parseFloat(process.env.REPLAY_MAX_OUTPUT_MB || '9.5') * 1024 * 1024); + +const events = new EventEmitter(); + +const workers = new Map(); // key -> worker +const segmentIndex = new Map(); // key -> [{filePath,startMs,endMs,mtimeMs,size,kind,sourceType,sourceId}] +let cleanupTimer = null; + +function sourceKey(source) { + return `${source.sourceType}__${source.kind}__${source.id}`; +} + +function sourceDirForKey(key) { + return path.join(SEGMENT_ROOT, key); +} + +function toSrtReadPath(streamId) { + return `srt://127.0.0.1:9000?streamid=read:${encodeURIComponent(streamId)}`; +} + +function getRoomCameraStream(camera) { + if (camera?.streamUrl) return camera.streamUrl; + const url = String(camera?.url || ''); + if (url.includes('.mjpg') || url.includes('mjpeg') || url.includes('stream')) return url; + return null; +} + +function listDesiredSources() { + const sources = []; + for (const rover of roverManager.getRoster()) { + if (!roverManager.canReplayRoverId(rover.id)) continue; + const roverId = String(rover.id); + sources.push({ + id: roverId, + sourceType: 'rover', + kind: 'video', + label: rover.name || roverId, + inputUrl: toSrtReadPath(roverId), + }); + if (rover?.media?.audioPublishUrl) { + sources.push({ + id: `${roverId}-audio`, + sourceType: 'rover', + roverId, + kind: 'audio', + label: `${rover.name || roverId} audio`, + inputUrl: toSrtReadPath(`${roverId}-audio`), + }); + } + } + + for (const camera of getRoomCameras()) { + const streamUrl = getRoomCameraStream(camera); + if (!streamUrl) continue; + sources.push({ + id: String(camera.id), + sourceType: 'room', + kind: 'video', + label: camera.name || camera.id, + inputUrl: streamUrl, + }); + } + + return sources; +} + +function buildWorkerArgs(source) { + const dir = sourceDirForKey(sourceKey(source)); + const pattern = path.join(dir, 'seg-%06d.mp4'); + const listPath = path.join(dir, 'index.csv'); + const listSize = Math.ceil((BUFFER_SECONDS + 10) / SEGMENT_SECONDS); + + const common = [ + '-hide_banner', + '-loglevel', + 'warning', + '-y', + '-fflags', + '+genpts', + '-use_wallclock_as_timestamps', + '1', + '-i', + source.inputUrl, + ]; + + if (source.kind === 'audio') { + return [ + ...common, + '-vn', + '-ac', + '1', + '-ar', + '48000', + '-af', + 'aresample=48000', + '-c:a', + 'aac', + '-b:a', + '96k', + '-f', + 'segment', + '-segment_time', + String(SEGMENT_SECONDS), + '-segment_atclocktime', + '1', + '-segment_list', + listPath, + '-segment_list_type', + 'csv', + '-segment_list_size', + String(listSize), + '-segment_list_flags', + '+live', + '-reset_timestamps', + '1', + pattern, + ]; + } + + return [ + ...common, + '-an', + '-vf', + `fps=${TARGET_FPS}`, + '-c:v', + 'libx264', + '-preset', + 'veryfast', + '-tune', + 'zerolatency', + '-pix_fmt', + 'yuv420p', + '-g', + String(TARGET_FPS * SEGMENT_SECONDS), + '-keyint_min', + String(TARGET_FPS * SEGMENT_SECONDS), + '-sc_threshold', + '0', + '-f', + 'segment', + '-segment_time', + String(SEGMENT_SECONDS), + '-segment_atclocktime', + '1', + '-segment_list', + listPath, + '-segment_list_type', + 'csv', + '-segment_list_size', + String(listSize), + '-segment_list_flags', + '+live', + '-reset_timestamps', + '1', + pattern, + ]; +} + +async function ensureDir(dir) { + await fsp.mkdir(dir, { recursive: true }); +} + +function startWorker(source) { + const key = sourceKey(source); + if (workers.has(key)) return; + const dir = sourceDirForKey(key); + ensureDir(dir) + .then(() => { + const proc = spawn(FFMPEG_BIN, buildWorkerArgs(source), { stdio: ['ignore', 'ignore', 'pipe'] }); + const worker = { + key, + source, + proc, + startedAtMs: Date.now(), + firstSegmentStartSec: null, + restarting: false, + }; + workers.set(key, worker); + proc.stderr.on('data', (chunk) => { + const text = String(chunk || '').trim(); + if (!text) return; + logger.warn('worker stderr', { key, text: text.slice(0, 500) }); + }); + proc.on('exit', (code, signal) => { + const current = workers.get(key); + if (current?.proc === proc) { + workers.delete(key); + } + logger.warn('worker exited', { key, code, signal }); + setTimeout(() => { + const desired = listDesiredSources().find((entry) => sourceKey(entry) === key); + if (desired && !workers.has(key)) { + startWorker(desired); + } + }, 1500); + }); + }) + .catch((err) => { + logger.warn('failed to start worker', { key, error: err.message }); + }); +} + +function stopWorker(key) { + const worker = workers.get(key); + if (!worker) return; + try { + worker.proc.kill('SIGTERM'); + } catch { + // noop + } + workers.delete(key); +} + +async function syncWorkers() { + const desired = listDesiredSources(); + const desiredKeys = new Set(desired.map(sourceKey)); + + for (const source of desired) { + const key = sourceKey(source); + if (!workers.has(key)) { + startWorker(source); + } + } + + for (const key of Array.from(workers.keys())) { + if (!desiredKeys.has(key)) { + stopWorker(key); + } + } +} + +function parseCsvLine(line) { + const trimmed = String(line || '').trim(); + if (!trimmed) return null; + const parts = trimmed.split(','); + if (parts.length < 3) return null; + return { + filename: parts[0], + startSec: Number(parts[1]), + endSec: Number(parts[2]), + }; +} + +async function refreshIndexForWorker(worker) { + const key = worker.key; + const dir = sourceDirForKey(key); + const csvPath = path.join(dir, 'index.csv'); + let csv; + try { + csv = await fsp.readFile(csvPath, 'utf8'); + } catch { + return; + } + + const lines = csv.split(/\r?\n/).map(parseCsvLine).filter(Boolean); + if (!lines.length) return; + const first = lines[0]; + if (!Number.isFinite(first.startSec) || !Number.isFinite(first.endSec)) return; + + if (worker.firstSegmentStartSec == null) { + worker.firstSegmentStartSec = first.startSec; + } + + const baseMs = worker.startedAtMs - Math.max(0, (worker.firstSegmentStartSec || 0) * 1000); + const cutoffMs = Date.now() - BUFFER_SECONDS * 1000 - 5000; + + const entries = []; + for (const row of lines) { + if (!Number.isFinite(row.startSec) || !Number.isFinite(row.endSec)) continue; + const startMs = Math.round(baseMs + row.startSec * 1000); + const endMs = Math.round(baseMs + row.endSec * 1000); + const filePath = path.join(dir, row.filename); + let stat; + try { + stat = await fsp.stat(filePath); + } catch { + continue; + } + if (!stat.isFile() || stat.size < 4096) continue; + if (endMs < cutoffMs) continue; + entries.push({ + filePath, + startMs, + endMs, + mtimeMs: stat.mtimeMs, + size: stat.size, + kind: worker.source.kind, + sourceType: worker.source.sourceType, + sourceId: worker.source.id, + roverId: worker.source.roverId || null, + }); + } + + segmentIndex.set(key, entries.sort((a, b) => a.startMs - b.startMs)); +} + +async function refreshSegmentIndex() { + const list = Array.from(workers.values()); + for (const worker of list) { + await refreshIndexForWorker(worker); + } +} + +async function cleanupOldFiles() { + const cutoff = Date.now() - BUFFER_SECONDS * 1000; + try { + await ensureDir(SEGMENT_ROOT); + const dirs = await fsp.readdir(SEGMENT_ROOT, { withFileTypes: true }); + for (const dirent of dirs) { + if (!dirent.isDirectory()) continue; + const dirPath = path.join(SEGMENT_ROOT, dirent.name); + let files; + try { + files = await fsp.readdir(dirPath, { withFileTypes: true }); + } catch { + continue; + } + for (const file of files) { + if (!file.isFile() || !file.name.endsWith('.mp4')) continue; + const filePath = path.join(dirPath, file.name); + try { + const stat = await fsp.stat(filePath); + if (stat.mtimeMs < cutoff) { + await fsp.unlink(filePath); + } + } catch { + // noop + } + } + } + } catch (err) { + logger.warn('cleanup failed', err.message); + } +} + +function getVideoEntriesForSource(source) { + const key = sourceKey({ sourceType: String(source.type), kind: 'video', id: String(source.id) }); + return segmentIndex.get(key) || []; +} + +function getAudioEntriesForRover(roverId) { + const key = sourceKey({ sourceType: 'rover', kind: 'audio', id: `${String(roverId)}-audio` }); + return segmentIndex.get(key) || []; +} + +function overlapping(entries, startMs, endMs) { + return entries.filter((entry) => entry.endMs > startMs && entry.startMs < endMs); +} + +function buildGridLayout(count) { + const cols = Math.ceil(Math.sqrt(count)); + const rows = Math.ceil(count / cols); + return { cols, rows }; +} + +function clampEven(value) { + return Math.max(2, Math.floor(value / 2) * 2); +} + +function scalePadFilter(tileWidth, tileHeight) { + return `scale=${tileWidth}:${tileHeight}:force_original_aspect_ratio=decrease:flags=lanczos,pad=${tileWidth}:${tileHeight}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1`; +} + +async function concatFiles(inputPaths, outPath) { + const listPath = `${outPath}.concat.txt`; + const body = inputPaths.map((file) => `file '${file.replace(/'/g, "'\\''")}'`).join('\n'); + await fsp.writeFile(listPath, `${body}\n`, 'utf8'); + await execFileAsync(FFMPEG_BIN, [ + '-y', + '-hide_banner', + '-loglevel', + 'error', + '-f', + 'concat', + '-safe', + '0', + '-i', + listPath, + '-c', + 'copy', + outPath, + ]); +} + +async function probeMaxFrameSize(paths) { + let maxWidth = 0; + let maxHeight = 0; + for (const filePath of paths) { + try { + const { stdout } = await execFileAsync('ffprobe', [ + '-v', + 'error', + '-select_streams', + 'v:0', + '-show_entries', + 'stream=width,height', + '-of', + 'csv=p=0', + filePath, + ]); + const [wRaw, hRaw] = stdout.trim().split(','); + const w = Number(wRaw); + const h = Number(hRaw); + if (Number.isFinite(w) && Number.isFinite(h)) { + maxWidth = Math.max(maxWidth, w); + maxHeight = Math.max(maxHeight, h); + } + } catch { + // noop + } + } + return { maxWidth, maxHeight }; +} + +async function buildReplayVideo({ sources = [] } = {}) { + if (!Array.isArray(sources) || !sources.length) { + throw new Error('No replay sources selected'); + } + + const tEnd = Date.now() - BUILD_GUARD_MS; + const tStart = tEnd - BUILD_DURATION_MS; + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'mrr-replay-v2-')); + + try { + await refreshSegmentIndex(); + + const usedSources = []; + const missingSources = []; + const normalizedVideos = []; + const normalizedAudios = []; + + for (let i = 0; i < sources.length; i += 1) { + const source = sources[i]; + const sourceId = String(source.id); + const videoEntries = overlapping(getVideoEntriesForSource({ id: sourceId }), tStart, tEnd); + if (!videoEntries.length) { + missingSources.push({ ...source, reason: 'no video coverage in replay window' }); + continue; + } + + const videoConcat = path.join(tmpDir, `video-${i}.mp4`); + await concatFiles(videoEntries.map((entry) => entry.filePath), videoConcat); + const videoTrimmed = path.join(tmpDir, `video-${i}.trim.mp4`); + const firstStartMs = videoEntries[0].startMs; + const ss = Math.max(0, (tStart - firstStartMs) / 1000); + const to = Math.max(ss + 0.1, (tEnd - firstStartMs) / 1000); + await execFileAsync(FFMPEG_BIN, [ + '-y', + '-hide_banner', + '-loglevel', + 'error', + '-ss', + ss.toFixed(3), + '-to', + to.toFixed(3), + '-i', + videoConcat, + '-an', + '-c:v', + 'libx264', + '-preset', + 'veryfast', + '-pix_fmt', + 'yuv420p', + '-r', + String(TARGET_FPS), + videoTrimmed, + ]); + + normalizedVideos.push({ path: videoTrimmed, source }); + usedSources.push(source); + + if (source.type === 'rover') { + const audioEntries = overlapping(getAudioEntriesForRover(sourceId), tStart, tEnd); + if (audioEntries.length) { + const audioConcat = path.join(tmpDir, `audio-${i}.m4a`); + await concatFiles(audioEntries.map((entry) => entry.filePath), audioConcat); + const audioTrimmed = path.join(tmpDir, `audio-${i}.trim.m4a`); + const firstAudioStartMs = audioEntries[0].startMs; + const ass = Math.max(0, (tStart - firstAudioStartMs) / 1000); + const ato = Math.max(ass + 0.1, (tEnd - firstAudioStartMs) / 1000); + await execFileAsync(FFMPEG_BIN, [ + '-y', + '-hide_banner', + '-loglevel', + 'error', + '-ss', + ass.toFixed(3), + '-to', + ato.toFixed(3), + '-i', + audioConcat, + '-vn', + '-ac', + '1', + '-ar', + '48000', + '-c:a', + 'aac', + '-b:a', + '96k', + audioTrimmed, + ]); + normalizedAudios.push(audioTrimmed); + } + } + } + + if (!normalizedVideos.length) { + throw new Error('No replay segments available for selected sources'); + } + + const layout = buildGridLayout(normalizedVideos.length); + const { maxWidth, maxHeight } = await probeMaxFrameSize(normalizedVideos.map((v) => v.path)); + let tileWidth = maxWidth || 640; + let tileHeight = maxHeight || 360; + let outWidth = tileWidth * layout.cols; + let outHeight = tileHeight * layout.rows; + if (outWidth > MAX_WIDTH || outHeight > MAX_HEIGHT) { + const scale = Math.min(MAX_WIDTH / outWidth, MAX_HEIGHT / outHeight); + tileWidth *= scale; + tileHeight *= scale; + outWidth = tileWidth * layout.cols; + outHeight = tileHeight * layout.rows; + } + tileWidth = clampEven(tileWidth); + tileHeight = clampEven(tileHeight); + + const inputArgs = []; + const filterParts = []; + const layoutParts = []; + + for (let i = 0; i < normalizedVideos.length; i += 1) { + inputArgs.push('-i', normalizedVideos[i].path); + filterParts.push(`[${i}:v]${scalePadFilter(tileWidth, tileHeight)}[v${i}]`); + const x = (i % layout.cols) * tileWidth; + const y = Math.floor(i / layout.cols) * tileHeight; + layoutParts.push(`${x}_${y}`); + } + + const audioInputStart = normalizedVideos.length; + for (const audioPath of normalizedAudios) { + inputArgs.push('-i', audioPath); + } + + if (normalizedVideos.length === 1) { + filterParts.push('[v0]null[vout]'); + } else { + filterParts.push( + `${normalizedVideos.map((_, i) => `[v${i}]`).join('')}` + + `xstack=inputs=${normalizedVideos.length}:layout=${layoutParts.join('|')}:fill=black[vout]`, + ); + } + + if (normalizedAudios.length) { + const audioRefs = normalizedAudios.map((_, idx) => `[${audioInputStart + idx}:a]`).join(''); + filterParts.push(`${audioRefs}amix=inputs=${normalizedAudios.length}:normalize=0,alimiter=limit=0.9[aout]`); + } + + const durationSec = BUILD_DURATION_MS / 1000; + const targetBitrateKbps = Math.max(400, Math.floor((MAX_BYTES * 8) / durationSec / 1000)); + const outPath = path.join(tmpDir, 'replay.mp4'); + + const args = [ + '-y', + '-hide_banner', + '-loglevel', + 'error', + ...inputArgs, + '-filter_complex', + filterParts.join(';'), + '-map', + '[vout]', + '-c:v', + 'libx264', + '-preset', + 'veryfast', + '-pix_fmt', + 'yuv420p', + '-r', + String(TARGET_FPS), + '-b:v', + `${targetBitrateKbps}k`, + '-maxrate', + `${Math.floor(targetBitrateKbps * 1.15)}k`, + '-bufsize', + `${Math.floor(targetBitrateKbps * 2)}k`, + ]; + + if (normalizedAudios.length) { + args.push('-map', '[aout]', '-c:a', 'aac', '-b:a', '128k'); + } + + args.push(outPath); + + await execFileAsync(FFMPEG_BIN, args); + const buffer = await fsp.readFile(outPath); + return { buffer, usedSources, missingSources }; + } finally { + try { + await fsp.rm(tmpDir, { recursive: true, force: true }); + } catch { + // noop + } + } +} + +function getReplayHealthSnapshot() { + const now = Date.now(); + const neededWindowMs = BUILD_DURATION_MS; + const sources = []; + let readyCount = 0; + + const replaySources = []; + for (const rover of roverManager.getRoster().filter((entry) => roverManager.canReplayRoverId(entry.id))) { + replaySources.push({ type: 'rover', id: String(rover.id), label: rover.name || rover.id }); + } + for (const camera of getRoomCameras()) { + replaySources.push({ type: 'room', id: String(camera.id), label: camera.name || camera.id }); + } + + for (const source of replaySources) { + const entries = getVideoEntriesForSource({ id: source.id }); + const inWindow = entries.filter((entry) => now - entry.endMs <= BUFFER_SECONDS * 1000); + const newest = inWindow[inWindow.length - 1] || null; + const oldest = inWindow[0] || null; + const coveredMs = newest && oldest ? Math.max(0, newest.endMs - oldest.startMs) : 0; + const ready = coveredMs >= neededWindowMs; + if (ready) readyCount += 1; + sources.push({ + type: source.type, + id: source.id, + label: source.label, + recentCount: inWindow.length, + neededCount: Math.ceil(neededWindowMs / 1000), + lastSegmentAt: newest?.endMs || null, + ready, + }); + } + + return { + sources, + readyCount, + totalCount: sources.length, + }; +} + +async function bootstrapIndexFromDisk() { + await ensureDir(SEGMENT_ROOT); + const dirs = await fsp.readdir(SEGMENT_ROOT, { withFileTypes: true }); + for (const dirent of dirs) { + if (!dirent.isDirectory()) continue; + segmentIndex.set(dirent.name, []); + } +} + +async function tick() { + await syncWorkers(); + await refreshSegmentIndex(); + await cleanupOldFiles(); + events.emit('health', getReplayHealthSnapshot()); +} + +async function start() { + await ensureDir(SEGMENT_ROOT); + await bootstrapIndexFromDisk(); + await tick(); + if (cleanupTimer) clearInterval(cleanupTimer); + cleanupTimer = setInterval(() => { + tick().catch((err) => logger.warn('tick failed', err.message)); + }, CLEANUP_INTERVAL_MS); +} + +roomCameraEvents.on('update', () => { + tick().catch((err) => logger.warn('room camera update tick failed', err.message)); +}); +roverManager.managerEvents.on('rover', () => { + tick().catch((err) => logger.warn('rover update tick failed', err.message)); +}); +roverManager.managerEvents.on('private', () => { + tick().catch((err) => logger.warn('private update tick failed', err.message)); +}); + +start().catch((err) => { + logger.warn('replay engine startup failed', err.message); +}); + +module.exports = { + buildReplayVideo, + getReplayHealthSnapshot, + replayEngineEvents: events, + replaySegmentRootDir: SEGMENT_ROOT, + replaySegmentSeconds: SEGMENT_SECONDS, + replayBufferSeconds: BUFFER_SECONDS, +}; diff --git a/server/src/services/replaySegmentManager.js b/server/src/services/replaySegmentManager.js deleted file mode 100644 index 7c82079d..00000000 --- a/server/src/services/replaySegmentManager.js +++ /dev/null @@ -1,325 +0,0 @@ -const { spawn } = require('child_process'); -const fsp = require('fs/promises'); -const path = require('path'); -const logger = require('../globals/logger').child('replaySegments'); -const { getRoomCameras, roomCameraEvents } = require('./roomCameraService'); -const roverManager = require('./roverManager'); - -const SEGMENT_DIR = process.env.REPLAY_SEGMENT_DIR || '/var/lib/replay-segments'; -const SEGMENT_SECONDS = 2; -const BUFFER_SECONDS = 40; -const CLEANUP_INTERVAL_MS = 20000; -const FPS = 15; -const SCALE_WIDTH = 640; -const MAX_BYTES = Number.parseInt(process.env.REPLAY_SEGMENT_MAX_BYTES || '0', 10); -const FFMPEG_BIN = process.env.FFMPEG_BIN || 'ffmpeg'; -const ROVER_SNAPSHOT_DIR = process.env.ROVER_SNAPSHOT_DIR || '/var/lib/rover-snapshots'; -const ROVER_SNAPSHOT_FPS = 3; - -const recorders = new Map(); // key -> { proc, source } -let cleanupTimer = null; - -function sourceKey(source) { - return `${source.type}__${source.id}`; -} - -function getRoomCameraStream(camera) { - if (camera.streamUrl) return camera.streamUrl; - const url = String(camera.url || ''); - if (url.includes('.mjpg') || url.includes('mjpeg') || url.includes('stream')) { - return url; - } - return null; -} - -function listSources() { - const rooms = getRoomCameras().map((camera) => ({ - type: 'room', - id: String(camera.id), - label: camera.name || camera.id, - streamUrl: getRoomCameraStream(camera), - })); - const rovers = roverManager.getRoster().map((rover) => ({ - type: 'rover', - id: String(rover.id), - label: rover.name || rover.id, - })); - return [...rooms, ...rovers]; -} - -function buildInputUrl(source) { - if (source.type === 'room') { - return source.streamUrl || null; - } - return `srt://127.0.0.1:9000?streamid=read:${encodeURIComponent(source.id)}`; -} - -async function ensureDir(dir) { - await fsp.mkdir(dir, { recursive: true }); -} - -function buildOutputPattern(key) { - return path.join(SEGMENT_DIR, key, '%Y%m%d-%H%M%S.mp4'); -} - -function spawnRecorder(source) { - const key = sourceKey(source); - const inputUrl = buildInputUrl(source); - if (!inputUrl) { - logger.warn('Replay recorder missing input URL', { key, source }); - return; - } - const outPattern = buildOutputPattern(key); - const outputSnapshot = source.type === 'rover' - ? path.join(ROVER_SNAPSHOT_DIR, `${source.id}.jpg`) - : null; - ensureDir(path.dirname(outPattern)) - .then(async () => { - if (outputSnapshot) { - await ensureDir(ROVER_SNAPSHOT_DIR); - } - const args = [ - '-hide_banner', - '-loglevel', - 'info', - '-y', - '-fflags', - 'nobuffer', - '-flags', - 'low_delay', - '-i', - inputUrl, - ]; - if (outputSnapshot) { - args.push( - '-filter_complex', - `[0:v]fps=${FPS},scale=${SCALE_WIDTH}:-1[vseg];` + - `[0:v]fps=${ROVER_SNAPSHOT_FPS},scale=${SCALE_WIDTH}:-1[vjpg]`, - '-map', - '[vseg]', - '-an', - '-c:v:0', - 'libx264', - '-preset', - 'veryfast', - '-tune', - 'zerolatency', - '-g', - String(FPS), - '-keyint_min', - String(FPS), - '-sc_threshold', - '0', - '-f', - 'segment', - '-segment_time', - String(SEGMENT_SECONDS), - '-segment_format', - 'mp4', - '-reset_timestamps', - '1', - '-strftime', - '1', - outPattern, - '-map', - '[vjpg]', - '-an', - '-q:v:1', - '8', - '-f', - 'image2', - '-update', - '1', - outputSnapshot, - ); - } else { - args.push( - '-r', - String(FPS), - '-vf', - `fps=${FPS},scale=${SCALE_WIDTH}:-1`, - '-an', - '-c:v', - 'libx264', - '-preset', - 'veryfast', - '-tune', - 'zerolatency', - '-g', - String(FPS), - '-keyint_min', - String(FPS), - '-sc_threshold', - '0', - '-f', - 'segment', - '-segment_time', - String(SEGMENT_SECONDS), - '-segment_format', - 'mp4', - '-reset_timestamps', - '1', - '-strftime', - '1', - outPattern, - ); - } - const proc = spawn(FFMPEG_BIN, args, { stdio: ['ignore', 'ignore', 'pipe'] }); - const stderrChunks = []; - let stderrSize = 0; - proc.stderr.on('data', (chunk) => { - if (!chunk || stderrSize > 8192) return; - stderrChunks.push(chunk); - stderrSize += chunk.length; - }); - recorders.set(key, { proc, source }); - proc.on('exit', (code, signal) => { - recorders.delete(key); - const stderrBuffer = stderrChunks.length ? Buffer.concat(stderrChunks) : null; - if (stderrBuffer && stderrBuffer.length) { - const preview = stderrBuffer.toString('utf8', 0, 600).trim(); - logger.warn('Replay recorder stderr', { - key, - stderrBytes: stderrBuffer.length, - stderrPreview: preview || '', - }); - } else { - logger.warn('Replay recorder stderr', { key, stderrBytes: 0 }); - } - if (!shouldRecord(source)) { - return; - } - const delay = 2000; - logger.warn('Replay recorder exited; restarting', { key, code, signal }); - setTimeout(() => { - if (!recorders.has(key) && shouldRecord(source)) { - spawnRecorder(source); - } - }, delay); - }); - }) - .catch((err) => { - logger.warn('Replay recorder setup failed', { key, err: err.message }); - }); -} - -function stopRecorder(key) { - const entry = recorders.get(key); - if (!entry) return; - entry.proc.kill('SIGTERM'); - recorders.delete(key); -} - -async function removeSourceArtifacts(key, source) { - try { - await fsp.rm(path.join(SEGMENT_DIR, key), { recursive: true, force: true }); - } catch (err) { - logger.warn('Failed to clear replay segments', { key, error: err.message }); - } - if (source?.type === 'rover' && source?.id) { - try { - await fsp.rm(path.join(ROVER_SNAPSHOT_DIR, `${source.id}.jpg`), { force: true }); - } catch (err) { - logger.warn('Failed to clear rover snapshot artifact', { roverId: source.id, error: err.message }); - } - } -} - -function shouldRecord(source) { - if (source.type === 'room') { - return Boolean(source.streamUrl); - } - return roverManager.canReplayRoverId(source.id); -} - -function syncRecorders() { - const sources = listSources(); - const desiredKeys = new Set(); - sources.forEach((source) => { - if (!shouldRecord(source)) return; - const key = sourceKey(source); - desiredKeys.add(key); - if (!recorders.has(key)) { - spawnRecorder(source); - } - }); - Array.from(recorders.keys()).forEach((key) => { - if (!desiredKeys.has(key)) { - const entry = recorders.get(key); - stopRecorder(key); - removeSourceArtifacts(key, entry?.source).catch(() => {}); - } - }); -} - -async function cleanupSegments() { - try { - await ensureDir(SEGMENT_DIR); - const cutoff = Date.now() - BUFFER_SECONDS * 1000; - const entries = await fsp.readdir(SEGMENT_DIR, { withFileTypes: true }); - let totalBytes = 0; - const files = []; - for (const entry of entries) { - if (!entry.isDirectory()) continue; - const dirPath = path.join(SEGMENT_DIR, entry.name); - const inner = await fsp.readdir(dirPath, { withFileTypes: true }); - for (const file of inner) { - if (!file.isFile() || !file.name.endsWith('.mp4')) continue; - const filePath = path.join(dirPath, file.name); - const stats = await fsp.stat(filePath); - totalBytes += stats.size; - files.push({ filePath, mtimeMs: stats.mtimeMs, size: stats.size }); - if (stats.mtimeMs < cutoff) { - await fsp.unlink(filePath); - } - } - const remaining = await fsp.readdir(dirPath); - if (!remaining.length) { - await fsp.rmdir(dirPath); - } - } - if (MAX_BYTES > 0 && totalBytes > MAX_BYTES) { - const overBy = totalBytes - MAX_BYTES; - let freed = 0; - files.sort((a, b) => a.mtimeMs - b.mtimeMs); - for (const file of files) { - if (freed >= overBy) break; - try { - await fsp.unlink(file.filePath); - freed += file.size; - } catch { - // ignore - } - } - } - } catch (err) { - logger.warn('Replay cleanup failed', err.message); - } -} - -function start() { - syncRecorders(); - cleanupSegments(); - if (cleanupTimer) clearInterval(cleanupTimer); - cleanupTimer = setInterval(cleanupSegments, CLEANUP_INTERVAL_MS); -} - -roomCameraEvents.on('update', () => { - syncRecorders(); -}); - -roverManager.managerEvents.on('rover', () => { - syncRecorders(); -}); - -roverManager.managerEvents.on('private', () => { - syncRecorders(); -}); - -start(); - -module.exports = { - replaySegmentsDir: SEGMENT_DIR, - segmentSeconds: SEGMENT_SECONDS, - bufferSeconds: BUFFER_SECONDS, -};