josh/ytplayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cache every played or saved video on the server via background jobs with a validation gate and a Broken re-download button

Browse Source
This commit is contained in:
Jonathan Sykes
2026-09-13 09:45:01 +08:00
parent 292a0c7e29
commit 5fb863ba25
8 changed files with 1240 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
CLAUDE.md
View File

@@ -15,15 +15,44 @@ The PWA is what runs in production; `legacy/` holds the old native-only docs.
| `frontend/fingerprint.js` | Browser fingerprint used as the sync key |
| `frontend/async-guard.js` | Stale-async-response guard (unit-tested) |
| `server/server.js` | Bun + Hono backend — endpoint list is in its header comment |
| `server/db.js` | libsql schema/queries (`users`, `profiles`, `video_history`…) |
| `server/db.js` | libsql schema/queries (`users`, `profiles`, `video_history`, `media_cache`…) |
| `server/media-cache.js` | Server-side video cache: background fetch jobs, validation gate, x264 lane, LRU |
| `bin/yt-dlp` | Downloaded by `npm run setup`, gitignored |
| `scripts/` | icon generation, yt-dlp setup, push helper |
| `tests/` | Playwright e2e specs · unit tests live next to sources in `frontend/*.test.js` |
| `legacy/` | Old native-shell docs — do not treat as current |
API endpoints: `GET /api/search|channel|streams|download/:id|version|user/data|profile/load`,
`GET /api/media/:id|media/:id/status|media/stats`, `POST /api/media/:id/redownload`,
`POST /api/user/sync|profile/create|profile/save`, `GET /sw.js` (BUILD_TAG-injected), `GET /*` static.
JSON shapes mirror the Tauri Rust bridge exactly — don't change one side alone.
JSON shapes mirror the Tauri Rust bridge exactly — don't change one side alone
(`/api/streams`' `data.serverCached` is additive and web-only).
## Server media cache (`server/media-cache.js`)
- Every played (`/api/streams`, LOW priority, ≤ `MEDIA_AUTO_MAX_SECONDS`) or saved
(`/api/download`, HIGH, ≤ 3 h) video gets ONE copy: `$MEDIA_DIR/<id>.<gen>.mp4`
(≤720p H.264 8-bit + AAC, faststart) + `<id>.<gen>.m4a` audio sidecar for audio-only
mode. `MEDIA_DIR` defaults to `./data/media` (the `ytplayer-data` volume in prod).
- Jobs are server-owned — **never pass a request AbortSignal into them**; a client
closing its tab must not kill a fetch. Downloads still run inside `withSaveSlot`.
- **Validation gate** (`validateMedia`): h264 yuv420p + AAC, container/stream durations
cover the source, a full `-c copy` demux pass with zero stderr, head + tail decode.
The demux pass is what catches a faststart file whose index is intact but whose
data is cut. Nothing reaches the cache dir without passing it; failures back off
(15 min · 2^n, max 24 h) and saves fall back to the legacy per-request path.
- `gen` is in the filename and URL (`/api/media/<id>?g=<gen>`): a URL never serves
different bytes. A replaced gen is kept 30 min so in-flight Range playback survives.
- **Compression lane**: niced x264 `-preset slow -crf 28 -fpsmax 30`, audio copied,
kept only if ≥15% smaller. Measured: YouTube's 720p avc1 is already ~0.6–0.7 Mbps, so
CRF 24 made files *bigger*; CRF 28 ≈ −20% video at SSIM 0.994. Lyric videos usually
keep the original. HEVC/AV1/10-bit are smaller but don't play in iOS `<video>`.
- **Budget**: `MEDIA_CACHE_MAX_BYTES` (10 GiB) LRU by `last_access` (anything played in
the last 10 min is protected) + `MEDIA_MIN_FREE_BYTES` (5 GiB) disk guard → skip, stream.
- **Broken**: `POST /api/media/:id/redownload` (now-playing ⚠ Broken button). When a
client falls back from a cached copy (`/api/streams?nocache=1`) the server also
re-validates its copy and refetches it if it fails — never trusts one device's error.
- Ops: `curl https://worship.hesed.sbs/api/media/stats`; files under
`/app/data/media` in the container.
## Local dev
@@ -129,6 +158,8 @@ Local DB file: `server/data/ytplayer.db` (gitignored). `BUILD_TAG` is computed f
## Testing
- Unit: **`node --test frontend/*.test.js`** (sw, sw-update, async-guard, video-edit).
Server: **`cd server && bun test`** (media-cache: real ffmpeg fixtures + a temp libsql DB).
Local e2e needs a CURRENT yt-dlp — a 2-month-old one 403s on every download.
The directory form `node --test frontend/` fails on Node 22 with
`Cannot find module .../frontend` — it resolves the dir as a module, not a test
glob. That's the harness, not the tests.

9
docker-compose.yml
View File

@@ -11,6 +11,15 @@ services:
APP_VERSION: "1.0.0"
# Optional: override yt-dlp binary path if you mount a custom one
# YTDLP_PATH: "/usr/local/bin/yt-dlp"
# Server media cache (server/media-cache.js) — defaults shown.
# MEDIA_DIR: "/app/data/media" # on the ytplayer-data volume
# MEDIA_CACHE_MAX_BYTES: "10737418240" # 10 GiB, LRU eviction past this
# MEDIA_MIN_FREE_BYTES: "5368709120" # skip caching below 5 GiB free disk
# MEDIA_AUTO_MAX_SECONDS: "3600" # longest video auto-cached on play
# MEDIA_TRANSCODE: "1" # 0 disables the x264 lane
# MEDIA_CRF: "28"
# MEDIA_PRESET: "slow"
# MEDIA_THREADS: "2"
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:3000/api/version"]
interval: 30s

78
frontend/app.js
View File

@@ -196,8 +196,10 @@ const API = {
expandPlaylist: (url) => WEB
? webFetch(`/api/playlist/expand?url=${encodeURIComponent(url)}`)
: call('yt.expand_playlist', 'yt_expand_playlist', { url }),
getStreams: (videoId) => WEB
? webFetch(`/api/streams?v=${encodeURIComponent(videoId)}`)
// nocache=1 skips the server's cached copy and resolves fresh YouTube
// streams (used when the cached copy won't play on this device).
getStreams: (videoId, { nocache = false } = {}) => WEB
? webFetch(`/api/streams?v=${encodeURIComponent(videoId)}${nocache ? '&nocache=1' : ''}`)
: call('yt.streams', 'yt_streams', { videoId }),
loadData: () => WEB
? Promise.resolve(loadDataFromStorage())
@@ -333,6 +335,7 @@ const els = {
audioOnlyToggle: $('audioOnlyToggle'),
saveBtn: $('saveBtn'),
editBtn: $('editBtn'),
brokenBtn: $('brokenBtn'),
addPlaylistBtn: $('addPlaylistBtn'),
};
@@ -1377,7 +1380,7 @@ const Player = {
return this.mode === 'dual' ? els.audio : this.master;
},
async loadVideo(videoObj, { preferStream = false, resume = true, reveal = true } = {}) {
async loadVideo(videoObj, { preferStream = false, resume = true, reveal = true, nocache = false } = {}) {
// When false (auto-advance / prev), skip resuming the saved timestamp and
// start from the beginning (or the A marker, if an A-B loop is set).
this._resumeOnLoad = resume;
@@ -1449,7 +1452,7 @@ const Player = {
}
}
const res = await API.getStreams(videoObj.id);
const res = await API.getStreams(videoObj.id, { nocache });
if (seq !== this._loadSeq) return; // a newer load took over meanwhile
if (!res || !res.ok) throw new Error(res?.error || 'Could not load streams');
current = res.data;
@@ -1471,7 +1474,7 @@ const Player = {
btn.textContent = '↻ Retry';
btn.addEventListener('click', () => {
btn.remove();
Player.loadVideo(videoObj, { preferStream, resume, reveal });
Player.loadVideo(videoObj, { preferStream, resume, reveal, nocache });
});
els.playerPane.appendChild(btn);
}
@@ -1572,6 +1575,16 @@ const Player = {
this.loadVideo(meta, { preferStream: true, resume: this._resumeOnLoad, reveal: this._revealOnLoad });
return;
}
// The server's cached copy failed on this device — stream from YouTube
// instead. Never auto-reported: a decode failure can be specific to one
// device, so the ⚠ Broken button is how a user flags a bad server copy.
if (current && current.serverCached) {
const meta = current.meta;
current.serverCached = false;
toast('Server copy failed — streaming from YouTube. Tap ⚠ Broken to re-download it.', { duration: 5000 });
this.loadVideo(meta, { preferStream: true, nocache: true, resume: this._resumeOnLoad, reveal: this._revealOnLoad });
return;
}
// Advance to the next candidate stream; give up with a clear message at the end.
if (this.fbIndex < this.fallbackQueue.length - 1) {
this.fbIndex++;
@@ -2027,6 +2040,43 @@ function updatePlayBtn() {
const spBtn = $('servicePlayBtn');
if (spBtn) spBtn.textContent = Player.master.paused ? '▶' : '⏸';
}
// "⚠ Broken": ask the server to drop its cached copy and fetch a fresh one.
// Playback switches to live YouTube streams meanwhile (resuming where it
// was), and a device copy — most likely made from the same bad file — is
// replaced: the re-save waits on the server's refetch job.
const reportingBroken = new Set();
async function reportBroken(video) {
const id = video.id;
if (!id || video.custom || reportingBroken.has(id)) return;
reportingBroken.add(id);
updateNowPlayingActions();
try {
const res = await fetch(`/api/media/${encodeURIComponent(id)}/redownload`, { method: 'POST' });
const j = await res.json().catch(() => ({}));
if (!res.ok || !j.ok) throw new Error(j.error || `HTTP ${res.status}`);
toast('Re-downloading on the server…');
const hadLocal = cachedIds.has(id);
if (hadLocal) {
try { await API.cacheDelete(id); } catch { /* best-effort */ }
cachedIds.delete(id);
markCardCacheState(id, 'none');
}
if (current && current.meta && current.meta.id === id) {
const at = Player.master ? Player.master.currentTime || 0 : 0;
await Player.loadVideo(current.meta, { preferStream: true, nocache: true, resume: false, reveal: false });
if (at > 1 && current && current.meta && current.meta.id === id) {
Player.master.addEventListener('loadedmetadata', () => Player.seek(at), { once: true });
}
}
if (hadLocal) preload(video);
} catch (err) {
toast('⚠ Could not report: ' + err.message);
} finally {
reportingBroken.delete(id);
updateNowPlayingActions();
}
}
// Reflect cache state on the now-playing Save button.
function updateNowPlayingActions() {
if (!current || !current.meta) return;
@@ -2036,6 +2086,12 @@ function updateNowPlayingActions() {
const isCustom = !!current.meta.custom;
if (els.editBtn) els.editBtn.hidden = isCustom;
if (els.saveBtn) els.saveBtn.hidden = isCustom;
// The server cache exists only behind the web PWA backend.
if (els.brokenBtn) {
els.brokenBtn.hidden = isCustom || !WEB;
els.brokenBtn.disabled = reportingBroken.has(id);
els.brokenBtn.textContent = reportingBroken.has(id) ? '⏳ Reporting…' : '⚠ Broken';
}
const btn = els.saveBtn;
if (!btn) return;
if (removing.has(id)) {
@@ -4501,6 +4557,18 @@ document.querySelectorAll('.chip').forEach((c) => {
if (els.editBtn) els.editBtn.addEventListener('click', () => {
if (current && current.meta) openVideoEditor(current.meta);
});
if (els.brokenBtn) els.brokenBtn.addEventListener('click', () => {
if (!current || !current.meta || current.meta.custom) return;
const video = { ...current.meta };
const body = document.createElement('p');
body.textContent = `Re-download “${video.title || video.id}” on the server? Its saved copy is deleted `
+ 'and fetched again from YouTube; playback continues from YouTube meanwhile'
+ (cachedIds.has(video.id) ? ', and this device’s offline copy is replaced with the fresh one.' : '.');
showModal('Video broken?', body, [
{ label: 'Cancel', onClick: closeModal },
{ label: 'Re-download', primary: true, onClick: () => { closeModal(); reportBroken(video); } },
]);
});
els.queueBtn.addEventListener('click', () => {
if (current && current.meta) addToQueue(current.meta);
});

1
frontend/index.html
View File

@@ -202,6 +202,7 @@
<button id="saveBtn" class="np-btn" title="Save this video for offline playback">⬇ Save</button>
<button id="editBtn" class="np-btn" title="Cut parts out and save a custom edited copy">✂ Edit &amp; download</button>
<button id="addPlaylistBtn" class="np-btn" title="Add to a playlist">＋ Playlist</button>
<button id="brokenBtn" class="np-btn" title="Video won't play or looks broken? Re-download the server's copy" hidden>⚠ Broken</button>
</div>
</div>

102
server/db.js
View File

@@ -9,6 +9,7 @@
* users — one row per fingerprint; tracks last app version seen
* playlists — one row per fingerprint; full playlist JSON blob
* video_history — one row per (fingerprint, video_id); recent 200 entries
* media_cache — one row per server-cached video file (see media-cache.js)
* ========================================================================== */
import { createClient } from '@libsql/client';
@@ -81,9 +82,110 @@ export async function initDb() {
CREATE INDEX IF NOT EXISTS idx_inbox_to
ON playlist_inbox (to_name, created_at DESC);
-- Server-side media cache: one validated mp4 per video under MEDIA_DIR.
-- The file on disk is <video_id>.<gen>.mp4 (+ .m4a audio sidecar); gen
-- bumps on every replacement so a URL carrying ?g= always maps to the
-- same bytes. Timestamps here are ms epochs (backoff math needs ms).
CREATE TABLE IF NOT EXISTS media_cache (
video_id TEXT PRIMARY KEY,
status TEXT NOT NULL DEFAULT 'queued', -- queued|downloading|validating|ready|failed
gen INTEGER NOT NULL DEFAULT 0,
size INTEGER NOT NULL DEFAULT 0, -- mp4 + m4a bytes
height INTEGER NOT NULL DEFAULT 0,
vcodec TEXT,
acodec TEXT,
duration REAL NOT NULL DEFAULT 0,
optimized INTEGER NOT NULL DEFAULT 0,
meta TEXT NOT NULL DEFAULT '{}',
priority INTEGER NOT NULL DEFAULT 1,
auto INTEGER NOT NULL DEFAULT 0,
attempts INTEGER NOT NULL DEFAULT 0,
error TEXT,
retry_at INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL DEFAULT 0,
last_access INTEGER NOT NULL DEFAULT 0,
hits INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_media_lru
ON media_cache (status, last_access);
`);
}
// ---- Media cache -------------------------------------------------------------
const MEDIA_COLS = new Set([
'status', 'gen', 'size', 'height', 'vcodec', 'acodec', 'duration', 'optimized',
'meta', 'priority', 'auto', 'attempts', 'error', 'retry_at', 'created_at',
'updated_at', 'last_access', 'hits',
]);
function rowToObj(res, row) {
const o = {};
res.columns.forEach((c, i) => { o[c] = row[i]; });
return o;
}
export async function getMedia(videoId) {
const r = await db.execute({ sql: 'SELECT * FROM media_cache WHERE video_id = ?', args: [videoId] });
return r.rows[0] ? rowToObj(r, r.rows[0]) : null;
}
// Insert or partially update a media row. Unknown keys are ignored.
export async function upsertMedia(videoId, fields) {
const cols = Object.keys(fields).filter((k) => MEDIA_COLS.has(k));
const all = ['video_id', ...cols];
const set = cols.length ? 'DO UPDATE SET ' + cols.map((c) => `${c} = excluded.${c}`).join(', ') : 'DO NOTHING';
await db.execute({
sql: `INSERT INTO media_cache (${all.join(', ')}) VALUES (${all.map(() => '?').join(', ')})
ON CONFLICT(video_id) ${set}`,
args: [videoId, ...cols.map((c) => fields[c] ?? null)],
});
}
export async function deleteMedia(videoId) {
await db.execute({ sql: 'DELETE FROM media_cache WHERE video_id = ?', args: [videoId] });
}
export async function listMedia() {
const r = await db.execute('SELECT * FROM media_cache');
return r.rows.map((row) => rowToObj(r, row));
}
// Ready rows, least-recently-played first — the eviction order.
export async function listMediaLru() {
const r = await db.execute(
"SELECT video_id, size, last_access FROM media_cache WHERE status = 'ready' ORDER BY last_access ASC",
);
return r.rows.map((row) => rowToObj(r, row));
}
export async function touchMedia(videoId, now) {
await db.execute({
sql: 'UPDATE media_cache SET last_access = ?, hits = hits + 1 WHERE video_id = ?',
args: [now, videoId],
});
}
export async function mediaStats() {
const r = await db.execute(
`SELECT
SUM(CASE WHEN status = 'ready' THEN 1 ELSE 0 END) AS count,
SUM(CASE WHEN status = 'ready' THEN size ELSE 0 END) AS bytes,
SUM(CASE WHEN status IN ('queued','downloading','validating') THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed,
SUM(CASE WHEN status = 'ready' AND optimized = 1 THEN 1 ELSE 0 END) AS optimized
FROM media_cache`,
);
const o = r.rows[0] ? rowToObj(r, r.rows[0]) : {};
return {
count: o.count || 0, bytes: o.bytes || 0, pending: o.pending || 0,
failed: o.failed || 0, optimized: o.optimized || 0,
};
}
// ---- Profiles (named cross-device sync; the name acts as the passkey) ------
// Insert a new profile. Returns false when the name is already taken.

556
server/media-cache.js Normal file
View File

@@ -0,0 +1,556 @@
/* ============================================================================
* media-cache.js — server-side video cache fed by background jobs
*
* Every video that is played or saved gets ONE validated copy on disk:
* <dir>/<videoId>.<gen>.mp4 ≤720p H.264 (8-bit 4:2:0) + AAC, faststart
* <dir>/<videoId>.<gen>.m4a audio-only sidecar (stream copy) for audio mode
*
* - Jobs are owned by the server, not by an HTTP request: a client closing
* its tab never kills a download, the copy just lands for next time.
* - Nothing is promoted without passing validateMedia() — a truncated or
* undecodable file must never be served "forever".
* - Copies never expire by time. They leave only through LRU eviction at the
* byte budget, or through redownload() (the client's "Broken" button).
* - `gen` bumps on every replacement and is part of the filename, so a URL
* carrying ?g=<gen> always maps to the same bytes; superseded files are kept
* for a grace period so in-flight Range playback never splices two encodes.
* - After a copy is ready, a separate low-priority lane re-encodes it with
* x264 CRF and keeps the result only when it is meaningfully smaller.
* ========================================================================== */
import { spawn } from 'node:child_process';
import {
copyFileSync, existsSync, linkSync, mkdirSync, readdirSync, renameSync,
statSync, statfsSync, unlinkSync,
} from 'node:fs';
import { join } from 'node:path';
export const HIGH = 0; // explicit save / Broken re-download
export const LOW = 1; // auto-cache after a play
const ID_RE = /^[A-Za-z0-9_-]{6,64}$/;
export const isMediaId = (id) => typeof id === 'string' && ID_RE.test(id);
const MIN_BYTES = 100 * 1024;
const OLD_GEN_GRACE_MS = 30 * 60_000; // keep superseded files for in-flight playback
const EVICT_PROTECT_MS = 10 * 60_000; // never evict something played this recently
const TOUCH_DEBOUNCE_MS = 60_000;
const EST_BYTES_PER_SEC = 250 * 1024; // ~2 Mbps: pre-compression 720p + 128k AAC
const MIN_SAVING = 0.15; // keep an x264 re-encode only if ≥15% smaller
const DAY_MS = 24 * 3600_000;
// A condition that means "don't cache this right now, just stream it" — not a
// broken download. code: SKIPPED (live, too long, disk/budget) | BACKOFF.
export class MediaSkip extends Error {
constructor(message, code = 'SKIPPED') { super(message); this.code = code; }
}
const tail = (s, n = 400) => String(s || '').trim().slice(-n);
const sizeOf = (p) => { try { return statSync(p).size; } catch { return 0; } };
const fmtMB = (b) => (b / 1048576).toFixed(1) + ' MB';
// Spawn a process and collect its output. Never rejects.
function run(cmd, args) {
return new Promise((resolve) => {
let child;
try { child = spawn(cmd, args, { stdio: ['ignore', 'pipe', 'pipe'] }); }
catch (e) { resolve({ code: -1, stdout: '', stderr: e.message }); return; }
let out = '';
let err = '';
child.stdout.setEncoding('utf8');
child.stderr.setEncoding('utf8');
child.stdout.on('data', (d) => { out += d; });
child.stderr.on('data', (d) => { err = (err + d).slice(-8000); });
child.on('error', (e) => resolve({ code: -1, stdout: out, stderr: e.message }));
child.on('close', (code) => resolve({ code, stdout: out, stderr: err }));
});
}
// ----------------------------------------------------------------------------
// Validation gate. Throws with a reason; resolves to probe facts on success.
// 1. file exists and is not tiny
// 2. exactly one H.264 8-bit 4:2:0 video stream + an AAC audio stream
// (AV1/VP9/10-bit play on almost nothing Apple — see SAVE_MUX_FORMAT)
// 3. container AND per-stream durations cover the expected length
// 4. full demux pass reports no errors (catches corrupt/truncated boxes)
// 5. the first and last seconds actually decode
// ----------------------------------------------------------------------------
export async function validateMedia(path, expected, { ffmpeg = 'ffmpeg', ffprobe = 'ffprobe' } = {}) {
const size = sizeOf(path);
if (!size) throw new Error('validation: output file missing');
if (size < MIN_BYTES) throw new Error(`validation: file too small (${size} bytes)`);
const pr = await run(ffprobe, [
'-v', 'error',
'-show_entries', 'format=duration:stream=codec_type,codec_name,height,pix_fmt,duration:stream_disposition=attached_pic',
'-of', 'json', path,
]);
if (pr.code !== 0) throw new Error('validation: ffprobe failed: ' + tail(pr.stderr));
let j;
try { j = JSON.parse(pr.stdout); } catch { throw new Error('validation: unreadable ffprobe output'); }
const streams = Array.isArray(j.streams) ? j.streams : [];
const video = streams.filter((s) => s.codec_type === 'video' && !(s.disposition && s.disposition.attached_pic));
const audio = streams.filter((s) => s.codec_type === 'audio');
if (video.length !== 1) throw new Error(`validation: expected 1 video stream, found ${video.length}`);
const v = video[0];
if (v.codec_name !== 'h264') throw new Error(`validation: video codec ${v.codec_name} is not h264`);
if (v.pix_fmt && !/^yuvj?420p$/.test(v.pix_fmt)) throw new Error(`validation: pixel format ${v.pix_fmt} is not 8-bit 4:2:0`);
const a = audio.find((s) => s.codec_name === 'aac');
if (!a) throw new Error('validation: no AAC audio stream');
const duration = Number(j.format && j.format.duration) || 0;
if (duration <= 0) throw new Error('validation: zero duration');
if (expected > 0) {
const min = expected - Math.max(3, expected * 0.03);
for (const [label, d] of [['file', duration], ['video', Number(v.duration)], ['audio', Number(a.duration)]]) {
if (Number.isFinite(d) && d > 0 && d < min) {
throw new Error(`validation: ${label} truncated (${d.toFixed(1)}s of ${expected}s)`);
}
}
}
const demux = await run(ffmpeg, ['-v', 'error', '-nostdin', '-i', path, '-map', '0', '-c', 'copy', '-f', 'null', '-']);
if (demux.code !== 0 || demux.stderr.trim()) throw new Error('validation: demux errors: ' + tail(demux.stderr));
const head = await run(ffmpeg, ['-v', 'error', '-nostdin', '-i', path, '-t', '3', '-f', 'null', '-']);
if (head.code !== 0 || head.stderr.trim()) throw new Error('validation: head decode failed: ' + tail(head.stderr));
const end = await run(ffmpeg, ['-v', 'error', '-nostdin', '-sseof', '-5', '-i', path, '-f', 'null', '-']);
if (end.code !== 0 || end.stderr.trim()) throw new Error('validation: tail decode failed: ' + tail(end.stderr));
return { size, height: v.height || 0, vcodec: v.codec_name, acodec: a.codec_name, duration };
}
function metaFromInfo(info, duration) {
const s = (...keys) => {
for (const k of keys) if (typeof info[k] === 'string' && info[k].trim()) return info[k].trim();
return '';
};
return {
title: s('title') || '(untitled)',
channel: s('channel', 'uploader'),
channelId: s('channel_id', 'uploader_id'),
channelUrl: s('channel_url', 'uploader_url'),
duration: typeof info.duration === 'number' ? info.duration : Math.round(duration || 0),
};
}
// ----------------------------------------------------------------------------
// createMediaCache(opts)
// dir cache directory (tmp work dir lives in <dir>/.tmp, same fs)
// db { getMedia, upsertMedia, deleteMedia, listMedia, listMediaLru, touchMedia, mediaStats }
// download async (videoId, outPath) => path — yt-dlp into outPath
// getInfo async (videoId) => yt-dlp -J info — duration, live flags, meta
// ----------------------------------------------------------------------------
export function createMediaCache({
dir, db, download, getInfo,
ffmpeg = 'ffmpeg', ffprobe = 'ffprobe', nice = 'nice',
maxBytes = 10 * 1024 ** 3,
minFreeBytes = 5 * 1024 ** 3,
autoMaxSeconds = 3600,
saveMaxSeconds = 3 * 3600,
transcode = { enabled: true, crf: 28, preset: 'slow', threads: 2 },
freeBytes = () => { const s = statfsSync(dir); return s.bavail * s.bsize; },
now = () => Date.now(),
log = console,
} = {}) {
const tmpDir = join(dir, '.tmp');
const fileFor = (id, gen, kind = 'mp4') => join(dir, `${id}.${gen}.${kind}`);
const jobs = new Map(); // videoId -> fetch job (queued or running)
const fetchQueue = [];
let fetchActive = null; // job currently downloading/validating
let seq = 0;
const optQueue = [];
const optPending = new Set();
let optActive = null; // videoId currently being re-encoded
const lastTouch = new Map();
function sweepTmp(prefix) {
let names = [];
try { names = readdirSync(tmpDir); } catch { return; }
for (const n of names) {
if (!prefix || n.startsWith(prefix)) { try { unlinkSync(join(tmpDir, n)); } catch { /* gone */ } }
}
}
// Delete this video's files, optionally keeping one generation. Superseded
// generations can be deleted after a delay so playback in flight finishes.
function removeFiles(id, keepGen = null, delayMs = 0) {
let names = [];
try { names = readdirSync(dir); } catch { return; }
for (const n of names) {
const m = /^(.+)\.(\d+)\.(mp4|m4a)$/.exec(n);
if (!m || m[1] !== id || (keepGen !== null && Number(m[2]) === keepGen)) continue;
const p = join(dir, n);
const rm = () => { try { unlinkSync(p); } catch { /* gone */ } };
if (delayMs > 0) { const t = setTimeout(rm, delayMs); if (t && t.unref) t.unref(); }
else rm();
}
}
function readyFilesExist(row) {
return existsSync(fileFor(row.video_id, row.gen)) && existsSync(fileFor(row.video_id, row.gen, 'm4a'));
}
function safeFree() {
try { return freeBytes(); } catch { return null; }
}
async function evict(id, why) {
await db.deleteMedia(id);
removeFiles(id);
log.info?.(`[media] evicted ${id} (${why})`);
}
// Make room for ~est bytes under the budget (LRU), then check the disk guard.
async function makeRoom(id, est) {
let { bytes } = await db.mediaStats();
if (bytes + est > maxBytes) {
const cutoff = now() - EVICT_PROTECT_MS;
for (const r of await db.listMediaLru()) {
if (bytes + est <= maxBytes) break;
if (r.video_id === id || r.last_access > cutoff || jobs.has(r.video_id) || optActive === r.video_id) continue;
await evict(r.video_id, 'budget');
bytes -= r.size || 0;
}
}
if (bytes + est > maxBytes) throw new MediaSkip('cache budget is full of recently played videos');
const free = safeFree();
if (free !== null && free - est < minFreeBytes) {
throw new MediaSkip(`disk guard: only ${fmtMB(free)} free on the cache volume`);
}
}
async function setStatus(job, status) {
job.status = status;
await db.upsertMedia(job.id, { status, updated_at: now() });
}
function bump(job, priority, auto) {
if (priority < job.priority) job.priority = priority;
if (!auto) job.auto = false; // an explicit request lifts the auto length limit
}
function startJob(id, priority, auto, row) {
let resolve, reject;
const promise = new Promise((res, rej) => { resolve = res; reject = rej; });
promise.catch(() => {}); // auto jobs have nobody awaiting them
const job = { id, priority, auto, seq: seq++, status: 'queued', promise, resolve, reject };
job.persisted = db.upsertMedia(id, {
status: 'queued', priority, auto: auto ? 1 : 0,
gen: (row && row.gen) || 0,
created_at: (row && row.created_at) || now(),
updated_at: now(),
}).catch((e) => log.warn?.(`[media] ${id} queue write failed: ${e.message}`));
jobs.set(id, job);
fetchQueue.push(job);
pumpFetch();
return promise;
}
function pumpFetch() {
if (fetchActive || !fetchQueue.length) return;
fetchQueue.sort((a, b) => a.priority - b.priority || a.seq - b.seq);
const job = fetchQueue.shift();
fetchActive = job;
runFetch(job).then(job.resolve, job.reject).finally(() => {
jobs.delete(job.id);
fetchActive = null;
pumpFetch();
});
}
async function runFetch(job) {
const { id } = job;
const prefix = `${id}-${now()}`;
const base = join(tmpDir, prefix);
await job.persisted;
try {
await setStatus(job, 'downloading');
const info = await getInfo(id);
const live = info.is_live || ['is_live', 'post_live', 'is_upcoming'].includes(info.live_status);
if (live) throw new MediaSkip('live streams are not cached');
const expected = typeof info.duration === 'number' ? info.duration : 0;
const limit = job.auto ? autoMaxSeconds : saveMaxSeconds;
if (expected > limit) {
throw new MediaSkip(`too long to cache (${Math.round(expected / 60)} min, limit ${Math.round(limit / 60)} min)`);
}
await makeRoom(id, Math.max(expected, 60) * EST_BYTES_PER_SEC);
const raw = await download(id, `${base}.dl.mp4`);
await setStatus(job, 'validating');
// Normalise: first video + first audio, moov atom up front so the copy
// starts instantly under Range requests. Stream copy — no quality loss.
const norm = `${base}.norm.mp4`;
const rm = await run(ffmpeg, ['-v', 'error', '-nostdin', '-y', '-i', raw,
'-map', '0:v:0', '-map', '0:a:0', '-c', 'copy', '-movflags', '+faststart', '-f', 'mp4', norm]);
if (rm.code !== 0) throw new Error('remux failed: ' + tail(rm.stderr));
const probe = await validateMedia(norm, expected, { ffmpeg, ffprobe });
const m4a = `${base}.norm.m4a`;
const ex = await run(ffmpeg, ['-v', 'error', '-nostdin', '-y', '-i', norm,
'-map', '0:a:0', '-c', 'copy', '-movflags', '+faststart', '-f', 'mp4', m4a]);
if (ex.code !== 0 || sizeOf(m4a) < 1024) throw new Error('audio sidecar failed: ' + tail(ex.stderr));
const prev = await db.getMedia(id);
const gen = ((prev && prev.gen) || 0) + 1;
renameSync(norm, fileFor(id, gen));
renameSync(m4a, fileFor(id, gen, 'm4a'));
const t = now();
const fields = {
status: 'ready', gen,
size: probe.size + sizeOf(fileFor(id, gen, 'm4a')),
height: probe.height, vcodec: probe.vcodec, acodec: probe.acodec, duration: probe.duration,
optimized: 0, meta: JSON.stringify(metaFromInfo(info, probe.duration)),
attempts: 0, error: null, retry_at: 0, updated_at: t, last_access: t,
};
await db.upsertMedia(id, fields);
removeFiles(id, gen);
job.status = 'ready';
log.info?.(`[media] cached ${id} ${probe.height}p ${fmtMB(fields.size)}`);
enqueueOptimize(id);
return { ...(prev || {}), video_id: id, ...fields };
} catch (err) {
job.status = 'failed';
if (err instanceof MediaSkip) {
await db.deleteMedia(id).catch(() => {});
log.info?.(`[media] skipped ${id}: ${err.message}`);
} else {
const prev = await db.getMedia(id).catch(() => null);
const attempts = ((prev && prev.attempts) || 0) + 1;
await db.upsertMedia(id, {
status: 'failed', attempts, error: String(err.message || err).slice(0, 500),
retry_at: now() + Math.min(DAY_MS, 15 * 60_000 * 2 ** (attempts - 1)),
updated_at: now(),
}).catch(() => {});
log.warn?.(`[media] ${id} failed (attempt ${attempts}): ${err.message}`);
}
throw err;
} finally {
sweepTmp(prefix);
}
}
// ---- Compression lane ----------------------------------------------------
// CRF (quality-targeted, not a fixed bitrate) at a slow preset is where
// small files come from — static lyric/slide videos collapse to a fraction
// of YouTube's bitrate. 60 fps is capped to 30. Audio is copied untouched.
// Output stays 8-bit H.264: 10-bit / HEVC / AV1 are smaller but do not
// play in iOS Safari's <video>.
function enqueueOptimize(id) {
if (!transcode || !transcode.enabled || optPending.has(id)) return;
optPending.add(id);
optQueue.push(id);
pumpOpt();
}
function pumpOpt() {
if (optActive || !optQueue.length) return;
const id = optQueue.shift();
optActive = id;
runOptimize(id)
.catch((err) => log.warn?.(`[media] optimize ${id} failed: ${err.message}`))
.finally(() => { optPending.delete(id); optActive = null; pumpOpt(); });
}
async function runOptimize(id) {
const row = await db.getMedia(id);
if (!row || row.status !== 'ready' || row.optimized) return;
const src = fileFor(id, row.gen);
if (!existsSync(src)) return;
const prefix = `${id}-${now()}-x264`;
const out = join(tmpDir, `${prefix}.mp4`);
const markDone = async () => {
const cur = await db.getMedia(id);
if (cur && cur.status === 'ready' && cur.gen === row.gen) await db.upsertMedia(id, { optimized: 1 });
};
try {
const args = ['-v', 'error', '-nostdin', '-y', '-i', src,
'-map', '0:v:0', '-map', '0:a:0',
'-vf', 'scale=-2:min(720\\,ih)',
'-c:v', 'libx264', '-preset', String(transcode.preset), '-crf', String(transcode.crf),
'-profile:v', 'high', '-level:v', '4.0', '-pix_fmt', 'yuv420p',
'-threads', String(transcode.threads), '-fpsmax', '30',
'-c:a', 'copy', '-movflags', '+faststart', '-f', 'mp4', out];
const r = nice ? await run(nice, ['-n', '19', ffmpeg, ...args]) : await run(ffmpeg, args);
if (r.code !== 0) throw new Error('x264 encode failed: ' + tail(r.stderr));
const probe = await validateMedia(out, row.duration, { ffmpeg, ffprobe });
const cur = await db.getMedia(id);
if (!cur || cur.status !== 'ready' || cur.gen !== row.gen) return; // replaced or evicted meanwhile
const oldSize = sizeOf(src);
if (probe.size > oldSize * (1 - MIN_SAVING)) {
await db.upsertMedia(id, { optimized: 1 });
log.info?.(`[media] kept original ${id}: x264 ${fmtMB(probe.size)} vs ${fmtMB(oldSize)}`);
return;
}
const gen = row.gen + 1;
renameSync(out, fileFor(id, gen));
const oldAudio = fileFor(id, row.gen, 'm4a');
const newAudio = fileFor(id, gen, 'm4a');
try { linkSync(oldAudio, newAudio); } catch { copyFileSync(oldAudio, newAudio); }
await db.upsertMedia(id, {
gen, size: probe.size + sizeOf(newAudio), height: probe.height,
duration: probe.duration, optimized: 1, updated_at: now(),
});
removeFiles(id, gen, OLD_GEN_GRACE_MS);
log.info?.(`[media] optimized ${id}: ${fmtMB(oldSize)} → ${fmtMB(probe.size)}`);
} catch (err) {
await markDone().catch(() => {}); // don't retry a source x264 can't improve
throw err;
} finally {
sweepTmp(prefix);
}
}
// ---- Public API ----------------------------------------------------------
// Resolve to the ready row, fetching it first if needed. Rejects with a
// MediaSkip (stream instead) or the download/validation error.
async function ensureCached(id, { priority = LOW, auto = false, force = false } = {}) {
if (!isMediaId(id)) throw new MediaSkip('invalid video id');
const existing = jobs.get(id);
if (existing) { bump(existing, priority, auto); return existing.promise; }
const row = await db.getMedia(id);
const again = jobs.get(id);
if (again) { bump(again, priority, auto); return again.promise; }
if (row && row.status === 'ready') {
if (readyFilesExist(row)) return row;
await db.deleteMedia(id);
removeFiles(id);
} else if (row && row.status === 'failed' && !force && now() < row.retry_at) {
throw new MediaSkip(`cache fetch failed recently (${row.error || 'unknown error'})`, 'BACKOFF');
}
if (jobs.get(id)) { bump(jobs.get(id), priority, auto); return jobs.get(id).promise; }
return startJob(id, priority, auto, row);
}
// The "Broken" button: drop the copy and fetch it again at high priority.
async function redownload(id) {
if (!isMediaId(id)) throw new MediaSkip('invalid video id');
const existing = jobs.get(id);
if (existing) { bump(existing, HIGH, false); return { status: existing.status }; }
const row = await db.getMedia(id);
if (jobs.get(id)) return { status: jobs.get(id).status };
removeFiles(id);
await db.upsertMedia(id, {
status: 'queued', gen: ((row && row.gen) || 0) + 1, size: 0, optimized: 0,
attempts: 0, error: null, retry_at: 0, updated_at: now(),
created_at: (row && row.created_at) || now(),
});
log.info?.(`[media] ${id} reported broken — re-downloading`);
startJob(id, HIGH, false, { ...(row || {}), gen: ((row && row.gen) || 0) + 1 });
return { status: 'queued' };
}
// Ready row whose files are on disk, else null. Hot path for /api/streams.
async function getReady(id) {
if (!isMediaId(id)) return null;
const row = await db.getMedia(id);
return row && row.status === 'ready' && readyFilesExist(row) ? row : null;
}
// Path for /api/media. With a gen, ONLY that generation is served (never
// different bytes under the same URL); without one, the current copy.
async function filePath(id, gen, kind = 'mp4') {
if (!isMediaId(id)) return null;
let g = gen !== undefined && gen !== null && gen !== '' ? Number(gen) : null;
if (g !== null && !Number.isInteger(g)) return null;
if (g === null) {
const row = await db.getMedia(id);
if (!row || row.status !== 'ready') return null;
g = row.gen;
}
const p = fileFor(id, g, kind);
return existsSync(p) ? p : null;
}
// A client fell back from our copy (?nocache=1). Its decode failure may be
// device-specific, so re-run the server-side gate instead of trusting it:
// a copy that fails is dropped and refetched; one that passes stays.
const lastVerify = new Map();
async function verify(id) {
if (!isMediaId(id)) return null;
const t = now();
if (t - (lastVerify.get(id) || 0) < 10 * 60_000) return null;
lastVerify.set(id, t);
if (lastVerify.size > 5000) lastVerify.clear();
const row = await getReady(id);
if (!row || jobs.has(id)) return null;
try {
await validateMedia(fileFor(id, row.gen), row.duration, { ffmpeg, ffprobe });
return true;
} catch (err) {
const cur = await db.getMedia(id);
if (!cur || cur.gen !== row.gen) return null; // replaced meanwhile
log.warn?.(`[media] ${id} copy failed re-validation (${err.message}) — re-downloading`);
await redownload(id);
return false;
}
}
function touch(id) {
const t = now();
if (t - (lastTouch.get(id) || 0) < TOUCH_DEBOUNCE_MS) return;
lastTouch.set(id, t);
if (lastTouch.size > 5000) lastTouch.clear();
db.touchMedia(id, t).catch(() => {});
}
async function status(id) {
if (!isMediaId(id)) return { status: 'none' };
const job = jobs.get(id);
const row = await db.getMedia(id);
return {
status: job ? job.status : row ? row.status : 'none',
size: row ? row.size : 0,
height: row ? row.height : 0,
optimized: !!(row && row.optimized),
optimizing: optActive === id,
error: row && row.status === 'failed' ? row.error : null,
retryAt: row && row.status === 'failed' ? row.retry_at : 0,
};
}
async function stats() {
return {
...(await db.mediaStats()),
maxBytes,
freeBytes: safeFree(),
queued: fetchQueue.length,
active: fetchActive ? fetchActive.id : null,
optimizing: optActive,
optimizeQueued: optQueue.length,
};
}
// Boot: sweep partial work, drop files/rows that disagree, resume jobs.
async function init() {
mkdirSync(tmpDir, { recursive: true });
sweepTmp();
const rows = await db.listMedia();
const byId = new Map(rows.map((r) => [r.video_id, r]));
for (const name of readdirSync(dir)) {
if (name === '.tmp') continue;
const m = /^(.+)\.(\d+)\.(mp4|m4a)$/.exec(name);
const r = m && byId.get(m[1]);
if (!m || !r || r.status !== 'ready' || r.gen !== Number(m[2])) {
try { unlinkSync(join(dir, name)); } catch { /* gone */ }
}
}
let resumed = 0;
for (const r of rows) {
if (r.status === 'ready') {
if (!readyFilesExist(r)) { await db.deleteMedia(r.video_id); removeFiles(r.video_id); continue; }
if (!r.optimized) enqueueOptimize(r.video_id);
} else if (r.status === 'queued' || r.status === 'downloading' || r.status === 'validating') {
startJob(r.video_id, r.priority === HIGH ? HIGH : LOW, !!r.auto, r);
resumed++;
}
}
const s = await db.mediaStats();
log.info?.(`[media] ${s.count} cached (${fmtMB(s.bytes)}), ${resumed} job(s) resumed, dir ${dir}`);
}
return { init, ensureCached, redownload, verify, getReady, filePath, touch, status, stats, isMediaId };
}

249
server/media-cache.test.js Normal file
View File

@@ -0,0 +1,249 @@
// bun test media-cache.test.js — validation gate, job queue, eviction, redownload.
// Uses real ffmpeg fixtures and the real db.js against a throwaway DB file.
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { spawnSync } from 'node:child_process';
import { mkdtempSync, rmSync, copyFileSync, writeFileSync, readFileSync, existsSync, readdirSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
const root = mkdtempSync(join(tmpdir(), 'ytp-media-test-'));
process.env.DB_PATH = join(root, 'test.db');
const dbmod = await import('./db.js');
const { createMediaCache, validateMedia, MediaSkip, HIGH, LOW } = await import('./media-cache.js');
const fx = (name) => join(root, name);
const ff = (...args) => {
const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args]);
if (r.status !== 0) throw new Error(String(r.stderr));
};
const silentLog = { info() {}, warn() {} };
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
async function until(fn, ms = 20000) {
const end = Date.now() + ms;
while (Date.now() < end) { if (await fn()) return true; await sleep(50); }
return false;
}
beforeAll(async () => {
await dbmod.initDb();
// 10 s 640x360 H.264 + AAC — the shape of a good cached copy.
ff('-f', 'lavfi', '-i', 'testsrc=size=640x360:rate=30:duration=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=10',
'-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', fx('good.mp4'));
ff('-f', 'lavfi', '-i', 'testsrc=size=640x360:rate=30:duration=10',
'-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', fx('noaudio.mp4'));
ff('-f', 'lavfi', '-i', 'testsrc=size=640x360:rate=30:duration=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=10',
'-c:v', 'mpeg4', '-c:a', 'aac', '-shortest', fx('mpeg4.mp4'));
const buf = readFileSync(fx('good.mp4'));
writeFileSync(fx('truncated.mp4'), buf.subarray(0, Math.floor(buf.length * 0.6)));
// The nasty case: moov up front (index says 10 s) but the media data is cut.
ff('-i', fx('good.mp4'), '-c', 'copy', '-movflags', '+faststart', fx('good-fs.mp4'));
const fs = readFileSync(fx('good-fs.mp4'));
writeFileSync(fx('truncated-fs.mp4'), fs.subarray(0, Math.floor(fs.length * 0.6)));
});
afterAll(() => rmSync(root, { recursive: true, force: true }));
describe('validateMedia', () => {
test('accepts a good H.264 + AAC file', async () => {
const p = await validateMedia(fx('good.mp4'), 10);
expect(p.vcodec).toBe('h264');
expect(p.acodec).toBe('aac');
expect(p.height).toBe(360);
});
test('rejects a truncated file', async () => {
await expect(validateMedia(fx('truncated.mp4'), 10)).rejects.toThrow(/validation/);
});
test('rejects a faststart file whose media data is cut short', async () => {
await expect(validateMedia(fx('truncated-fs.mp4'), 10)).rejects.toThrow(/validation/);
});
test('rejects a file without audio', async () => {
await expect(validateMedia(fx('noaudio.mp4'), 10)).rejects.toThrow(/AAC/);
});
test('rejects a non-h264 file', async () => {
await expect(validateMedia(fx('mpeg4.mp4'), 10)).rejects.toThrow(/not h264/);
});
test('rejects a file shorter than the source video', async () => {
await expect(validateMedia(fx('good.mp4'), 60)).rejects.toThrow(/truncated/);
});
});
let dirN = 0;
function makeCache({ fixture = 'good.mp4', duration = 10, ...opts } = {}) {
const dir = join(root, 'media' + dirN++);
const calls = [];
const db = {
getMedia: dbmod.getMedia, upsertMedia: dbmod.upsertMedia, deleteMedia: dbmod.deleteMedia,
listMedia: dbmod.listMedia, listMediaLru: dbmod.listMediaLru, touchMedia: dbmod.touchMedia,
mediaStats: dbmod.mediaStats,
};
const state = { fixture };
const cache = createMediaCache({
dir, db, log: silentLog,
getInfo: async (id) => ({ id, title: 'T ' + id, channel: 'C', duration }),
download: async (id, out) => { calls.push(id); await sleep(30); copyFileSync(fx(state.fixture), out); return out; },
transcode: { enabled: false },
freeBytes: () => 100 * 1024 ** 3,
...opts,
});
return { cache, dir, calls, state };
}
async function clearDb() {
for (const r of await dbmod.listMedia()) await dbmod.deleteMedia(r.video_id);
}
describe('media cache jobs', () => {
test('fetches once, dedupes concurrent requests, then serves from disk', async () => {
await clearDb();
const { cache, dir, calls } = makeCache();
await cache.init();
const [a, b] = await Promise.all([
cache.ensureCached('vidAAAAAAA1', { priority: LOW, auto: true }),
cache.ensureCached('vidAAAAAAA1', { priority: HIGH }),
]);
expect(calls).toEqual(['vidAAAAAAA1']);
expect(a.status).toBe('ready');
expect(b.gen).toBe(a.gen);
expect(existsSync(join(dir, `vidAAAAAAA1.${a.gen}.mp4`))).toBe(true);
expect(existsSync(join(dir, `vidAAAAAAA1.${a.gen}.m4a`))).toBe(true);
expect(await cache.getReady('vidAAAAAAA1')).not.toBeNull();
await cache.ensureCached('vidAAAAAAA1');
expect(calls.length).toBe(1);
expect(JSON.parse((await cache.getReady('vidAAAAAAA1')).meta).title).toBe('T vidAAAAAAA1');
});
test('HIGH request jumps ahead of queued LOW jobs', async () => {
await clearDb();
const { cache, calls } = makeCache();
await cache.init();
const p = [
cache.ensureCached('lowAAAAAAA1', { auto: true }),
cache.ensureCached('lowAAAAAAA2', { auto: true }),
cache.ensureCached('lowAAAAAAA3', { auto: true }),
cache.ensureCached('highAAAAAA1', { priority: HIGH }),
];
await Promise.all(p);
expect(calls[0]).toBe('lowAAAAAAA1'); // already running when HIGH arrived
expect(calls[1]).toBe('highAAAAAA1');
});
test('a broken download is never cached and backs off', async () => {
await clearDb();
const { cache, dir, calls } = makeCache({ fixture: 'truncated.mp4' });
await cache.init();
await expect(cache.ensureCached('brokenAAAA1', { priority: HIGH })).rejects.toThrow(/validation|remux/);
const row = await dbmod.getMedia('brokenAAAA1');
expect(row.status).toBe('failed');
expect(row.retry_at).toBeGreaterThan(Date.now());
expect(readdirSync(dir).filter((n) => n !== '.tmp')).toEqual([]);
expect(readdirSync(join(dir, '.tmp'))).toEqual([]);
await expect(cache.ensureCached('brokenAAAA1', { priority: HIGH })).rejects.toMatchObject({ code: 'BACKOFF' });
expect(calls.length).toBe(1);
});
test('a faststart download with cut media data is rejected too', async () => {
await clearDb();
const { cache, dir } = makeCache({ fixture: 'truncated-fs.mp4' });
await cache.init();
await expect(cache.ensureCached('brokenAAAA2', { priority: HIGH })).rejects.toThrow();
expect((await dbmod.getMedia('brokenAAAA2')).status).toBe('failed');
expect(readdirSync(dir).filter((n) => n !== '.tmp')).toEqual([]);
});
test('redownload drops the copy and fetches a fresh one', async () => {
await clearDb();
const { cache, dir, calls } = makeCache();
await cache.init();
const first = await cache.ensureCached('redoAAAAAA1', { priority: HIGH });
const res = await cache.redownload('redoAAAAAA1');
expect(res.status).toBe('queued');
expect(existsSync(join(dir, `redoAAAAAA1.${first.gen}.mp4`))).toBe(false);
expect(await cache.getReady('redoAAAAAA1')).toBeNull();
expect(await until(async () => !!(await cache.getReady('redoAAAAAA1')))).toBe(true);
const second = await cache.getReady('redoAAAAAA1');
expect(second.gen).toBeGreaterThan(first.gen);
expect(calls.length).toBe(2);
});
test('auto jobs respect the auto length limit; explicit saves do not', async () => {
await clearDb();
const { cache } = makeCache({ duration: 7200, autoMaxSeconds: 3600 });
await cache.init();
await expect(cache.ensureCached('longAAAAAA1', { auto: true })).rejects.toMatchObject({ code: 'SKIPPED' });
expect(await dbmod.getMedia('longAAAAAA1')).toBeNull();
});
test('evicts least-recently-played copies to stay under the budget', async () => {
await clearDb();
const size = readFileSync(fx('good.mp4')).length;
// Budget fits ~2 copies (+ sidecars) but not 3 once the new-download estimate is added.
const { cache } = makeCache({ maxBytes: size * 2 + 60 * 250 * 1024 + 50_000, duration: 10 });
await cache.init();
await cache.ensureCached('lruAAAAAAA1', { priority: HIGH });
await cache.ensureCached('lruAAAAAAA2', { priority: HIGH });
// Age both beyond the eviction protection window; #1 is the older one.
await dbmod.upsertMedia('lruAAAAAAA1', { last_access: 1000 });
await dbmod.upsertMedia('lruAAAAAAA2', { last_access: 2000 });
await cache.ensureCached('lruAAAAAAA3', { priority: HIGH });
expect(await dbmod.getMedia('lruAAAAAAA1')).toBeNull();
expect(await cache.getReady('lruAAAAAAA3')).not.toBeNull();
});
test('free-disk guard skips caching', async () => {
await clearDb();
const { cache, calls } = makeCache({ freeBytes: () => 1024 ** 3, minFreeBytes: 5 * 1024 ** 3 });
await cache.init();
await expect(cache.ensureCached('diskAAAAAA1', { priority: HIGH })).rejects.toMatchObject({ code: 'SKIPPED' });
expect(calls.length).toBe(0);
});
test('boot resumes a job left queued and sweeps stray files', async () => {
await clearDb();
const { cache, dir, calls } = makeCache();
await dbmod.upsertMedia('bootAAAAAA1', { status: 'downloading', priority: HIGH });
writeFileSync(join(dir + '_never'), '');
const { mkdirSync } = await import('node:fs');
mkdirSync(join(dir, '.tmp'), { recursive: true });
writeFileSync(join(dir, '.tmp', 'partial.mp4.part'), 'x');
writeFileSync(join(dir, 'orphanAAAA1.3.mp4'), 'x');
await cache.init();
expect(existsSync(join(dir, 'orphanAAAA1.3.mp4'))).toBe(false);
expect(await until(async () => !!(await cache.getReady('bootAAAAAA1')))).toBe(true);
expect(calls).toEqual(['bootAAAAAA1']);
});
test('x264 compression lane swaps in a smaller copy under a new gen', async () => {
await clearDb();
const { cache } = makeCache({ transcode: { enabled: true, crf: 35, preset: 'ultrafast', threads: 2 } });
await cache.init();
const first = await cache.ensureCached('optAAAAAAA1', { priority: HIGH });
expect(await until(async () => (await dbmod.getMedia('optAAAAAAA1')).optimized === 1)).toBe(true);
const after = await dbmod.getMedia('optAAAAAAA1');
if (after.gen !== first.gen) {
expect(after.size).toBeLessThan(first.size);
// old generation is kept for in-flight playback (grace period)
expect(await cache.filePath('optAAAAAAA1', first.gen)).not.toBeNull();
}
expect(await cache.filePath('optAAAAAAA1', after.gen, 'm4a')).not.toBeNull();
});
test('verify() drops and refetches a copy that went bad on disk, keeps a good one', async () => {
await clearDb();
let clock = Date.now();
const { cache, dir, calls } = makeCache({ now: () => clock });
await cache.init();
const row = await cache.ensureCached('verAAAAAAA1', { priority: HIGH });
expect(await cache.verify('verAAAAAAA1')).toBe(true);
writeFileSync(join(dir, `verAAAAAAA1.${row.gen}.mp4`), Buffer.alloc(300_000, 7));
expect(await cache.verify('verAAAAAAA1')).toBeNull(); // debounced for 10 min
clock += 11 * 60_000;
expect(await cache.verify('verAAAAAAA1')).toBe(false);
expect(await until(async () => !!(await cache.getReady('verAAAAAAA1')))).toBe(true);
expect((await cache.getReady('verAAAAAAA1')).gen).toBeGreaterThan(row.gen);
expect(calls.length).toBe(2);
});
test('rejects path-like ids', async () => {
const { cache } = makeCache();
await expect(cache.ensureCached('../../etc')).rejects.toBeInstanceOf(MediaSkip);
expect(await cache.filePath('..', 1)).toBeNull();
});
});

230
server/server.js
View File

@@ -10,7 +10,11 @@
* GET /api/channel?c=<channel> yt-dlp channel uploads → slim card array
* GET /api/streams?v=<videoId> yt-dlp stream info → {meta, audioUrl, qualities} (proxied URLs)
* GET /api/play?v=<id>&f=<fmt> same-origin playback proxy (Range-aware) → media bytes
* GET /api/download/:videoId proxy best progressive stream → binary
* GET /api/download/:videoId server-cached copy (fetched by a background job) → binary
* GET /api/media/:id?g=<gen>[&a=1] Range-aware server-cached mp4 (or m4a audio sidecar)
* GET /api/media/:id/status { status, size, height, optimized, error }
* POST /api/media/:id/redownload "Broken" button: drop the copy, fetch it again
* GET /api/media/stats cache totals, budget, free disk, queue
* GET /api/version { version }
* POST /api/user/sync upsert user playlists + last-seen version
* GET /api/user/data?fp=<fp> retrieve stored playlists + history
@@ -28,11 +32,13 @@ import { serveStatic } from 'hono/bun';
import { logger } from 'hono/logger';
import { spawn } from 'node:child_process';
import { createServer } from 'node:http';
import { readFileSync, readdirSync, existsSync, statSync, openSync, unlinkSync, createReadStream } from 'node:fs';
import { readFileSync, readdirSync, existsSync, statSync, openSync, unlinkSync, createReadStream, mkdirSync } from 'node:fs';
import { Readable } from 'node:stream';
import { tmpdir } from 'node:os';
import { createHash } from 'node:crypto';
import { initDb, upsertUser, recordVideoAccess, getUserData, createProfile, getProfile, saveProfile, createSharedPlaylist, getSharedPlaylist, queueInboxPlaylist, listInbox, deleteInboxItem, countInbox } from './db.js';
import { initDb, upsertUser, recordVideoAccess, getUserData, createProfile, getProfile, saveProfile, createSharedPlaylist, getSharedPlaylist, queueInboxPlaylist, listInbox, deleteInboxItem, countInbox,
getMedia, upsertMedia, deleteMedia, listMedia, listMediaLru, touchMedia, mediaStats } from './db.js';
import { createMediaCache, HIGH, LOW } from './media-cache.js';
// A media proxy must not die because one client's stream hit an edge case
// (see /api/play cancel()): log and keep serving instead of crash-looping.
@@ -494,6 +500,24 @@ app.get('/api/streams', async (c) => {
const videoId = (c.req.query('v') || '').replace(/[/\\:?<>|*"]/g, '').trim();
if (!videoId) return c.json({ ok: false, error: 'missing videoId' }, 400);
// Server already holds a validated copy → answer from the DB alone, no
// yt-dlp round trip. ?nocache=1 (the client's fallback when a cached copy
// won't play on its device) forces the YouTube path below.
if (c.req.query('nocache') !== '1') {
try {
const row = await media.getReady(videoId);
if (row) {
media.touch(videoId);
return c.json({ ok: true, data: cachedStreamsPayload(videoId, row) });
}
} catch (err) {
console.warn(`[media] cache lookup failed for ${videoId}:`, err.message);
}
} else {
// A client couldn't play our copy — re-check it server-side (background).
media.verify(videoId).catch(() => {});
}
try {
const { info, formats } = await resolveStreams(videoId);
@@ -536,6 +560,10 @@ app.get('/api/streams', async (c) => {
return c.json({ ok: false, error: live ? 'Live streams are not supported' : 'No playable formats for this video' }, 422);
}
// Played but not cached yet: fetch a copy in the background so the next
// play is served from disk. Runs server-side; the client never waits.
media.ensureCached(videoId, { priority: LOW, auto: true }).catch(() => {});
return c.json({
ok: true,
data: {
@@ -852,15 +880,16 @@ async function ytdlpDownloadResponse(videoId, fp, formatArgs, signal) {
// concatenate them into one continuous mp4 — the user's custom cut. The result
// is streamed to the browser exactly like a normal save, so OPFS stores it
// under the caller-chosen custom id. Every temp file is swept afterwards.
async function ytdlpEditedDownloadResponse(videoId, fp, keep, signal) {
await assertNotLive(videoId);
async function ytdlpEditedDownloadResponse(videoId, fp, keep, signal, cachedSrc = null) {
if (!cachedSrc) await assertNotLive(videoId);
const tmpBase = `ytp-edit-${videoId}-${Date.now()}`;
const srcTmp = `${tmpdir()}/${tmpBase}.src.mp4`;
const srcTmp = cachedSrc || `${tmpdir()}/${tmpBase}.src.mp4`;
const outTmp = `${tmpdir()}/${tmpBase}.out.mp4`;
let size, fd;
try {
// 1) Grab the full source (video+audio merged) so ffmpeg has both streams.
await withSaveSlot(() => runYtdlpResilient([
// 1) Grab the full source (video+audio merged) so ffmpeg has both streams
// — unless the server cache already holds it.
if (!cachedSrc) await withSaveSlot(() => runYtdlpResilient([
`https://www.youtube.com/watch?v=${videoId}`,
'--no-warnings', '--no-playlist',
'-f', SAVE_MUX_FORMAT,
@@ -900,10 +929,168 @@ async function ytdlpEditedDownloadResponse(videoId, fp, keep, signal) {
});
}
// ============================================================================
// Server-side media cache (see media-cache.js). One validated ≤720p H.264 +
// AAC copy per played/saved video under $MEDIA_DIR, fetched by server-owned
// jobs that survive the client going away. Budget-bounded LRU; never expires
// by time. Downloads still go through withSaveSlot so they stay serialized
// with any legacy fallback save (the homelab downlink is the constraint).
// ============================================================================
const envNum = (k, d) => (process.env[k] !== undefined && process.env[k] !== '' ? Number(process.env[k]) : d);
const MEDIA_DIR = process.env.MEDIA_DIR || './data/media';
mkdirSync(MEDIA_DIR, { recursive: true });
const media = createMediaCache({
dir: MEDIA_DIR,
db: { getMedia, upsertMedia, deleteMedia, listMedia, listMediaLru, touchMedia, mediaStats },
getInfo: async (videoId) => (await resolveStreams(videoId)).info,
download: async (videoId, out) => {
await withSaveSlot(() => runYtdlpResilient([
`https://www.youtube.com/watch?v=${videoId}`,
'--no-warnings', '--no-playlist',
'-f', SAVE_MUX_FORMAT,
'--merge-output-format', 'mp4',
'--limit-rate', DOWNLOAD_RATE,
'-o', out,
]));
return out;
},
ffmpeg: FFMPEG,
ffprobe: process.env.FFPROBE_PATH || 'ffprobe',
maxBytes: envNum('MEDIA_CACHE_MAX_BYTES', 10 * 1024 ** 3),
minFreeBytes: envNum('MEDIA_MIN_FREE_BYTES', 5 * 1024 ** 3),
autoMaxSeconds: envNum('MEDIA_AUTO_MAX_SECONDS', 3600),
saveMaxSeconds: MAX_SAVE_SECONDS,
transcode: {
enabled: process.env.MEDIA_TRANSCODE !== '0',
crf: envNum('MEDIA_CRF', 28),
preset: process.env.MEDIA_PRESET || 'slow',
threads: envNum('MEDIA_THREADS', 2),
},
});
// /api/streams response for a server-cached video: one progressive quality
// (the mp4 carries its own audio) plus the m4a sidecar for audio-only mode.
// Same shape as the YouTube path; `serverCached` is additive.
function cachedStreamsPayload(videoId, row) {
let meta = {};
try { meta = JSON.parse(row.meta || '{}'); } catch { /* corrupt — use defaults */ }
const url = `/api/media/${videoId}?g=${row.gen}`;
return {
meta: {
id: videoId,
title: meta.title || '(untitled)',
channel: meta.channel || '',
channelId: meta.channelId || '',
channelUrl: meta.channelUrl || '',
duration: meta.duration || Math.round(row.duration || 0),
thumbnail: `https://i.ytimg.com/vi/${videoId}/hqdefault.jpg`,
},
audioUrl: `${url}&a=1`,
qualities: [{
label: (row.height || 720) + 'p',
height: row.height || 720,
hasAudio: true,
url,
ext: 'mp4',
}],
serverCached: true,
};
}
// Serve a file with byte-Range support (the <video> element seeks with it).
function rangeFileResponse(c, path, contentType, cacheControl) {
const file = Bun.file(path);
const total = file.size;
const headers = {
'Content-Type': contentType,
'Accept-Ranges': 'bytes',
'Cache-Control': cacheControl,
};
const range = c.req.header('range');
if (!range) return new Response(file, { status: 200, headers });
const m = /^bytes=(\d*)-(\d*)$/.exec(range.trim());
let start, end;
if (m && m[1] !== '') {
start = Number(m[1]);
end = m[2] !== '' ? Math.min(Number(m[2]), total - 1) : total - 1;
} else if (m && m[2] !== '') {
start = Math.max(0, total - Number(m[2]));
end = total - 1;
}
if (start === undefined || start > end || start >= total) {
return new Response(null, { status: 416, headers: { ...headers, 'Content-Range': `bytes */${total}` } });
}
return new Response(file.slice(start, end + 1), {
status: 206,
headers: { ...headers, 'Content-Range': `bytes ${start}-${end}/${total}` },
});
}
// GET /api/media/stats — registered before /api/media/:id so it isn't an id.
app.get('/api/media/stats', async (c) => {
try {
return c.json({ ok: true, ...(await media.stats()) }, 200, { 'Cache-Control': 'no-store' });
} catch (err) {
return c.json({ ok: false, error: err.message }, 500);
}
});
// GET /api/media/:id?g=<gen>[&a=1] — the cached mp4 (or its m4a sidecar).
// A URL with ?g= is immutable: a replaced copy gets a new gen, and an old gen
// is served only while its file still exists (never different bytes).
app.get('/api/media/:id', async (c) => {
const id = c.req.param('id');
const g = c.req.query('g');
const audio = c.req.query('a') === '1';
const path = await media.filePath(id, g, audio ? 'm4a' : 'mp4');
if (!path) return c.json({ ok: false, error: 'not cached' }, 404);
media.touch(id);
return rangeFileResponse(c, path, audio ? 'audio/mp4' : 'video/mp4',
g ? 'public, max-age=31536000, immutable' : 'no-cache');
});
// GET /api/media/:id/status
app.get('/api/media/:id/status', async (c) => {
try {
return c.json({ ok: true, ...(await media.status(c.req.param('id'))) }, 200, { 'Cache-Control': 'no-store' });
} catch (err) {
return c.json({ ok: false, error: err.message }, 500);
}
});
// POST /api/media/:id/redownload — the client's "Broken" button. Idempotent:
// a job already queued/running for this id is just reported back.
app.post('/api/media/:id/redownload', async (c) => {
try {
return c.json({ ok: true, ...(await media.redownload(c.req.param('id'))) });
} catch (err) {
return c.json({ ok: false, error: err.message }, 400);
}
});
// Stream the server-cached copy as a save (OPFS writes it on the device).
function cachedDownloadResponse(videoId, fp, row) {
const file = Bun.file(`${MEDIA_DIR}/${videoId}.${row.gen}.mp4`);
if (fp) recordVideoAccess(fp, { id: videoId }).catch(() => {});
media.touch(videoId);
return new Response(file, {
status: 200,
headers: {
'Content-Type': 'video/mp4',
'Content-Length': String(file.size),
'Content-Disposition': `attachment; filename="${videoId}.mp4"`,
'Cache-Control': 'no-store',
'Access-Control-Allow-Origin': '*',
},
});
}
// GET /api/download/:videoId
// Downloads the video server-side via yt-dlp and streams the finished file
// to the browser so OPFS can store it. The browser never contacts YouTube
// CDN directly (CORS would block it).
// Saves go through the server cache: the fetch runs as a server-owned job
// (a disconnecting client no longer kills it — the copy lands for next
// time), then the validated file is streamed so OPFS can store it. The
// browser never contacts YouTube CDN directly (CORS would block it).
app.get('/api/download/:videoId', async (c) => {
const videoId = (c.req.param('videoId') || '').replace(/[/\\:?<>|*"]/g, '').trim();
if (!videoId) return c.json({ ok: false, error: 'missing videoId' }, 400);
@@ -917,17 +1104,33 @@ app.get('/api/download/:videoId', async (c) => {
const keep = parseKeepParam(c.req.query('keep') || '');
if (!keep.length) return c.json({ ok: false, error: 'missing or invalid keep segments' }, 400);
try {
return await ytdlpEditedDownloadResponse(videoId, fp, keep, c.req.raw.signal);
const row = await media.ensureCached(videoId, { priority: HIGH }).catch(() => null);
const src = row ? await media.filePath(videoId, row.gen) : null;
return await ytdlpEditedDownloadResponse(videoId, fp, keep, c.req.raw.signal, src);
} catch (err) {
return c.json({ ok: false, error: err.message }, 500);
}
}
// Server cache first — plain saves and ?mux=1 both resolve to the same
// canonical ≤720p H.264 + AAC copy.
let cacheErr = null;
try {
const row = await media.ensureCached(videoId, { priority: HIGH });
return cachedDownloadResponse(videoId, fp, row);
} catch (err) {
cacheErr = err;
console.warn(`[ytplayer] cache save unavailable for ${videoId} (${err.code || 'error'}): ${err.message}`);
}
// Legacy per-request saves — only when the cache can't hold this video.
// A cache job that downloaded and FAILED validation already tried the mux
// format, so go straight to the progressive-first default in that case.
// ?mux=1 — "Save before playing" path: bestvideo up to 720p PLUS bestaudio
// compiled into one mp4 with ffmpeg on the server. Falls back to the
// progressive single-file save below when ffmpeg is missing or the merge
// fails.
if (c.req.query('mux') === '1') {
if (c.req.query('mux') === '1' && cacheErr && cacheErr.code === 'SKIPPED') {
try {
return await ytdlpDownloadResponse(videoId, fp, [
'-f', SAVE_MUX_FORMAT,
@@ -1388,6 +1591,7 @@ app.get('/*', serveStatic({ path: './public/index.html' }));
async function main() {
await initDb();
console.log(`[ytplayer] DB ready`);
await media.init();
console.log(`[ytplayer] Starting on port ${PORT}`);
// Bun.serve is the native Bun HTTP server