ytplayer/server/admin-analytics.js

import { randomUUID } from 'node:crypto';
import { statfs, stat } from 'node:fs/promises';
import { db } from './db.js';
import { ingest } from './video-catalog.js';

const LEASE = 120000;
const n = v => Number(v) || 0;
const clean = value => String(value || '').trim().slice(0, 200);
const parse = value => { try { return JSON.parse(value) || {}; } catch { return {}; } };
function publicJob(row) {
  const state = parse(row.state);
  return { id: row.id, query: row.query, maxVideos: n(row.max_videos), depth: n(row.depth), status: row.status,
    collected: state.ids?.length || 0, enriched: state.enriched || 0, failed: state.failed || 0,
    searches: state.searches || 0, current: state.current || '', error: row.error, errors: state.errors || [],
    createdAt: n(row.created_at), updatedAt: n(row.updated_at) };
}
export function descriptiveMetadata(raw) {
  const out = {};
  for (const key of ['id','title','fulltitle','description','channel','channel_id','channel_url','uploader','uploader_id','uploader_url','upload_date','release_date','timestamp','duration','view_count','like_count','comment_count','tags','categories','language','live_status','availability','age_limit','license','chapters','thumbnails','thumbnail','webpage_url','artist','artists','album','track','release_year']) {
    if (raw[key] != null) out[key] = raw[key];
  }
  // Retain format specs, excluding expiring media URLs and request headers.
  if (Array.isArray(raw.formats)) out.formats = raw.formats.map(f => Object.fromEntries(['format_id','format_note','ext','width','height','fps','vcodec','acodec','abr','tbr','filesize','filesize_approx','audio_channels','asr'].filter(k => f[k] != null).map(k => [k, f[k]])));
  if (JSON.stringify(out).length > 500000) throw Error('Extracted metadata exceeds 500 KB.');
  return out;
}
function card(raw) {
  return { ...raw, channel: raw.channel || raw.uploader || '', channelId: raw.channel_id,
    channelUrl: raw.channel_url, thumbnail: raw.thumbnail || raw.thumbnails?.at(-1)?.url };
}
export async function storageAnalytics(paths = []) {
  const queries = [
    "SELECT status,COUNT(*) AS count,COALESCE(SUM(size),0) AS bytes,COALESCE(SUM(duration),0) AS seconds FROM media_cache GROUP BY status",
    "SELECT kind,COUNT(*) AS count,COALESCE(SUM(size),0) AS bytes FROM uploads GROUP BY kind",
  ];
  // Keep each aggregate on its own table, so thumbnail bytes and plays never multiply.
  const [media, uploads, catalog, thumbs, details, sources, plays, top] = await Promise.all([
    db.execute(queries[0]), db.execute(queries[1]),
    db.execute('SELECT COUNT(*) AS videos,COALESCE(SUM(length(CAST(card AS BLOB))),0) AS bytes,MAX(updated_at) AS lastSeen FROM video_meta'),
    db.execute('SELECT COUNT(*) AS total,SUM(CASE WHEN data IS NOT NULL THEN 1 ELSE 0 END) AS saved,COALESCE(SUM(size),0) AS bytes,SUM(CASE WHEN data IS NULL AND retry_at<9007199254740991 THEN 1 ELSE 0 END) AS pending FROM video_thumbnails'),
    db.execute('SELECT COUNT(*) AS videos,COALESCE(SUM(length(CAST(metadata AS BLOB))),0) AS bytes FROM video_details'),
    db.execute('SELECT source,COUNT(*) AS videos,SUM(discoveries) AS discoveries,MAX(last_seen) AS lastSeen FROM video_meta_sources GROUP BY source ORDER BY videos DESC'),
    db.execute('SELECT COALESCE(SUM(plays),0) AS plays,COUNT(DISTINCT video_id) AS videos FROM listening_daily'),
    db.execute('SELECT l.video_id,SUM(l.plays) AS plays,m.card FROM listening_daily l LEFT JOIN video_meta m ON m.id=l.video_id GROUP BY l.video_id ORDER BY plays DESC LIMIT 10'),
  ]);
  const volumes = await Promise.all(paths.map(async ({ label, path }) => {
    try { const [fs, file] = await Promise.all([statfs(path), stat(path)]); return { label, device: String(file.dev), total: n(fs.blocks) * n(fs.bsize), free: n(fs.bavail) * n(fs.bsize) }; }
    catch { return { label, unavailable: true }; }
  }));
  return { media: media.rows, uploads: uploads.rows, catalog: catalog.rows[0], thumbnails: thumbs.rows[0], details: details.rows[0], sources: sources.rows, listening: plays.rows[0], topPlayed: top.rows.map(r => ({ id: r.video_id, plays: n(r.plays), title: parse(r.card).title || r.video_id })), volumes };
}

