whisper_web/src/server/transcribe.ts

/**
 * Server-side transcription engine (M5, Decisions.md D-013): bridges the
 * faster-whisper Python sidecar (server-py/sidecar.py) to the streamed
 * JSON-lines HTTP contract (D-011, wire format in src/lib/engine.ts).
 *
 * Routes (wired in src/index.ts, auth-guarded — the server engine is the one
 * logged-in-only feature, Architecture.md §5):
 *   GET  /api/transcribe/status → 200 {available, reason?} | 401
 *   POST /api/transcribe?model=<catalog id>[&language=..][&task=..]
 *        body: raw float32 mono 16 kHz PCM (application/octet-stream)
 *        → 200 application/x-ndjson (segment* then result | error lines)
 *        | 400 | 401 | 413 | 503 {error}
 *
 * The sidecar is spawned lazily on first use and supervised: a crash rejects
 * the requests it was carrying and the next request respawns it, while a
 * missing runtime (no python3, faster-whisper not installed) is remembered as
 * a permanent "unavailable" reason — fixing that requires a deployment
 * change, so we never busy-respawn. Either way the browser/anonymous path is
 * untouched: this module holds the only Python dependency in the app.
 *
 * Warm-up strategy (WorkBreakdown M5): the sidecar keeps the last-used model
 * loaded between requests, so only the first request for a model (or a model
 * switch) pays the load; weights download once into the sidecar's cache.
 */

import { tmpdir } from "node:os";
import { join } from "node:path";
import type { TranscribeStreamLine } from "../lib/engine";
import { HINT_MAX_CHARS, normalizeHint } from "../lib/hint";
import { getFasterWhisperModel } from "../lib/models";
import type { TranscriptChunk, WhisperTask } from "../worker/protocol";
import { getSessionUser as appGetSessionUser, type SessionUser } from "./auth.ts";

// ---------------------------------------------------------------------------
// Sidecar interface (injectable — tests use a fake, see tests/transcribe.test.ts)
// ---------------------------------------------------------------------------

export interface SidecarSegment {
  start: number;
  end: number;
  text: string;
}

export interface SidecarTranscribeRequest {
  /** faster-whisper model name (catalog mapping, src/lib/models.ts). */
  model: string;
  /** Path to a file of raw float32 mono 16 kHz PCM. */
  audioPath: string;
  language?: string;
  task?: WhisperTask;
  /**
   * "file" (default): whole uploads — batched inference over VAD speech
   * chunks. "live": short server-live passes — sequential, no VAD (D-WS5a).
   */
  mode?: SidecarMode;
  /**
   * Normalized "Names and terms" hint (D-WS2a); the sidecar truncates it to
   * Whisper's prompt budget with faster-whisper's own tokenizer.
   */
  prompt?: string;
}

export type SidecarMode = "file" | "live";

export interface SidecarStatus {
  available: boolean;
  reason?: string;
}

export interface Sidecar {
  /** Spawns the process if needed; reports whether inference can run. */
  status(): Promise<SidecarStatus>;
  /**
   * Run one transcription, streaming segments as they decode. Resolves after
   * the sidecar's "done" (cancelled: true when `signal` fired first);
   * rejects on sidecar errors or crashes.
   */
  transcribe(
    request: SidecarTranscribeRequest,
    onSegment: (segment: SidecarSegment) => void,
    signal?: AbortSignal,
  ): Promise<{ cancelled: boolean }>;
}

// ---------------------------------------------------------------------------
// The real sidecar: Bun.spawn + JSON-lines over stdio (D-013)
// ---------------------------------------------------------------------------

const SIDECAR_SCRIPT = join(import.meta.dir, "../../server-py/sidecar.py");

/** How long the spawned process gets to import faster-whisper and report. */
const READY_TIMEOUT_MS = 60_000;

interface SidecarProcess {
  proc: ReturnType<typeof Bun.spawn>;
  ready: Promise<void>;
}

interface PendingTranscription {
  onSegment: (segment: SidecarSegment) => void;
  resolve: (outcome: { cancelled: boolean }) => void;
  reject: (error: Error) => void;
}

