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

Add upload library management and ops diagnostics

Browse Source
This commit is contained in:
Jonathan Sykes
2026-10-14 03:09:12 +08:00
parent 5177a4d441
commit 27c694901d
15 changed files with 467 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
server/db.js
View File

@@ -234,7 +234,9 @@ export async function initDb() {
size INTEGER NOT NULL DEFAULT 0,
art TEXT, -- 'embedded' | 'file' | NULL
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
plays INTEGER NOT NULL DEFAULT 0
plays INTEGER NOT NULL DEFAULT 0,
gen INTEGER NOT NULL DEFAULT 1,
art_gen INTEGER NOT NULL DEFAULT 1
);
-- "This lyric line is wrong" reports from service mode. One open row per
@@ -294,6 +296,9 @@ export async function initDb() {
for (const col of ['owner TEXT', 'listed INTEGER NOT NULL DEFAULT 1']) {
try { await db.execute(`ALTER TABLE uploads ADD COLUMN ${col}`); } catch { /* already there */ }
}
try { await db.execute('ALTER TABLE uploads ADD COLUMN gen INTEGER NOT NULL DEFAULT 1'); } catch { /* already migrated */ }
try { await db.execute('ALTER TABLE uploads ADD COLUMN art_gen INTEGER NOT NULL DEFAULT 1'); } catch { /* already migrated */ }
await db.execute(`CREATE TABLE IF NOT EXISTS upload_owner_quotas (owner TEXT PRIMARY KEY,max_bytes INTEGER NOT NULL CHECK(max_bytes>=0),updated_at INTEGER NOT NULL)`);
// Integrated loudness (EBU R128 LUFS) measured by devices in the browser and
// shared, so each song is measured once for everyone.
await db.execute(`CREATE TABLE IF NOT EXISTS loudness (
@@ -956,12 +961,13 @@ const uploadRow = (r) => ({
duration: Number(r.duration) || 0, ext: r.ext, mime: r.mime, size: Number(r.size) || 0,
art: r.art || null, createdAt: Number(r.created_at), plays: Number(r.plays) || 0,
owner: r.owner || null, listed: r.listed === undefined || r.listed === null ? true : Number(r.listed) === 1,
gen: Number(r.gen) || 1, artGen: Number(r.art_gen) || 1,
});
export async function createUpload(u) {
await db.execute({
sql: `INSERT INTO uploads (id, kind, title, artist, album, duration, ext, mime, size, art, owner, listed, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, unixepoch())`,
sql: `INSERT INTO uploads (id, kind, title, artist, album, duration, ext, mime, size, art, owner, listed, gen, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, unixepoch())`,
args: [u.id, u.kind, u.title, u.artist || null, u.album || null, u.duration || 0, u.ext, u.mime, u.size || 0, u.art || null,
u.owner || null, u.listed === false ? 0 : 1],
});
@@ -980,7 +986,7 @@ export async function sharedUploadUsage(owner) {
}
export async function setUploadArt(id, art) {
await db.execute({ sql: 'UPDATE uploads SET art = ? WHERE id = ?', args: [art, id] });
await db.execute({ sql: 'UPDATE uploads SET art = ?, art_gen = art_gen + 1 WHERE id = ?', args: [art, id] });
}
export async function getUpload(id) {
@@ -988,6 +994,45 @@ export async function getUpload(id) {
return r.rows[0] ? uploadRow(r.rows[0]) : null;
}
export async function updateUploadMeta(id, patch = {}) {
const fields = ['title', 'artist', 'album', 'listed'].filter(k => patch[k] !== undefined);
if (!fields.length) return !!(await getUpload(id));
const values = fields.map(k => k === 'listed' ? (patch[k] ? 1 : 0) : String(patch[k] ?? '').slice(0, k === 'title' ? 300 : 200) || null);
const result = await db.execute({ sql: `UPDATE uploads SET ${fields.map(k => `${k}=?`).join(',')} WHERE id=?`, args: [...values, id] });
return !!result.rowsAffected;
}
export async function replaceUploadFile(id, upload) {
const result = await db.execute({ sql: `UPDATE uploads SET kind=?,title=?,artist=?,album=?,duration=?,ext=?,mime=?,size=?,art=?,gen=gen+1 WHERE id=?`,
args: [upload.kind, upload.title, upload.artist || null, upload.album || null, upload.duration || 0, upload.ext, upload.mime, upload.size || 0, upload.art || null, id] });
return !!result.rowsAffected;
}
export async function listAdminUploads({ q = '', limit = 100, offset = 0 } = {}) {
const args = [], where = q ? "WHERE lower(title || ' ' || ifnull(artist,'') || ' ' || ifnull(album,'')) LIKE ?" : '';
if (q) args.push(`%${String(q).toLowerCase().replace(/[\\%_]/g, '\\$&')}%`);
const rows = await db.execute({ sql: `SELECT * FROM uploads ${where} ORDER BY created_at DESC LIMIT ? OFFSET ?`, args: [...args, Math.max(1, Math.min(200, limit)), Math.max(0, offset)] });
const count = await db.execute({ sql: `SELECT COUNT(*) AS n FROM uploads ${where}`, args });
const owners = await db.execute("SELECT ifnull(owner,'(admin)') AS owner,COUNT(*) AS count,COALESCE(SUM(size),0) AS bytes FROM uploads GROUP BY owner ORDER BY bytes DESC");
return { uploads: rows.rows.map(uploadRow), total: Number(count.rows[0]?.n) || 0, owners: owners.rows.map(r => ({ owner: r.owner, count: Number(r.count), bytes: Number(r.bytes) })) };
}
export async function listUploadFileNames() {
const r = await db.execute('SELECT id,ext FROM uploads');
return r.rows.flatMap(row => [`${row.id}.${row.ext}`, `${row.id}.art.jpg`]);
}
export async function listUploadQuotas() {
const r = await db.execute('SELECT owner,max_bytes,updated_at FROM upload_owner_quotas ORDER BY owner');
return r.rows.map(x => ({ owner: x.owner, maxBytes: Number(x.max_bytes), updatedAt: Number(x.updated_at) }));
}
export async function getUploadQuota(owner) {
const r = await db.execute({ sql: 'SELECT max_bytes FROM upload_owner_quotas WHERE owner=?', args: [owner] });
return r.rows[0] ? Number(r.rows[0].max_bytes) : null;
}
export async function setUploadQuota(owner, maxBytes) {
if (maxBytes == null || Number(maxBytes) <= 0) return db.execute({ sql: 'DELETE FROM upload_owner_quotas WHERE owner=?', args: [owner] });
return db.execute({ sql: 'INSERT INTO upload_owner_quotas(owner,max_bytes,updated_at) VALUES(?,?,unixepoch()) ON CONFLICT(owner) DO UPDATE SET max_bytes=excluded.max_bytes,updated_at=excluded.updated_at', args: [owner, Math.floor(Number(maxBytes))] });
}
// Newest first; `q` matches title/artist/album (case-insensitive).
// `listedOnly`: public search shows admin uploads, never device-shared ones.
export async function listUploads({ q = '', limit = 100, listedOnly = false } = {}) {

5
server/p2p-admin.js
View File

@@ -3,8 +3,9 @@ import { audit } from './audit-log.js';
export function registerP2pAdminRoutes(app, { db, adminAuth, p2pDb, hub, stats = async () => ({}) }) {
app.get('/api/admin/p2p/devices', adminAuth, async c => {
const offset = Math.max(0, Number(c.req.query('offset')) || 0);
const devices = await p2pDb.listAdminDevices({ limit: 100, offset });
return c.json({ ok: true, devices: devices.map(d => ({
const page = await p2pDb.listAdminDevices({ limit: 101, offset });
const devices = page.slice(0, 100);
return c.json({ ok: true, hasMore: page.length > 100, devices: devices.map(d => ({
id: d.device_id, fingerprint: d.fingerprint || '', fingerprintShort: (d.fingerprint || d.device_id).slice(0, 12),
profile: d.profile || '', userAgent: d.user_agent || '', share: !!Number(d.share), banned: !!d.banned_at,
lastSeenAt: Number(d.last_seen_at), online: hub.isOnline(d.device_id), holdings: Number(d.holdings), holdingBytes: Number(d.holding_bytes),

26
server/p2p-admin.test.js Normal file
View File

@@ -0,0 +1,26 @@
import { test, expect } from 'bun:test';
import { Hono } from 'hono';
import { registerP2pAdminRoutes } from './p2p-admin.js';
test('P2P admin actions require auth, log mutations, and control online devices', async () => {
const events = [], calls = [];
const db = { execute: async query => { events.push(query.args); return { rowsAffected: 1 }; } };
const p2pDb = {
listAdminDevices: async ({ limit }) => { expect(limit).toBe(101); return [{ device_id: 'dev_0123456789abcdef', fingerprint: 'fingerprint', profile: 'home', user_agent: 'iPhone', share: 1, banned_at: null, last_seen_at: 42, holdings: 2, holding_bytes: 80, bytes_up: 4, bytes_down: 3 }]; },
setDeviceBanned: async (id, banned) => { calls.push(['ban', id, banned]); return true; },
removeDeviceHolders: async id => { calls.push(['remove', id]); return 2; },
};
const hub = { isOnline: () => true, onlineCount: () => 1, send: (...args) => { calls.push(['send', ...args]); return true; }, disconnect: id => calls.push(['disconnect', id]) };
const app = new Hono();
registerP2pAdminRoutes(app, { db, p2pDb, hub, stats: async () => ({ content: 5 }), adminAuth: async (c, next) => { if (c.req.header('x-admin') !== 'yes') return c.json({ ok:false },401); await next(); } });
expect((await app.request('/api/admin/p2p/devices')).status).toBe(401);
const list = await app.request('/api/admin/p2p/devices', { headers: { 'x-admin': 'yes' } });
expect(await list.json()).toMatchObject({ ok: true, network: { online: 1, content: 5 }, devices: [{ id: 'dev_0123456789abcdef', online: true, holdingBytes: 80 }] });
const headers = { 'x-admin': 'yes', 'content-type': 'application/json' };
expect((await app.request('/api/admin/p2p/devices/dev_0123456789abcdef/ban', { method:'POST',headers,body:'{"banned":true}' })).status).toBe(200);
expect((await app.request('/api/admin/p2p/devices/dev_0123456789abcdef/reannounce', { method:'POST',headers,body:'{}' })).status).toBe(200);
expect((await app.request('/api/admin/p2p/devices/dev_0123456789abcdef/holdings', { method:'DELETE',headers })).status).toBe(200);
expect(calls.some(x => x[0] === 'disconnect')).toBe(true);
expect(calls.some(x => x[0] === 'send' && x[2].type === 'force-sync')).toBe(true);
expect(events).toHaveLength(3);
});

2
server/package.json
View File

@@ -6,7 +6,7 @@
"scripts": {
"start": "bun server.js",
"dev": "bun --hot server.js",
"test": "bun test ./recommendations.test.js && bun test --timeout 60000 ./media-cache.test.js && bun test ./notes.test.js && bun test ./audit-log.test.js && bun test ./admin-tools.test.js && bun test ./transcriptions.test.js && bun test ./admin-analytics.test.js && bun test ./remote.test.js && bun test ./party.test.js && bun test ./uploads.test.js && bun test ./innertube.test.js && bun test ./related.test.js && bun test ./ytdlp-pool.test.js && bun test ./warm-queue.test.js && bun test ./p2p-db.test.js && bun test ./p2p-admit.test.js && bun test ./p2p-retention.test.js && bun test ./p2p-routes.test.js && bun test ./p2p-hub.test.js && bun test --timeout 60000 ./p2p-intake.test.js && bun test ./flags.test.js && bun test ./asset-manifest.test.js && bun test ./static-delivery.test.js && bun test ./video-catalog.test.js && bun test ./search-cache.test.js && bun test --timeout 60000 ./deep-search.test.js"
"test": "bun test ./recommendations.test.js && bun test --timeout 60000 ./media-cache.test.js && bun test ./notes.test.js && bun test ./audit-log.test.js && bun test ./request-monitor.test.js && bun test ./admin-tools.test.js && bun test ./transcriptions.test.js && bun test ./admin-analytics.test.js && bun test ./remote.test.js && bun test ./party.test.js && bun test ./uploads.test.js && bun test ./innertube.test.js && bun test ./related.test.js && bun test ./ytdlp-pool.test.js && bun test ./warm-queue.test.js && bun test ./p2p-db.test.js && bun test ./p2p-admin.test.js && bun test ./p2p-admit.test.js && bun test ./p2p-retention.test.js && bun test ./p2p-routes.test.js && bun test ./p2p-hub.test.js && bun test --timeout 60000 ./p2p-intake.test.js && bun test ./flags.test.js && bun test ./asset-manifest.test.js && bun test ./static-delivery.test.js && bun test ./video-catalog.test.js && bun test ./search-cache.test.js && bun test --timeout 60000 ./deep-search.test.js"
},
"dependencies": {
"@hono/node-server": "^1.14.0",

17
server/request-monitor.js
View File

@@ -1,4 +1,4 @@
import { appendFile, mkdir, readFile } from 'node:fs/promises';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { dirname } from 'node:path';
import { statfs } from 'node:fs/promises';
@@ -43,7 +43,12 @@ export function createRequestMonitor({ snapshotFile = null, diskPaths = [], adap
const value = await sample(); snapshots.push(value);
while (snapshots.length && now() - snapshots[0].at > 24 * 3600_000) snapshots.shift();
if (snapshotFile) {
try { await mkdir(dirname(snapshotFile), { recursive: true }); await appendFile(snapshotFile, JSON.stringify(value) + '\n'); }
try {
await mkdir(dirname(snapshotFile), { recursive: true });
const prior = (await readFile(snapshotFile, 'utf8').catch(() => '')).split('\n').filter(Boolean).map(line => { try { return JSON.parse(line); } catch { return null; } }).filter(x => x?.at >= now() - 24 * 3600_000);
const unique = new Map([...prior, value].map(item => [item.at, item]));
await writeFile(snapshotFile, [...unique.values()].sort((a,b)=>a.at-b.at).map(item => JSON.stringify(item)).join('\n') + '\n');
}
catch (error) { console.warn('[monitor] snapshot write failed:', error.message); }
}
return value;
@@ -54,7 +59,8 @@ export function createRequestMonitor({ snapshotFile = null, diskPaths = [], adap
if (snapshotFile) {
try {
const persisted = (await readFile(snapshotFile, 'utf8')).split('\n').filter(Boolean).map(line => { try { return JSON.parse(line); } catch { return null; } }).filter(x => x?.at >= cutoff);
return [...persisted, ...out].slice(-1440);
const unique = new Map([...persisted, ...out].map(item => [item.at, item]));
return [...unique.values()].sort((a,b)=>a.at-b.at).slice(-1440);
} catch { /* no file yet */ }
}
return out.slice(-1440);
@@ -67,10 +73,11 @@ export function createRequestMonitor({ snapshotFile = null, diskPaths = [], adap
try { await next(); status = c.res.status; }
finally {
inflight--;
const key = routeKey(c.req.path);
let key = routeKey(c.req.path);
if (!routes.has(key) && routes.size >= 300) key = 'other';
let row = routes.get(key);
if (!row) { row = { latencies: [], requests: 0, errors: 0 }; routes.set(key, row); }
row.requests++; if (status >= 500) row.errors++;
row.requests++; if (status >= 400) row.errors++;
row.latencies.push(Math.max(0, now() - started)); if (row.latencies.length > 1000) row.latencies.shift();
}
};

19
server/request-monitor.test.js
View File

@@ -1,6 +1,8 @@
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createRequestMonitor } from './request-monitor.js';
import { Hono } from 'hono';
import { registerRequestMonitorRoutes } from './request-monitor.js';
test('request monitor records route latency, status errors, and current process health', async () => {
let t = 100;
@@ -8,12 +10,25 @@ test('request monitor records route latency, status errors, and current process
const middleware = monitor.middleware();
const ctx = (path, status) => ({ req: { path }, res: { status } });
await middleware(ctx('/api/media/abc/status', 200), async () => { t += 8; });
await middleware(ctx('/api/media/abc/status', 404), async () => { t += 4; });
await middleware(ctx('/api/media/xyz/status', 503), async () => { t += 12; });
const sample = await monitor.sample();
assert.equal(sample.routes['/api/media/:id/status'].requests, 2);
assert.equal(sample.routes['/api/media/:id/status'].errors, 1);
assert.equal(sample.routes['/api/media/:id/status'].requests, 3);
assert.equal(sample.routes['/api/media/:id/status'].errors, 2);
assert.equal(sample.routes['/api/media/:id/status'].p50, 8);
assert.equal(sample.media.queued, 2);
assert.ok(sample.rssBytes > 0);
monitor.close();
});
test('monitor snapshot history is retained and its endpoint is admin protected', async () => {
let t = 10_000;
const monitor = createRequestMonitor({ now: () => t, snapshotIntervalMs: 0 });
await monitor.snapshot(); t += 60_000; await monitor.snapshot();
const app = new Hono();
registerRequestMonitorRoutes(app, { monitor, adminAuth: async (c,next) => c.req.header('x-admin') ? next() : c.json({ok:false},401) });
assert.equal((await app.request('/api/admin/ops/metrics')).status, 401);
const response = await app.request('/api/admin/ops/metrics', { headers: { 'x-admin': 'yes' } });
const data = await response.json(); assert.equal(data.history.length, 2); assert.ok(data.current.rssBytes > 0);
monitor.close();
});

9
server/server.js
View File

@@ -350,15 +350,20 @@ const app = new Hono();
app.use('*', logger());
const requestMonitor = createRequestMonitor({
snapshotFile: pathJoin(dirname(process.env.DB_PATH || './data/ytplayer.db'), 'ops-metrics.jsonl'),
snapshotIntervalMs: process.env.OPS_MONITOR === '0' ? 0 : 60_000,
diskPaths: [{ name: 'app-data', path: dirname(process.env.DB_PATH || './data/ytplayer.db') }, { name: 'media', path: process.env.MEDIA_DIR || './data/media' }],
adapters: {
ytdlp: () => ytdlpPool ? ytdlpPool.stats() : null,
media: () => media?.stats ? media.stats() : null,
transcription: () => notesDb.transcriptionStats?.() || null,
transcription: async () => {
try { const rows = (await notesDb.db.execute("SELECT status,COUNT(*) AS n FROM lyric_transcriptions WHERE status IN ('queued','running') GROUP BY status")).rows;
const counts = Object.fromEntries(rows.map(row => [row.status, Number(row.n)])); return { queued: counts.queued || 0, active: counts.running || 0 };
} catch { return null; }
},
cache: () => ({ ...searchCacheCounters, memoryEntries: searchCache.size }),
},
});
app.use('*', requestMonitor.middleware());
if (process.env.OPS_MONITOR !== '0') app.use('*', requestMonitor.middleware());
registerCatalogRoutes(app, { resolveListener: recommendationListener });
// ============================================================================

108
server/uploads.js
View File

@@ -23,9 +23,10 @@
* ========================================================================== */
import { spawn } from 'node:child_process';
import { createWriteStream, existsSync, mkdirSync, unlinkSync } from 'node:fs';
import { createWriteStream, existsSync, mkdirSync, unlinkSync, readdirSync, renameSync } from 'node:fs';
import { join } from 'node:path';
import { randomBytes } from 'node:crypto';
import { audit } from './audit-log.js';
export const UPLOAD_ID_RE = /^upl_[a-f0-9]{12}$/;
const MAX_BYTES = 4 * 1024 * 1024 * 1024; // 4 GB
@@ -104,6 +105,7 @@ export function registerUploadRoutes(app, deps) {
const artPath = (id) => locate(`${id}.art.jpg`);
const newArtPath = (id) => join(writeDir(), `${id}.art.jpg`);
const removeEverywhere = (name) => { for (const d of [uploadDir, backupDir].filter(Boolean)) { try { unlinkSync(join(d, name)); } catch { /* gone */ } } };
const removeStaleCopies = (name, currentPath) => { for (const d of [uploadDir, backupDir].filter(Boolean)) { const p = join(d, name); if (p !== currentPath) try { unlinkSync(p); } catch { /* no alternate copy */ } } };
async function probeFile(path) {
const out = await run(ffprobe, ['-v', 'error', '-print_format', 'json', '-show_format', '-show_streams', path]);
@@ -181,6 +183,7 @@ export function registerUploadRoutes(app, deps) {
try {
await Bun.write(target, file);
const r = await ingest({ id, ext, target, size: file.size, filename: file.name, fields: body, artFile: body.art });
await audit(db, 'admin', 'upload.create', id, { size: file.size, kind: r.upload.kind });
return c.json({ ok: true, ...r });
} catch (err) {
cleanup(id, target);
@@ -188,6 +191,85 @@ export function registerUploadRoutes(app, deps) {
}
});
app.get('/api/admin/uploads/manage', requireAdminOrToken, async c => {
const q = String(c.req.query('q') || '').trim().slice(0, 120), offset = Math.max(0, Number(c.req.query('offset')) || 0);
const result = await db.listAdminUploads({ q, offset, limit: 100 });
result.quotas = await db.listUploadQuotas?.() || [];
const known = new Set(await db.listUploadFileNames());
const orphans = [];
for (const dir of dirs()) for (const name of readdirSync(dir).slice(0, 10000)) {
if (!name.includes('.')) continue;
const id = name.split('.')[0];
if (UPLOAD_ID_RE.test(id) && !known.has(name) && !name.includes('.replace.')) orphans.push({ name, dir });
}
return c.json({ ok: true, ...result, offset, orphans: orphans.slice(0, 200) }, 200, { 'Cache-Control': 'no-store' });
});
app.put('/api/admin/uploads/quotas/:owner', requireAdminOrToken, async c => {
const owner = c.req.param('owner'); if (!/^[\w-]{8,128}$/.test(owner)) return c.json({ ok: false, error: 'invalid owner id' }, 400);
const body = await c.req.json().catch(() => ({})), maxBytes = Number(body.maxBytes);
if (!Number.isFinite(maxBytes) || maxBytes < 0 || maxBytes > 10 * 1024 ** 4) return c.json({ ok: false, error: 'quota must be zero (unlimited) or a valid byte count' }, 400);
await db.setUploadQuota?.(owner, maxBytes || null);
await audit(db, 'admin', 'upload.quota.set', owner, { maxBytes: maxBytes || null });
return c.json({ ok: true, maxBytes: maxBytes || null });
});
app.patch('/api/admin/uploads/:id', requireAdminOrToken, async c => {
const id = c.req.param('id'); if (!UPLOAD_ID_RE.test(id)) return c.json({ ok: false, error: 'invalid id' }, 400);
const body = await c.req.json().catch(() => ({})), before = await db.getUpload(id);
if (!before) return c.json({ ok: false, error: 'not found' }, 404);
const patch = {};
for (const key of ['title','artist','album']) if (body[key] !== undefined) patch[key] = String(body[key]).trim();
if (body.listed !== undefined) patch.listed = body.listed === true;
await db.updateUploadMeta(id, patch);
await audit(db, 'admin', 'upload.update', id, patch);
return c.json({ ok: true, upload: await db.getUpload(id) });
});
app.post('/api/admin/uploads/bulk', requireAdminOrToken, async c => {
const body = await c.req.json().catch(() => ({})), ids = [...new Set((Array.isArray(body.ids) ? body.ids : []).filter(id => UPLOAD_ID_RE.test(String(id))))].slice(0, 200);
if (!ids.length || !['delete','list','unlist'].includes(body.action)) return c.json({ ok: false, error: 'choose uploads and delete, list, or unlist' }, 400);
let changed = 0;
for (const id of ids) {
const u = await db.getUpload(id); if (!u) continue;
if (body.action === 'delete') { await db.deleteUpload(id); removeEverywhere(`${u.id}.${u.ext}`); removeEverywhere(`${id}.art.jpg`); }
else await db.updateUploadMeta(id, { listed: body.action === 'list' });
await audit(db, 'admin', `upload.bulk.${body.action}`, id); changed++;
}
return c.json({ ok: true, changed });
});
app.put('/api/admin/uploads/:id/replace', requireAdminOrToken, async c => {
const id = c.req.param('id'), old = UPLOAD_ID_RE.test(id) ? await db.getUpload(id) : null;
if (!old) return c.json({ ok: false, error: 'not found' }, 404);
const filename = String(c.req.query('name') || 'replacement.' + old.ext), ext = extOf(filename);
if (!VIDEO_EXT.has(ext) && !AUDIO_EXT.has(ext)) return c.json({ ok: false, error: extError(ext) }, 415);
const declared = Number(c.req.header('content-length')); if (declared > MAX_BYTES) return c.json({ ok: false, error: 'file is larger than 4 GB' }, 413);
const reader = c.req.raw.body?.getReader(); if (!reader) return c.json({ ok: false, error: 'empty body' }, 400);
const dir = writeDir(), temp = join(dir, `${id}.replace.${randomBytes(4).toString('hex')}.${ext}`), target = join(dir, `${id}.${ext}`);
let got = 0; const out = createWriteStream(temp);
try {
try { for (;;) { const { done, value } = await reader.read(); if (done) break; got += value.byteLength; if (got > MAX_BYTES) throw Object.assign(Error('file is larger than 4 GB'), { status: 413 }); if (!out.write(value)) await new Promise(r => out.once('drain', r)); } }
finally { await new Promise(r => out.end(r)); }
if (!got || (declared > 0 && got !== declared)) throw Object.assign(Error('empty or incomplete replacement'), { status: 400 });
const probe = await probeFile(temp), info = describeProbe(probe, filename);
if (!info.hasAudio) throw Object.assign(Error('replacement has no audio stream'), { status: 415 });
const backup = `${target}.previous`;
const existingTarget = existsSync(target);
if (existingTarget) renameSync(target, backup);
try {
renameSync(temp, target);
await db.replaceUploadFile(id, { kind: info.kind, title: String(c.req.query('title') ?? old.title).slice(0,300), artist: String(c.req.query('artist') ?? old.artist).slice(0,200), album: String(c.req.query('album') ?? old.album).slice(0,200), duration: info.duration, ext, mime: MIME[ext] || (info.kind === 'video' ? 'video/mp4' : 'audio/mpeg'), size: got, art: old.art });
} catch (error) { try { unlinkSync(target); } catch {} if (existingTarget) renameSync(backup, target); throw error; }
if (existingTarget) try { unlinkSync(backup); } catch {}
removeStaleCopies(`${id}.${ext}`, target);
if (old.ext !== ext) removeEverywhere(`${id}.${old.ext}`);
await audit(db, 'admin', 'upload.replace', id, { oldExt: old.ext, ext, size: got });
return c.json({ ok: true, upload: await db.getUpload(id) });
} catch (error) { return c.json({ ok: false, error: error.status ? error.message : 'replacement failed validation' }, error.status || 422); }
finally { try { unlinkSync(temp); } catch {} }
});
// Streaming upload: the raw file is the request body, copied to disk chunk by
// chunk so a multi-GB (hour-long) video never sits in memory.
// PUT /api/admin/uploads/stream?name=talk.mp4&title=&artist=&album=
@@ -219,6 +301,7 @@ export function registerUploadRoutes(app, deps) {
if (declared > 0 && got !== declared) throw Object.assign(new Error(`upload cut short (${got} of ${declared} bytes)`), { status: 400 });
const q = (k) => String(c.req.query(k) || '');
const r = await ingest({ id, ext, target, size: got, filename: name, fields: { title: q('title'), artist: q('artist'), album: q('album') } });
await audit(db, 'admin', 'upload.create', id, { size: got, kind: r.upload.kind });
return c.json({ ok: true, ...r });
} catch (err) {
cleanup(id, target);
@@ -252,6 +335,8 @@ export function registerUploadRoutes(app, deps) {
if (declared > SHARED.fileBytes) return c.json({ ok: false, error: `shared files can be up to ${mb(SHARED.fileBytes)}` }, 413);
if (sharedActive.has(fp)) return c.json({ ok: false, error: 'this device is already uploading — wait for it to finish' }, 429);
const use = await db.sharedUploadUsage(fp);
const ownerQuota = await db.getUploadQuota?.(fp);
if (ownerQuota && use.bytes + declared > ownerQuota) return c.json({ ok: false, error: `owner storage quota of ${mb(ownerQuota)} reached` }, 413);
if (use.today >= SHARED.perDay) return c.json({ ok: false, error: `limit of ${SHARED.perDay} shared uploads a day reached` }, 429);
if (use.bytes + declared > SHARED.deviceBytes) return c.json({ ok: false, error: `this device's ${mb(SHARED.deviceBytes)} of shared uploads is used up` }, 413);
if (use.totalShared + declared > SHARED.totalBytes) return c.json({ ok: false, error: 'the server has no room for more shared uploads' }, 507);
@@ -268,8 +353,8 @@ export function registerUploadRoutes(app, deps) {
const { done, value } = await reader.read();
if (done) break;
got += value.byteLength;
if (got > SHARED.fileBytes || use.bytes + got > SHARED.deviceBytes) {
throw Object.assign(new Error(`shared files can be up to ${mb(Math.min(SHARED.fileBytes, SHARED.deviceBytes - use.bytes))}`), { status: 413 });
if (got > SHARED.fileBytes || use.bytes + got > SHARED.deviceBytes || (ownerQuota && use.bytes + got > ownerQuota)) {
throw Object.assign(new Error(`shared storage quota reached (owner cap ${ownerQuota ? mb(ownerQuota) : mb(SHARED.deviceBytes)})`), { status: 413 });
}
if (!out.write(value)) await new Promise((r) => out.once('drain', r));
}
@@ -306,6 +391,8 @@ export function registerUploadRoutes(app, deps) {
await Bun.write(tmp, buf);
await run(ffmpeg, ['-v', 'error', '-y', '-i', tmp, '-frames:v', '1', '-vf', "scale='min(800,iw)':-2", newArtPath(id)]);
await db.setUploadArt(id, 'file');
removeStaleCopies(`${id}.art.jpg`, newArtPath(id));
await audit(db, 'admin', 'upload.art.replace', id, { size: buf.byteLength });
return c.json({ ok: true });
} catch (err) {
return c.json({ ok: false, error: 'unusable image: ' + err.message }, 400);
@@ -320,6 +407,7 @@ export function registerUploadRoutes(app, deps) {
const u = await db.getUpload(id);
if (!u) return c.json({ ok: false, error: 'not found' }, 404);
await db.deleteUpload(id);
await audit(db, 'admin', 'upload.delete', id, { size: u.size });
removeEverywhere(`${u.id}.${u.ext}`);
removeEverywhere(`${id}.art.jpg`);
return c.json({ ok: true });
@@ -335,6 +423,8 @@ export function registerUploadRoutes(app, deps) {
const id = c.req.param('id');
if (!UPLOAD_ID_RE.test(id)) return c.json({ ok: false, error: 'invalid id' }, 400);
const u = await db.getUpload(id);
const requestedGen = Number(c.req.query('g'));
if (requestedGen && requestedGen !== Number(u?.gen || 1)) return c.json({ ok: false, error: 'upload generation changed' }, 409, { 'Cache-Control': 'no-store' });
if (!u || !existsSync(filePath(u))) return c.json({ ok: false, error: 'not found' }, 404);
db.touchUpload(id);
return rangeFileResponse(c, filePath(u), u.mime, 'public, max-age=31536000, immutable');
@@ -343,6 +433,8 @@ export function registerUploadRoutes(app, deps) {
app.get('/api/uploads/:id/art', async (c) => {
const id = c.req.param('id');
if (!UPLOAD_ID_RE.test(id)) return c.json({ ok: false, error: 'invalid id' }, 400);
const row = await db.getUpload(id), requestedGen = Number(c.req.query('g'));
if (requestedGen && requestedGen !== Number(row?.artGen || 1)) return c.json({ ok: false, error: 'upload artwork generation changed' }, 409, { 'Cache-Control': 'no-store' });
const p = artPath(id);
if (!existsSync(p)) return c.json({ ok: false, error: 'no cover art' }, 404);
return rangeFileResponse(c, p, 'image/jpeg', 'public, max-age=31536000, immutable');
@@ -351,20 +443,20 @@ export function registerUploadRoutes(app, deps) {
// /api/streams payload for an upload: one "Original" quality for video,
// audio + cover art for audio (the app shows the art where the video goes).
function streamsPayload(u) {
const art = u.art ? `/api/uploads/${u.id}/art` : '';
const url = `/api/uploads/${u.id}`;
const art = u.art ? `/api/uploads/${u.id}/art?g=${u.artGen || 1}` : '';
const url = `/api/uploads/${u.id}?g=${u.gen || 1}`;
return {
meta: { id: u.id, title: u.title, channel: u.artist || u.album || 'Uploaded', channelId: '', channelUrl: '', duration: u.duration, thumbnail: art },
audioUrl: u.kind === 'audio' ? url : '',
qualities: u.kind === 'video' ? [{ label: 'Original', height: 0, hasAudio: true, url }] : [],
upload: true,
upload: true, gen: u.gen || 1,
kind: u.kind,
art,
};
}
const card = (u) => ({
id: u.id, title: u.title, channel: u.artist || u.album || 'Uploaded', channelId: '', channelUrl: '',
duration: u.duration, thumbnail: u.art ? `/api/uploads/${u.id}/art` : '', upload: true, kind: u.kind,
id: u.id, title: u.title, channel: u.artist || u.album || 'Uploaded', channelId: '', channelUrl: '', gen: u.gen || 1,
duration: u.duration, thumbnail: u.art ? `/api/uploads/${u.id}/art?g=${u.artGen || 1}` : '', upload: true, kind: u.kind,
});
return { streamsPayload, card, filePath, artPath, isUploadId: (id) => UPLOAD_ID_RE.test(id) };

48
server/uploads.test.js
View File

@@ -1,7 +1,7 @@
// Admin uploads: real ffmpeg-made files through the real routes.
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { spawnSync } from 'node:child_process';
import { mkdtempSync, rmSync, existsSync, readFileSync } from 'node:fs';
import { mkdtempSync, rmSync, existsSync, readFileSync, copyFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Hono } from 'hono';
@@ -32,7 +32,7 @@ beforeAll(async () => {
'-shortest', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', fx('clip.mp4'));
app = new Hono();
uploads = registerUploadRoutes(app, {
db, uploadDir: join(root, 'uploads'), parseLrc, sanitizeLyrics,
db, uploadDir: join(root, 'uploads'), backupDir: join(root, 'backup-uploads'), parseLrc, sanitizeLyrics,
requireAdminOrToken: async (c, next) => { if (c.req.header('x-admin') === 'yes') await next(); else return c.json({ ok: false, error: 'admin only' }, 401); },
rangeFileResponse: (c, path, type) => {
const body = readFileSync(path);
@@ -108,11 +108,11 @@ describe('uploads', () => {
expect(hit.uploads.map((u) => u.id)).toEqual([audioId]);
const u = await db.getUpload(audioId);
expect(uploads.streamsPayload(u)).toMatchObject({
audioUrl: `/api/uploads/${audioId}`, kind: 'audio', upload: true, art: `/api/uploads/${audioId}/art`, qualities: [],
audioUrl: `/api/uploads/${audioId}?g=1`, kind: 'audio', upload: true, art: `/api/uploads/${audioId}/art?g=1`, qualities: [],
});
expect(uploads.card(u)).toMatchObject({ id: audioId, title: 'Test Song', channel: 'Test Artist', kind: 'audio' });
const v = uploads.streamsPayload(await db.getUpload(videoId));
expect(v.qualities[0]).toMatchObject({ label: 'Original', hasAudio: true, url: `/api/uploads/${videoId}` });
expect(v.qualities[0]).toMatchObject({ label: 'Original', hasAudio: true, url: `/api/uploads/${videoId}?g=1` });
});
test('serving supports Range, and delete removes the files', async () => {
@@ -162,7 +162,45 @@ describe('streaming upload (large / long files)', () => {
const up = await (await put('/api/admin/uploads/stream?name=song.mp3', readFileSync(fx('song.mp3')))).json();
const r = await put(`/api/admin/uploads/${up.upload.id}/art`, readFileSync(fx('cover.png')));
expect((await r.json()).ok).toBe(true);
expect((await db.getUpload(up.upload.id)).art).toBeTruthy();
const updated = await db.getUpload(up.upload.id);
expect(updated).toMatchObject({ art: 'file', gen: 1, artGen: 2 });
expect(uploads.streamsPayload(updated).audioUrl).toBe(`/api/uploads/${up.upload.id}?g=1`);
expect(uploads.streamsPayload(updated).art).toBe(`/api/uploads/${up.upload.id}/art?g=2`);
expect((await app.request(`/api/uploads/${up.upload.id}/art?g=1`)).status).toBe(409);
});
test('admin manager edits metadata, replaces bytes in place with a new generation, and pages orphan reports', async () => {
const up = await (await put('/api/admin/uploads/stream?name=original.mp3', readFileSync(fx('song.mp3')))).json();
const id = up.upload.id;
expect(up.upload.gen).toBe(1);
const edited = await app.request(`/api/admin/uploads/${id}`, { method: 'PATCH', headers: { 'x-admin': 'yes', 'content-type': 'application/json' }, body: JSON.stringify({ title: 'Changed title', artist: 'New artist', listed: false }) });
expect((await edited.json()).upload).toMatchObject({ id, title: 'Changed title', artist: 'New artist', listed: false });
const replaceBytes = readFileSync(fx('clip.mp4'));
const replaced = await put(`/api/admin/uploads/${id}/replace?name=concert.mp4`, replaceBytes);
const payload = await replaced.json();
expect(replaced.status).toBe(200);
expect(payload.upload).toMatchObject({ id, ext: 'mp4', kind: 'video', gen: 2, listed: false });
expect((await app.request(`/api/uploads/${id}?g=1`)).status).toBe(409);
expect((await app.request(`/api/uploads/${id}`, { headers: { Range: 'bytes=0-9' } })).status).toBe(206);
expect(uploads.streamsPayload(payload.upload).qualities[0].url).toContain('?g=2');
const listed = await app.request('/api/admin/uploads/bulk', { method: 'POST', headers: { 'x-admin': 'yes', 'content-type': 'application/json' }, body: JSON.stringify({ ids: [id], action: 'list' }) });
expect(await listed.json()).toMatchObject({ ok: true, changed: 1 });
const manage = await app.request('/api/admin/uploads/manage?q=Changed', { headers: { 'x-admin': 'yes' } });
expect(await manage.json()).toMatchObject({ ok: true, total: 1, uploads: [expect.objectContaining({ id, listed: true, gen: 2 })] });
const quota = await app.request('/api/admin/uploads/quotas/owner-device1', { method: 'PUT', headers: { 'x-admin': 'yes', 'content-type': 'application/json' }, body: JSON.stringify({ maxBytes: 5_000_000 }) });
expect(await quota.json()).toMatchObject({ ok: true, maxBytes: 5_000_000 });
expect(await db.getUploadQuota('owner-device1')).toBe(5_000_000);
await db.setUploadQuota('owner-device1', null);
});
test('same-extension replacement removes a stale fallback copy before publishing its new generation', async () => {
const up = await (await put('/api/admin/uploads/stream?name=mirror.mp3', readFileSync(fx('song.mp3')))).json();
const staleBackup = join(root, 'backup-uploads', `${up.upload.id}.mp3`);
copyFileSync(join(root, 'uploads', `${up.upload.id}.mp3`), staleBackup);
expect(existsSync(staleBackup)).toBe(true);
const replaced = await put(`/api/admin/uploads/${up.upload.id}/replace?name=updated.mp3`, readFileSync(fx('song.mp3')));
expect((await replaced.json()).upload.gen).toBe(2);
expect(existsSync(staleBackup)).toBe(false);
});
});