music: SSE reindex progress stream + reindex-music CLI

Adds a live progress channel for the library index:
- indexer.ts: progress subscribers (onIndexProgress) + throttled emit during the
  walk, and buildReport() for a final summary.
- sidecar: GET /reindex/stream (SSE) — triggers a build if idle (?trigger=0 to
  watch only), streams `progress` events, ends with a `done` event carrying the
  report; auto-proxied at /api/music/reindex/stream for the app. Sidecar also
  writes DATA_PATH/music/.server (its port) for local tooling.
- scripts/reindex-music.ts: CLI that reads the port file, follows the SSE, prints
  live progress + a final report. Run: bun scripts/reindex-music.ts

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-26 03:18:41 +00:00
co-authored by Claude Opus 4.8
parent b6b9e21b72
commit 6224f6be51
3 changed files with 234 additions and 0 deletions
+62
View File
@@ -1,3 +1,5 @@
import { mkdirSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import type { SidecarCommand, SidecarEvent } from '../protocol';
import { createSidecarConnector } from '../connect';
import { streamAudioFile } from './stream-audio';
@@ -8,8 +10,12 @@ import {
albumVersion,
metaFilePath,
coverFilePath,
onIndexProgress,
buildReport,
} from './indexer';
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
// 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
@@ -56,6 +62,54 @@ const server = Bun.serve({
}
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 && !getIndexStatus().running) void buildMusicIndex();
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') return json(await getManifest());
@@ -83,6 +137,14 @@ const server = Bun.serve({
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;
+54
View File
@@ -101,6 +101,57 @@ export function getIndexStatus(): IndexStatus {
return { ...status };
}
export type IndexReport = {
albums: number; // albums with content (built + skipped)
built: number;
skipped: number;
foldersScanned: number;
tracksIndexed: number;
coversSaved: number;
elapsedSec: number;
error: string | null;
};
export function buildReport(s: IndexStatus = status): IndexReport {
const elapsed = s.startedAt && s.finishedAt ? (s.finishedAt - s.startedAt) / 1000 : 0;
return {
albums: s.albumsBuilt + s.albumsSkipped,
built: s.albumsBuilt,
skipped: s.albumsSkipped,
foldersScanned: s.foldersScanned,
tracksIndexed: s.tracksIndexed,
coversSaved: s.coversSaved,
elapsedSec: Math.round(elapsed * 10) / 10,
error: s.error,
};
}
// ── Progress subscribers (for SSE) ──
type ProgressListener = (s: IndexStatus) => void;
const listeners = new Set<ProgressListener>();
let lastEmit = 0;
export function onIndexProgress(cb: ProgressListener): () => void {
listeners.add(cb);
return () => listeners.delete(cb);
}
/** Notify subscribers of progress; throttled to ~200ms unless `force` (e.g. terminal 'done'). */
function emitProgress(force = false): void {
const now = Date.now();
if (!force && now - lastEmit < 200) return;
lastEmit = now;
const snap = getIndexStatus();
for (const cb of listeners) {
try {
cb(snap);
} catch {
/* ignore listener errors */
}
}
}
// ── Helpers ──
function isAudio(name: string): boolean {
@@ -226,6 +277,7 @@ export async function buildMusicIndex(): Promise<IndexStatus> {
status.running = false;
status.finishedAt = Date.now();
status.currentPath = '';
emitProgress(true); // final push — signals 'done' to SSE subscribers
}
return getIndexStatus();
}
@@ -233,6 +285,7 @@ export async function buildMusicIndex(): Promise<IndexStatus> {
async function walk(dirAbs: string, prev: Manifest, next: Manifest): Promise<void> {
status.currentPath = relative(MUSIC_ROOT, dirAbs) || '.';
status.foldersScanned += 1;
emitProgress();
let entries: import('node:fs').Dirent[];
try {
@@ -276,6 +329,7 @@ async function walk(dirAbs: string, prev: Manifest, next: Manifest): Promise<voi
const tracks = await mapPool(audio, TRACK_CONCURRENCY, async (name) => {
const t = await ffprobeTrack(join(dirAbs, name), name);
status.tracksIndexed += 1;
emitProgress();
return t;
});
const meta: IndexMeta = { path: rel, cover: coverName ? 'cover.jpg' : undefined, tracks };