export function createSidecar(command?: readonly string[]): Sidecar {
  // "python3" resolved per spawn — SIDECAR_PYTHON overrides for exotic setups.
  const cmd = command ?? [process.env.SIDECAR_PYTHON || "python3", SIDECAR_SCRIPT];

  let current: SidecarProcess | null = null;
  /** Deployment-level failure (no python, deps missing): never respawn. */
  let permanentFailure: string | null = null;
  const pending = new Map<string, PendingTranscription>();
  let nextRequestId = 0;

  function failAllPending(error: Error): void {
    const requests = [...pending.values()];
    pending.clear();
    for (const request of requests) request.reject(error);
  }

  function routeLine(rawLine: string): void {
    let message: Record<string, unknown>;
    try {
      message = JSON.parse(rawLine) as Record<string, unknown>;
    } catch {
      return; // stray non-JSON output must not kill the bridge
    }
    const request = pending.get(String(message.id));
    if (!request) return;
    switch (message.type) {
      case "segment":
        if (
          typeof message.start === "number" &&
          typeof message.end === "number" &&
          typeof message.text === "string"
        ) {
          request.onSegment({
            start: message.start,
            end: message.end,
            text: message.text,
          });
        }
        break;
      case "done":
        pending.delete(String(message.id));
        request.resolve({ cancelled: message.cancelled === true });
        break;
      case "error":
        pending.delete(String(message.id));
        request.reject(new Error(String(message.message ?? "sidecar error")));
        break;
    }
  }

  function spawn(): SidecarProcess {
    const proc = Bun.spawn({
      cmd: [...cmd],
      stdin: "pipe",
      stdout: "pipe",
      stderr: "inherit", // model download progress / tracebacks → server log
    });

    const ready = new Promise<void>((resolve, reject) => {
      let settled = false;
      const timeout = setTimeout(() => {
        if (settled) return;
        settled = true;
        proc.kill();
        reject(new Error("the transcription sidecar did not start in time"));
      }, READY_TIMEOUT_MS);

      void (async () => {
        // First line decides readiness; everything after is request traffic.
        const decoder = new TextDecoder();
        let buffer = "";
        for await (const chunk of proc.stdout as ReadableStream<Uint8Array>) {
          buffer += decoder.decode(chunk, { stream: true });
          let newline: number;
          while ((newline = buffer.indexOf("\n")) !== -1) {
            const line = buffer.slice(0, newline);
            buffer = buffer.slice(newline + 1);
            if (!settled) {
              settled = true;
              clearTimeout(timeout);
              let handshake: Record<string, unknown> = {};
              try {
                handshake = JSON.parse(line) as Record<string, unknown>;
              } catch {
                // fall through — treated as a bad handshake below
              }
              if (handshake.type === "ready") {
                resolve();
              } else {
                // "fatal": faster-whisper missing — a deployment problem.
                permanentFailure = String(
                  handshake.message ?? "the transcription sidecar failed to start",
                );
                reject(new Error(permanentFailure));
              }
            } else {
              routeLine(line);
            }
          }
        }
        // stdout closed: the process exited (crash or shutdown).
        if (!settled) {
          settled = true;
          clearTimeout(timeout);
          reject(new Error("the transcription sidecar exited during startup"));
        }
        failAllPending(new Error("the transcription sidecar exited unexpectedly"));
        if (current?.proc === proc) current = null; // next request respawns
      })();
    });
    // A failed startup must not surface as an unhandled rejection when nobody
    // is awaiting yet (e.g. crash between requests).
    ready.catch(() => {});
    return { proc, ready };
  }

  async function ensure(): Promise<SidecarProcess> {
    if (permanentFailure) throw new Error(permanentFailure);
    if (current) return current;
    try {
      current = spawn();
    } catch (error) {
      // Bun.spawn throws synchronously when the binary is missing.
      permanentFailure = `python3 is not available on this server (${
        error instanceof Error ? error.message : String(error)
      })`;
      throw new Error(permanentFailure);
    }
    return current;
  }

  function send(proc: ReturnType<typeof Bun.spawn>, message: unknown): void {
    const stdin = proc.stdin as { write(data: string): unknown; flush(): unknown };
    stdin.write(`${JSON.stringify(message)}\n`);
    stdin.flush();
  }

  return {
    async status() {
      try {
        const sidecar = await ensure();
        await sidecar.ready;
        return { available: true };
      } catch (error) {
        return {
          available: false,
          reason: error instanceof Error ? error.message : String(error),
        };
      }
    },

    async transcribe(request, onSegment, signal) {
      const sidecar = await ensure();
      await sidecar.ready;
      const id = `req-${++nextRequestId}`;
      return new Promise<{ cancelled: boolean }>((resolve, reject) => {
        pending.set(id, { onSegment, resolve, reject });
        signal?.addEventListener(
          "abort",
          () => {
            // The sidecar checks between segments and still answers "done"
            // (cancelled: true), which resolves this promise normally.
            if (pending.has(id) && current?.proc === sidecar.proc) {
              send(sidecar.proc, { type: "cancel", id });
            }
          },
          { once: true },
        );
        try {
          send(sidecar.proc, {
            type: "transcribe",
            id,
            model: request.model,
            audio_path: request.audioPath,
            language: request.language ?? null,
            task: request.task ?? "transcribe",
            mode: request.mode ?? "file",
            prompt: request.prompt ?? null,
          });
        } catch (error) {
          pending.delete(id);
          reject(error instanceof Error ? error : new Error(String(error)));
        }
      });
    },
  };
}

