Expose Theo's latest completed stream for post-stream automation

docs/public-stream-api.md

@@ -0,0 +1,11 @@
+# Public stream workflow API
+
+`GET /api/streams/theo/latest` is an intentionally public, read-only endpoint for Theo's post-stream processing utility. No client authentication or new environment variables are required. It uses the creator's existing Clerk-managed Twitch connection on the server. Other creators are not opted in.
+
+The endpoint checks that the channel is offline, selects the newest Twitch archive (`sort=time`, `first=1`), retrieves every marker page, then checks offline status again. Failed upstream reads, duplicate markers, cursor loops, and pagination beyond ten pages return an error instead of partial data. Responses are uncached.
+
+Successful JSON has `schema_version: 1`, `channel: "theo"`, `offline: true`, `markers_complete: true`, `fetched_at`, `id`, `user_login`, `type`, `title`, `created_at`, `duration`, and `markers`. Each marker includes only `id`, `created_at`, `description`, and `position_seconds`. Timestamps use Twitch's UTC strings and duration retains Twitch's `5h34m44s` format. `OFFSET HH:MM:SS` remains a marker description; clients apply the existing MarkerThing parsing rules.
+
+Errors: `409 channel_live`, `404 no_archive`, or `503 upstream_unavailable`. Upstream error details and tokens are never returned. An expired creator connection still needs reconnecting through the existing MarkerThing flow; callers never provide Twitch or Clerk credentials.
+
+This endpoint exposes metadata only. It does not export media, mutate markers, access the user's disks, or claim that a locally recorded session is a Twitch stream. A consuming utility must check local media duration/continuity before delivery, and may retry while Twitch finalizes a just-ended archive.

src/app/api/streams/theo/latest/route.test.ts

@@ -0,0 +1,41 @@
+import { afterEach, expect, it, vi } from "vitest";
+import { GET } from "./route";
+import {
+  getLatestPublicTheoStream,
+  PublicStreamError,
+} from "~/utils/twitch-server";
+
+vi.mock("~/utils/twitch-server", () => ({
+  getLatestPublicTheoStream: vi.fn(),
+  PublicStreamError: class extends Error {
+    constructor(public code: string, public status: number, message: string) {
+      super(message);
+    }
+  },
+}));
+afterEach(() => vi.resetAllMocks());
+it("returns public data without caller authentication and forbids caching", async () => {
+  vi.mocked(getLatestPublicTheoStream).mockResolvedValue({
+    id: "123",
+  } as Awaited<ReturnType<typeof getLatestPublicTheoStream>>);
+  const response = await GET();
+  expect(response.status).toBe(200);
+  expect(response.headers.get("Cache-Control")).toBe("no-store");
+  expect(await response.json()).toEqual({ id: "123" });
+});
+it("preserves expected offline/no-archive errors", async () => {
+  vi.mocked(getLatestPublicTheoStream).mockRejectedValue(
+    new PublicStreamError("channel_live", 409, "The channel is still live.")
+  );
+  const response = await GET();
+  expect(response.status).toBe(409);
+  expect(await response.json()).toMatchObject({ error: "channel_live" });
+});
+it("does not leak upstream tokens or Clerk error details", async () => {
+  vi.mocked(getLatestPublicTheoStream).mockRejectedValue(
+    new Error("private-token and Clerk details")
+  );
+  const response = await GET();
+  expect(response.status).toBe(503);
+  expect(await response.text()).not.toMatch(/private-token|Clerk/);
+});

src/app/api/streams/theo/latest/route.ts

@@ -0,0 +1,30 @@
+import {
+  getLatestPublicTheoStream,
+  PublicStreamError,
+} from "~/utils/twitch-server";
+
+export const dynamic = "force-dynamic";
+
+// Intentionally public, read-only, and limited to the channel that opted in.
+export async function GET() {
+  const headers = { "Cache-Control": "no-store" };
+  try {
+    return Response.json(await getLatestPublicTheoStream(), { headers });
+  } catch (error) {
+    if (error instanceof PublicStreamError) {
+      return Response.json(
+        { error: error.code, message: error.message },
+        { status: error.status, headers }
+      );
+    }
+    // Upstream/Clerk errors may contain private details. Never serialize them.
+    return Response.json(
+      {
+        error: "upstream_unavailable",
+        message:
+          "Could not read the latest stream from Twitch. Try again shortly.",
+      },
+      { status: 503, headers }
+    );
+  }
+}

src/utils/public-stream.test.ts

