3225 lines
82 KiB
JavaScript
3225 lines
82 KiB
JavaScript
import { spawn } from 'node:child_process';
|
|
import { randomUUID } from 'node:crypto';
|
|
import fs from 'node:fs/promises';
|
|
import { createServer } from 'node:http';
|
|
import path from 'node:path';
|
|
import { Readable } from 'node:stream';
|
|
import { fileURLToPath } from 'node:url';
|
|
|
|
import express from 'express';
|
|
import { WebSocket, WebSocketServer } from 'ws';
|
|
|
|
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
|
const publicDir = path.join(__dirname, '..', 'public');
|
|
|
|
const app = express();
|
|
const server = createServer(app);
|
|
const wss = new WebSocketServer({ noServer: true, perMessageDeflate: false });
|
|
|
|
const PORT = Number(process.env.PORT ?? 3000);
|
|
const FFMPEG_PATH = process.env.FFMPEG_PATH ?? 'ffmpeg';
|
|
const FFPROBE_PATH = process.env.FFPROBE_PATH ?? 'ffprobe';
|
|
const YT_DLP_PATH = process.env.YT_DLP_PATH ?? 'yt-dlp';
|
|
const YT_DLP_FORMAT = process.env.YT_DLP_FORMAT ?? 'best[ext=mp4][vcodec!=none][acodec!=none]/best[vcodec!=none][acodec!=none]/best';
|
|
const YT_DLP_TIMEOUT_MS = clampInteger(process.env.YT_DLP_TIMEOUT_MS, 45 * 1000, 5 * 1000, 120 * 1000);
|
|
const FFMPEG_LOG_LEVEL = process.env.FFMPEG_LOG_LEVEL ?? 'warning';
|
|
const FFMPEG_INPUT_SEEKABLE = process.env.FFMPEG_INPUT_SEEKABLE ?? '0';
|
|
const FFMPEG_HTTP_RECONNECT = parseBoolean(process.env.FFMPEG_HTTP_RECONNECT, true);
|
|
const FFMPEG_HTTP_RECONNECT_DELAY_MAX = clampInteger(process.env.FFMPEG_HTTP_RECONNECT_DELAY_MAX, 2, 0, 30);
|
|
const FFMPEG_HTTP_RECONNECT_MAX_RETRIES = clampInteger(process.env.FFMPEG_HTTP_RECONNECT_MAX_RETRIES, 4, 0, 20);
|
|
const FFMPEG_HTTP_RECONNECT_ON_HTTP_ERROR = process.env.FFMPEG_HTTP_RECONNECT_ON_HTTP_ERROR ?? '5xx';
|
|
const FFMPEG_CAPABILITY_TIMEOUT_MS = 3000;
|
|
const PLAYBACK_CONNECTION_MODE = parsePlaybackConnectionMode(process.env.PLAYBACK_CONNECTION_MODE ?? process.env.PLAYBACK_MODE);
|
|
const METADATA_PROBE_ENABLED = parseBoolean(process.env.METADATA_PROBE_ENABLED, PLAYBACK_CONNECTION_MODE !== 'relay');
|
|
const METADATA_PROBE_TIMEOUT_MS = clampInteger(process.env.METADATA_PROBE_TIMEOUT_MS, 4 * 1000, 1000, 30 * 1000);
|
|
const RECENT_URLS_PATH = process.env.RECENT_URLS_PATH ?? path.join(__dirname, '..', 'data', 'recent-urls.json');
|
|
const RECENT_URL_LIMIT = clampInteger(process.env.RECENT_URL_LIMIT, 12, 1, 50);
|
|
const FAVORITES_PATH = process.env.FAVORITES_PATH ?? path.join(__dirname, '..', 'data', 'favorites.json');
|
|
const FAVORITES_LIMIT = clampInteger(process.env.FAVORITES_LIMIT, 50, 1, 200);
|
|
const LOCAL_VIDEOS_ROOT = parseLocalVideosRoot(process.env.LOCAL_VIDEOS);
|
|
const SESSION_TTL_MS = 60 * 60 * 1000;
|
|
const PLAYBACK_READY_TIMEOUT_MS = 15 * 1000;
|
|
const CLIENT_CLOCK_FRAME_LATE_GRACE_SECONDS = 0.25;
|
|
const FRAME_CONDITION_LOG_INTERVAL_MS = 5000;
|
|
const MAX_WS_BUFFER_BYTES = clampInteger(process.env.MAX_WS_BUFFER_BYTES, 2 * 1024 * 1024, 128 * 1024, 64 * 1024 * 1024);
|
|
const MAX_AUDIO_QUEUE_BYTES = clampInteger(process.env.MAX_AUDIO_QUEUE_BYTES, 4 * 1024 * 1024, 256 * 1024, 128 * 1024 * 1024);
|
|
const MAX_RELAY_BRANCH_QUEUE_BYTES = clampInteger(process.env.MAX_RELAY_BRANCH_QUEUE_BYTES, 8 * 1024 * 1024, 512 * 1024, 256 * 1024 * 1024);
|
|
const MAX_YT_DLP_OUTPUT_BYTES = 8 * 1024 * 1024;
|
|
const RELAY_BRANCH_PAUSE_BYTES = Math.floor(MAX_RELAY_BRANCH_QUEUE_BYTES / 2);
|
|
const JPEG_SOI = Buffer.from([0xff, 0xd8]);
|
|
const JPEG_EOI = Buffer.from([0xff, 0xd9]);
|
|
const BEST_EFFORT_RESUME_MAX_SECONDS = 30 * 24 * 60 * 60;
|
|
const RECORDED_MEDIA_EXTENSIONS = new Set([
|
|
'.avi',
|
|
'.flv',
|
|
'.m4v',
|
|
'.mkv',
|
|
'.mov',
|
|
'.mp4',
|
|
'.mpeg',
|
|
'.mpg',
|
|
'.ogv',
|
|
'.webm',
|
|
'.wmv',
|
|
]);
|
|
|
|
const defaults = {
|
|
fps: clampInteger(process.env.DEFAULT_FPS, 24, 1, 30),
|
|
width: clampInteger(process.env.DEFAULT_FRAME_WIDTH, 960, 160, 1920),
|
|
quality: clampInteger(process.env.JPEG_QUALITY, 7, 2, 18),
|
|
audioBitrate: parseAudioBitrate(process.env.DEFAULT_AUDIO_BITRATE, '160k'),
|
|
audioChannels: clampInteger(process.env.DEFAULT_AUDIO_CHANNELS, 2, 1, 2),
|
|
audioSampleRate: clampInteger(process.env.DEFAULT_AUDIO_SAMPLE_RATE, 48000, 22050, 48000),
|
|
};
|
|
|
|
const sessions = new Map();
|
|
const sourceTokens = new Map();
|
|
const playbacks = new Map();
|
|
let recentUrls = [];
|
|
let recentWrite = Promise.resolve();
|
|
let favorites = [];
|
|
let favoritesWrite = Promise.resolve();
|
|
let ffmpegSupportsReconnectMaxRetries = false;
|
|
let localVideosRealRootPromise = null;
|
|
|
|
app.disable('x-powered-by');
|
|
app.use(express.json({ limit: '256kb' }));
|
|
app.use(express.static(publicDir));
|
|
|
|
app.get('/api/health', (_request, response) => {
|
|
response.json({ ok: true, ffmpeg: FFMPEG_PATH, playbackConnectionMode: PLAYBACK_CONNECTION_MODE });
|
|
});
|
|
|
|
app.get('/api/recent-urls', (_request, response) => {
|
|
response.json({
|
|
urls: recentUrls.map((item) => ({
|
|
url: item.url,
|
|
displayUrl: redactSecrets(item.url),
|
|
lastPlayedAt: item.lastPlayedAt,
|
|
})),
|
|
});
|
|
});
|
|
|
|
app.post('/api/recent-urls', async (request, response) => {
|
|
let url;
|
|
|
|
try {
|
|
url = parseStreamUrl(request.body?.url);
|
|
} catch (error) {
|
|
response.status(400).json({ error: error.message });
|
|
return;
|
|
}
|
|
|
|
try {
|
|
await addRecentUrl(url);
|
|
} catch (error) {
|
|
console.warn(`Failed to store recent URL: ${error.message}`);
|
|
response.status(500).json({ error: 'Failed to store recent URL.' });
|
|
return;
|
|
}
|
|
|
|
response.status(201).json({
|
|
urls: recentUrls.map((item) => ({
|
|
url: item.url,
|
|
displayUrl: redactSecrets(item.url),
|
|
lastPlayedAt: item.lastPlayedAt,
|
|
})),
|
|
});
|
|
});
|
|
|
|
app.get('/api/favorites', (_request, response) => {
|
|
response.json({ favorites: formatFavoritesPayload() });
|
|
});
|
|
|
|
app.get('/api/local-videos', async (_request, response) => {
|
|
if (!LOCAL_VIDEOS_ROOT) {
|
|
response.json({ enabled: false, videos: [] });
|
|
return;
|
|
}
|
|
|
|
try {
|
|
response.json({
|
|
enabled: true,
|
|
rootName: formatLocalVideosRootName(),
|
|
videos: await listLocalVideos(),
|
|
});
|
|
} catch (error) {
|
|
logWarn(`local videos listing failed error=${oneLine(error.message)}`);
|
|
response.json({
|
|
enabled: true,
|
|
rootName: formatLocalVideosRootName(),
|
|
videos: [],
|
|
error: 'Failed to read local videos.',
|
|
});
|
|
}
|
|
});
|
|
|
|
app.put('/api/favorites', async (request, response) => {
|
|
let nextFavorites;
|
|
|
|
try {
|
|
nextFavorites = parseFavoritesPayload(request.body?.favorites);
|
|
} catch (error) {
|
|
response.status(400).json({ error: error.message });
|
|
return;
|
|
}
|
|
|
|
favorites = nextFavorites;
|
|
favoritesWrite = favoritesWrite.catch(() => {}).then(() => saveFavorites());
|
|
|
|
try {
|
|
await favoritesWrite;
|
|
} catch (error) {
|
|
console.warn(`Failed to store favorites: ${error.message}`);
|
|
response.status(500).json({ error: 'Failed to store favorites.' });
|
|
return;
|
|
}
|
|
|
|
response.json({ favorites: formatFavoritesPayload() });
|
|
});
|
|
|
|
app.post('/api/session', async (request, response) => {
|
|
let requestedSource;
|
|
let source;
|
|
|
|
try {
|
|
requestedSource = parseSessionSource(request.body);
|
|
} catch (error) {
|
|
response.status(400).json({ error: error.message });
|
|
return;
|
|
}
|
|
|
|
try {
|
|
source = requestedSource.type === 'local'
|
|
? await resolveLocalPlaybackSource(requestedSource.path)
|
|
: await resolvePlaybackSource(requestedSource.url);
|
|
} catch (error) {
|
|
response.status(400).json({ error: error.message });
|
|
return;
|
|
}
|
|
|
|
const options = parsePlaybackOptions(request.body);
|
|
const id = randomUUID();
|
|
const shouldProbeMetadata = METADATA_PROBE_ENABLED || canBestEffortResumeWithoutDuration({
|
|
sourceKind: source.kind,
|
|
originalUrl: source.originalUrl,
|
|
url: source.url,
|
|
});
|
|
const metadataStatus = shouldProbeMetadata ? 'pending' : 'disabled';
|
|
const session = {
|
|
id,
|
|
url: source.url,
|
|
originalUrl: source.originalUrl,
|
|
sourceKind: source.kind,
|
|
sourceHeaders: source.headers,
|
|
options,
|
|
duration: null,
|
|
metadataStatus,
|
|
metadataProbePromise: null,
|
|
seekSeconds: 0,
|
|
seekGeneration: 0,
|
|
forceSplitPlayback: false,
|
|
createdAt: Date.now(),
|
|
lastUsedAt: Date.now(),
|
|
};
|
|
|
|
sessions.set(id, session);
|
|
|
|
logInfo(`session created id=${shortId(id)} source=${source.kind} mode=${getSessionPlaybackConnectionMode(session)} fps=${options.fps} width=${options.width} quality=${options.quality}`);
|
|
|
|
if (shouldProbeMetadata) {
|
|
startSessionMetadataProbe(session);
|
|
}
|
|
|
|
if (requestedSource.type === 'url') {
|
|
try {
|
|
await addRecentUrl(requestedSource.url);
|
|
} catch (error) {
|
|
console.warn(`Failed to store recent URL: ${error.message}`);
|
|
}
|
|
}
|
|
|
|
response.status(201).json(formatSessionPayload(sessions.get(id)));
|
|
});
|
|
|
|
app.get('/api/session/:sessionId', (request, response) => {
|
|
const session = getSession(request.params.sessionId);
|
|
|
|
if (!session) {
|
|
response.status(404).json({ error: 'Unknown or expired session.' });
|
|
return;
|
|
}
|
|
|
|
response.json(formatSessionPayload(session));
|
|
});
|
|
|
|
app.post('/api/session/:sessionId/seek', async (request, response) => {
|
|
const session = getSession(request.params.sessionId);
|
|
|
|
if (!session) {
|
|
response.status(404).json({ error: 'Unknown or expired session.' });
|
|
return;
|
|
}
|
|
|
|
const requestedSeconds = Number(request.body?.time);
|
|
|
|
if (!Number.isFinite(requestedSeconds)) {
|
|
response.status(400).json({ error: 'A numeric seek time is required.' });
|
|
return;
|
|
}
|
|
|
|
const allowBestEffort = request.body?.allowBestEffort === true;
|
|
|
|
if (!isSessionSeekable(session) && allowBestEffort && session.metadataStatus !== 'unavailable') {
|
|
await ensureSessionDuration(session);
|
|
}
|
|
|
|
const canSeekExactly = isSessionSeekable(session);
|
|
const canResumeBestEffort = allowBestEffort && canBestEffortResumeWithoutDuration(session);
|
|
|
|
if (!canSeekExactly && !canResumeBestEffort) {
|
|
response.status(409).json({ error: 'This stream is not seekable.' });
|
|
return;
|
|
}
|
|
|
|
const seekSeconds = canSeekExactly
|
|
? clampNumber(requestedSeconds, 0, session.duration)
|
|
: clampNumber(requestedSeconds, 0, BEST_EFFORT_RESUME_MAX_SECONDS);
|
|
const playback = playbacks.get(session.id);
|
|
const stopReason = allowBestEffort ? 'client_recovery_seek' : 'client_seek';
|
|
|
|
session.seekSeconds = seekSeconds;
|
|
session.seekGeneration += 1;
|
|
session.lastUsedAt = Date.now();
|
|
|
|
if (!canSeekExactly && canResumeBestEffort && PLAYBACK_CONNECTION_MODE === 'relay') {
|
|
session.forceSplitPlayback = true;
|
|
}
|
|
|
|
if (playback && !playback.closed) {
|
|
playback.stop(stopReason);
|
|
if (playbacks.get(session.id) === playback) {
|
|
playbacks.delete(session.id);
|
|
}
|
|
}
|
|
|
|
logInfo(`session seeked id=${shortId(session.id)} reason=${stopReason} time=${seekSeconds.toFixed(3)} duration=${Number.isFinite(session.duration) ? session.duration.toFixed(3) : 'unknown'} exact=${canSeekExactly}`);
|
|
response.json(formatSessionPayload(session));
|
|
});
|
|
|
|
app.all('/_source/:token', async (request, response) => {
|
|
const source = sourceTokens.get(request.params.token);
|
|
const session = source ? getSession(source.sessionId) : null;
|
|
const startedAt = Date.now();
|
|
const sourceLabel = source ? `${source.kind}:${shortId(source.sessionId)}` : `unknown:${request.params.token.slice(0, 8)}`;
|
|
let bytes = 0;
|
|
let upstreamEnded = false;
|
|
let upstreamStatus = 'pending';
|
|
|
|
if (!session) {
|
|
response.status(404).end();
|
|
return;
|
|
}
|
|
|
|
const controller = new AbortController();
|
|
const logClose = once(() => {
|
|
logInfo(`source closed kind=${sourceLabel} status=${upstreamStatus} bytes=${bytes} upstreamEnded=${upstreamEnded} durationMs=${Date.now() - startedAt}`);
|
|
});
|
|
const cleanup = once(() => {
|
|
controller.abort();
|
|
logClose();
|
|
});
|
|
response.on('close', cleanup);
|
|
|
|
try {
|
|
const headers = createSourceRequestHeaders(session);
|
|
const range = request.get('range');
|
|
|
|
if (range) {
|
|
headers.set('range', range);
|
|
}
|
|
|
|
headers.set('accept-encoding', 'identity');
|
|
|
|
const upstream = await fetch(session.url, {
|
|
method: request.method === 'HEAD' ? 'HEAD' : 'GET',
|
|
headers,
|
|
redirect: 'follow',
|
|
signal: controller.signal,
|
|
});
|
|
|
|
upstreamStatus = String(upstream.status);
|
|
logInfo(`source connected kind=${sourceLabel} status=${upstream.status} contentType=${upstream.headers.get('content-type') ?? 'unknown'}`);
|
|
|
|
response.status(upstream.status);
|
|
copyUpstreamHeaders(upstream.headers, response);
|
|
|
|
if (request.method === 'HEAD' || !upstream.body) {
|
|
response.end();
|
|
return;
|
|
}
|
|
|
|
Readable.fromWeb(upstream.body).on('data', (chunk) => {
|
|
bytes += chunk.length;
|
|
}).on('end', () => {
|
|
upstreamEnded = true;
|
|
}).on('error', (error) => {
|
|
if (!controller.signal.aborted) {
|
|
logWarn(`source stream error kind=${sourceLabel} error=${oneLine(error.message)}`);
|
|
}
|
|
|
|
if (!response.destroyed) {
|
|
response.destroy(error);
|
|
}
|
|
}).pipe(response);
|
|
} catch (error) {
|
|
if (controller.signal.aborted) {
|
|
return;
|
|
}
|
|
|
|
logWarn(`source failed kind=${sourceLabel} error=${oneLine(error.message)}`);
|
|
|
|
if (!response.headersSent) {
|
|
response.status(502).json({ error: 'Failed to fetch source stream.' });
|
|
} else {
|
|
response.destroy(error);
|
|
}
|
|
}
|
|
});
|
|
|
|
app.get('/audio/:sessionId', (request, response) => {
|
|
const session = getSession(request.params.sessionId);
|
|
|
|
if (!session) {
|
|
response.status(404).json({ error: 'Unknown or expired session.' });
|
|
return;
|
|
}
|
|
|
|
const playbackMode = getSessionPlaybackConnectionMode(session);
|
|
|
|
if (playbackMode === 'single' || playbackMode === 'relay') {
|
|
getOrCreatePlayback(session, playbackMode).attachAudio(request, response);
|
|
return;
|
|
}
|
|
|
|
stopSharedPlaybackIfNeeded(session, playbackMode);
|
|
streamSplitAudio(request, response, session);
|
|
});
|
|
|
|
server.on('upgrade', (request, socket, head) => {
|
|
const host = request.headers.host ?? 'localhost';
|
|
const { pathname } = new URL(request.url ?? '/', `http://${host}`);
|
|
const match = pathname.match(/^\/frames\/([0-9a-f-]+)$/i);
|
|
|
|
if (!match || !getSession(match[1])) {
|
|
socket.write('HTTP/1.1 404 Not Found\r\n\r\n');
|
|
socket.destroy();
|
|
return;
|
|
}
|
|
|
|
wss.handleUpgrade(request, socket, head, (websocket) => {
|
|
wss.emit('connection', websocket, request, match[1]);
|
|
});
|
|
});
|
|
|
|
wss.on('connection', (websocket, _request, sessionId) => {
|
|
const session = getSession(sessionId);
|
|
|
|
if (!session) {
|
|
websocket.close(1008, 'Unknown session');
|
|
return;
|
|
}
|
|
|
|
const playbackMode = getSessionPlaybackConnectionMode(session);
|
|
|
|
if (playbackMode === 'single' || playbackMode === 'relay') {
|
|
getOrCreatePlayback(session, playbackMode).attachFrames(websocket);
|
|
return;
|
|
}
|
|
|
|
stopSharedPlaybackIfNeeded(session, playbackMode);
|
|
streamSplitFrames(websocket, session);
|
|
});
|
|
|
|
setInterval(() => {
|
|
const now = Date.now();
|
|
|
|
for (const [id, session] of sessions) {
|
|
if (now - session.lastUsedAt > SESSION_TTL_MS) {
|
|
sessions.delete(id);
|
|
}
|
|
}
|
|
|
|
for (const [token, source] of sourceTokens) {
|
|
if (now - source.createdAt > SESSION_TTL_MS) {
|
|
sourceTokens.delete(token);
|
|
}
|
|
}
|
|
}, 60 * 1000).unref();
|
|
|
|
await Promise.all([
|
|
loadRecentUrls(),
|
|
loadFavorites(),
|
|
detectFfmpegHttpReconnectSupport(),
|
|
]);
|
|
|
|
server.listen(PORT, () => {
|
|
console.log(`Frame stream app listening at http://localhost:${PORT} mode=${PLAYBACK_CONNECTION_MODE}`);
|
|
});
|
|
|
|
function parseSessionSource(body) {
|
|
if (typeof body?.localPath === 'string' && body.localPath.trim()) {
|
|
return {
|
|
type: 'local',
|
|
path: parseLocalVideoPath(body.localPath),
|
|
};
|
|
}
|
|
|
|
return {
|
|
type: 'url',
|
|
url: parseStreamUrl(body?.url),
|
|
};
|
|
}
|
|
|
|
function parseStreamUrl(value) {
|
|
if (typeof value !== 'string' || value.trim().length === 0) {
|
|
throw new Error('A stream URL is required.');
|
|
}
|
|
|
|
if (value.length > 4096) {
|
|
throw new Error('The stream URL is too long.');
|
|
}
|
|
|
|
const parsed = new URL(value.trim());
|
|
|
|
if (!['http:', 'https:'].includes(parsed.protocol)) {
|
|
throw new Error('Only http and https stream URLs are supported.');
|
|
}
|
|
|
|
return parsed.toString();
|
|
}
|
|
|
|
function parseLocalVideosRoot(value) {
|
|
if (typeof value !== 'string' || value.trim().length === 0) {
|
|
return null;
|
|
}
|
|
|
|
return path.resolve(value.trim());
|
|
}
|
|
|
|
function parseLocalVideoPath(value) {
|
|
if (typeof value !== 'string' || value.trim().length === 0) {
|
|
throw new Error('A local file is required.');
|
|
}
|
|
|
|
const trimmed = value.trim();
|
|
|
|
if (trimmed.length > 4096) {
|
|
throw new Error('The local file path is too long.');
|
|
}
|
|
|
|
if (trimmed.includes('\0')) {
|
|
throw new Error('The local file path is invalid.');
|
|
}
|
|
|
|
if (path.isAbsolute(trimmed)) {
|
|
throw new Error('Local file paths must be relative.');
|
|
}
|
|
|
|
const segments = trimmed.split(/[\\/]+/);
|
|
|
|
if (segments.some((segment) => !segment || segment === '.' || segment === '..')) {
|
|
throw new Error('The local file path is invalid.');
|
|
}
|
|
|
|
return path.join(...segments);
|
|
}
|
|
|
|
async function resolveLocalPlaybackSource(relativePath) {
|
|
const { absolutePath, displayPath } = await resolveLocalVideoPath(relativePath);
|
|
|
|
return {
|
|
kind: 'local',
|
|
url: absolutePath,
|
|
originalUrl: displayPath,
|
|
headers: {},
|
|
};
|
|
}
|
|
|
|
async function resolveLocalVideoPath(relativePath) {
|
|
if (!LOCAL_VIDEOS_ROOT) {
|
|
throw new Error('Local video playback is not enabled.');
|
|
}
|
|
|
|
const parsedPath = parseLocalVideoPath(relativePath);
|
|
const rootRealPath = await getLocalVideosRealRoot();
|
|
const candidatePath = path.resolve(LOCAL_VIDEOS_ROOT, parsedPath);
|
|
|
|
if (!isPathInside(candidatePath, LOCAL_VIDEOS_ROOT)) {
|
|
throw new Error('The local file path is invalid.');
|
|
}
|
|
|
|
let realPath;
|
|
let stats;
|
|
|
|
try {
|
|
realPath = await fs.realpath(candidatePath);
|
|
stats = await fs.stat(realPath);
|
|
} catch {
|
|
throw new Error('Local file was not found.');
|
|
}
|
|
|
|
if (!isPathInside(realPath, rootRealPath)) {
|
|
throw new Error('The local file path is outside the configured library.');
|
|
}
|
|
|
|
if (!stats.isFile()) {
|
|
throw new Error('Local selection is not a file.');
|
|
}
|
|
|
|
return {
|
|
absolutePath: realPath,
|
|
displayPath: formatLocalVideoPath(parsedPath),
|
|
};
|
|
}
|
|
|
|
async function listLocalVideos() {
|
|
if (!LOCAL_VIDEOS_ROOT) {
|
|
return [];
|
|
}
|
|
|
|
const rootRealPath = await getLocalVideosRealRoot();
|
|
const videos = [];
|
|
|
|
await walkLocalVideosDirectory(LOCAL_VIDEOS_ROOT, '', rootRealPath, videos);
|
|
|
|
return videos.sort((first, second) => first.path.localeCompare(second.path, undefined, {
|
|
numeric: true,
|
|
sensitivity: 'base',
|
|
}));
|
|
}
|
|
|
|
async function walkLocalVideosDirectory(directoryPath, relativeDirectory, rootRealPath, videos) {
|
|
let entries;
|
|
|
|
try {
|
|
entries = await fs.readdir(directoryPath, { withFileTypes: true });
|
|
} catch (error) {
|
|
const label = relativeDirectory ? formatLocalVideoPath(relativeDirectory) : '.';
|
|
logWarn(`local videos directory unreadable path=${label} error=${oneLine(error.message)}`);
|
|
return;
|
|
}
|
|
|
|
for (const entry of entries) {
|
|
const relativePath = relativeDirectory ? path.join(relativeDirectory, entry.name) : entry.name;
|
|
const absolutePath = path.join(directoryPath, entry.name);
|
|
|
|
if (entry.isDirectory()) {
|
|
await walkLocalVideosDirectory(absolutePath, relativePath, rootRealPath, videos);
|
|
continue;
|
|
}
|
|
|
|
if (entry.isFile() || entry.isSymbolicLink()) {
|
|
await addLocalVideoEntry(absolutePath, relativePath, rootRealPath, videos);
|
|
}
|
|
}
|
|
}
|
|
|
|
async function addLocalVideoEntry(absolutePath, relativePath, rootRealPath, videos) {
|
|
let realPath;
|
|
let stats;
|
|
|
|
try {
|
|
realPath = await fs.realpath(absolutePath);
|
|
stats = await fs.stat(realPath);
|
|
} catch {
|
|
return;
|
|
}
|
|
|
|
if (!stats.isFile() || !isPathInside(realPath, rootRealPath)) {
|
|
return;
|
|
}
|
|
|
|
const displayPath = formatLocalVideoPath(relativePath);
|
|
const folder = formatLocalVideoPath(path.dirname(relativePath));
|
|
|
|
videos.push({
|
|
path: displayPath,
|
|
title: path.basename(relativePath),
|
|
folder: folder === '.' ? '' : folder,
|
|
size: stats.size,
|
|
modifiedAt: stats.mtime.toISOString(),
|
|
});
|
|
}
|
|
|
|
function getLocalVideosRealRoot() {
|
|
if (!LOCAL_VIDEOS_ROOT) {
|
|
return Promise.reject(new Error('Local video playback is not enabled.'));
|
|
}
|
|
|
|
if (!localVideosRealRootPromise) {
|
|
localVideosRealRootPromise = fs.realpath(LOCAL_VIDEOS_ROOT).catch((error) => {
|
|
localVideosRealRootPromise = null;
|
|
throw new Error(`Local videos directory is unavailable: ${error.message}`);
|
|
});
|
|
}
|
|
|
|
return localVideosRealRootPromise;
|
|
}
|
|
|
|
function isPathInside(candidatePath, rootPath) {
|
|
const relativePath = path.relative(rootPath, candidatePath);
|
|
return relativePath === '' || (!relativePath.startsWith('..') && !path.isAbsolute(relativePath));
|
|
}
|
|
|
|
function formatLocalVideoPath(value) {
|
|
return value.split(path.sep).join('/');
|
|
}
|
|
|
|
function formatLocalVideosRootName() {
|
|
return path.basename(LOCAL_VIDEOS_ROOT) || LOCAL_VIDEOS_ROOT;
|
|
}
|
|
|
|
function parseResolvedMediaUrl(value) {
|
|
if (typeof value !== 'string' || value.trim().length === 0) {
|
|
throw new Error('Resolved media URL is empty.');
|
|
}
|
|
|
|
if (value.length > 32768) {
|
|
throw new Error('Resolved media URL is too long.');
|
|
}
|
|
|
|
const parsed = new URL(value.trim());
|
|
|
|
if (!['http:', 'https:'].includes(parsed.protocol)) {
|
|
throw new Error('Resolved media URL is not an HTTP(S) stream.');
|
|
}
|
|
|
|
return parsed.toString();
|
|
}
|
|
|
|
async function resolvePlaybackSource(url) {
|
|
if (!isYouTubeUrl(url)) {
|
|
return {
|
|
kind: 'direct',
|
|
url,
|
|
originalUrl: url,
|
|
headers: {},
|
|
};
|
|
}
|
|
|
|
const resolved = await resolveYouTubeSource(url);
|
|
return {
|
|
kind: 'youtube',
|
|
url: resolved.url,
|
|
originalUrl: url,
|
|
headers: resolved.headers,
|
|
};
|
|
}
|
|
|
|
function isYouTubeUrl(value) {
|
|
let parsed;
|
|
|
|
try {
|
|
parsed = new URL(value);
|
|
} catch {
|
|
return false;
|
|
}
|
|
|
|
const hostname = parsed.hostname.toLowerCase();
|
|
|
|
return hostname === 'youtu.be'
|
|
|| hostname.endsWith('.youtu.be')
|
|
|| hostname === 'youtube.com'
|
|
|| hostname.endsWith('.youtube.com')
|
|
|| hostname === 'youtube-nocookie.com'
|
|
|| hostname.endsWith('.youtube-nocookie.com');
|
|
}
|
|
|
|
async function resolveYouTubeSource(url) {
|
|
const startedAt = Date.now();
|
|
logInfo(`yt-dlp resolving source host=${new URL(url).hostname}`);
|
|
|
|
let payload;
|
|
|
|
try {
|
|
payload = await runYtDlpJson([
|
|
'--no-playlist',
|
|
'--no-warnings',
|
|
'--no-progress',
|
|
'--dump-single-json',
|
|
'--format',
|
|
YT_DLP_FORMAT,
|
|
url,
|
|
]);
|
|
} catch (error) {
|
|
throw new Error(`yt-dlp failed to resolve YouTube URL: ${error.message}`);
|
|
}
|
|
|
|
const mediaUrl = typeof payload?.url === 'string' ? payload.url : '';
|
|
|
|
if (!mediaUrl) {
|
|
throw new Error('yt-dlp did not return a playable media URL.');
|
|
}
|
|
|
|
const parsed = parseResolvedMediaUrl(mediaUrl);
|
|
logInfo(`yt-dlp resolved source host=${new URL(parsed).hostname} durationMs=${Date.now() - startedAt}`);
|
|
|
|
return {
|
|
url: parsed,
|
|
headers: normalizeSourceHeaders(payload?.http_headers),
|
|
};
|
|
}
|
|
|
|
function runYtDlpJson(args) {
|
|
return new Promise((resolve, reject) => {
|
|
const ytDlp = spawn(YT_DLP_PATH, args, {
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
});
|
|
let stdout = '';
|
|
let stdoutBytes = 0;
|
|
let stderr = '';
|
|
let timedOut = false;
|
|
let settled = false;
|
|
|
|
const finish = (callback) => {
|
|
if (settled) {
|
|
return;
|
|
}
|
|
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
callback();
|
|
};
|
|
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
stopProcess(ytDlp);
|
|
}, YT_DLP_TIMEOUT_MS);
|
|
timer.unref();
|
|
|
|
ytDlp.stdout.on('data', (chunk) => {
|
|
stdoutBytes += chunk.length;
|
|
|
|
if (stdoutBytes > MAX_YT_DLP_OUTPUT_BYTES) {
|
|
finish(() => {
|
|
stopProcess(ytDlp);
|
|
reject(new Error('yt-dlp output was too large.'));
|
|
});
|
|
return;
|
|
}
|
|
|
|
stdout += chunk.toString('utf8');
|
|
});
|
|
|
|
ytDlp.stderr.on('data', (chunk) => {
|
|
stderr = appendTail(stderr, chunk);
|
|
});
|
|
|
|
ytDlp.on('error', (error) => {
|
|
finish(() => {
|
|
if (error.code === 'ENOENT') {
|
|
reject(new Error(`yt-dlp was not found at "${YT_DLP_PATH}".`));
|
|
return;
|
|
}
|
|
|
|
reject(error);
|
|
});
|
|
});
|
|
|
|
ytDlp.on('close', (code, signal) => {
|
|
finish(() => {
|
|
if (timedOut) {
|
|
reject(new Error('yt-dlp timed out.'));
|
|
return;
|
|
}
|
|
|
|
if (code !== 0) {
|
|
const detail = redactSecrets(stderr).trim();
|
|
reject(new Error(detail ? `yt-dlp exited with code ${code}: ${detail}` : `yt-dlp exited with code ${code ?? 'null'} signal ${signal ?? 'none'}.`));
|
|
return;
|
|
}
|
|
|
|
try {
|
|
resolve(JSON.parse(stdout));
|
|
} catch (error) {
|
|
reject(new Error(`yt-dlp returned invalid JSON: ${error.message}`));
|
|
}
|
|
});
|
|
});
|
|
});
|
|
}
|
|
|
|
function normalizeSourceHeaders(value) {
|
|
if (!value || typeof value !== 'object' || Array.isArray(value)) {
|
|
return {};
|
|
}
|
|
|
|
const headers = {};
|
|
|
|
for (const [name, rawValue] of Object.entries(value)) {
|
|
const normalizedName = name.toLowerCase();
|
|
|
|
if (!isAllowedSourceHeader(normalizedName) || typeof rawValue !== 'string') {
|
|
continue;
|
|
}
|
|
|
|
headers[normalizedName] = rawValue;
|
|
}
|
|
|
|
return headers;
|
|
}
|
|
|
|
function isAllowedSourceHeader(name) {
|
|
return ![
|
|
'accept-encoding',
|
|
'connection',
|
|
'content-length',
|
|
'host',
|
|
'range',
|
|
'transfer-encoding',
|
|
].includes(name);
|
|
}
|
|
|
|
function parsePlaybackOptions(body) {
|
|
return {
|
|
fps: clampInteger(body?.fps, defaults.fps, 1, 30),
|
|
width: clampInteger(body?.width, defaults.width, 160, 1920),
|
|
quality: clampInteger(body?.quality, defaults.quality, 2, 18),
|
|
audioBitrate: parseAudioBitrate(body?.audioBitrate, defaults.audioBitrate),
|
|
audioChannels: clampInteger(body?.audioChannels, defaults.audioChannels, 1, 2),
|
|
audioSampleRate: clampInteger(body?.audioSampleRate, defaults.audioSampleRate, 22050, 48000),
|
|
};
|
|
}
|
|
|
|
function parsePlaybackConnectionMode(value) {
|
|
if (value === 'relay' || value === 'tee') {
|
|
return 'relay';
|
|
}
|
|
|
|
if (value === 'single' || value === 'unified') {
|
|
return 'single';
|
|
}
|
|
|
|
if (value === 'split' || value === 'separate' || value === undefined || value === '') {
|
|
return 'split';
|
|
}
|
|
|
|
console.warn(`Unknown PLAYBACK_CONNECTION_MODE "${value}", using split.`);
|
|
return 'split';
|
|
}
|
|
|
|
function clampInteger(value, fallback, min, max) {
|
|
const parsed = Number(value);
|
|
|
|
if (!Number.isFinite(parsed)) {
|
|
return fallback;
|
|
}
|
|
|
|
return Math.min(max, Math.max(min, Math.round(parsed)));
|
|
}
|
|
|
|
function parseBoolean(value, fallback) {
|
|
if (value === undefined || value === '') {
|
|
return fallback;
|
|
}
|
|
|
|
if (['1', 'true', 'yes', 'on'].includes(String(value).toLowerCase())) {
|
|
return true;
|
|
}
|
|
|
|
if (['0', 'false', 'no', 'off'].includes(String(value).toLowerCase())) {
|
|
return false;
|
|
}
|
|
|
|
return fallback;
|
|
}
|
|
|
|
function clampNumber(value, min, max) {
|
|
return Math.min(max, Math.max(min, value));
|
|
}
|
|
|
|
function parseAudioBitrate(value, fallback = '64k') {
|
|
if (typeof value !== 'string') {
|
|
return fallback;
|
|
}
|
|
|
|
return /^\d{2,3}k$/i.test(value) ? value.toLowerCase() : fallback;
|
|
}
|
|
|
|
function formatSessionPayload(session) {
|
|
return {
|
|
id: session.id,
|
|
options: session.options,
|
|
duration: Number.isFinite(session.duration) ? session.duration : null,
|
|
metadataStatus: session.metadataStatus,
|
|
seekable: isSessionSeekable(session),
|
|
seekSeconds: session.seekSeconds,
|
|
seekGeneration: session.seekGeneration,
|
|
};
|
|
}
|
|
|
|
function isSessionSeekable(session) {
|
|
return hasRecordedDuration(session);
|
|
}
|
|
|
|
function hasRecordedDuration(session) {
|
|
return Number.isFinite(session.duration) && session.duration > 0;
|
|
}
|
|
|
|
function canBestEffortResumeWithoutDuration(session) {
|
|
return session.sourceKind === 'local'
|
|
|| session.sourceKind === 'youtube'
|
|
|| isLikelyRecordedMediaUrl(session.originalUrl)
|
|
|| isLikelyRecordedMediaUrl(session.url);
|
|
}
|
|
|
|
function isLikelyRecordedMediaUrl(value) {
|
|
try {
|
|
const parsed = new URL(value);
|
|
return RECORDED_MEDIA_EXTENSIONS.has(path.extname(parsed.pathname).toLowerCase());
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function getSessionPlaybackConnectionMode(session) {
|
|
if (session.sourceKind === 'local') {
|
|
return 'split';
|
|
}
|
|
|
|
if (session.forceSplitPlayback) {
|
|
return 'split';
|
|
}
|
|
|
|
if (PLAYBACK_CONNECTION_MODE === 'relay' && hasRecordedDuration(session)) {
|
|
return 'split';
|
|
}
|
|
|
|
return PLAYBACK_CONNECTION_MODE;
|
|
}
|
|
|
|
function startSessionMetadataProbe(session) {
|
|
void ensureSessionDuration(session);
|
|
}
|
|
|
|
function ensureSessionDuration(session) {
|
|
if (hasRecordedDuration(session)) {
|
|
return Promise.resolve(session.duration);
|
|
}
|
|
|
|
if (session.metadataProbePromise) {
|
|
return session.metadataProbePromise;
|
|
}
|
|
|
|
session.metadataStatus = 'pending';
|
|
|
|
const probePromise = probeSessionDuration(session).then((duration) => {
|
|
const current = sessions.get(session.id);
|
|
|
|
if (current !== session) {
|
|
return duration;
|
|
}
|
|
|
|
session.duration = duration;
|
|
session.metadataStatus = Number.isFinite(duration) ? 'ready' : 'unavailable';
|
|
return duration;
|
|
}).catch((error) => {
|
|
const current = sessions.get(session.id);
|
|
|
|
if (current === session) {
|
|
session.duration = null;
|
|
session.metadataStatus = 'unavailable';
|
|
logWarn(`duration probe failed id=${shortId(session.id)} error=${oneLine(error.message)}`);
|
|
}
|
|
|
|
return null;
|
|
}).finally(() => {
|
|
if (session.metadataProbePromise === probePromise) {
|
|
session.metadataProbePromise = null;
|
|
}
|
|
});
|
|
|
|
session.metadataProbePromise = probePromise;
|
|
return probePromise;
|
|
}
|
|
|
|
async function probeSessionDuration(session) {
|
|
const source = createSourceInput(session.id, 'probe');
|
|
|
|
try {
|
|
const output = await runFfprobe([
|
|
'-v',
|
|
'error',
|
|
'-show_entries',
|
|
'format=duration',
|
|
'-of',
|
|
'json',
|
|
source.url,
|
|
]);
|
|
const payload = JSON.parse(output);
|
|
const duration = Number(payload?.format?.duration);
|
|
|
|
return Number.isFinite(duration) && duration > 0 ? duration : null;
|
|
} finally {
|
|
source.release();
|
|
}
|
|
}
|
|
|
|
function runFfprobe(args) {
|
|
return new Promise((resolve, reject) => {
|
|
const ffprobe = spawn(FFPROBE_PATH, args, {
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
});
|
|
let stdout = '';
|
|
let stderr = '';
|
|
let timedOut = false;
|
|
let settled = false;
|
|
|
|
const finish = (callback) => {
|
|
if (settled) {
|
|
return;
|
|
}
|
|
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
callback();
|
|
};
|
|
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
stopProcess(ffprobe);
|
|
}, METADATA_PROBE_TIMEOUT_MS);
|
|
timer.unref();
|
|
|
|
ffprobe.stdout.on('data', (chunk) => {
|
|
stdout = appendTail(stdout, chunk);
|
|
});
|
|
|
|
ffprobe.stderr.on('data', (chunk) => {
|
|
stderr = appendTail(stderr, chunk);
|
|
});
|
|
|
|
ffprobe.on('error', (error) => {
|
|
finish(() => reject(error));
|
|
});
|
|
|
|
ffprobe.on('close', (code, signal) => {
|
|
finish(() => {
|
|
if (timedOut) {
|
|
reject(new Error('ffprobe timed out.'));
|
|
return;
|
|
}
|
|
|
|
if (code === 0) {
|
|
resolve(stdout);
|
|
return;
|
|
}
|
|
|
|
const detail = redactSecrets(stderr).trim();
|
|
reject(new Error(detail ? `ffprobe exited with code ${code}: ${detail}` : `ffprobe exited with code ${code ?? 'null'} signal ${signal ?? 'none'}.`));
|
|
});
|
|
});
|
|
});
|
|
}
|
|
|
|
async function detectFfmpegHttpReconnectSupport() {
|
|
if (!FFMPEG_HTTP_RECONNECT || FFMPEG_HTTP_RECONNECT_MAX_RETRIES <= 0) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const output = await runFfmpegCapabilityCheck(['-hide_banner', '-h', 'protocol=http']);
|
|
ffmpegSupportsReconnectMaxRetries = /^\s*-reconnect_max_retries\b/m.test(output);
|
|
|
|
if (!ffmpegSupportsReconnectMaxRetries) {
|
|
logWarn('ffmpeg http option reconnect_max_retries is unsupported; omitting FFMPEG_HTTP_RECONNECT_MAX_RETRIES');
|
|
}
|
|
} catch (error) {
|
|
ffmpegSupportsReconnectMaxRetries = false;
|
|
logWarn(`ffmpeg http option detection failed; omitting FFMPEG_HTTP_RECONNECT_MAX_RETRIES error=${oneLine(error.message)}`);
|
|
}
|
|
}
|
|
|
|
function runFfmpegCapabilityCheck(args) {
|
|
return new Promise((resolve, reject) => {
|
|
const ffmpeg = spawn(FFMPEG_PATH, args, {
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
});
|
|
let output = '';
|
|
let timedOut = false;
|
|
let settled = false;
|
|
|
|
const finish = (callback) => {
|
|
if (settled) {
|
|
return;
|
|
}
|
|
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
callback();
|
|
};
|
|
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
stopProcess(ffmpeg);
|
|
}, FFMPEG_CAPABILITY_TIMEOUT_MS);
|
|
timer.unref();
|
|
|
|
ffmpeg.stdout.on('data', (chunk) => {
|
|
output = appendTail(output, chunk);
|
|
});
|
|
|
|
ffmpeg.stderr.on('data', (chunk) => {
|
|
output = appendTail(output, chunk);
|
|
});
|
|
|
|
ffmpeg.on('error', (error) => {
|
|
finish(() => reject(error));
|
|
});
|
|
|
|
ffmpeg.on('close', (code, signal) => {
|
|
finish(() => {
|
|
if (timedOut) {
|
|
reject(new Error('ffmpeg capability check timed out.'));
|
|
return;
|
|
}
|
|
|
|
if (code === 0) {
|
|
resolve(output);
|
|
return;
|
|
}
|
|
|
|
const detail = redactSecrets(output).trim();
|
|
reject(new Error(detail ? `ffmpeg capability check exited with code ${code}: ${detail}` : `ffmpeg capability check exited with code ${code ?? 'null'} signal ${signal ?? 'none'}.`));
|
|
});
|
|
});
|
|
});
|
|
}
|
|
|
|
async function loadRecentUrls() {
|
|
try {
|
|
const raw = await fs.readFile(RECENT_URLS_PATH, 'utf8');
|
|
const parsed = JSON.parse(raw);
|
|
const items = Array.isArray(parsed?.urls) ? parsed.urls : [];
|
|
|
|
recentUrls = items
|
|
.filter((item) => typeof item?.url === 'string' && typeof item?.lastPlayedAt === 'string')
|
|
.slice(0, RECENT_URL_LIMIT);
|
|
} catch (error) {
|
|
if (error.code !== 'ENOENT') {
|
|
console.warn(`Failed to read recent URLs: ${error.message}`);
|
|
}
|
|
|
|
recentUrls = [];
|
|
}
|
|
}
|
|
|
|
async function loadFavorites() {
|
|
try {
|
|
const raw = await fs.readFile(FAVORITES_PATH, 'utf8');
|
|
const parsed = JSON.parse(raw);
|
|
const items = Array.isArray(parsed?.favorites) ? parsed.favorites : [];
|
|
|
|
favorites = items
|
|
.map(normalizeFavorite)
|
|
.filter(Boolean)
|
|
.slice(0, FAVORITES_LIMIT);
|
|
} catch (error) {
|
|
if (error.code !== 'ENOENT') {
|
|
console.warn(`Failed to read favorites: ${error.message}`);
|
|
}
|
|
|
|
favorites = [];
|
|
}
|
|
}
|
|
|
|
async function addRecentUrl(url) {
|
|
const lastPlayedAt = new Date().toISOString();
|
|
recentUrls = [
|
|
{ url, lastPlayedAt },
|
|
...recentUrls.filter((item) => item.url !== url),
|
|
].slice(0, RECENT_URL_LIMIT);
|
|
|
|
recentWrite = recentWrite.catch(() => {}).then(() => saveRecentUrls());
|
|
await recentWrite;
|
|
}
|
|
|
|
async function saveRecentUrls() {
|
|
const directory = path.dirname(RECENT_URLS_PATH);
|
|
const temporaryPath = `${RECENT_URLS_PATH}.${process.pid}.tmp`;
|
|
const payload = `${JSON.stringify({ urls: recentUrls }, null, 2)}\n`;
|
|
|
|
await fs.mkdir(directory, { recursive: true });
|
|
await fs.writeFile(temporaryPath, payload, 'utf8');
|
|
await fs.rename(temporaryPath, RECENT_URLS_PATH);
|
|
}
|
|
|
|
async function saveFavorites() {
|
|
const directory = path.dirname(FAVORITES_PATH);
|
|
const temporaryPath = `${FAVORITES_PATH}.${process.pid}.tmp`;
|
|
const payload = `${JSON.stringify({ favorites }, null, 2)}\n`;
|
|
|
|
await fs.mkdir(directory, { recursive: true });
|
|
await fs.writeFile(temporaryPath, payload, 'utf8');
|
|
await fs.rename(temporaryPath, FAVORITES_PATH);
|
|
}
|
|
|
|
function formatFavoritesPayload() {
|
|
return favorites.map((item) => ({
|
|
title: item.title,
|
|
url: item.url,
|
|
}));
|
|
}
|
|
|
|
function parseFavoritesPayload(value) {
|
|
if (!Array.isArray(value)) {
|
|
throw new Error('Favorites must be a list.');
|
|
}
|
|
|
|
if (value.length > FAVORITES_LIMIT) {
|
|
throw new Error(`Favorites are limited to ${FAVORITES_LIMIT} items.`);
|
|
}
|
|
|
|
return value.map((item, index) => parseFavorite(item, index));
|
|
}
|
|
|
|
function parseFavorite(item, index) {
|
|
if (!item || typeof item !== 'object') {
|
|
throw new Error(`Favorite ${index + 1} is invalid.`);
|
|
}
|
|
|
|
const title = parseFavoriteTitle(item.title, index);
|
|
let url;
|
|
|
|
try {
|
|
url = parseStreamUrl(item.url);
|
|
} catch (error) {
|
|
throw new Error(`Favorite ${index + 1}: ${error.message}`);
|
|
}
|
|
|
|
return { title, url };
|
|
}
|
|
|
|
function normalizeFavorite(item) {
|
|
if (!item || typeof item !== 'object') {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
return {
|
|
title: parseFavoriteTitle(item.title, 0),
|
|
url: parseStreamUrl(item.url),
|
|
};
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function parseFavoriteTitle(value, index) {
|
|
if (typeof value !== 'string') {
|
|
throw new Error(`Favorite ${index + 1} needs a title.`);
|
|
}
|
|
|
|
const title = value.trim();
|
|
|
|
if (!title) {
|
|
throw new Error(`Favorite ${index + 1} needs a title.`);
|
|
}
|
|
|
|
if (title.length > 120) {
|
|
throw new Error(`Favorite ${index + 1} title is too long.`);
|
|
}
|
|
|
|
return title;
|
|
}
|
|
|
|
function getSession(sessionId) {
|
|
const session = sessions.get(sessionId);
|
|
|
|
if (!session) {
|
|
return null;
|
|
}
|
|
|
|
session.lastUsedAt = Date.now();
|
|
return session;
|
|
}
|
|
|
|
function getOrCreatePlayback(session, playbackMode = getSessionPlaybackConnectionMode(session)) {
|
|
const existing = playbacks.get(session.id);
|
|
|
|
if (existing && !existing.closed && existing.mode === playbackMode) {
|
|
return existing;
|
|
}
|
|
|
|
stopSharedPlaybackIfNeeded(session, playbackMode);
|
|
|
|
const playback = playbackMode === 'relay' ? createRelayPlayback(session) : createPlayback(session);
|
|
playbacks.set(session.id, playback);
|
|
return playback;
|
|
}
|
|
|
|
function stopSharedPlaybackIfNeeded(session, playbackMode) {
|
|
const existing = playbacks.get(session.id);
|
|
|
|
if (!existing || existing.closed || existing.mode === playbackMode) {
|
|
return;
|
|
}
|
|
|
|
existing.stop('playback_mode_changed');
|
|
|
|
if (playbacks.get(session.id) === existing) {
|
|
playbacks.delete(session.id);
|
|
}
|
|
}
|
|
|
|
function streamSplitAudio(request, response, session) {
|
|
const worker = createAudioWorker(session);
|
|
const releaseWorker = once(worker.release);
|
|
const ffmpeg = spawnFfmpeg(worker.args);
|
|
const logger = createFfmpegLogger('audio', session.id, ffmpeg.pid);
|
|
const stderrTail = createFfmpegStderrTail();
|
|
let responseFinished = false;
|
|
let stopReason = 'process_exit';
|
|
|
|
logger.start();
|
|
|
|
response.set({
|
|
'Cache-Control': 'no-store',
|
|
'Content-Type': 'audio/mpeg',
|
|
'X-Content-Type-Options': 'nosniff',
|
|
});
|
|
response.flushHeaders();
|
|
|
|
ffmpeg.stderr.on('data', (chunk) => {
|
|
stderrTail.write(chunk);
|
|
logger.stderr(chunk);
|
|
});
|
|
|
|
ffmpeg.stdout.on('error', (error) => {
|
|
logWarn(`audio output error kind=audio:${shortId(session.id)} error=${oneLine(error.message)}`);
|
|
stopReason = 'audio_output_error';
|
|
stopProcess(ffmpeg);
|
|
});
|
|
ffmpeg.stdout.pipe(response);
|
|
|
|
ffmpeg.on('error', (error) => {
|
|
stopReason = 'start_error';
|
|
releaseWorker();
|
|
logger.error(error);
|
|
if (!response.writableEnded) {
|
|
response.end();
|
|
}
|
|
});
|
|
|
|
ffmpeg.on('close', (code, signal) => {
|
|
stderrTail.flush();
|
|
releaseWorker();
|
|
logger.close(code, signal, stopReason, stderrTail.value);
|
|
if (!response.writableEnded) {
|
|
response.end();
|
|
}
|
|
});
|
|
|
|
response.on('finish', () => {
|
|
responseFinished = true;
|
|
});
|
|
|
|
const cleanup = once(() => {
|
|
if (responseFinished || response.writableEnded) {
|
|
return;
|
|
}
|
|
|
|
stopReason = 'client_disconnect';
|
|
stopProcess(ffmpeg);
|
|
});
|
|
request.on('close', cleanup);
|
|
response.on('close', cleanup);
|
|
}
|
|
|
|
function streamSplitFrames(websocket, session) {
|
|
const worker = createFrameWorker(session);
|
|
const releaseWorker = once(worker.release);
|
|
const ffmpeg = spawnFfmpeg(worker.args);
|
|
const logger = createFfmpegLogger('frames', session.id, ffmpeg.pid);
|
|
const stderrTail = createFfmpegStderrTail();
|
|
const { fps, quality, width } = session.options;
|
|
const label = `frames:${shortId(session.id)}`;
|
|
let frameIndex = 0;
|
|
let skippedFrames = 0;
|
|
let stopReason = 'process_exit';
|
|
let serverClosingWebSocket = false;
|
|
let clientAudioTime = null;
|
|
const frameConditions = createFrameConditionLogger(label);
|
|
const frameSender = createLatestFrameSender(websocket, {
|
|
onError: () => {
|
|
stopReason = 'websocket_send_error';
|
|
stopProcess(ffmpeg);
|
|
},
|
|
onSkip: () => {
|
|
skippedFrames += 1;
|
|
},
|
|
onQueue: (event) => {
|
|
frameConditions.senderQueue(event);
|
|
},
|
|
});
|
|
const frameParser = createJpegFrameParser(handleJpegFrame);
|
|
|
|
logger.start();
|
|
|
|
sendJson(websocket, {
|
|
type: 'ready',
|
|
codec: 'jpeg',
|
|
fps,
|
|
quality,
|
|
width,
|
|
duration: session.duration,
|
|
seekable: isSessionSeekable(session),
|
|
seekSeconds: session.seekSeconds,
|
|
});
|
|
|
|
ffmpeg.stdout.on('data', handleFrameData);
|
|
ffmpeg.stdout.on('error', (error) => {
|
|
logWarn(`frame output error kind=${label} error=${oneLine(error.message)}`);
|
|
stopReason = 'frame_output_error';
|
|
stopProcess(ffmpeg);
|
|
});
|
|
|
|
ffmpeg.stderr.on('data', (chunk) => {
|
|
stderrTail.write(chunk);
|
|
logger.stderr(chunk);
|
|
});
|
|
|
|
ffmpeg.on('error', (error) => {
|
|
stopReason = 'start_error';
|
|
releaseWorker();
|
|
logger.error(error);
|
|
sendJson(websocket, {
|
|
type: 'error',
|
|
message: `Failed to start ffmpeg: ${error.message}`,
|
|
});
|
|
websocket.close(1011, 'ffmpeg start failed');
|
|
});
|
|
|
|
ffmpeg.on('close', (code, signal) => {
|
|
stderrTail.flush();
|
|
releaseWorker();
|
|
logger.close(code, signal, stopReason, stderrTail.value);
|
|
|
|
if (isWebSocketOpen(websocket) && stopReason !== 'client_disconnect') {
|
|
sendJson(websocket, {
|
|
type: 'end',
|
|
code,
|
|
signal,
|
|
skippedFrames,
|
|
message: summarizeFfmpegExit(code, signal, stderrTail.value),
|
|
});
|
|
serverClosingWebSocket = true;
|
|
websocket.close(1000, 'ffmpeg exited');
|
|
}
|
|
|
|
logInfo(`frames closed kind=${label} reason=${stopReason} frames=${frameIndex} skippedFrames=${skippedFrames}`);
|
|
frameConditions.flush('close', true);
|
|
});
|
|
|
|
websocket.on('close', () => {
|
|
if (serverClosingWebSocket) {
|
|
return;
|
|
}
|
|
|
|
stopReason = 'client_disconnect';
|
|
stopProcess(ffmpeg);
|
|
});
|
|
|
|
websocket.on('error', () => {
|
|
stopReason = 'websocket_error';
|
|
stopProcess(ffmpeg);
|
|
});
|
|
|
|
websocket.on('message', (data, isBinary) => {
|
|
clientAudioTime = handleFrameClientMessage(data, isBinary, label, clientAudioTime);
|
|
});
|
|
|
|
function handleFrameData(chunk) {
|
|
frameParser.write(chunk);
|
|
}
|
|
|
|
function handleJpegFrame(jpeg) {
|
|
const timestamp = frameIndex / fps;
|
|
frameIndex += 1;
|
|
|
|
if (isFrameLateForClient(timestamp, clientAudioTime)) {
|
|
skippedFrames += 1;
|
|
frameConditions.clientClockSkip(timestamp, clientAudioTime);
|
|
return true;
|
|
}
|
|
|
|
const sendResult = frameSender.send(timestamp, jpeg);
|
|
|
|
if (sendResult === 'closed') {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
}
|
|
|
|
function createRelayPlayback(session) {
|
|
const { fps, quality, width } = session.options;
|
|
const label = `relay:${shortId(session.id)}`;
|
|
let audioResponse = null;
|
|
let websocket = null;
|
|
let audioFfmpeg = null;
|
|
let frameFfmpeg = null;
|
|
let audioInput = null;
|
|
let frameInput = null;
|
|
let audioLogger = null;
|
|
let frameLogger = null;
|
|
let sourceController = null;
|
|
let sourceBytes = 0;
|
|
let frameIndex = 0;
|
|
let skippedFrames = 0;
|
|
let stopReason = 'process_exit';
|
|
let started = false;
|
|
let stopping = false;
|
|
let closed = false;
|
|
let sourceEnded = false;
|
|
let audioClosed = false;
|
|
let frameClosed = false;
|
|
let frameExitCode = null;
|
|
let frameExitSignal = null;
|
|
let frameEndSent = false;
|
|
let frameSender = null;
|
|
let clientAudioTime = null;
|
|
let audioResponseFinished = false;
|
|
let serverClosingWebSocket = false;
|
|
let readyTimer = null;
|
|
const frameConditions = createFrameConditionLogger(label);
|
|
const audioStderr = createFfmpegStderrTail();
|
|
const frameStderr = createFfmpegStderrTail();
|
|
const frameParser = createJpegFrameParser(handleJpegFrame);
|
|
|
|
const playback = {
|
|
mode: 'relay',
|
|
get closed() {
|
|
return closed;
|
|
},
|
|
stop(reason = 'playback_stopped') {
|
|
stop(reason);
|
|
},
|
|
attachAudio(request, response) {
|
|
if (closed) {
|
|
response.status(410).end();
|
|
return;
|
|
}
|
|
|
|
if (audioResponse && !audioResponse.writableEnded) {
|
|
response.status(409).json({ error: 'Audio stream already attached for this session.' });
|
|
return;
|
|
}
|
|
|
|
audioResponse = response;
|
|
response.set({
|
|
'Cache-Control': 'no-store',
|
|
'Content-Type': 'audio/mpeg',
|
|
'X-Content-Type-Options': 'nosniff',
|
|
});
|
|
response.flushHeaders();
|
|
|
|
response.on('finish', () => {
|
|
audioResponseFinished = true;
|
|
});
|
|
|
|
const cleanup = once(() => {
|
|
if (!closed && !audioResponseFinished && !response.writableEnded) {
|
|
stop('client_disconnect');
|
|
}
|
|
});
|
|
|
|
request.on('close', cleanup);
|
|
response.on('close', cleanup);
|
|
maybeStart();
|
|
},
|
|
attachFrames(nextWebsocket) {
|
|
if (closed) {
|
|
nextWebsocket.close(1011, 'Playback closed');
|
|
return;
|
|
}
|
|
|
|
if (websocket && websocket.readyState === WebSocket.OPEN) {
|
|
nextWebsocket.close(1013, 'Frame stream already attached for this session');
|
|
return;
|
|
}
|
|
|
|
websocket = nextWebsocket;
|
|
frameSender = createLatestFrameSender(websocket, {
|
|
onError: (error) => {
|
|
if (error && !closed) {
|
|
stop('websocket_send_error');
|
|
}
|
|
},
|
|
onSkip: () => {
|
|
skippedFrames += 1;
|
|
},
|
|
onQueue: (event) => {
|
|
frameConditions.senderQueue(event);
|
|
},
|
|
});
|
|
|
|
sendJson(websocket, {
|
|
type: 'ready',
|
|
codec: 'jpeg',
|
|
fps,
|
|
quality,
|
|
width,
|
|
duration: session.duration,
|
|
seekable: isSessionSeekable(session),
|
|
seekSeconds: session.seekSeconds,
|
|
});
|
|
|
|
websocket.on('close', () => {
|
|
if (!closed && !serverClosingWebSocket) {
|
|
stop('client_disconnect');
|
|
}
|
|
});
|
|
|
|
websocket.on('error', () => {
|
|
if (!closed) {
|
|
stop('websocket_error');
|
|
}
|
|
});
|
|
|
|
websocket.on('message', (data, isBinary) => {
|
|
clientAudioTime = handleFrameClientMessage(data, isBinary, label, clientAudioTime);
|
|
});
|
|
|
|
maybeStart();
|
|
},
|
|
};
|
|
|
|
readyTimer = setTimeout(() => {
|
|
if (!started && !closed) {
|
|
stop('consumer_attach_timeout');
|
|
}
|
|
}, PLAYBACK_READY_TIMEOUT_MS).unref();
|
|
|
|
return playback;
|
|
|
|
function maybeStart() {
|
|
if (started || closed || !audioResponse || !isWebSocketOpen(websocket)) {
|
|
return;
|
|
}
|
|
|
|
started = true;
|
|
clearReadyTimer();
|
|
|
|
audioFfmpeg = spawnFfmpeg(buildRelayAudioArgs(session), ['pipe', 'pipe', 'pipe']);
|
|
frameFfmpeg = spawnFfmpeg(buildRelayFrameArgs(session), ['pipe', 'pipe', 'pipe']);
|
|
audioLogger = createFfmpegLogger('relay-audio', session.id, audioFfmpeg.pid);
|
|
frameLogger = createFfmpegLogger('relay-frames', session.id, frameFfmpeg.pid);
|
|
audioInput = createRelayInputBranch(audioFfmpeg.stdin, handleRelayInputError);
|
|
frameInput = createRelayInputBranch(frameFfmpeg.stdin, handleRelayInputError);
|
|
|
|
audioLogger.start();
|
|
frameLogger.start();
|
|
|
|
audioFfmpeg.stdout.on('error', (error) => {
|
|
logWarn(`audio output error kind=${label} error=${oneLine(error.message)}`);
|
|
stop('audio_output_error');
|
|
});
|
|
audioFfmpeg.stdout.pipe(audioResponse);
|
|
|
|
frameFfmpeg.stdout.on('error', (error) => {
|
|
logWarn(`frame output error kind=${label} error=${oneLine(error.message)}`);
|
|
stop('frame_output_error');
|
|
});
|
|
frameFfmpeg.stdout.on('data', handleFrameData);
|
|
|
|
audioFfmpeg.stderr.on('data', (chunk) => {
|
|
audioStderr.write(chunk);
|
|
audioLogger.stderr(chunk);
|
|
});
|
|
frameFfmpeg.stderr.on('data', (chunk) => {
|
|
frameStderr.write(chunk);
|
|
frameLogger.stderr(chunk);
|
|
});
|
|
|
|
audioFfmpeg.on('error', (error) => {
|
|
audioLogger.error(error);
|
|
stop('audio_start_error');
|
|
});
|
|
frameFfmpeg.on('error', (error) => {
|
|
frameLogger.error(error);
|
|
stop('frame_start_error');
|
|
});
|
|
|
|
audioFfmpeg.on('close', (code, signal) => {
|
|
audioClosed = true;
|
|
audioStderr.flush();
|
|
audioLogger.close(code, signal, stopReason, audioStderr.value);
|
|
maybeFinishRelay();
|
|
});
|
|
frameFfmpeg.on('close', (code, signal) => {
|
|
frameClosed = true;
|
|
frameExitCode = code;
|
|
frameExitSignal = signal;
|
|
frameStderr.flush();
|
|
frameLogger.close(code, signal, stopReason, frameStderr.value);
|
|
sendFrameEndIfNeeded();
|
|
maybeFinishRelay();
|
|
});
|
|
|
|
void startRelaySource();
|
|
}
|
|
|
|
async function startRelaySource() {
|
|
const startedAt = Date.now();
|
|
let upstreamStatus = 'pending';
|
|
let upstreamEnded = false;
|
|
|
|
sourceController = new AbortController();
|
|
|
|
try {
|
|
const headers = createSourceRequestHeaders(session);
|
|
headers.set('accept-encoding', 'identity');
|
|
|
|
const upstream = await fetch(session.url, {
|
|
headers,
|
|
redirect: 'follow',
|
|
signal: sourceController.signal,
|
|
});
|
|
|
|
upstreamStatus = String(upstream.status);
|
|
logInfo(`relay source connected kind=${label} status=${upstream.status} contentType=${upstream.headers.get('content-type') ?? 'unknown'}`);
|
|
|
|
if (!upstream.ok || !upstream.body) {
|
|
throw new Error(`Source returned HTTP ${upstream.status}`);
|
|
}
|
|
|
|
for await (const chunk of Readable.fromWeb(upstream.body)) {
|
|
if (closed || stopping) {
|
|
return;
|
|
}
|
|
|
|
const buffer = toBufferView(chunk);
|
|
sourceBytes += buffer.length;
|
|
|
|
if (!audioInput.write(buffer) || !frameInput.write(buffer)) {
|
|
stop('relay_input_queue_overflow');
|
|
return;
|
|
}
|
|
|
|
await waitForRelayCapacity([audioInput, frameInput], () => closed || stopping);
|
|
}
|
|
|
|
upstreamEnded = true;
|
|
sourceEnded = true;
|
|
audioInput.end();
|
|
frameInput.end();
|
|
} catch (error) {
|
|
if (sourceController.signal.aborted || closed) {
|
|
return;
|
|
}
|
|
|
|
logWarn(`relay source failed kind=${label} error=${oneLine(error.message)}`);
|
|
stop('relay_source_error');
|
|
} finally {
|
|
logInfo(`relay source closed kind=${label} status=${upstreamStatus} bytes=${sourceBytes} upstreamEnded=${upstreamEnded} durationMs=${Date.now() - startedAt}`);
|
|
}
|
|
}
|
|
|
|
function handleRelayInputError(error) {
|
|
if (!closed && !stopping) {
|
|
logWarn(`relay input error kind=${label} error=${oneLine(error.message)}`);
|
|
stop('relay_input_error');
|
|
}
|
|
}
|
|
|
|
function handleFrameData(chunk) {
|
|
frameParser.write(chunk);
|
|
}
|
|
|
|
function handleJpegFrame(jpeg) {
|
|
const timestamp = frameIndex / fps;
|
|
frameIndex += 1;
|
|
|
|
if (isFrameLateForClient(timestamp, clientAudioTime)) {
|
|
skippedFrames += 1;
|
|
frameConditions.clientClockSkip(timestamp, clientAudioTime);
|
|
return true;
|
|
}
|
|
|
|
const sendResult = frameSender?.send(timestamp, jpeg) ?? 'closed';
|
|
|
|
if (sendResult === 'closed') {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
function stop(reason) {
|
|
if (closed) {
|
|
return;
|
|
}
|
|
|
|
stopReason = reason;
|
|
stopping = true;
|
|
sourceController?.abort();
|
|
audioInput?.destroy();
|
|
frameInput?.destroy();
|
|
|
|
if (isChildRunning(audioFfmpeg)) {
|
|
stopProcess(audioFfmpeg);
|
|
}
|
|
|
|
if (isChildRunning(frameFfmpeg)) {
|
|
stopProcess(frameFfmpeg);
|
|
}
|
|
|
|
if (!isChildRunning(audioFfmpeg) && !isChildRunning(frameFfmpeg)) {
|
|
finishRelay(1000, 'relay stopped');
|
|
}
|
|
}
|
|
|
|
function maybeFinishRelay() {
|
|
if (closed) {
|
|
return;
|
|
}
|
|
|
|
if (!stopping && !sourceEnded) {
|
|
stop('relay_process_exit');
|
|
return;
|
|
}
|
|
|
|
if (audioClosed && frameClosed) {
|
|
finishRelay(1000, 'relay ended');
|
|
}
|
|
}
|
|
|
|
function sendFrameEndIfNeeded() {
|
|
if (frameEndSent || !isWebSocketOpen(websocket)) {
|
|
return;
|
|
}
|
|
|
|
frameEndSent = true;
|
|
sendJson(websocket, {
|
|
type: 'end',
|
|
code: frameExitCode,
|
|
signal: frameExitSignal,
|
|
skippedFrames,
|
|
message: summarizeFfmpegExit(frameExitCode, frameExitSignal, frameStderr.value),
|
|
});
|
|
serverClosingWebSocket = true;
|
|
websocket.close(1000, 'ffmpeg exited');
|
|
}
|
|
|
|
function finishRelay(websocketCode, websocketReason) {
|
|
if (closed) {
|
|
return;
|
|
}
|
|
|
|
closed = true;
|
|
clearReadyTimer();
|
|
sourceController?.abort();
|
|
audioInput?.destroy();
|
|
frameInput?.destroy();
|
|
if (playbacks.get(session.id) === playback) {
|
|
playbacks.delete(session.id);
|
|
}
|
|
|
|
if (audioResponse && !audioResponse.writableEnded) {
|
|
audioResponse.end();
|
|
}
|
|
|
|
if (isWebSocketOpen(websocket) && frameEndSent) {
|
|
serverClosingWebSocket = true;
|
|
websocket.close(websocketCode, websocketReason);
|
|
} else {
|
|
sendFrameEndIfNeeded();
|
|
}
|
|
|
|
logInfo(`relay closed kind=${label} reason=${stopReason} sourceBytes=${sourceBytes} frames=${frameIndex} skippedFrames=${skippedFrames} audioInputPeakBytes=${audioInput?.peakBytes ?? 0} frameInputPeakBytes=${frameInput?.peakBytes ?? 0}`);
|
|
frameConditions.flush('close', true);
|
|
}
|
|
|
|
function clearReadyTimer() {
|
|
if (readyTimer) {
|
|
clearTimeout(readyTimer);
|
|
readyTimer = null;
|
|
}
|
|
}
|
|
}
|
|
|
|
function createPlayback(session) {
|
|
const { fps, quality, width } = session.options;
|
|
const label = `playback:${shortId(session.id)}`;
|
|
let audioResponse = null;
|
|
let audioQueue = [];
|
|
let audioQueueBytes = 0;
|
|
let audioQueuePeakBytes = 0;
|
|
let audioBackpressureCount = 0;
|
|
let audioDrainCount = 0;
|
|
let audioDrainPending = false;
|
|
let websocket = null;
|
|
let ffmpeg = null;
|
|
let logger = null;
|
|
let releaseSource = () => {};
|
|
let stderr = '';
|
|
let frameIndex = 0;
|
|
let skippedFrames = 0;
|
|
let stopReason = 'process_exit';
|
|
let started = false;
|
|
let closed = false;
|
|
let readyTimer = null;
|
|
let frameSender = null;
|
|
let clientAudioTime = null;
|
|
const frameConditions = createFrameConditionLogger(label);
|
|
const stderrTail = createFfmpegStderrTail();
|
|
const frameParser = createJpegFrameParser(handleJpegFrame);
|
|
|
|
const playback = {
|
|
mode: 'single',
|
|
get closed() {
|
|
return closed;
|
|
},
|
|
stop(reason = 'playback_stopped') {
|
|
stop(reason);
|
|
},
|
|
attachAudio(request, response) {
|
|
if (closed) {
|
|
response.status(410).end();
|
|
return;
|
|
}
|
|
|
|
if (audioResponse && !audioResponse.writableEnded) {
|
|
response.status(409).json({ error: 'Audio stream already attached for this session.' });
|
|
return;
|
|
}
|
|
|
|
audioResponse = response;
|
|
response.set({
|
|
'Cache-Control': 'no-store',
|
|
'Content-Type': 'audio/mpeg',
|
|
'X-Content-Type-Options': 'nosniff',
|
|
});
|
|
response.flushHeaders();
|
|
|
|
const cleanup = once(() => {
|
|
if (!closed) {
|
|
stop('client_disconnect');
|
|
}
|
|
});
|
|
|
|
request.on('close', cleanup);
|
|
response.on('close', cleanup);
|
|
maybeStart();
|
|
},
|
|
attachFrames(nextWebsocket) {
|
|
if (closed) {
|
|
nextWebsocket.close(1011, 'Playback closed');
|
|
return;
|
|
}
|
|
|
|
if (websocket && websocket.readyState === WebSocket.OPEN) {
|
|
nextWebsocket.close(1013, 'Frame stream already attached for this session');
|
|
return;
|
|
}
|
|
|
|
websocket = nextWebsocket;
|
|
frameSender = createLatestFrameSender(websocket, {
|
|
onError: (error) => {
|
|
if (error && !closed) {
|
|
stop('websocket_send_error');
|
|
}
|
|
},
|
|
onSkip: () => {
|
|
skippedFrames += 1;
|
|
},
|
|
onQueue: (event) => {
|
|
frameConditions.senderQueue(event);
|
|
},
|
|
});
|
|
|
|
sendJson(websocket, {
|
|
type: 'ready',
|
|
codec: 'jpeg',
|
|
fps,
|
|
quality,
|
|
width,
|
|
duration: session.duration,
|
|
seekable: isSessionSeekable(session),
|
|
seekSeconds: session.seekSeconds,
|
|
});
|
|
|
|
websocket.on('close', () => {
|
|
if (!closed) {
|
|
stop('client_disconnect');
|
|
}
|
|
});
|
|
|
|
websocket.on('error', () => {
|
|
if (!closed) {
|
|
stop('websocket_error');
|
|
}
|
|
});
|
|
|
|
websocket.on('message', (data, isBinary) => {
|
|
clientAudioTime = handleFrameClientMessage(data, isBinary, label, clientAudioTime);
|
|
});
|
|
|
|
maybeStart();
|
|
},
|
|
};
|
|
|
|
readyTimer = setTimeout(() => {
|
|
if (!started && !closed) {
|
|
stop('consumer_attach_timeout');
|
|
}
|
|
}, PLAYBACK_READY_TIMEOUT_MS).unref();
|
|
|
|
return playback;
|
|
|
|
function maybeStart() {
|
|
if (started || closed || !audioResponse || !isWebSocketOpen(websocket)) {
|
|
return;
|
|
}
|
|
|
|
started = true;
|
|
clearReadyTimer();
|
|
|
|
const source = createSourceInput(session.id, 'playback');
|
|
releaseSource = once(source.release);
|
|
ffmpeg = spawnFfmpeg(buildPlaybackArgs(session, source.url), ['ignore', 'pipe', 'pipe', 'pipe']);
|
|
logger = createFfmpegLogger('playback', session.id, ffmpeg.pid);
|
|
logger.start();
|
|
|
|
const frameStream = ffmpeg.stdio[3];
|
|
|
|
ffmpeg.stdout.on('data', handleAudioData);
|
|
ffmpeg.stdout.on('error', (error) => {
|
|
logWarn(`audio output error kind=${label} error=${oneLine(error.message)}`);
|
|
stop('audio_output_error');
|
|
});
|
|
frameStream.on('error', (error) => {
|
|
logWarn(`frame output error kind=${label} error=${oneLine(error.message)}`);
|
|
stop('frame_output_error');
|
|
});
|
|
frameStream.on('data', handleFrameData);
|
|
|
|
ffmpeg.stderr.on('data', (chunk) => {
|
|
stderrTail.write(chunk);
|
|
logger.stderr(chunk);
|
|
});
|
|
|
|
ffmpeg.on('error', (error) => {
|
|
stopReason = 'start_error';
|
|
logger.error(error);
|
|
finishPlayback(1011, `Failed to start ffmpeg: ${error.message}`, null, null);
|
|
});
|
|
|
|
ffmpeg.on('close', (code, signal) => {
|
|
stderrTail.flush();
|
|
stderr = stderrTail.value;
|
|
logger.close(code, signal, stopReason, stderr);
|
|
finishPlayback(1000, 'ffmpeg exited', code, signal);
|
|
});
|
|
}
|
|
|
|
function handleAudioData(chunk) {
|
|
if (closed) {
|
|
return;
|
|
}
|
|
|
|
if (!audioResponse || audioResponse.writableEnded) {
|
|
stop('audio_response_closed');
|
|
return;
|
|
}
|
|
|
|
audioQueue.push(chunk);
|
|
audioQueueBytes += chunk.length;
|
|
audioQueuePeakBytes = Math.max(audioQueuePeakBytes, audioQueueBytes);
|
|
|
|
if (audioQueueBytes > MAX_AUDIO_QUEUE_BYTES) {
|
|
logWarn(`audio queue overflow kind=${label} bytes=${audioQueueBytes} maxBytes=${MAX_AUDIO_QUEUE_BYTES}`);
|
|
stop('audio_queue_overflow');
|
|
return;
|
|
}
|
|
|
|
flushAudioQueue();
|
|
}
|
|
|
|
function flushAudioQueue() {
|
|
if (closed || !audioResponse || audioResponse.writableEnded) {
|
|
return;
|
|
}
|
|
|
|
while (audioQueue.length > 0) {
|
|
const chunk = audioQueue.shift();
|
|
audioQueueBytes -= chunk.length;
|
|
|
|
let canContinue = false;
|
|
|
|
try {
|
|
canContinue = audioResponse.write(chunk);
|
|
} catch (error) {
|
|
logWarn(`audio response write failed kind=${label} error=${oneLine(error.message)}`);
|
|
stop('audio_response_write_error');
|
|
return;
|
|
}
|
|
|
|
if (!canContinue) {
|
|
audioBackpressureCount += 1;
|
|
waitForAudioDrain();
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
function waitForAudioDrain() {
|
|
if (audioDrainPending || !audioResponse || audioResponse.writableEnded) {
|
|
return;
|
|
}
|
|
|
|
audioDrainPending = true;
|
|
audioResponse.once('drain', () => {
|
|
audioDrainPending = false;
|
|
audioDrainCount += 1;
|
|
flushAudioQueue();
|
|
});
|
|
}
|
|
|
|
function handleFrameData(chunk) {
|
|
frameParser.write(chunk);
|
|
}
|
|
|
|
function handleJpegFrame(jpeg) {
|
|
const timestamp = frameIndex / fps;
|
|
frameIndex += 1;
|
|
|
|
if (isFrameLateForClient(timestamp, clientAudioTime)) {
|
|
skippedFrames += 1;
|
|
frameConditions.clientClockSkip(timestamp, clientAudioTime);
|
|
return true;
|
|
}
|
|
|
|
const sendResult = frameSender?.send(timestamp, jpeg) ?? 'closed';
|
|
|
|
if (sendResult === 'closed') {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
function stop(reason) {
|
|
stopReason = reason;
|
|
|
|
if (ffmpeg && ffmpeg.exitCode === null && ffmpeg.signalCode === null) {
|
|
stopProcess(ffmpeg);
|
|
return;
|
|
}
|
|
|
|
finishPlayback(1000, 'playback stopped', null, null);
|
|
}
|
|
|
|
function finishPlayback(websocketCode, websocketReason, ffmpegCode, ffmpegSignal) {
|
|
if (closed) {
|
|
return;
|
|
}
|
|
|
|
closed = true;
|
|
clearReadyTimer();
|
|
releaseSource();
|
|
if (playbacks.get(session.id) === playback) {
|
|
playbacks.delete(session.id);
|
|
}
|
|
audioQueue = [];
|
|
audioQueueBytes = 0;
|
|
|
|
if (audioResponse && !audioResponse.writableEnded) {
|
|
audioResponse.end();
|
|
}
|
|
|
|
if (isWebSocketOpen(websocket)) {
|
|
sendJson(websocket, {
|
|
type: 'end',
|
|
code: ffmpegCode,
|
|
signal: ffmpegSignal,
|
|
skippedFrames,
|
|
message: summarizeFfmpegExit(ffmpegCode, ffmpegSignal, stderr),
|
|
});
|
|
websocket.close(websocketCode, websocketReason);
|
|
}
|
|
|
|
logInfo(`playback closed kind=${label} reason=${stopReason} frames=${frameIndex} skippedFrames=${skippedFrames} audioQueuePeakBytes=${audioQueuePeakBytes} audioBackpressureCount=${audioBackpressureCount} audioDrainCount=${audioDrainCount}`);
|
|
frameConditions.flush('close', true);
|
|
}
|
|
|
|
function clearReadyTimer() {
|
|
if (readyTimer) {
|
|
clearTimeout(readyTimer);
|
|
readyTimer = null;
|
|
}
|
|
}
|
|
}
|
|
|
|
function isWebSocketOpen(websocket) {
|
|
return websocket?.readyState === WebSocket.OPEN;
|
|
}
|
|
|
|
function createJpegFrameParser(onFrame) {
|
|
let collecting = false;
|
|
let pendingMarkerByte = false;
|
|
let parts = [];
|
|
let byteLength = 0;
|
|
|
|
return {
|
|
write(chunk) {
|
|
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
|
let offset = 0;
|
|
|
|
while (offset < buffer.length) {
|
|
if (!collecting) {
|
|
if (pendingMarkerByte && buffer[offset] === 0xd8) {
|
|
collecting = true;
|
|
pendingMarkerByte = false;
|
|
parts = [JPEG_SOI];
|
|
byteLength = JPEG_SOI.length;
|
|
offset += 1;
|
|
} else {
|
|
const start = buffer.indexOf(JPEG_SOI, offset);
|
|
|
|
if (start === -1) {
|
|
pendingMarkerByte = buffer[buffer.length - 1] === 0xff;
|
|
return true;
|
|
}
|
|
|
|
collecting = true;
|
|
pendingMarkerByte = false;
|
|
parts = [];
|
|
byteLength = 0;
|
|
offset = start;
|
|
}
|
|
}
|
|
|
|
if (pendingMarkerByte) {
|
|
pendingMarkerByte = false;
|
|
|
|
if (buffer[offset] === 0xd9) {
|
|
appendFramePart(buffer.subarray(offset, offset + 1));
|
|
offset += 1;
|
|
|
|
if (!emitFrame()) {
|
|
return false;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
}
|
|
|
|
const end = buffer.indexOf(JPEG_EOI, offset);
|
|
|
|
if (end === -1) {
|
|
appendFramePart(buffer.subarray(offset));
|
|
pendingMarkerByte = buffer[buffer.length - 1] === 0xff;
|
|
return true;
|
|
}
|
|
|
|
appendFramePart(buffer.subarray(offset, end + JPEG_EOI.length));
|
|
offset = end + JPEG_EOI.length;
|
|
|
|
if (!emitFrame()) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
return true;
|
|
},
|
|
};
|
|
|
|
function appendFramePart(part) {
|
|
if (part.length === 0) {
|
|
return;
|
|
}
|
|
|
|
parts.push(part);
|
|
byteLength += part.length;
|
|
}
|
|
|
|
function emitFrame() {
|
|
const frame = { parts, byteLength };
|
|
collecting = false;
|
|
pendingMarkerByte = false;
|
|
parts = [];
|
|
byteLength = 0;
|
|
return onFrame(frame) !== false;
|
|
}
|
|
}
|
|
|
|
function createLatestFrameSender(websocket, { onError = () => {}, onSkip = () => {}, onQueue = () => {} } = {}) {
|
|
let sending = false;
|
|
let latestFrame = null;
|
|
let pumpTimer = null;
|
|
|
|
return {
|
|
send(timestamp, jpeg) {
|
|
if (!isWebSocketOpen(websocket)) {
|
|
return 'closed';
|
|
}
|
|
|
|
const packetBytes = 8 + getJpegFrameByteLength(jpeg);
|
|
|
|
if (sending) {
|
|
queueLatestFrame(timestamp, jpeg, {
|
|
reason: 'send_in_progress',
|
|
packetBytes,
|
|
bufferedAmount: websocket.bufferedAmount,
|
|
});
|
|
return 'queued';
|
|
}
|
|
|
|
if (isWebSocketOverFrameBudget(websocket, packetBytes)) {
|
|
queueLatestFrame(timestamp, jpeg, {
|
|
reason: 'websocket_buffer_budget',
|
|
packetBytes,
|
|
bufferedAmount: websocket.bufferedAmount,
|
|
});
|
|
return 'queued';
|
|
}
|
|
|
|
return sendNow(timestamp, jpeg, packetBytes);
|
|
},
|
|
};
|
|
|
|
function queueLatestFrame(timestamp, jpeg, event) {
|
|
const replacing = Boolean(latestFrame);
|
|
|
|
if (replacing) {
|
|
onSkip();
|
|
}
|
|
|
|
latestFrame = { timestamp, jpeg };
|
|
onQueue({ ...event, replacing });
|
|
schedulePump();
|
|
}
|
|
|
|
function sendNow(timestamp, jpeg, packetBytes = 8 + getJpegFrameByteLength(jpeg)) {
|
|
if (!isWebSocketOpen(websocket)) {
|
|
return 'closed';
|
|
}
|
|
|
|
const packet = Buffer.allocUnsafe(packetBytes);
|
|
packet.writeDoubleLE(timestamp, 0);
|
|
copyJpegFrame(jpeg, packet, 8);
|
|
sending = true;
|
|
|
|
try {
|
|
websocket.send(packet, { binary: true, compress: false }, (error) => {
|
|
sending = false;
|
|
|
|
if (error) {
|
|
onError(error);
|
|
return;
|
|
}
|
|
|
|
pumpLatestFrame();
|
|
});
|
|
} catch (error) {
|
|
sending = false;
|
|
onError(error);
|
|
return 'closed';
|
|
}
|
|
|
|
return 'sent';
|
|
}
|
|
|
|
function pumpLatestFrame() {
|
|
if (!latestFrame || sending) {
|
|
return;
|
|
}
|
|
|
|
if (!isWebSocketOpen(websocket)) {
|
|
latestFrame = null;
|
|
return;
|
|
}
|
|
|
|
const packetBytes = 8 + getJpegFrameByteLength(latestFrame.jpeg);
|
|
|
|
if (isWebSocketOverFrameBudget(websocket, packetBytes)) {
|
|
schedulePump();
|
|
return;
|
|
}
|
|
|
|
const frame = latestFrame;
|
|
latestFrame = null;
|
|
sendNow(frame.timestamp, frame.jpeg, packetBytes);
|
|
}
|
|
|
|
function schedulePump() {
|
|
if (pumpTimer) {
|
|
return;
|
|
}
|
|
|
|
pumpTimer = setTimeout(() => {
|
|
pumpTimer = null;
|
|
pumpLatestFrame();
|
|
}, 10);
|
|
pumpTimer.unref?.();
|
|
}
|
|
}
|
|
|
|
function handleFrameClientMessage(data, isBinary, label, currentClientAudioTime) {
|
|
const message = parseFrameClientMessage(data, isBinary);
|
|
|
|
if (!message) {
|
|
return currentClientAudioTime;
|
|
}
|
|
|
|
if (message.type === 'clock') {
|
|
return message.currentTime;
|
|
}
|
|
|
|
if (message.type === 'telemetry') {
|
|
logClientTelemetry(label, message);
|
|
}
|
|
|
|
return currentClientAudioTime;
|
|
}
|
|
|
|
function parseFrameClientMessage(data, isBinary) {
|
|
if (isBinary) {
|
|
return null;
|
|
}
|
|
|
|
let message;
|
|
|
|
try {
|
|
const raw = typeof data === 'string' ? data : data.toString('utf8');
|
|
message = JSON.parse(raw);
|
|
} catch {
|
|
return null;
|
|
}
|
|
|
|
if (message?.type === 'clock') {
|
|
const currentTime = Number(message.currentTime);
|
|
|
|
if (!Number.isFinite(currentTime)) {
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
type: 'clock',
|
|
currentTime: clampNumber(currentTime, 0, Number.MAX_SAFE_INTEGER),
|
|
};
|
|
}
|
|
|
|
if (message?.type === 'telemetry') {
|
|
return {
|
|
type: 'telemetry',
|
|
reason: sanitizeTelemetryText(message.reason, 'periodic'),
|
|
currentTime: finiteTelemetryNumber(message.currentTime),
|
|
paused: Boolean(message.paused),
|
|
readyState: finiteTelemetryNumber(message.readyState),
|
|
pendingFrames: finiteTelemetryNumber(message.pendingFrames),
|
|
decodedFrames: finiteTelemetryNumber(message.decodedFrames),
|
|
paintedFrames: finiteTelemetryNumber(message.paintedFrames),
|
|
lastFramePacketAgeMs: finiteTelemetryNumber(message.lastFramePacketAgeMs),
|
|
lastFramePaintAgeMs: finiteTelemetryNumber(message.lastFramePaintAgeMs),
|
|
hidden: Boolean(message.hidden),
|
|
online: Boolean(message.online),
|
|
counters: sanitizeTelemetryMap(message.counters),
|
|
max: sanitizeTelemetryMap(message.max),
|
|
};
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function isFrameLateForClient(timestamp, clientAudioTime) {
|
|
return Number.isFinite(clientAudioTime) && timestamp < clientAudioTime - CLIENT_CLOCK_FRAME_LATE_GRACE_SECONDS;
|
|
}
|
|
|
|
function createFrameConditionLogger(label) {
|
|
let counters = {};
|
|
let max = {};
|
|
let lastLoggedAt = 0;
|
|
|
|
return {
|
|
senderQueue(event) {
|
|
incrementCounter(`senderQueue_${event.reason}`);
|
|
|
|
if (event.replacing) {
|
|
incrementCounter('senderReplacedLatest');
|
|
}
|
|
|
|
recordMax('wsBufferedBytes', event.bufferedAmount);
|
|
recordMax('framePacketBytes', event.packetBytes);
|
|
flush('periodic');
|
|
},
|
|
clientClockSkip(timestamp, clientAudioTime) {
|
|
incrementCounter('clientClockSkips');
|
|
recordMax('clientClockLagMs', (clientAudioTime - timestamp) * 1000);
|
|
flush('periodic');
|
|
},
|
|
flush,
|
|
};
|
|
|
|
function incrementCounter(name, count = 1) {
|
|
counters[name] = (counters[name] ?? 0) + count;
|
|
}
|
|
|
|
function recordMax(name, value) {
|
|
if (!Number.isFinite(value)) {
|
|
return;
|
|
}
|
|
|
|
max[name] = Math.max(max[name] ?? 0, Math.round(value));
|
|
}
|
|
|
|
function flush(reason = 'periodic', force = false) {
|
|
const now = Date.now();
|
|
|
|
if (!force && now - lastLoggedAt < FRAME_CONDITION_LOG_INTERVAL_MS) {
|
|
return;
|
|
}
|
|
|
|
if (Object.keys(counters).length === 0 && Object.keys(max).length === 0) {
|
|
return;
|
|
}
|
|
|
|
logInfo(`frame conditions kind=${label} reason=${reason} counters=${formatTelemetryMap(counters)} max=${formatTelemetryMap(max)}`);
|
|
counters = {};
|
|
max = {};
|
|
lastLoggedAt = now;
|
|
}
|
|
}
|
|
|
|
function logClientTelemetry(label, telemetry) {
|
|
logInfo(
|
|
`client telemetry kind=${label}` +
|
|
` reason=${telemetry.reason}` +
|
|
` currentTime=${formatTelemetryNumber(telemetry.currentTime)}` +
|
|
` paused=${telemetry.paused}` +
|
|
` readyState=${formatTelemetryNumber(telemetry.readyState)}` +
|
|
` pendingFrames=${formatTelemetryNumber(telemetry.pendingFrames)}` +
|
|
` decodedFrames=${formatTelemetryNumber(telemetry.decodedFrames)}` +
|
|
` paintedFrames=${formatTelemetryNumber(telemetry.paintedFrames)}` +
|
|
` lastFramePacketAgeMs=${formatTelemetryNumber(telemetry.lastFramePacketAgeMs)}` +
|
|
` lastFramePaintAgeMs=${formatTelemetryNumber(telemetry.lastFramePaintAgeMs)}` +
|
|
` hidden=${telemetry.hidden}` +
|
|
` online=${telemetry.online}` +
|
|
` counters=${formatTelemetryMap(telemetry.counters)}` +
|
|
` max=${formatTelemetryMap(telemetry.max)}`,
|
|
);
|
|
}
|
|
|
|
function sanitizeTelemetryMap(value) {
|
|
if (!value || typeof value !== 'object' || Array.isArray(value)) {
|
|
return {};
|
|
}
|
|
|
|
const sanitized = {};
|
|
|
|
for (const [key, rawValue] of Object.entries(value).slice(0, 40)) {
|
|
const name = sanitizeTelemetryText(key, '').replace(/[^a-zA-Z0-9_.-]/g, '_').slice(0, 80);
|
|
const number = finiteTelemetryNumber(rawValue);
|
|
|
|
if (name && number !== null) {
|
|
sanitized[name] = number;
|
|
}
|
|
}
|
|
|
|
return sanitized;
|
|
}
|
|
|
|
function sanitizeTelemetryText(value, fallback) {
|
|
const text = typeof value === 'string' ? value : fallback;
|
|
return oneLine(text).replace(/"/g, '').slice(0, 120);
|
|
}
|
|
|
|
function finiteTelemetryNumber(value) {
|
|
const number = Number(value);
|
|
return Number.isFinite(number) ? number : null;
|
|
}
|
|
|
|
function formatTelemetryMap(value) {
|
|
const entries = Object.entries(value);
|
|
|
|
if (entries.length === 0) {
|
|
return 'none';
|
|
}
|
|
|
|
return entries.map(([key, entryValue]) => `${key}=${formatTelemetryNumber(entryValue)}`).join(',');
|
|
}
|
|
|
|
function formatTelemetryNumber(value) {
|
|
if (!Number.isFinite(value)) {
|
|
return 'null';
|
|
}
|
|
|
|
return Math.round(value * 1000) / 1000;
|
|
}
|
|
|
|
function isWebSocketOverFrameBudget(websocket, packetBytes) {
|
|
return websocket.bufferedAmount > 0 && websocket.bufferedAmount + packetBytes > MAX_WS_BUFFER_BYTES;
|
|
}
|
|
|
|
function getJpegFrameByteLength(jpeg) {
|
|
return Buffer.isBuffer(jpeg) ? jpeg.length : jpeg.byteLength;
|
|
}
|
|
|
|
function copyJpegFrame(jpeg, packet, offset) {
|
|
if (Buffer.isBuffer(jpeg)) {
|
|
jpeg.copy(packet, offset);
|
|
return;
|
|
}
|
|
|
|
let writeOffset = offset;
|
|
|
|
for (const part of jpeg.parts) {
|
|
part.copy(packet, writeOffset);
|
|
writeOffset += part.length;
|
|
}
|
|
}
|
|
|
|
function toBufferView(chunk) {
|
|
if (Buffer.isBuffer(chunk)) {
|
|
return chunk;
|
|
}
|
|
|
|
if (chunk instanceof ArrayBuffer) {
|
|
return Buffer.from(chunk);
|
|
}
|
|
|
|
return Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
|
|
}
|
|
|
|
function createSourceInput(sessionId, kind) {
|
|
const session = getSession(sessionId);
|
|
|
|
if (session?.sourceKind === 'local') {
|
|
return {
|
|
url: session.url,
|
|
release: () => {},
|
|
};
|
|
}
|
|
|
|
const token = randomUUID();
|
|
sourceTokens.set(token, { sessionId, kind, createdAt: Date.now() });
|
|
|
|
return {
|
|
url: `http://127.0.0.1:${getListeningPort()}/_source/${token}`,
|
|
release: () => sourceTokens.delete(token),
|
|
};
|
|
}
|
|
|
|
function getListeningPort() {
|
|
const address = server.address();
|
|
return typeof address === 'object' && address ? address.port : PORT;
|
|
}
|
|
|
|
function createAudioWorker(session) {
|
|
const source = createSourceInput(session.id, 'audio');
|
|
|
|
return {
|
|
args: buildAudioArgs(session, source.url),
|
|
release: source.release,
|
|
};
|
|
}
|
|
|
|
function createFrameWorker(session) {
|
|
const source = createSourceInput(session.id, 'frames');
|
|
|
|
return {
|
|
args: buildFrameArgs(session, source.url),
|
|
release: source.release,
|
|
};
|
|
}
|
|
|
|
function buildRelayAudioArgs(session) {
|
|
return buildAudioArgs(session, 'pipe:0', { seekable: false, startTime: 0 });
|
|
}
|
|
|
|
function buildRelayFrameArgs(session) {
|
|
return buildFrameArgs(session, 'pipe:0', { seekable: false, startTime: 0 });
|
|
}
|
|
|
|
function buildPlaybackArgs(session, inputUrl) {
|
|
return [
|
|
...buildInputArgs(inputUrl, { startTime: session.seekSeconds }),
|
|
'-map',
|
|
'0:a:0?',
|
|
'-vn',
|
|
'-ac',
|
|
String(session.options.audioChannels),
|
|
'-ar',
|
|
String(session.options.audioSampleRate),
|
|
'-codec:a',
|
|
'libmp3lame',
|
|
'-b:a',
|
|
session.options.audioBitrate,
|
|
'-f',
|
|
'mp3',
|
|
'pipe:1',
|
|
'-map',
|
|
'0:v:0',
|
|
...buildFrameOutputArgs(session, 'pipe:3'),
|
|
];
|
|
}
|
|
|
|
function buildAudioArgs(session, inputUrl, inputOptions) {
|
|
return [
|
|
...buildInputArgs(inputUrl, { ...inputOptions, startTime: inputOptions?.startTime ?? session.seekSeconds }),
|
|
'-map',
|
|
'0:a:0?',
|
|
'-vn',
|
|
'-ac',
|
|
String(session.options.audioChannels),
|
|
'-ar',
|
|
String(session.options.audioSampleRate),
|
|
'-codec:a',
|
|
'libmp3lame',
|
|
'-b:a',
|
|
session.options.audioBitrate,
|
|
'-f',
|
|
'mp3',
|
|
'pipe:1',
|
|
];
|
|
}
|
|
|
|
function buildFrameArgs(session, inputUrl, inputOptions) {
|
|
return [
|
|
...buildInputArgs(inputUrl, { ...inputOptions, startTime: inputOptions?.startTime ?? session.seekSeconds }),
|
|
'-map',
|
|
'0:v:0',
|
|
...buildFrameOutputArgs(session, 'pipe:1'),
|
|
];
|
|
}
|
|
|
|
function buildInputArgs(inputUrl, { seekable = true, startTime = 0 } = {}) {
|
|
const args = [
|
|
'-hide_banner',
|
|
'-nostdin',
|
|
'-loglevel',
|
|
FFMPEG_LOG_LEVEL,
|
|
'-nostats',
|
|
];
|
|
|
|
if (seekable) {
|
|
args.push('-seekable', startTime > 0 ? '1' : FFMPEG_INPUT_SEEKABLE);
|
|
}
|
|
|
|
if (startTime > 0) {
|
|
args.push('-ss', formatFfmpegSeconds(startTime));
|
|
}
|
|
|
|
if (FFMPEG_HTTP_RECONNECT && isHttpInputUrl(inputUrl)) {
|
|
args.push(
|
|
'-reconnect',
|
|
'1',
|
|
'-reconnect_on_network_error',
|
|
'1',
|
|
'-reconnect_streamed',
|
|
'1',
|
|
'-reconnect_delay_max',
|
|
String(FFMPEG_HTTP_RECONNECT_DELAY_MAX),
|
|
);
|
|
|
|
if (FFMPEG_HTTP_RECONNECT_MAX_RETRIES > 0 && ffmpegSupportsReconnectMaxRetries) {
|
|
args.push('-reconnect_max_retries', String(FFMPEG_HTTP_RECONNECT_MAX_RETRIES));
|
|
}
|
|
|
|
if (FFMPEG_HTTP_RECONNECT_ON_HTTP_ERROR) {
|
|
args.push('-reconnect_on_http_error', FFMPEG_HTTP_RECONNECT_ON_HTTP_ERROR);
|
|
}
|
|
}
|
|
|
|
args.push('-re', '-i', inputUrl);
|
|
return args;
|
|
}
|
|
|
|
function isHttpInputUrl(inputUrl) {
|
|
return inputUrl.startsWith('http://') || inputUrl.startsWith('https://');
|
|
}
|
|
|
|
function formatFfmpegSeconds(seconds) {
|
|
return clampNumber(seconds, 0, Number.MAX_SAFE_INTEGER).toFixed(3);
|
|
}
|
|
|
|
function buildFrameOutputArgs(session, outputUrl) {
|
|
const { fps, quality, width } = session.options;
|
|
const videoFilter = `fps=${fps},scale=w='min(${width},iw)':h=-2:flags=bicubic:out_range=pc,format=yuvj420p`;
|
|
|
|
return [
|
|
'-an',
|
|
'-vf',
|
|
videoFilter,
|
|
'-codec:v',
|
|
'mjpeg',
|
|
'-pix_fmt',
|
|
'yuvj420p',
|
|
'-color_range',
|
|
'pc',
|
|
'-q:v',
|
|
String(quality),
|
|
'-f',
|
|
'image2pipe',
|
|
outputUrl,
|
|
];
|
|
}
|
|
|
|
function spawnFfmpeg(args, stdio = ['ignore', 'pipe', 'pipe']) {
|
|
return spawn(FFMPEG_PATH, args, {
|
|
stdio,
|
|
});
|
|
}
|
|
|
|
function createFfmpegLogger(kind, sessionId, pid) {
|
|
const startedAt = Date.now();
|
|
const label = `${kind}:${shortId(sessionId)}`;
|
|
const lineLogger = createLineLogger((line) => {
|
|
if (shouldSuppressFfmpegLogLine(line)) {
|
|
return;
|
|
}
|
|
|
|
logWarn(`ffmpeg stderr kind=${label} pid=${pid ?? 'unknown'} line="${oneLine(redactSecrets(line))}"`);
|
|
});
|
|
|
|
return {
|
|
start() {
|
|
logInfo(`ffmpeg started kind=${label} pid=${pid ?? 'unknown'} loglevel=${FFMPEG_LOG_LEVEL}`);
|
|
},
|
|
stderr(chunk) {
|
|
lineLogger.write(chunk);
|
|
},
|
|
error(error) {
|
|
logWarn(`ffmpeg start error kind=${label} pid=${pid ?? 'unknown'} error=${oneLine(error.message)}`);
|
|
},
|
|
close(code, signal, reason, stderr) {
|
|
lineLogger.flush();
|
|
const level = reason === 'client_disconnect' || code === 0 || signal === 'SIGTERM' ? 'info' : 'warn';
|
|
const tail = redactSecrets(stderr).trim();
|
|
const tailText = tail ? ` stderrTail="${oneLine(tail)}"` : '';
|
|
|
|
log(level, `ffmpeg exited kind=${label} pid=${pid ?? 'unknown'} code=${code ?? 'null'} signal=${signal ?? 'none'} reason=${reason} durationMs=${Date.now() - startedAt}${tailText}`);
|
|
},
|
|
};
|
|
}
|
|
|
|
function shouldSuppressFfmpegLogLine(line) {
|
|
return line.includes('deprecated pixel format used, make sure you did set range correctly');
|
|
}
|
|
|
|
function createFfmpegStderrTail() {
|
|
let value = '';
|
|
const lineLogger = createLineLogger((line) => {
|
|
if (!shouldSuppressFfmpegLogLine(line)) {
|
|
value = appendTail(value, `${line}\n`);
|
|
}
|
|
});
|
|
|
|
return {
|
|
get value() {
|
|
return value;
|
|
},
|
|
write(chunk) {
|
|
lineLogger.write(chunk);
|
|
},
|
|
flush() {
|
|
lineLogger.flush();
|
|
},
|
|
};
|
|
}
|
|
|
|
function createRelayInputBranch(stream, onError) {
|
|
let accepting = true;
|
|
let ending = false;
|
|
let drainPending = false;
|
|
let streamPendingBytes = 0;
|
|
let queue = [];
|
|
let queueBytes = 0;
|
|
let peakBytes = 0;
|
|
let waiters = [];
|
|
|
|
stream.on('drain', () => {
|
|
drainPending = false;
|
|
streamPendingBytes = 0;
|
|
updatePeakBytes();
|
|
notifyWaiters();
|
|
flush();
|
|
});
|
|
|
|
stream.on('error', (error) => {
|
|
accepting = false;
|
|
streamPendingBytes = 0;
|
|
notifyWaiters();
|
|
onError(error);
|
|
});
|
|
|
|
stream.on('close', () => {
|
|
accepting = false;
|
|
streamPendingBytes = 0;
|
|
notifyWaiters();
|
|
});
|
|
|
|
return {
|
|
get queueBytes() {
|
|
return getPendingBytes();
|
|
},
|
|
get peakBytes() {
|
|
return peakBytes;
|
|
},
|
|
write(chunk) {
|
|
if (!accepting || stream.destroyed) {
|
|
return false;
|
|
}
|
|
|
|
if (queue.length === 0 && !drainPending) {
|
|
return writeNow(chunk);
|
|
}
|
|
|
|
queue.push(chunk);
|
|
queueBytes += chunk.length;
|
|
updatePeakBytes();
|
|
notifyWaiters();
|
|
return getPendingBytes() <= MAX_RELAY_BRANCH_QUEUE_BYTES;
|
|
},
|
|
end() {
|
|
accepting = false;
|
|
ending = true;
|
|
flush();
|
|
},
|
|
destroy() {
|
|
accepting = false;
|
|
ending = false;
|
|
queue = [];
|
|
queueBytes = 0;
|
|
streamPendingBytes = 0;
|
|
notifyWaiters();
|
|
|
|
if (!stream.destroyed) {
|
|
stream.destroy();
|
|
}
|
|
},
|
|
waitForCapacity() {
|
|
if (getPendingBytes() <= RELAY_BRANCH_PAUSE_BYTES || !accepting || stream.destroyed) {
|
|
return Promise.resolve();
|
|
}
|
|
|
|
return new Promise((resolve) => {
|
|
waiters.push(resolve);
|
|
});
|
|
},
|
|
};
|
|
|
|
function writeNow(chunk) {
|
|
try {
|
|
if (!stream.write(chunk)) {
|
|
drainPending = true;
|
|
streamPendingBytes += chunk.length;
|
|
updatePeakBytes();
|
|
}
|
|
|
|
notifyWaiters();
|
|
return true;
|
|
} catch (error) {
|
|
accepting = false;
|
|
streamPendingBytes = 0;
|
|
notifyWaiters();
|
|
onError(error);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function flush() {
|
|
while (queue.length > 0 && !drainPending && !stream.destroyed) {
|
|
const chunk = queue.shift();
|
|
queueBytes -= chunk.length;
|
|
|
|
if (!writeNow(chunk)) {
|
|
return;
|
|
}
|
|
|
|
notifyWaiters();
|
|
}
|
|
|
|
if (ending && queue.length === 0 && !drainPending && !stream.destroyed) {
|
|
stream.end();
|
|
}
|
|
}
|
|
|
|
function notifyWaiters() {
|
|
if (getPendingBytes() > RELAY_BRANCH_PAUSE_BYTES && accepting && !stream.destroyed) {
|
|
return;
|
|
}
|
|
|
|
const currentWaiters = waiters;
|
|
waiters = [];
|
|
|
|
for (const resolve of currentWaiters) {
|
|
resolve();
|
|
}
|
|
}
|
|
|
|
function getPendingBytes() {
|
|
return queueBytes + streamPendingBytes;
|
|
}
|
|
|
|
function updatePeakBytes() {
|
|
peakBytes = Math.max(peakBytes, getPendingBytes());
|
|
}
|
|
}
|
|
|
|
async function waitForRelayCapacity(branches, shouldStop) {
|
|
while (!shouldStop()) {
|
|
const blockedBranches = branches.filter((branch) => branch.queueBytes > RELAY_BRANCH_PAUSE_BYTES);
|
|
|
|
if (blockedBranches.length === 0) {
|
|
return;
|
|
}
|
|
|
|
await Promise.race(blockedBranches.map((branch) => branch.waitForCapacity()));
|
|
}
|
|
}
|
|
|
|
function createLineLogger(callback) {
|
|
let buffer = '';
|
|
|
|
return {
|
|
write(chunk) {
|
|
buffer += chunk.toString('utf8');
|
|
const lines = buffer.split(/\r?\n/);
|
|
buffer = lines.pop() ?? '';
|
|
|
|
for (const line of lines) {
|
|
if (line.trim()) {
|
|
callback(line);
|
|
}
|
|
}
|
|
},
|
|
flush() {
|
|
if (buffer.trim()) {
|
|
callback(buffer);
|
|
}
|
|
|
|
buffer = '';
|
|
},
|
|
};
|
|
}
|
|
|
|
function createSourceRequestHeaders(session) {
|
|
const headers = new Headers();
|
|
|
|
for (const [name, value] of Object.entries(session.sourceHeaders ?? {})) {
|
|
headers.set(name, value);
|
|
}
|
|
|
|
return headers;
|
|
}
|
|
|
|
function copyUpstreamHeaders(upstreamHeaders, response) {
|
|
for (const header of [
|
|
'accept-ranges',
|
|
'content-length',
|
|
'content-range',
|
|
'content-type',
|
|
'etag',
|
|
'last-modified',
|
|
]) {
|
|
const value = upstreamHeaders.get(header);
|
|
|
|
if (value) {
|
|
response.setHeader(header, value);
|
|
}
|
|
}
|
|
|
|
response.setHeader('cache-control', 'no-store');
|
|
}
|
|
|
|
function sendJson(websocket, payload) {
|
|
if (websocket.readyState === WebSocket.OPEN) {
|
|
websocket.send(JSON.stringify(payload));
|
|
}
|
|
}
|
|
|
|
function appendTail(current, chunk) {
|
|
const next = current + chunk.toString('utf8');
|
|
return next.length > 4000 ? next.slice(-4000) : next;
|
|
}
|
|
|
|
function shortId(id) {
|
|
return id.slice(0, 8);
|
|
}
|
|
|
|
function oneLine(value) {
|
|
return String(value).replace(/\s+/g, ' ').trim().slice(0, 1200);
|
|
}
|
|
|
|
function logInfo(message) {
|
|
log('info', message);
|
|
}
|
|
|
|
function logWarn(message) {
|
|
log('warn', message);
|
|
}
|
|
|
|
function log(level, message) {
|
|
const output = `${new Date().toISOString()} ${message}`;
|
|
|
|
if (level === 'warn') {
|
|
console.warn(output);
|
|
return;
|
|
}
|
|
|
|
console.log(output);
|
|
}
|
|
|
|
function summarizeFfmpegExit(code, signal, stderr) {
|
|
if (signal) {
|
|
return `ffmpeg stopped by ${signal}.`;
|
|
}
|
|
|
|
if (code === 0 || code === null) {
|
|
return 'Frame stream ended.';
|
|
}
|
|
|
|
const detail = redactSecrets(stderr).trim();
|
|
return detail ? `ffmpeg exited with code ${code}: ${detail}` : `ffmpeg exited with code ${code}.`;
|
|
}
|
|
|
|
function redactSecrets(text) {
|
|
return text.replace(/([?&](?:api_key|apikey|access_token|token|key|sig|signature|lsig|n|expire|ip)=)[^&\s]+/gi, '$1[redacted]');
|
|
}
|
|
|
|
function isChildRunning(child) {
|
|
return Boolean(child && child.exitCode === null && child.signalCode === null);
|
|
}
|
|
|
|
function stopProcess(child) {
|
|
if (!child || child.exitCode !== null || child.signalCode !== null) {
|
|
return;
|
|
}
|
|
|
|
child.kill('SIGTERM');
|
|
|
|
setTimeout(() => {
|
|
if (child.exitCode === null && child.signalCode === null) {
|
|
child.kill('SIGKILL');
|
|
}
|
|
}, 1500).unref();
|
|
}
|
|
|
|
function once(callback) {
|
|
let called = false;
|
|
|
|
return (...args) => {
|
|
if (called) {
|
|
return;
|
|
}
|
|
|
|
called = true;
|
|
callback(...args);
|
|
};
|
|
}
|