// ---------------------------------------------------------------------------
// HTTP handlers
// ---------------------------------------------------------------------------

/** Default cap ≈ 2¼ hours of raw float32 16 kHz mono (64 kB/s). */
const DEFAULT_MAX_UPLOAD_BYTES = 512 * 1024 * 1024;

/**
 * Effective upload cap (env-overridable). Exported so src/index.ts can raise
 * Bun.serve's `maxRequestBodySize` to match — its 128 MB default would
 * otherwise reject uploads longer than ~35 minutes of audio before this
 * module ever sees them.
 */
export const MAX_UPLOAD_BYTES =
  Number(process.env.TRANSCRIBE_MAX_UPLOAD_BYTES) || DEFAULT_MAX_UPLOAD_BYTES;

/** Keep the ndjson stream visibly alive through quiet stretches (D-021). */
const HEARTBEAT_INTERVAL_MS = 5_000;
const HEARTBEAT = new TextEncoder().encode("\n");

const LANGUAGE_RE = /^[a-z]{2,3}$/;

export interface TranscribeHandlers {
  status(request: Request): Promise<Response>;
  post(request: Request): Promise<Response>;
}

export interface TranscribeDeps {
  sidecar: Sidecar;
  getSessionUser(request: Request): SessionUser | null;
  maxUploadBytes?: number;
  /** Directory for uploaded-PCM temp files. Default: the OS temp dir. */
  tempDir?: string;
  /** Blank-line heartbeat period on the ndjson stream (D-021). Default 5 s. */
  heartbeatIntervalMs?: number;
}

function json(body: unknown, status: number): Response {
  return new Response(JSON.stringify(body), {
    status,
    headers: { "Content-Type": "application/json" },
  });
}

interface TranscribeParams {
  modelId: string;
  fasterWhisperModel: string;
  language?: string;
  task: WhisperTask;
  prompt?: string;
}

/**
 * Parse + validate the `language`/`task`/`prompt` query params shared by both
 * the HTTP (`POST /api/transcribe`) and WebSocket (`GET /api/transcribe/live`,
 * src/server/live.ts) entry points, so a live session can't forward an
 * unvalidated string into the sidecar protocol just because it skips
 * `parseParams` below (D-032). A string return is the 400 error text.
 *
 * `prompt` is the raw "Names and terms" hint (D-WS2a): at most
 * HINT_MAX_CHARS (the UI's textarea limit), normalized exactly as the browser
 * worker does; empty after normalization = no prompt.
 */
export function parseLanguageAndTask(
  url: URL,
): { language?: string; task: WhisperTask; prompt?: string } | string {
  const rawLanguage = url.searchParams.get("language");
  let language: string | undefined;
  if (rawLanguage !== null && rawLanguage !== "" && rawLanguage !== "auto") {
    if (!LANGUAGE_RE.test(rawLanguage)) return `invalid language: ${rawLanguage}`;
    language = rawLanguage;
  }

  const rawTask = url.searchParams.get("task") ?? "transcribe";
  if (rawTask !== "transcribe" && rawTask !== "translate") {
    return `invalid task: ${rawTask}`;
  }

  const rawPrompt = url.searchParams.get("prompt") ?? "";
  if (rawPrompt.length > HINT_MAX_CHARS) {
    return `prompt too long (max ${HINT_MAX_CHARS} characters)`;
  }
  const prompt = normalizeHint(rawPrompt);

  return { language, task: rawTask, ...(prompt && { prompt }) };
}

/** Parse + validate query params; a string is the 400 error text. */
function parseParams(url: URL): TranscribeParams | string {
  const modelId = url.searchParams.get("model");
  if (!modelId) return "missing model parameter";
  const fasterWhisperModel = getFasterWhisperModel(modelId);
  // Catalog-only (D-006/D-013): never forward arbitrary ids to the sidecar.
  if (!fasterWhisperModel) return `unknown model: ${modelId}`;

  const options = parseLanguageAndTask(url);
  if (typeof options === "string") return options;

  return { modelId, fasterWhisperModel, ...options };
}

/**
 * Stream the request body to `path`, enforcing the size limit even for
 * chunked bodies. Returns bytes written, or a Response to answer with.
 */
