platform/src/servers/sidecar/music/index.ts

import { mkdirSync, writeFileSync } from 'node:fs';
import { join, basename } from 'node:path';
import type { SidecarCommand, SidecarEvent } from '../protocol';
import { createSidecarConnector } from '../connect';
import { streamAudioFile } from './stream-audio';
import { cliampUpgradeData, musicWebsocket } from './cliamp-ws';
import { ensurePulseAudio } from './pulse-audio';
import { startNightlyReindex, stopNightlyReindex } from './nightly-reindex';
import { startMusicWatcher, stopMusicWatcher } from './watcher';
import {
  reindexNow,
  reindexFull,
  ensureCacheSetup,
  MUSIC_ROOT,
  getIndexStatus,
  getManifest,
  albumVersion,
  metaFilePath,
  coverFilePath,
  discographyFilePath,
  posterFilePath,
  lyricsFilePath,
  onIndexProgress,
  buildReport,
} from './indexer';
import {
  getMusicFavorites,
  addMusicFavorite,
  removeMusicFavorite,
  getNowPlaying,
  setNowPlaying,
  clearNowPlaying,
  getPlaylists,
  getPlaylist,
  createPlaylist,
  renamePlaylist,
  deletePlaylist,
  addPlaylistItems,
  setPlaylistItems,
  type FavoriteKind,
} from 'officerdb';

const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');

// ── Per-user state validation ──
// The authenticated user id arrives in X-Officer-User (the platform proxy injects it after auth; we're
// loopback-only so we trust it). Favorite/playlist `key`s are opaque paths we never interpret.
const userIdOf = (req: Request): number | null => {
  const n = Number(req.headers.get('x-officer-user'));
  return Number.isInteger(n) && n > 0 ? n : null;
};
const FAVORITE_KINDS = new Set<FavoriteKind>(['track', 'album', 'artist']);
const isKind = (k: unknown): k is FavoriteKind => typeof k === 'string' && FAVORITE_KINDS.has(k as FavoriteKind);
const PLAYLIST_NAME_MAX = 200;
const cleanName = (v: unknown): string | null => {
  if (typeof v !== 'string') return null;
  const n = v.trim();
  return n && n.length <= PLAYLIST_NAME_MAX ? n : null;
};
const asKeys = (v: unknown): string[] | null =>
  Array.isArray(v) && v.every((k) => typeof k === 'string' && k) ? (v as string[]) : null;