@@ -0,0 +1,144 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { getLatestPublicTheoStream } from "./twitch-server";
+
+vi.mock("@clerk/nextjs/server", () => ({
+  clerkClient: async () => ({
+    users: {
+      getUserList: async () => ({ data: [{ id: "creator" }] }),
+      getUserOauthAccessToken: async () => ({
+        data: [{ token: "private-token" }],
+      }),
+    },
+  }),
+}));
+
+const vod = {
+  id: "123",
+  user_id: "42",
+  user_login: "theo",
+  type: "archive",
+  title: "Stream",
+  created_at: "2026-10-07T22:25:04Z",
+  duration: "5h34m44s",
+  private_field: "omit",
+};
+const marker = (id: string) => ({
+  id,
+  created_at: vod.created_at,
+  description: id === "1" ? "OFFSET 00:53:30" : "Topic",
+  position_seconds: 3211,
+  URL: "omit",
+});
+const page = (ids: string[], cursor?: string) => ({
+  data: [
+    { user_id: "42", videos: [{ video_id: "123", markers: ids.map(marker) }] },
+  ],
+  pagination: cursor ? { cursor } : {},
+});
+const stub = (...responses: unknown[]) => {
+  const fetcher = vi.fn();
+  for (const response of responses)
+    fetcher.mockResolvedValueOnce(
+      response instanceof Response ? response : Response.json(response)
+    );
+  vi.stubGlobal("fetch", fetcher);
+  return fetcher;
+};
+const user = { data: [{ id: "42" }] };
+const offline = { data: [] };
+afterEach(() => vi.unstubAllGlobals());
+
+describe("public latest stream", () => {
+  it("selects the latest archive, walks every marker page, and returns only public fields", async () => {
+    const fetcher = stub(
+      user,
+      offline,
+      { data: [vod] },
+      page(["1"], "next"),
+      page(["2"]),
+      offline
+    );
+    const result = await getLatestPublicTheoStream();
+    expect(result).toMatchObject({
+      schema_version: 1,
+      id: "123",
+      channel: "theo",
+      offline: true,
+      markers_complete: true,
+      markers: [marker("1"), marker("2")].map(({ URL: _url, ...m }) => m),
+    });
+    expect(JSON.stringify(result)).not.toMatch(
+      /private-token|private_field|URL/
+    );
+    expect(fetcher.mock.calls[2][0]).toContain(
+      "type=archive&sort=time&first=1"
+    );
+    expect(fetcher.mock.calls[4][0]).toContain("after=next");
+    for (const [, options] of fetcher.mock.calls.slice(1))
+      expect(options.cache).toBe("no-store");
+  });
+  it("refuses a live channel before fetching videos", async () => {
+    const fetcher = stub(user, { data: [{ id: "live" }] });
+    await expect(getLatestPublicTheoStream()).rejects.toMatchObject({
+      status: 409,
+      code: "channel_live",
+    });
+    expect(fetcher).toHaveBeenCalledTimes(2);
+  });
+  it("refuses a stream that goes live during pagination", async () => {
+    stub(user, offline, { data: [vod] }, page(["1"]), {
+      data: [{ id: "live" }],
+    });
+    await expect(getLatestPublicTheoStream()).rejects.toMatchObject({
+      status: 409,
+    });
+  });
+  it("reports no archive separately from an upstream failure", async () => {
+    stub(user, offline, { data: [] });
+    await expect(getLatestPublicTheoStream()).rejects.toMatchObject({
+      status: 404,
+    });
+  });
+  it.each([Response.json({}, { status: 401 }), {}, { data: null }])(
+    "does not mistake a failed live check for offline",
+    async (response) => {
+      stub(user, response);
+      await expect(getLatestPublicTheoStream()).rejects.toThrow();
+    }
+  );
+  it("never returns a partial marker response after a later page fails", async () => {
+    stub(
+      user,
+      offline,
+      { data: [vod] },
+      page(["1"], "next"),
+      Response.json({}, { status: 429 })
+    );
+    await expect(getLatestPublicTheoStream()).rejects.toThrow();
+  });
+  it("rejects repeated pagination cursors", async () => {
+    stub(
+      user,
+      offline,
+      { data: [vod] },
+      page(["1"], "loop"),
+      page(["2"], "loop")
+    );
+    await expect(getLatestPublicTheoStream()).rejects.toThrow("pagination");
+  });
+  it("rejects duplicate markers and other creators' archives", async () => {
+    stub(user, offline, { data: [vod] }, page(["1", "1"]));
+    await expect(getLatestPublicTheoStream()).rejects.toThrow("repeated");
+    stub(user, offline, { data: [{ ...vod, user_login: "someoneelse" }] });
+    await expect(getLatestPublicTheoStream()).rejects.toThrow("archive");
+  });
+  it("fails closed at the pagination cap", async () => {
+    stub(
+      user,
+      offline,
+      { data: [vod] },
+      ...Array.from({ length: 10 }, (_, i) => page([String(i)], `page${i}`))
+    );
+    await expect(getLatestPublicTheoStream()).rejects.toThrow("pagination");
+  });
+});

src/utils/twitch-server.ts

@@ -174,3 +174,152 @@ export const getLiveTopic = async (
     streamStart: Date.parse(stream.started_at),
   });
 };