async function writeBodyToFile(
  request: Request,
  path: string,
  maxBytes: number,
): Promise<number | Response> {
  const declared = Number(request.headers.get("content-length"));
  if (Number.isFinite(declared) && declared > maxBytes) {
    return json({ error: `audio exceeds the ${maxBytes}-byte upload limit` }, 413);
  }
  if (!request.body) return json({ error: "missing request body" }, 400);

  const sink = Bun.file(path).writer();
  let written = 0;
  try {
    for await (const chunk of request.body) {
      written += chunk.byteLength;
      if (written > maxBytes) {
        return json({ error: `audio exceeds the ${maxBytes}-byte upload limit` }, 413);
      }
      sink.write(chunk);
    }
  } finally {
    await sink.end();
  }
  if (written === 0) return json({ error: "missing request body" }, 400);
  if (written % 4 !== 0) {
    return json({ error: "body must be raw float32 PCM (length % 4 == 0)" }, 400);
  }
  return written;
}

export function createTranscribeHandlers(deps: TranscribeDeps): TranscribeHandlers {
  const maxUploadBytes = deps.maxUploadBytes ?? DEFAULT_MAX_UPLOAD_BYTES;
  const tempDir = deps.tempDir ?? tmpdir();
  const heartbeatIntervalMs = deps.heartbeatIntervalMs ?? HEARTBEAT_INTERVAL_MS;

  return {
    async status(request) {
      if (!deps.getSessionUser(request)) {
        return json({ error: "not authenticated" }, 401);
      }
      const status = await deps.sidecar.status();
      return json(status, 200);
    },

    async post(request) {
      if (!deps.getSessionUser(request)) {
        return json({ error: "not authenticated" }, 401);
      }
      const params = parseParams(new URL(request.url));
      if (typeof params === "string") return json({ error: params }, 400);

      // Refuse before consuming a potentially huge body when inference can't
      // run anyway (degrade gracefully — D-013).
      const status = await deps.sidecar.status();
      if (!status.available) {
        return json({ error: status.reason ?? "server engine unavailable" }, 503);
      }

      const audioPath = join(tempDir, `whisper-upload-${crypto.randomUUID()}.f32`);
      const written = await writeBodyToFile(request, audioPath, maxUploadBytes);
      if (written instanceof Response) {
        await Bun.file(audioPath).unlink().catch(() => {});
        return written;
      }

      // 200 with an ndjson stream (D-011); failures beyond this point travel
      // in-stream as an "error" line since the status is already committed.
      const encoder = new TextEncoder();
      const stream = new ReadableStream<Uint8Array>({
        async start(controller) {
          const enqueue = (line: TranscribeStreamLine) =>
            controller.enqueue(encoder.encode(`${JSON.stringify(line)}\n`));
          // Segment gaps can exceed any proxy/server idle timeout (model
          // load/download before the first segment; slow decode windows on
          // big models — 30 s+ observed on a 15-min file). Heartbeat blank
          // lines keep the connection visibly alive; ServerEngine's line
          // reader skips empty lines. A dead client makes the enqueue throw,
          // which is caught here (request.signal separately cancels the
          // sidecar pass).
          const heartbeat = setInterval(() => {
            try {
              controller.enqueue(HEARTBEAT);
            } catch {
              clearInterval(heartbeat);
            }
          }, heartbeatIntervalMs);
          const chunks: TranscriptChunk[] = [];
          try {
            await deps.sidecar.transcribe(
              {
                model: params.fasterWhisperModel,
                audioPath,
                language: params.language,
                task: params.task,
                prompt: params.prompt,
              },
              (segment) => {
                chunks.push({
                  text: segment.text,
                  timestamp: [segment.start, segment.end],
                });
                enqueue({
                  type: "segment",
                  text: segment.text,
                  start: segment.start,
                  end: segment.end,
                });
              },
              // Client gone (abort/refresh) → cancel inference between segments.
              request.signal,
            );
            enqueue({
              type: "result",
              text: chunks.map((chunk) => chunk.text).join("").trim(),
              chunks,
            });
          } catch (error) {
            enqueue({
              type: "error",
              message: error instanceof Error ? error.message : String(error),
            });
          } finally {
            clearInterval(heartbeat);
            await Bun.file(audioPath).unlink().catch(() => {});
            try {
              controller.close();
            } catch {
              // Already errored/cancelled — the temp file cleanup above is
              // what actually matters here.
            }
          }
        },
      });

      return new Response(stream, {
        headers: {
          "Content-Type": "application/x-ndjson; charset=utf-8",
          "Cache-Control": "no-store",
        },
      });
    },
  };
}

// ---------------------------------------------------------------------------
// Default instance (real sidecar + app sessions), wired in src/index.ts
// ---------------------------------------------------------------------------

export const sidecar = createSidecar();

export const transcribe: TranscribeHandlers = createTranscribeHandlers({
  sidecar,
  getSessionUser: appGetSessionUser,
  maxUploadBytes: MAX_UPLOAD_BYTES,
});