export function registerAnalyticsRoutes(app, { adminAuth, runYtdlp, paths = [], autoStart = true }) {
  let running = null;
  const owner = randomUUID();
  async function runNext() {
    if (running) return running;
    running = execute().finally(() => { running = null; });
    return running;
  }
  async function execute() {
    const time = Date.now();
    const job = (await db.execute({ sql: `UPDATE metadata_collections SET status='running',owner=?,lease_until=?,updated_at=?
      WHERE id=(SELECT id FROM metadata_collections WHERE status='queued' OR (status='running' AND lease_until<?) ORDER BY created_at LIMIT 1) RETURNING *`, args: [owner, time + LEASE, time, time] })).rows[0];
    if (!job) return false;
    const state = { ids: [], queries: [{ q: job.query, level: 0 }], pending: [], enriched: 0, failed: 0, searches: 0, errors: [], ...parse(job.state) };
    const seen = new Set(state.ids);
    async function checkpoint(status = 'running', error = null) {
      const result = await db.execute({ sql: "UPDATE metadata_collections SET state=?,status=?,error=?,lease_until=?,updated_at=? WHERE id=? AND owner=? AND status='running'", args: [JSON.stringify(state), status, error, Date.now() + LEASE, Date.now(), job.id, owner] });
      if (!result.rowsAffected) { const error = Error('Collection cancelled or reassigned.'); error.name = 'CollectionStopped'; throw error; }
    }
    try {
      while (state.pending.length || (state.queries.length && seen.size < job.max_videos && state.searches < 24)) {
        if (!state.pending.length) {
          const { q, level } = state.queries[0]; state.current = q; await checkpoint();
          const remaining = job.max_videos - seen.size;
          const count = Math.max(1, Math.ceil(remaining / (job.depth - level + 1)));
          const output = await runYtdlp([`ytsearch${count}:${q}`, '--dump-json', '--flat-playlist', '--skip-download', '--no-warnings', '--ignore-errors'], { signal: AbortSignal.timeout(60000) });
          await checkpoint(); // Recheck cancellation/ownership before storing discoveries.
          const unique = new Map();
          for (const raw of output.split('\n').filter(Boolean).map(parse)) {
            if (/^[\w-]{11}$/.test(raw.id) && !seen.has(raw.id)) unique.set(raw.id, raw);
          }
          const results = [...unique.values()].slice(0, remaining);
          await ingest(results.map(card), 'collector');
          for (const raw of results) { if (seen.has(raw.id)) continue; seen.add(raw.id); state.ids.push(raw.id); state.pending.push({ id: raw.id, level }); }
          state.queries.shift(); state.searches++; await checkpoint();
          continue;
        }
        const item = state.pending[0]; state.current = item.id; await checkpoint();
        try {
          const raw = JSON.parse(await runYtdlp(['-J', '--skip-download', '--no-playlist', '--no-warnings', `https://www.youtube.com/watch?v=${item.id}`], { signal: AbortSignal.timeout(60000) }));
          if (raw.id !== item.id) throw Error('Extractor returned a different video.');
          await checkpoint();
          const metadata = descriptiveMetadata(raw);
          await ingest([card(raw)], 'collector');
          await db.execute({ sql: 'INSERT INTO video_details (video_id,metadata,updated_at) VALUES (?,?,?) ON CONFLICT(video_id) DO UPDATE SET metadata=excluded.metadata,updated_at=excluded.updated_at', args: [item.id, JSON.stringify(metadata), Date.now()] });
          state.enriched++;
          if (item.level < job.depth && state.queries.length < 24) {
            for (const q of [raw.channel || raw.uploader, ...(raw.tags || []).slice(0, 2)]) {
              if (state.queries.length >= 24) break;
              const term = clean(q); if (!term) continue;
              state.visited ??= [job.query.toLowerCase()];
              if (state.visited.includes(term.toLowerCase())) continue;
              state.visited.push(term.toLowerCase()); state.queries.push({ q: term, level: item.level + 1 });
            }
          }
        } catch (error) { if (error.name === 'CollectionStopped') throw error; state.failed++; if (state.errors.length < 20) state.errors.push({ id: item.id, error: String(error.message).slice(0, 200) }); }
        state.pending.shift(); await checkpoint();
      }
      state.current = ''; await checkpoint('complete');
    } catch (error) {
      if (error.name === 'CollectionStopped') return true;
      await db.execute({ sql: "UPDATE metadata_collections SET status='failed',state=?,error=?,updated_at=? WHERE id=? AND owner=? AND status='running'", args: [JSON.stringify(state), String(error.message).slice(0, 300), Date.now(), job.id, owner] });
    }
    return true;
  }
  const kick = () => runNext().catch(error => console.error('[metadata collector]', error.message));
  if (autoStart) { const timer = setInterval(kick, 10000); timer.unref?.(); kick(); }
  app.get('/api/admin/analytics', adminAuth, async c => c.json({ ok: true, ...await storageAnalytics(paths) }, 200, { 'Cache-Control': 'no-store' }));
  app.get('/api/admin/metadata', adminAuth, async c => {
    const query = clean(c.req.query('q')), offset = Math.max(0, Math.min(500000, parseInt(c.req.query('offset')) || 0));
    const where = query ? 'WHERE m.hay LIKE ?' : '', args = query ? ['%' + query.replace(/[\\%_]/g, '\\$&') + '%'] : [];
    const filter = where ? where + " ESCAPE '\\'" : '';
    const rows = (await db.execute({ sql: `SELECT m.id,m.card,m.updated_at,d.updated_at AS enriched,t.size AS thumbnailBytes FROM video_meta m LEFT JOIN video_details d ON d.video_id=m.id LEFT JOIN video_thumbnails t ON t.video_id=m.id ${filter} ORDER BY m.updated_at DESC LIMIT 50 OFFSET ?`, args: [...args, offset] })).rows;
    const total = n((await db.execute({ sql: `SELECT COUNT(*) AS n FROM video_meta m ${filter}`, args })).rows[0].n);
    return c.json({ ok: true, total, offset, videos: rows.map(r => ({ ...parse(r.card), updatedAt: n(r.updated_at), enriched: !!r.enriched, thumbnailBytes: n(r.thumbnailBytes) })) });
  });
  app.get('/api/admin/metadata/:id', adminAuth, async c => {
    const row = (await db.execute({ sql: 'SELECT m.card,d.metadata FROM video_meta m LEFT JOIN video_details d ON d.video_id=m.id WHERE m.id=?', args: [c.req.param('id')] })).rows[0];
    return row ? c.json({ ok: true, metadata: row.metadata ? parse(row.metadata) : parse(row.card) }) : c.json({ ok: false, error: 'Not found.' }, 404);
  });
  app.get('/api/admin/collections', adminAuth, async c => c.json({ ok: true, jobs: (await db.execute('SELECT * FROM metadata_collections ORDER BY created_at DESC LIMIT 30')).rows.map(publicJob) }));
  app.post('/api/admin/collections', adminAuth, async c => {
    let body; try { body = await c.req.json(); } catch { return c.json({ ok: false, error: 'Invalid JSON.' }, 400); }
    const query = clean(body?.query), limit = Number(body?.maxVideos), depth = Number(body?.depth);
    if (!query || !Number.isInteger(limit) || limit < 1 || limit > 500 || !Number.isInteger(depth) || depth < 0 || depth > 3) return c.json({ ok: false, error: 'Enter a search, 1–500 videos and depth 0–3.' }, 400);
    const time = Date.now(), id = randomUUID();
    const result = await db.execute({ sql: "INSERT INTO metadata_collections (id,query,max_videos,depth,created_at,updated_at) SELECT ?,?,?,?,?,? WHERE (SELECT COUNT(*) FROM metadata_collections WHERE status IN ('queued','running'))<3", args: [id, query, limit, depth, time, time] });
    if (!result.rowsAffected) return c.json({ ok: false, error: 'Three collections are already active.' }, 429);
    if (autoStart) kick();
    return c.json({ ok: true, id }, 202);
  });
  app.post('/api/admin/collections/:id/cancel', adminAuth, async c => {
    const result = await db.execute({ sql: "UPDATE metadata_collections SET status='cancelled',updated_at=? WHERE id=? AND status IN ('queued','running')", args: [Date.now(), c.req.param('id')] });
    return c.json({ ok: !!result.rowsAffected });
  });
  return { runNext };
}