+
+export class PublicStreamError extends Error {
+  constructor(
+    public readonly code: "channel_live" | "no_archive",
+    public readonly status: 409 | 404,
+    message: string
+  ) {
+    super(message);
+  }
+}
+
+type PublicArchive = {
+  id: string;
+  user_id: string;
+  user_login: string;
+  type: string;
+  title: string;
+  created_at: string;
+  duration: string;
+};
+
+// Theo has opted into public markers. Keep this fixed to his channel: stored
+// creator tokens do not imply consent to publish other creators' markers.
+export const getLatestPublicTheoStream = async () => {
+  const channel = "theo";
+  const token = await getValidTokenForCreator(channel);
+  const userId = await getTwitchUserId(channel, token);
+  if (!userId) {
+    throw new PublicStreamError("no_archive", 404, "No archived stream found.");
+  }
+  const request = async <T extends { data: unknown[] }>(path: string) => {
+    const response = await fetch(`https://api.twitch.tv/helix${path}`, {
+      headers: generateTwitchRequestHeaders(token),
+      cache: "no-store",
+      signal: AbortSignal.timeout(10_000),
+    });
+    if (!response.ok)
+      throw new Error(`Twitch workflow request failed: ${response.status}`);
+    const value = (await response.json()) as T;
+    if (!value || !Array.isArray(value.data)) {
+      throw new Error("Twitch workflow response has no data array");
+    }
+    return value;
+  };
+  const requireOffline = async () => {
+    const live = await request<{ data: { id: string }[] }>(
+      `/streams?${new URLSearchParams({ user_id: userId })}`
+    );
+    if (live.data.length) {
+      throw new PublicStreamError(
+        "channel_live",
+        409,
+        "The channel is still live."
+      );
+    }
+  };
+  await requireOffline();
+  const archives = await request<{ data: PublicArchive[] }>(
+    `/videos?${new URLSearchParams({
+      user_id: userId,
+      type: "archive",
+      sort: "time",
+      first: "1",
+    })}`
+  );
+  const vod = archives.data[0];
+  if (!vod)
+    throw new PublicStreamError("no_archive", 404, "No archived stream found.");
+  if (
+    vod.user_id !== userId ||
+    vod.user_login !== channel ||
+    vod.type !== "archive" ||
+    !/^\d+$/.test(vod.id)
+  ) {
+    throw new Error("Twitch returned an unexpected archive");
+  }
+
+  type Marker = Pick<
+    VOD["markers"][number],
+    "id" | "created_at" | "description" | "position_seconds"
+  >;
+  type MarkerPage = {
+    data: {
+      user_id: string;
+      videos: { video_id: string; markers: Marker[] }[];
+    }[];
+    pagination?: { cursor?: string };
+  };
+  const markers: Marker[] = [];
+  const seenIds = new Set<string>();
+  const seenCursors = new Set<string>();
+  let cursor: string | undefined;
+  for (let page = 0; page < 10; page++) {
+    const params = new URLSearchParams({ video_id: vod.id, first: "100" });
+    if (cursor) params.set("after", cursor);
+    const result = await request<MarkerPage>(`/streams/markers?${params}`);
+    for (const creator of result.data) {
+      if (creator.user_id !== userId || !Array.isArray(creator.videos))
+        throw new Error("Unexpected marker owner");
+      for (const video of creator.videos) {
+        if (video.video_id !== vod.id || !Array.isArray(video.markers))
+          throw new Error("Unexpected marker video");
+        for (const marker of video.markers) {
+          if (
+            !marker ||
+            typeof marker.id !== "string" ||
+            seenIds.has(marker.id) ||
+            typeof marker.description !== "string" ||
+            typeof marker.created_at !== "string" ||
+            !Number.isFinite(marker.position_seconds) ||
+            marker.position_seconds < 0
+          ) {
+            throw new Error("Invalid or repeated stream marker");
+          }
+          seenIds.add(marker.id);
+          // Select fields explicitly; upstream credentials and future private
+          // Twitch response fields must never leak through this public API.
+          markers.push({
+            id: marker.id,
+            created_at: marker.created_at,
+            description: marker.description,
+            position_seconds: marker.position_seconds,
+          });
+        }
+      }
+    }
+    cursor = result.pagination?.cursor;
+    if (!cursor) break;
+    if (seenCursors.has(cursor) || page === 9)
+      throw new Error("Marker pagination is incomplete");
+    seenCursors.add(cursor);
+  }
+  // A stream could start while several pages of markers were being fetched.
+  await requireOffline();
+  return {
+    schema_version: 1,
+    channel,
+    offline: true,
+    markers_complete: true,
+    fetched_at: new Date().toISOString(),
+    id: vod.id,
+    user_login: vod.user_login,
+    type: vod.type,
+    title: vod.title,
+    created_at: vod.created_at,
+    duration: vod.duration,
+    markers,
+  };
+};