fix(ingest): report YouTube bot blocks accurately

.env.example

@@ -33,6 +33,7 @@ EVAL_USER_ID=""
 # so a fresh checkout can never pay; flip it per run when you intend to use it.
 ASSEMBLYAI_API_KEY=""
 TRANSCRIBE_FALLBACK="false"
+SUPADATA_API_KEY=""
 
 # Clerk — auth for the web app. Optional at build/typecheck time (see src/env.js) so a fresh
 # checkout builds without keys, but the authenticated app cannot run without them.

README.md

@@ -110,6 +110,12 @@ library unless ownership is assigned separately. Use the Sources page for the no
 AssemblyAI transcription is an opt-in fallback for videos without captions. Set
 `TRANSCRIBE_FALLBACK=true` only when you intend to use it.
 
+For hosted imports, set `SUPADATA_API_KEY` on the server. Nexus requests existing
+timestamped captions with Supadata's `mode=native` (one API credit per video);
+it does not request Supadata AI transcription. Without a key, local runs fetch
+captions directly from YouTube. Keep `TRANSCRIBE_FALLBACK=false` on Vercel while
+YouTube blocks its audio downloads.
+
 ## Quality checks
 
 ```bash

src/env.js

@@ -55,6 +55,8 @@ export const env = createEnv({
     TRANSCRIBE_FALLBACK: booleanFlag("false"),
     /** AssemblyAI key, direct — the AI Gateway does not route transcription. */
     ASSEMBLYAI_API_KEY: z.string().optional(),
+    /** Hosted native YouTube captions for serverless imports. */
+    SUPADATA_API_KEY: z.string().optional(),
     /**
      * Clerk secret key. Optional at build/typecheck time so a fresh checkout without Clerk
      * keys still builds and tests; the authed web app cannot actually run without it. Clerk's
@@ -89,6 +91,7 @@ export const env = createEnv({
     EVAL_USER_ID: process.env.EVAL_USER_ID,
     TRANSCRIBE_FALLBACK: process.env.TRANSCRIBE_FALLBACK,
     ASSEMBLYAI_API_KEY: process.env.ASSEMBLYAI_API_KEY,
+    SUPADATA_API_KEY: process.env.SUPADATA_API_KEY,
     CLERK_SECRET_KEY: process.env.CLERK_SECRET_KEY,
     NEXT_PUBLIC_CLERK_PUBLISHABLE_KEY:
       process.env.NEXT_PUBLIC_CLERK_PUBLISHABLE_KEY,

src/server/api/routers/imports.test.ts

@@ -52,7 +52,7 @@ describe("imports router ownership boundary", () => {
       channelId: "UC123",
       creatorHandle: "creator",
       displayName: "Creator",
-      importLimit: 50,
+      importLimit: 3,
     });
 
     await expect(caller("user-a").preview({ scope: " @creator " })).resolves.toMatchObject({

src/server/domain/channel-import.ts

@@ -6,7 +6,7 @@ import type {
 export type ImportJobStatus = ChannelImport["status"];
 export type ImportItemStatus = ChannelImportItem["status"];
 
-export const CHANNEL_IMPORT_LIMIT = 50;
+export const CHANNEL_IMPORT_LIMIT = 3;
 
 export type ImportSummary = {
   discovered: number;

src/server/imports/channel-import.test.ts

@@ -79,7 +79,7 @@ describe("channel import launch", () => {
       channelId: "UC123",
       creatorHandle: "creator",
       displayName: "Creator Name",
-      importLimit: 50,
+      importLimit: 3,
     });
     expect(mocks.discover).toHaveBeenCalledWith("@Creator", 1);
     expect(mocks.start).not.toHaveBeenCalled();

src/server/ingest/channel.test.ts

@@ -1,4 +1,7 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
+
+const mockEnv = vi.hoisted(() => ({ SUPADATA_API_KEY: undefined as string | undefined }));
+vi.mock("~/env", () => ({ env: mockEnv }));
 
 import type { SourceLoader, SourceRef } from "~/server/domain/types";
 import { IN_FLIGHT, ingestChannel, summarize } from "~/server/ingest/channel";
@@ -63,6 +66,27 @@ describe("ingestChannel — per-video isolation, discovery order", () => {
     expect(peak).toBe(IN_FLIGHT);
   });
 
+  it("runs one video at a time when Supadata is configured", async () => {
+    mockEnv.SUPADATA_API_KEY = "test-key";
+    let running = 0;
+    let peak = 0;
+    try {
+      await ingestChannel("@x", {
+        loader: loaderOf(["1", "2", "3"]),
+        ingest: async (r) => {
+          running++;
+          peak = Math.max(peak, running);
+          await new Promise((resolve) => setTimeout(resolve, 5));
+          running--;
+          return ingestedResult(r);
+        },
+      });
+      expect(peak).toBe(1);
+    } finally {
+      mockEnv.SUPADATA_API_KEY = undefined;
+    }
+  });
+
   it("passes --limit through to discovery", async () => {
     const outcomes = await ingestChannel("@x", {
       limit: 2,

src/server/ingest/channel.ts

@@ -1,5 +1,6 @@
 import pMap from "p-map";
 
+import { env } from "~/env";
 import type { ChannelScope, SourceLoader, SourceRef } from "~/server/domain/types";
 import type { IngestResult } from "~/server/ingest/pipeline";
 
@@ -27,7 +28,7 @@ export async function ingestChannel(
         return { status: "failed", ref, error };
       }
     },
-    { concurrency: IN_FLIGHT },
+    { concurrency: env.SUPADATA_API_KEY ? 1 : IN_FLIGHT },
   );
 }
 

src/server/ingest/stt.test.ts

@@ -1,30 +1,6 @@
-import {
-  YoutubeTranscriptDisabledError,
-  YoutubeTranscriptError,
-  YoutubeTranscriptNotAvailableError,
-  YoutubeTranscriptNotAvailableLanguageError,
-  YoutubeTranscriptTooManyRequestError,
-  YoutubeTranscriptVideoUnavailableError,
-} from "youtube-transcript";
 import { describe, expect, it } from "vitest";
 
-import { assertWithinCap, isNoCaptions, MAX_STT_MINUTES, toSegments } from "~/server/ingest/stt";
-
-describe("STT trigger — only genuine no-captions may lead to a paid call", () => {
-  it("fires on disabled / no track / no usable language", () => {
-    expect(isNoCaptions(new YoutubeTranscriptDisabledError("v"))).toBe(true);
-    expect(isNoCaptions(new YoutubeTranscriptNotAvailableError("v"))).toBe(true);
-    expect(isNoCaptions(new YoutubeTranscriptNotAvailableLanguageError("en", ["hi"], "v"))).toBe(true);
-  });
-
-  it("never fires on throttle, private/removed, or anything unknown", () => {
-    expect(isNoCaptions(new YoutubeTranscriptTooManyRequestError())).toBe(false);
-    expect(isNoCaptions(new YoutubeTranscriptVideoUnavailableError("v"))).toBe(false);
-    expect(isNoCaptions(new YoutubeTranscriptError("boom"))).toBe(false);
-    expect(isNoCaptions(new TypeError("fetch failed"))).toBe(false);
-    expect(isNoCaptions(undefined)).toBe(false);
-  });
-});
+import { assertWithinCap, MAX_STT_MINUTES, toSegments } from "~/server/ingest/stt";
 
 describe("STT duration cap", () => {
   it("rejects a known duration over the cap, names the cap", () => {

src/server/ingest/stt.ts

@@ -1,19 +1,13 @@
 import { createAssemblyAI } from "@ai-sdk/assemblyai";
 import { transcribe } from "ai";
-import {
-  YoutubeTranscriptDisabledError,
-  YoutubeTranscriptNotAvailableError,
-  YoutubeTranscriptNotAvailableLanguageError,
-} from "youtube-transcript";
 
 import { env } from "~/env";
 import type { Segment, SttProvider } from "~/server/domain/types";
 
 /**
  * Speech-to-text for caption-less videos. This file is the whole provider choice: swapping
- * AssemblyAI for another `transcribe()` provider touches nothing else. The decisions that
- * guard the bill (which caption failures may trigger a paid call, how long a video may be)
- * live here as pure functions so they are tested without any network.
+ * AssemblyAI for another `transcribe()` provider touches nothing else. The duration guard
+ * lives here as a pure function so it is tested without any network.
  */
 
 export const STT_PROVIDER: SttProvider = "assemblyai";
@@ -27,20 +21,6 @@ const STT_MODEL = "universal-3-5-pro";
  */
 export const MAX_STT_MINUTES = 180;
 
-/**
- * The fallback trigger. Only "these captions genuinely do not exist" may lead to a paid
- * transcription: disabled by the owner, no track at all, or no track in a language we can use.
- * Everything else — rate limit, private/removed video, network — rethrows so the video is
- * `failed` and a throttle can never fan out into paid calls. Rerunning is the retry.
- */
-export function isNoCaptions(error: unknown): boolean {
-  return (
-    error instanceof YoutubeTranscriptDisabledError ||
-    error instanceof YoutubeTranscriptNotAvailableError ||
-    error instanceof YoutubeTranscriptNotAvailableLanguageError
-  );
-}
-
 export function assertWithinCap(videoId: string, durationSec: number | undefined): void {
   if (durationSec === undefined) return;
   const minutes = Math.round(durationSec / 60);

src/server/ingest/supadata.test.ts

@@ -0,0 +1,113 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { fetchSupadataCaptions } from "~/server/ingest/supadata";
+
+afterEach(() => vi.unstubAllGlobals());
+
+describe("Supadata native captions", () => {
+  it("requests existing captions only and converts milliseconds to seconds", async () => {
+    const fetchMock = vi.fn(async (_url: URL, _init: RequestInit) => Response.json({
+      content: [
+        { text: " Hello ", offset: 12500, duration: 2500, lang: "en" },
+        { text: " ", offset: 15000, duration: 1000, lang: "en" },
+      ],
+      lang: "en",
+      availableLangs: ["en"],
+    }));
+    vi.stubGlobal("fetch", fetchMock);
+
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toEqual([
+      { text: "Hello", startSec: 12.5, endSec: 15 },
+    ]);
+    const [url, init] = fetchMock.mock.calls[0]!;
+    expect(new URL(url).searchParams.get("mode")).toBe("native");
+    expect(new URL(url).searchParams.get("url")).toBe("https://www.youtube.com/watch?v=5e37ZT3SQbk");
+    expect(init.headers).toEqual({ "x-api-key": "secret" });
+  });
+
+  it("reports unavailable captions without attempting AI generation", async () => {
+    vi.stubGlobal("fetch", vi.fn(async () => Response.json({ error: "transcript-unavailable" }, { status: 206 })));
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toBeUndefined();
+  });
+
+  it("does not accept an empty or malformed transcript as success", async () => {
+    vi.stubGlobal("fetch", vi.fn(async () => Response.json({ content: [{ text: "  ", offset: 0, duration: 1000 }] })));
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).rejects.toThrow(/no text/);
+    vi.stubGlobal("fetch", vi.fn(async () => Response.json({ content: "wrong shape" })));
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).rejects.toThrow(/invalid response/);
+  });
+
+  it("surfaces provider authentication and quota errors", async () => {
+    vi.stubGlobal("fetch", vi.fn(async () => Response.json({ message: "bad key" }, { status: 401 })));
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).rejects.toThrow(/Supadata HTTP 401/);
+  });
+
+  it("retries a rate limit and then returns captions", async () => {
+    const fetchMock = vi.fn()
+      .mockResolvedValueOnce(new Response(null, { status: 429, headers: { "retry-after": "0.001" } }))
+      .mockResolvedValueOnce(Response.json({ content: [
+        { text: "Recovered", offset: 0, duration: 1000 },
+      ] }));
+    vi.stubGlobal("fetch", fetchMock);
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toEqual([
+      { text: "Recovered", startSec: 0, endSec: 1 },
+    ]);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("uses an abort deadline and retries a temporary server failure", async () => {
+    const fetchMock = vi.fn()
+      .mockResolvedValueOnce(new Response(null, { status: 503, headers: { "retry-after": "0.001" } }))
+      .mockResolvedValueOnce(Response.json({ content: [
+        { text: "Available", offset: 1000, duration: 1000 },
+      ] }));
+    vi.stubGlobal("fetch", fetchMock);
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toEqual([
+      { text: "Available", startSec: 1, endSec: 2 },
+    ]);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(fetchMock.mock.calls[0]![1].signal).toBeInstanceOf(AbortSignal);
+  });
+
+  it("retries a transient network failure without repeating authentication errors", async () => {
+    const fetchMock = vi.fn()
+      .mockRejectedValueOnce(new TypeError("fetch failed"))
+      .mockResolvedValueOnce(Response.json({ content: [
+        { text: "Recovered", offset: 0, duration: 1000 },
+      ] }));
+    vi.stubGlobal("fetch", fetchMock);
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toHaveLength(1);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("polls an asynchronous native transcript job without resubmitting it", async () => {
+    const fetchMock = vi.fn()
+      .mockResolvedValueOnce(Response.json({ jobId: "job-1" }, { status: 202 }))
+      .mockResolvedValueOnce(Response.json({ status: "queued" }))
+      .mockResolvedValueOnce(Response.json({ status: "completed", content: [
+        { text: "Later", offset: 2000, duration: 500 },
+      ] }));
+    vi.stubGlobal("fetch", fetchMock);
+    await expect(fetchSupadataCaptions("5e37ZT3SQbk", "secret")).resolves.toEqual([
+      { text: "Later", startSec: 2, endSec: 2.5 },
+    ]);
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(String(fetchMock.mock.calls[1]![0])).toBe("https://api.supadata.ai/v1/transcript/job-1");
+    expect(String(fetchMock.mock.calls[2]![0])).toBe("https://api.supadata.ai/v1/transcript/job-1");
+  });
+
+  it("stops a stalled request at the shared deadline", async () => {
+    const controller = new AbortController();
+    const timeout = vi.spyOn(AbortSignal, "timeout").mockReturnValue(controller.signal);
+    vi.stubGlobal("fetch", vi.fn((_url: URL, init: RequestInit) => new Promise<Response>((_resolve, reject) => {
+      init.signal?.addEventListener("abort", () => reject(init.signal?.reason));
+    })));
+    try {
+      const pending = fetchSupadataCaptions("5e37ZT3SQbk", "secret");
+      controller.abort(new DOMException("deadline", "TimeoutError"));
+      await expect(pending).rejects.toThrow(/Supadata request timed out/);
+    } finally {
+      timeout.mockRestore();
+    }
+  });
+});

src/server/ingest/supadata.ts

@@ -0,0 +1,88 @@
+import { setTimeout as delay } from "node:timers/promises";
+import { z } from "zod";
+
+import type { Segment } from "~/server/domain/types";
+
+const transcriptResponse = z.object({
+  content: z.array(z.object({
+    text: z.string(),
+    offset: z.number().nonnegative(),
+    duration: z.number().nonnegative(),
+  })),
+});
+const jobStatus = z.object({ status: z.enum(["queued", "active", "completed", "failed"]) });
+
+async function request(url: URL, apiKey: string, signal: AbortSignal): Promise<Response> {
+  for (let attempt = 0; attempt < 4; attempt++) {
+    let response: Response;
+    try {
+      response = await fetch(url, { headers: { "x-api-key": apiKey }, signal });
+    } catch (error) {
+      if (signal.aborted || attempt === 3) throw error;
+      await delay(1000 * (attempt + 1), undefined, { signal });
+      continue;
+    }
+    if ((response.status === 429 || response.status >= 500) && attempt < 3) {
+      const retryAfter = Number(response.headers.get("retry-after"));
+      const delayMs = Number.isFinite(retryAfter) && retryAfter > 0
+        ? Math.min(retryAfter * 1000, 10_000)
+        : 1000 * (attempt + 1);
+      await delay(delayMs, undefined, { signal });
+      continue;
+    }
+    return response;
+  }
+  throw new Error("Supadata retry limit reached");
+}
+
+/** Fetch only existing captions. `mode=native` never starts paid AI transcription. */
+export async function fetchSupadataCaptions(
+  videoId: string,
+  apiKey: string,
+): Promise<Segment[] | undefined> {
+  const signal = AbortSignal.timeout(120_000);
+  const url = new URL("https://api.supadata.ai/v1/transcript");
+  url.searchParams.set("url", `https://www.youtube.com/watch?v=${videoId}`);
+  url.searchParams.set("lang", "en");
+  url.searchParams.set("text", "false");
+  url.searchParams.set("mode", "native");
+
+  try {
+    let response = await request(url, apiKey, signal);
+    if (response.status === 206) return undefined;
+    if (!response.ok) throw new Error(`${videoId}: Supadata HTTP ${response.status}`);
+    let payload: unknown = await response.json();
+
+    if (response.status === 202) {
+      const job = z.object({ jobId: z.string().min(1) }).safeParse(payload);
+      if (!job.success) throw new Error(`${videoId}: Supadata returned an invalid job ID`);
+      const jobUrl = new URL(`https://api.supadata.ai/v1/transcript/${encodeURIComponent(job.data.jobId)}`);
+      for (;;) {
+        response = await request(jobUrl, apiKey, signal);
+        if (response.status === 206) return undefined;
+        if (!response.ok) throw new Error(`${videoId}: Supadata job HTTP ${response.status}`);
+        payload = await response.json();
+        const status = jobStatus.safeParse(payload);
+        if (!status.success) throw new Error(`${videoId}: Supadata returned an invalid job status`);
+        if (status.data.status === "completed") break;
+        if (status.data.status === "failed") throw new Error(`${videoId}: Supadata transcript job failed`);
+        await delay(1000, undefined, { signal });
+      }
+    }
+
+    const parsed = transcriptResponse.safeParse(payload);
+    if (!parsed.success) throw new Error(`${videoId}: Supadata returned an invalid response`);
+    const segments = parsed.data.content
+      .filter((entry) => entry.text.trim())
+      .map((entry) => ({
+        text: entry.text.trim(),
+        startSec: entry.offset / 1000,
+        endSec: (entry.offset + entry.duration) / 1000,
+      }));
+    if (segments.length === 0) throw new Error(`${videoId}: Supadata caption track returned no text`);
+    return segments;
+  } catch (error) {
+    if (signal.aborted) throw new Error(`${videoId}: Supadata request timed out`, { cause: error });
+    throw error;
+  }
+}