// The music sidecar (officer-music). Same philosophy as the other officer-* sidecars: a singleton
// process that registers with the API server. It OWNS an audio-streaming HTTP server (all path
// resolution + streaming + ffprobe duration happens here); the platform API is just a thin proxy that
// authenticates and forwards to us. The server listens on a random loopback port, reported to the API
// on connect so it can route `/api/music/*` here.
//
// ─────────────────────────────────────────────────────────────────────────────────────────────────
// HTTP CONTRACT — the full `/api/music/*` surface (this fetch handler is the source of truth; the
// platform side is an opaque catch-all proxy). All routes are reached as `/api/music/<name>`, authed
// upstream by userMiddleware (Bearer header or `?token=` for media). Data shapes are the exported
// `IndexStatus` / `IndexReport` / `IndexMeta` types in indexer.ts.
//
//   GET  /stream?path=<home-relative>      audio with Range→206 (Content-Range/Length/Accept-Ranges)
//                                          + `X-Audio-Duration` (seconds, ffprobe). 400/404/416.
//   GET  /manifest                         pure read of the last completed index (NO build triggered) —
//                                          { version, generatedAt, albums: { "<rel>": { v, cover, tracks, videos?, disco? } } }
//   GET  /meta?path=<rel>                  album meta.json (IndexMeta). ETag: <v>; If-None-Match → 304.
//   GET  /cover?path=<rel>                 compressed cover.jpg.       ETag: <v>; If-None-Match → 304.
//   GET  /poster?path=<rel>&file=<video>   compressed video poster (frame grab). ETag: <v>; 304. 404 if none.
//   GET  /lyrics?path=<rel>&file=<track>   track lyrics text (X-Lyrics-Format: lrc|txt). ETag: <v>; 304. 404 if none.
//   GET  /image?path=<rel>&file=<img>      loose folder image bytes (image/*, the ORIGINAL). ETag: <v>; 304. 404 if none.
//   GET  /discography?path=<artist rel>    artist discography.json (only where manifest entry has
//                                          disco:true) = { artist, albums: { "<[year] album folder>":
//                                          "<Type>" } }, Type ∈ Studio/Live/Compilation/Single/EP/…
//                                          ETag: <v>; If-None-Match → 304. Source: each artist folder's
//                                          _discography.md (normalized; the md itself is never modified).
//   POST /reindex[?full=1]                 run the build to COMPLETION, then return the final IndexStatus.
//                                          default = incremental (skips unchanged); ?full=1 = full staged
//                                          rebuild + atomic swap (backfill a meta-format change).
//   GET  /reindex/status                   IndexStatus snapshot.
//   GET  /reindex/stream                   SSE. Triggers a build if idle (`?trigger=0` = watch-only).
//                                          `event: progress` (IndexStatus) throttled ~200ms, then one
//                                          `event: done` (IndexReport) and the stream closes.
//   GET  /health                           "ok".
//
//   ── Per-user state (USER data in Postgres, not library data). User id in X-Officer-User, injected by
//      the platform proxy after auth; `key`s are opaque paths (track homePath / album|artist rel). ──
//   GET    /favorites                       { tracks[], albums[], artists[] } (keys, newest first).
//   POST   /favorites { kind, key }         add (idempotent). kind ∈ track|album|artist.
//   DELETE /favorites?kind=&key=            remove.
//   GET    /now-playing[?device=]           last snapshot for that device, or null. device '' = default/phone, 'web' = browser.
//   PUT    /now-playing[?device=] { homePath, dir?, title?, artist?, album?, durationSec?, positionSec? }  upsert.
//   DELETE /now-playing[?device=]           clear that device's snapshot.
//   GET    /playlists                       [{ id, name, count, createdAt, updatedAt }] (recent first).
//   POST   /playlists { name }              create → 201 row; 409 if name taken.
//   GET    /playlists/:id                   { id, name, items:[keys], … }; 404 if not the user's.
//   PATCH  /playlists/:id { name }          rename; 404 / 409.
//   DELETE /playlists/:id                   delete (items cascade); 404.
//   POST   /playlists/:id/items { keys[] }  append → { count }; 404.
//   PUT    /playlists/:id/items { keys[] }  replace whole list (reorder/remove) → { count }; 404.
//
// `<rel>` = album folder path relative to the Music root (e.g. "Albums/AC-DC/[1980] Back in Black").
// `v` = per-album version stamp; unchanged `v` ⇒ nothing changed ⇒ the phone can skip re-downloading.
// ─────────────────────────────────────────────────────────────────────────────────────────────────

const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;

// ── Audio-streaming HTTP server ──

/** Grab an ephemeral free port by briefly binding one and releasing it. */
function getFreePort(): number {
  const probe = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: () => new Response('') });
  const port = probe.port;
  probe.stop(true);
  if (port == null) throw new Error('failed to acquire a free port');
  return port;
}

const port = getFreePort();

// Ensure the cache is a symlink-to-slot before serving/building, so full reindexes can swap atomically.
await ensureCacheSetup();

// Nightly full reindex at 3am (staged + atomic swap).
startNightlyReindex();

// Recursive watcher on ~/Music → localized reindex on any change.
startMusicWatcher();

// PulseAudio daemon + the `virtual_out` null sink both cliamp halves depend on. Officer used to do this at
// its own boot, which meant every restart of a process with no audio responsibilities re-checked the sink.
ensurePulseAudio();

