ytplayer/server/transcriptions.js

// Explicit Whisper requests produce reviewable drafts, never published notes.
import { randomUUID } from 'node:crypto';
import { db as sql } from './db.js';
const ID = /^([\w-]{11}|upl_[a-f0-9]{12})$/;
const LEASE_MS = 120000;
function publicJob(row) {
  if (!row) return null;
  return { id: row.id, videoId: row.video_id, status: row.status, stage: row.stage,
    baseRev: Number(row.base_rev), result: row.result ? JSON.parse(row.result) : null,
    error: row.error, createdAt: Number(row.created_at), updatedAt: Number(row.updated_at) };
}

export function registerTranscriptionRoutes(app, { db, adminAuth, workerAuth, workerEnabled, sanitizeLyrics, now = Date.now }) {
  const seen = () => sql.execute({ sql: 'INSERT INTO lyrics_worker_state (id,last_seen) VALUES (1,?) ON CONFLICT(id) DO UPDATE SET last_seen=excluded.last_seen', args: [now()] });
  async function source(id) {
    if (id.startsWith('upl_')) return db.getUpload(id);
    const row = await db.getMedia(id); return row?.status === 'ready' ? row : null;
  }
  app.post('/api/admin/transcriptions/:video', adminAuth, async c => {
    const id = c.req.param('video');
    if (!ID.test(id)) return c.json({ ok: false, error: 'invalid video id' }, 400);
    if (!workerEnabled) return c.json({ ok: false, error: 'Whisper is unavailable: the lyrics worker is not configured.' }, 503);
    const audio = await source(id);
    if (!audio) return c.json({ ok: false, error: 'Save this video on the server before transcribing its audio.' }, 409);
    if (Number(audio.duration) > 3600) return c.json({ ok: false, error: 'Whisper requests are limited to one hour of audio.' }, 422);
    const notes = await db.getNotes(id);
    // One active job per song and a bounded queue, including across restarts.
    const existing = (await sql.execute({ sql: "SELECT * FROM lyric_transcriptions WHERE video_id=? AND status IN ('queued','running')", args: [id] })).rows[0];
    if (existing) return c.json({ ok: true, job: publicJob(existing) });
    const active = Number((await sql.execute("SELECT COUNT(*) AS n FROM lyric_transcriptions WHERE status IN ('queued','running')")).rows[0].n);
    if (active >= 25) return c.json({ ok: false, error: 'The transcription queue is full. Try again after a job finishes.' }, 429);
    const time = now();
    await sql.execute({ sql: `INSERT OR IGNORE INTO lyric_transcriptions (id,video_id,base_rev,created_at,updated_at)
      SELECT ?,?,?,?,? WHERE (SELECT COUNT(*) FROM lyric_transcriptions WHERE status IN ('queued','running')) < 25`,
      args: [randomUUID(), id, notes.lyrics?.rev || 0, time, time] });
    const row = (await sql.execute({ sql: "SELECT * FROM lyric_transcriptions WHERE video_id=? AND status IN ('queued','running')", args: [id] })).rows[0];
    return row ? c.json({ ok: true, job: publicJob(row) }, 202) : c.json({ ok: false, error: 'The transcription queue is full.' }, 429);
  });
  app.get('/api/admin/transcriptions/:video', adminAuth, async c => {
    const id = c.req.param('video');
    if (!ID.test(id)) return c.json({ ok: false, error: 'invalid video id' }, 400);
    const row = (await sql.execute({ sql: 'SELECT * FROM lyric_transcriptions WHERE video_id=? ORDER BY created_at DESC,rowid DESC LIMIT 1', args: [id] })).rows[0];
    const worker = (await sql.execute('SELECT last_seen FROM lyrics_worker_state WHERE id=1')).rows[0];
    return c.json({ ok: true, job: publicJob(row), workerOnline: !!worker && now() - Number(worker.last_seen) < 90000, enabled: !!workerEnabled }, 200, { 'Cache-Control': 'no-store' });
  });
  app.post('/api/lyrics-worker/claim', workerAuth, async c => {
    await seen(); const time = now();
    await sql.execute({ sql: "UPDATE lyric_transcriptions SET status='failed',stage='failed',error='The worker stopped repeatedly. Please retry.',updated_at=? WHERE status='running' AND lease_until<? AND attempts>=3", args: [time, time] });
    // UPDATE RETURNING makes claiming atomic for multiple worker processes.
    const row = (await sql.execute({ sql: `UPDATE lyric_transcriptions SET status='running',stage='loading-model',attempts=attempts+1,
      lease_token=?,lease_until=?,updated_at=? WHERE id=(SELECT id FROM lyric_transcriptions
      WHERE (status='queued' OR (status='running' AND lease_until<?)) AND attempts<3 ORDER BY created_at LIMIT 1) RETURNING *`,
      args: [randomUUID(), time + LEASE_MS, time, time] })).rows[0];
    return c.json({ ok: true, job: row ? { ...publicJob(row), lease: row.lease_token,
      audioPath: row.video_id.startsWith('upl_') ? `/api/uploads/${row.video_id}` : `/api/media/${row.video_id}?a=1` } : null });
  });
  app.post('/api/lyrics-worker/jobs/:job', workerAuth, async c => {
    let body;
    try { const raw = await c.req.text(); if (raw.length > 400000) return c.json({ ok: false, error: 'transcript too large' }, 413); body = JSON.parse(raw); }
    catch { return c.json({ ok: false, error: 'invalid JSON' }, 400); }
    if (!body || typeof body.lease !== 'string') return c.json({ ok: false, error: 'missing lease' }, 400);
    const time = now();
    let result = null, status = 'running', error = null;
    let stage = ['loading-model', 'downloading-audio', 'transcribing'].includes(body.stage) ? body.stage : 'transcribing';
    if (body.status === 'complete') {
      try { result = sanitizeLyrics(body.result); if (!result.lines.length) throw Error('No sung lyrics were found.'); }
      catch (e) { return c.json({ ok: false, error: e.message }, 400); }
      result.tags = ['auto-transcribed (whisper)']; status = 'complete'; stage = 'complete';
    } else if (body.status === 'failed') { status = 'failed'; stage = 'failed'; error = String(body.error || 'Transcription failed.').slice(0, 300); }
    const changed = await sql.execute({ sql: `UPDATE lyric_transcriptions SET status=?,stage=?,result=?,error=?,lease_until=?,updated_at=?
      WHERE id=? AND status='running' AND lease_token=? AND lease_until>=?`,
      args: [status, stage, result ? JSON.stringify(result) : null, error, time + LEASE_MS, time, c.req.param('job'), body.lease, time] });
    if (!changed.rowsAffected) return c.json({ ok: false, error: 'This job lease has expired or finished.' }, 409);
    await seen();
    // Bound completed draft retention; active jobs are never deleted.
    await sql.execute({ sql: "DELETE FROM lyric_transcriptions WHERE status IN ('complete','failed') AND updated_at<?", args: [time - 30 * 86400000] });
    return c.json({ ok: true });
  });
}