const server = Bun.serve({
  port,
  hostname: '127.0.0.1',
  // Bun caps the server-level idleTimeout at 255s. Keep it there as the baseline; the build/stream
  // endpoints (which can idle for a whole from-scratch rebuild) extend it per-request via server.timeout.
  idleTimeout: 255,
  async fetch(req, server) {
    const url = new URL(req.url);

    // The two cliamp sockets. Officer has already authenticated the browser and is relaying frames; the
    // player and the capture themselves live here (cliamp-ws.ts).
    const wsData = cliampUpgradeData(url.pathname, url.searchParams);
    if (wsData) {
      if (server.upgrade(req, { data: wsData })) return undefined as unknown as Response;
      return new Response('Expected a WebSocket upgrade', { status: 400 });
    }

    // A from-scratch reindex can take many minutes with no bytes flowing on the triggering request.
    // Give the build endpoints a 30-min idle timeout so they aren't dropped (/manifest is a pure read now).
    if (url.pathname === '/reindex' || url.pathname === '/reindex/stream') {
      server.timeout(req, 1800);
    }
    const json = (data: unknown, init?: ResponseInit) =>
      new Response(JSON.stringify(data), {
        ...init,
        headers: { 'Content-Type': 'application/json', ...init?.headers },
      });
    // Version-stamped artifacts (cover/meta/…) and the manifest carry an ETag(=v) but must be REVALIDATED,
    // not served blind from the browser cache — otherwise a changed cover keeps showing the old image at
    // the same URL. `no-cache` = cache but always revalidate; the ETag/If-None-Match then makes it a cheap
    // 304 when nothing changed. (The platform proxy forwards If-None-Match so this works end-to-end.)
    const NO_CACHE = { 'Cache-Control': 'no-cache' } as const;

    if (url.pathname === '/health') return new Response('ok');

    // ── Streaming ──
    if (url.pathname === '/stream') {
      const path = url.searchParams.get('path');
      if (!path) return new Response('path is required', { status: 400 });
      return streamAudioFile(path, req.headers.get('range'));
    }

    // ── Index build ──
    if (url.pathname === '/reindex') {
      if (req.method !== 'POST') return new Response('Method not allowed', { status: 405 });
      // Run to completion, THEN respond — so the caller's manifest read right after is fresh. Both join an
      // in-flight build rather than starting a second.
      //   default : incremental — near-instant, skips unchanged albums by version stamp.
      //   ?full=1 : full from-scratch rebuild into a fresh slot, swapped in atomically (staged + safe) —
      //             use to backfill a meta-format change (e.g. a new track field) across the WHOLE library.
      const full = url.searchParams.get('full') === '1' || url.searchParams.get('full') === 'true';
      const result = await (full ? reindexFull() : reindexNow());
      return json(result);
    }
    if (url.pathname === '/reindex/status') return json(getIndexStatus());

    // SSE progress stream (for the app + the CLI). Triggers a build if idle (unless ?trigger=0), then
    // streams `progress` events until the build finishes, ending with a `done` event carrying the report.
    if (url.pathname === '/reindex/stream') {
      const trigger = url.searchParams.get('trigger') !== '0';
      if (trigger) void reindexNow();

      const encoder = new TextEncoder();
      const stream = new ReadableStream({
        start(controller) {
          let closed = false;
          let unsub = () => {};
          const send = (event: string, data: unknown) => {
            if (closed) return;
            try {
              controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`));
            } catch {
              /* stream closed */
            }
          };
          const finish = (s: ReturnType<typeof getIndexStatus>) => {
            send('done', buildReport(s));
            unsub();
            closed = true;
            try {
              controller.close();
            } catch {
              /* already closed */
            }
          };

          send('progress', getIndexStatus());
          const cur = getIndexStatus();
          if (!cur.running) {
            finish(cur); // nothing running → emit the last report and close
            return;
          }
          unsub = onIndexProgress((s) => {
            send('progress', s);
            if (!s.running && s.finishedAt) finish(s);
          });
        },
      });

      return new Response(stream, {
        headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' },
      });
    }

    // ── Sync surface ──
    if (url.pathname === '/manifest') {
      // Pure read — returns the last completed index. It does NOT trigger a build (that could kick off a
      // long/full rebuild on a plain app refresh); use POST /reindex explicitly to pick up disk changes.
      return json(await getManifest(), { headers: NO_CACHE });
    }

    // Video poster (a compressed frame grab). Keyed by the video's folder rel + its filename.
    if (url.pathname === '/poster') {
      const rel = url.searchParams.get('path');
      const file = url.searchParams.get('file');
      if (rel === null || !file) return new Response('path and file are required', { status: 400 });
      const posterPath = posterFilePath(rel, file);
      if (!posterPath) return new Response('Invalid path', { status: 400 });
      if (!(await Bun.file(posterPath).exists())) return new Response('Not found', { status: 404 });
      const v = await albumVersion(rel);
      if (v && req.headers.get('if-none-match') === v) return new Response(null, { status: 304, headers: NO_CACHE });
      return new Response(Bun.file(posterPath), {
        headers: { 'Content-Type': 'image/jpeg', ...NO_CACHE, ...(v ? { ETag: v } : {}) },
      });
    }

    // Track lyrics (external .lrc/.txt sidecar or embedded, cached during indexing). Try synced first.
    if (url.pathname === '/lyrics') {
      const rel = url.searchParams.get('path');
      const file = url.searchParams.get('file');
      if (rel === null || !file) return new Response('path and file are required', { status: 400 });
      for (const fmt of ['lrc', 'txt'] as const) {
        const p = lyricsFilePath(rel, file, fmt);
        if (p && (await Bun.file(p).exists())) {
          const v = await albumVersion(rel);
          if (v && req.headers.get('if-none-match') === v)
            return new Response(null, { status: 304, headers: NO_CACHE });
          return new Response(Bun.file(p), {
            headers: {
              'Content-Type': 'text/plain; charset=utf-8',
              'X-Lyrics-Format': fmt,
              ...NO_CACHE,
              ...(v ? { ETag: v } : {}),
            },
          });
        }
      }
      return new Response('Not found', { status: 404 });
    }

    // Folder image (band photo / booklet scan) — served as the ORIGINAL file from the library folder
    // (no cached artifact). `path` = music-relative folder, `file` = image name (basename'd for safety).
    if (url.pathname === '/image') {
      const rel = url.searchParams.get('path') ?? '';
      const file = basename(url.searchParams.get('file') ?? '');
      if (!file) return new Response('file is required', { status: 400 });
      const abs = join(MUSIC_ROOT, rel, file);
      if (abs !== MUSIC_ROOT && !abs.startsWith(MUSIC_ROOT + '/')) return new Response('Invalid path', { status: 400 });
      if (!(await Bun.file(abs).exists())) return new Response('Not found', { status: 404 });
      const v = await albumVersion(rel);
      if (v && req.headers.get('if-none-match') === v) return new Response(null, { status: 304, headers: NO_CACHE });
      const ext = file.slice(file.lastIndexOf('.') + 1).toLowerCase();
      const type =
        ext === 'png' ? 'image/png' : ext === 'webp' ? 'image/webp' : ext === 'gif' ? 'image/gif' : 'image/jpeg';
      return new Response(Bun.file(abs), { headers: { 'Content-Type': type, ...NO_CACHE, ...(v ? { ETag: v } : {}) } });
    }

    if (url.pathname === '/meta' || url.pathname === '/cover' || url.pathname === '/discography') {
      const rel = url.searchParams.get('path');
      if (rel === null) return new Response('path is required', { status: 400 });
      const spec = {
        '/meta': { file: metaFilePath(rel), type: 'application/json' },
        '/cover': { file: coverFilePath(rel), type: 'image/jpeg' },
        '/discography': { file: discographyFilePath(rel), type: 'application/json' },
      }[url.pathname]!;
      if (!spec.file) return new Response('Invalid path', { status: 400 });
      if (!(await Bun.file(spec.file).exists())) return new Response('Not found', { status: 404 });

      const v = await albumVersion(rel);
      if (v && req.headers.get('if-none-match') === v) return new Response(null, { status: 304, headers: NO_CACHE });
      return new Response(Bun.file(spec.file), {
        headers: { 'Content-Type': spec.type, ...NO_CACHE, ...(v ? { ETag: v } : {}) },
      });
    }

    // ── Per-user state (favorites / now-playing / playlists) ──────────────────────────────────────
    // USER data backed by Postgres (NOT library data). The user id comes from X-Officer-User (see above).
    const P = url.pathname;
    if (P === '/favorites' || P === '/now-playing' || P === '/playlists' || P.startsWith('/playlists/')) {
      const uid = userIdOf(req);
      if (uid === null) return json({ error: 'unauthenticated' }, { status: 401 });
      const m = req.method;

      if (P === '/favorites') {
        if (m === 'GET') return json(await getMusicFavorites(uid));
        if (m === 'POST') {
          const { kind, key } = (await req.json().catch(() => ({}))) as { kind?: unknown; key?: unknown };
          if (!isKind(kind) || typeof key !== 'string' || !key)
            return json({ error: 'kind and key required' }, { status: 400 });
          await addMusicFavorite(uid, kind, key);
          return json({ ok: true });
        }
        if (m === 'DELETE') {
          const kind = url.searchParams.get('kind');
          const key = url.searchParams.get('key');
          if (!isKind(kind) || !key) return json({ error: 'kind and key required' }, { status: 400 });
          await removeMusicFavorite(uid, kind, key);
          return json({ ok: true });
        }
        return new Response('Method not allowed', { status: 405 });
      }

      if (P === '/now-playing') {
        // Per-client resume state: `?device=` tags the caller ('' = default/phone, 'web' = the browser)
        // so each keeps its own now-playing instead of sharing one row. Missing → '' (back-compat).
        const device = url.searchParams.get('device') ?? '';
        if (m === 'GET') return json(await getNowPlaying(uid, device));
        if (m === 'PUT') {
          const b = (await req.json().catch(() => ({}))) as Record<string, unknown>;
          if (typeof b.homePath !== 'string' || !b.homePath)
            return json({ error: 'homePath required' }, { status: 400 });
          const str = (v: unknown) => (typeof v === 'string' ? v : undefined);
          const num = (v: unknown) => (typeof v === 'number' && Number.isFinite(v) ? v : undefined);
          await setNowPlaying(uid, device, {
            homePath: b.homePath,
            dir: str(b.dir),
            title: str(b.title),
            artist: str(b.artist),
            album: str(b.album),
            durationSec: num(b.durationSec),
            positionSec: num(b.positionSec),
          });
          return json({ ok: true });
        }
        if (m === 'DELETE') {
          await clearNowPlaying(uid, device);
          return json({ ok: true });
        }
        return new Response('Method not allowed', { status: 405 });
      }

      if (P === '/playlists') {
        if (m === 'GET') return json(await getPlaylists(uid));
        if (m === 'POST') {
          const { name } = (await req.json().catch(() => ({}))) as { name?: unknown };
          const n = cleanName(name);
          if (!n) return json({ error: 'name required (1-200 chars)' }, { status: 400 });
          const row = await createPlaylist(uid, n);
          return row
            ? json(row, { status: 201 })
            : json({ error: 'a playlist with that name already exists' }, { status: 409 });
        }
        return new Response('Method not allowed', { status: 405 });
      }

      // /playlists/:id  and  /playlists/:id/items
      const match = P.match(/^\/playlists\/(\d+)(\/items)?$/);
      if (!match) return json({ error: 'not found' }, { status: 404 });
      const id = Number(match[1]);

      if (match[2]) {
        // /playlists/:id/items — POST append, PUT replace (reorder/remove)
        if (m !== 'POST' && m !== 'PUT') return new Response('Method not allowed', { status: 405 });
        const { keys } = (await req.json().catch(() => ({}))) as { keys?: unknown };
        const ks = asKeys(keys);
        if (!ks) return json({ error: 'keys[] required' }, { status: 400 });
        const count = await (m === 'POST' ? addPlaylistItems(uid, id, ks) : setPlaylistItems(uid, id, ks));
        return count === null ? json({ error: 'not found' }, { status: 404 }) : json({ count });
      }

      if (m === 'GET') {
        const pl = await getPlaylist(uid, id);
        return pl ? json(pl) : json({ error: 'not found' }, { status: 404 });
      }
      if (m === 'PATCH') {
        const { name } = (await req.json().catch(() => ({}))) as { name?: unknown };
        const n = cleanName(name);
        if (!n) return json({ error: 'name required (1-200 chars)' }, { status: 400 });
        if (!(await getPlaylist(uid, id))) return json({ error: 'not found' }, { status: 404 });
        const ok = await renamePlaylist(uid, id, n);
        return ok ? json({ ok: true }) : json({ error: 'a playlist with that name already exists' }, { status: 409 });
      }
      if (m === 'DELETE') {
        const ok = await deletePlaylist(uid, id);
        return ok ? json({ ok: true }) : json({ error: 'not found' }, { status: 404 });
      }
      return new Response('Method not allowed', { status: 405 });
    }

    return new Response('Not found', { status: 404 });
  },
  websocket: musicWebsocket,
});

console.log(`[music] audio server listening on http://127.0.0.1:${port}`);

// Write the port to a well-known file so local tooling (scripts/reindex-music.ts) can find the server.
try {
  mkdirSync(join(DATA_PATH, 'music'), { recursive: true });
  writeFileSync(join(DATA_PATH, 'music', '.server'), String(port));
} catch {
  /* best-effort */
}

// ── Command handlers ──

type ReplyFn = (msg: SidecarEvent) => void;

function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
  switch (cmd.type) {
    case 'ping':
      reply({ type: 'pong', id: cmd.id });
      break;

    default:
      reply({
        type: 'error',
        id: (cmd as SidecarCommand).id,
        error: `Unknown command type: ${(cmd as Record<string, unknown>).type}`,
      });
  }
}

// ── Connect to API server ──

const connection = createSidecarConnector({
  apiUrl: `${API_URL}/api/sidecar/register`,
  name: 'music',
  capabilities: ['music'],
  onCommand(cmd, reply) {
    handleCommand(cmd as SidecarCommand, reply as ReplyFn);
  },
  onConnected() {
    // Tell the API where our audio server is listening, so it can proxy /api/music/* here.
    connection.send({ type: 'music:server', port });
    console.log(`[music] reported audio server port ${port} to API`);
  },
});

// ── Graceful shutdown ──

function shutdown(signal: string) {
  console.log(`[music] ${signal} received, shutting down...`);
  stopNightlyReindex();
  stopMusicWatcher();
  try {
    server.stop(true);
  } catch {
    /* already stopped */
  }
  connection.destroy();
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));