diff --git a/app/api/chat/[chatId]/stream/route.ts b/app/api/chat/[chatId]/stream/route.ts new file mode 100644 index 000000000..289f56bdc --- /dev/null +++ b/app/api/chat/[chatId]/stream/route.ts @@ -0,0 +1,43 @@ +import type { NextRequest } from "next/server"; +import { NextResponse } from "next/server"; +import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; +import { handleResumeChatStream } from "@/lib/chat/handleResumeChatStream"; + +// Matches POST /api/chat: a resumed stream stays open as long as the turn it +// is following, so it needs the same ceiling rather than the route default. +export const maxDuration = 800; +export const dynamic = "force-dynamic"; + +/** + * OPTIONS handler for CORS preflight requests. + * + * @returns A NextResponse with CORS headers. + */ +export async function OPTIONS() { + return new NextResponse(null, { + status: 200, + headers: getCorsHeaders(), + }); +} + +/** + * GET /api/chat/{chatId}/stream — reconnect to an in-progress chat response. + * + * The resume counterpart to `POST /api/chat`, which only resumes as a side + * effect of starting a turn. Pass `startIndex` to continue from the chunk + * after the last one received; omit it to read the response from the start. + * + * Contract: https://docs.recoupable.dev/api-reference/chat/workflow-stream + * + * @param request - The incoming NextRequest. + * @param options - Route options containing the async params. + * @param options.params - Route params containing the chat id. + * @returns A streaming 200, 204 when there is nothing to resume, or an error. + */ +export async function GET( + request: NextRequest, + options: { params: Promise<{ chatId: string }> }, +): Promise { + const { chatId } = await options.params; + return handleResumeChatStream(request, chatId); +} diff --git a/lib/chat/__tests__/handleResumeChatStream.test.ts b/lib/chat/__tests__/handleResumeChatStream.test.ts new file mode 100644 index 000000000..21a984ccb --- /dev/null +++ b/lib/chat/__tests__/handleResumeChatStream.test.ts @@ -0,0 +1,166 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { NextRequest, NextResponse } from "next/server"; +import { handleResumeChatStream } from "@/lib/chat/handleResumeChatStream"; +import { validateChatOwnership } from "@/lib/chat/validateChatOwnership"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; +import { getRun } from "workflow/api"; + +vi.mock("@/lib/chat/validateChatOwnership", () => ({ validateChatOwnership: vi.fn() })); +vi.mock("@/lib/chat/compareAndSetChatActiveStreamId", () => ({ + compareAndSetChatActiveStreamId: vi.fn(), +})); +vi.mock("workflow/api", () => ({ getRun: vi.fn() })); + +const CHAT_ID = "11111111-2222-3333-4444-555555555555"; +const RUN_ID = "wrun_01ABC"; + +const request = (qs = "") => + new NextRequest(`https://api.test/api/chat/${CHAT_ID}/stream${qs}`, { method: "GET" }); + +/** Validator resolves with a chat carrying the given active_stream_id. */ +function withChat(activeStreamId: string | null) { + vi.mocked(validateChatOwnership).mockResolvedValue({ + auth: { accountId: "acc-1" }, + chat: { id: CHAT_ID, active_stream_id: activeStreamId }, + } as never); +} + +function withRun( + status: string, + getReadable = vi.fn(() => Object.assign(new ReadableStream(), { getTailIndex: async () => 41 })), +) { + vi.mocked(getRun).mockReturnValue({ + get status() { + return Promise.resolve(status); + }, + getReadable, + } as never); + return getReadable; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValue({ + ok: true, + claimed: true, + } as never); +}); + +describe("handleResumeChatStream", () => { + it("returns 204 when the chat has no active stream", async () => { + withChat(null); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(204); + expect(getRun).not.toHaveBeenCalled(); + }); + + it("returns 204 and clears the stale id when the run is already terminal", async () => { + withChat(RUN_ID); + withRun("completed"); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(204); + expect(compareAndSetChatActiveStreamId).toHaveBeenCalledWith(CHAT_ID, RUN_ID, null); + }); + + it("streams the run and advertises the run id when the run is live", async () => { + withChat(RUN_ID); + withRun("running"); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(200); + expect(res.headers.get("x-workflow-run-id")).toBe(RUN_ID); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + }); + + it("forwards startIndex to getReadable so a reconnect skips chunks already seen", async () => { + withChat(RUN_ID); + const getReadable = withRun("running"); + + await handleResumeChatStream(request("?startIndex=12"), CHAT_ID); + + expect(getReadable).toHaveBeenCalledWith(expect.objectContaining({ startIndex: 12 })); + }); + + it("omits startIndex when absent so a fresh reader gets the whole turn", async () => { + withChat(RUN_ID); + const getReadable = withRun("running"); + + await handleResumeChatStream(request(), CHAT_ID); + + expect(getReadable).toHaveBeenCalledWith(expect.objectContaining({ startIndex: undefined })); + }); + + it("returns 400 for a malformed startIndex without touching the run", async () => { + withChat(RUN_ID); + withRun("running"); + + const res = await handleResumeChatStream(request("?startIndex=-3"), CHAT_ID); + + expect(res.status).toBe(400); + expect(getRun).not.toHaveBeenCalled(); + }); + + it("propagates the validator's response (401/403/404) unchanged", async () => { + vi.mocked(validateChatOwnership).mockResolvedValue( + NextResponse.json({ error: "Forbidden" }, { status: 403 }) as never, + ); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(403); + expect(getRun).not.toHaveBeenCalled(); + }); + + // A transient workflow-API failure must not be reported as "nothing to + // resume" — that would tell a client with a live run to stop reconnecting. + it("returns 502 rather than 204 when the run status lookup throws", async () => { + withChat(RUN_ID); + vi.mocked(getRun).mockReturnValue({ + get status() { + return Promise.reject(new Error("workflow api down")); + }, + getReadable: vi.fn(), + } as never); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(502); + expect(compareAndSetChatActiveStreamId).not.toHaveBeenCalled(); + }); + + // Upstream open-agents returns this so the client knows which startIndex to + // send on its next reconnect; the SDK's WorkflowChatTransport reads it to + // compute absolute chunk positions. Without it a reconnect replays from 0. + it("advertises the stream tail index so the client can resume precisely", async () => { + withChat(RUN_ID); + withRun("running"); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.headers.get("x-workflow-stream-tail-index")).toBe("41"); + }); + + it("still streams when the tail index cannot be read", async () => { + withChat(RUN_ID); + withRun( + "running", + vi.fn(() => + Object.assign(new ReadableStream(), { + getTailIndex: async () => { + throw new Error("unsupported"); + }, + }), + ), + ); + + const res = await handleResumeChatStream(request(), CHAT_ID); + + expect(res.status).toBe(200); + expect(res.headers.get("x-workflow-stream-tail-index")).toBeNull(); + }); +}); diff --git a/lib/chat/__tests__/parseStreamStartIndex.test.ts b/lib/chat/__tests__/parseStreamStartIndex.test.ts new file mode 100644 index 000000000..208d6ec2b --- /dev/null +++ b/lib/chat/__tests__/parseStreamStartIndex.test.ts @@ -0,0 +1,43 @@ +import { describe, it, expect } from "vitest"; +import { NextResponse } from "next/server"; +import { parseStreamStartIndex } from "@/lib/chat/parseStreamStartIndex"; + +const url = (qs: string) => new URL(`https://api.test/api/chat/abc/stream${qs}`); + +describe("parseStreamStartIndex", () => { + it("returns undefined when startIndex is absent — a fresh reader gets the whole turn", () => { + expect(parseStreamStartIndex(url(""))).toBeUndefined(); + }); + + it("parses a valid non-negative integer", () => { + expect(parseStreamStartIndex(url("?startIndex=0"))).toBe(0); + expect(parseStreamStartIndex(url("?startIndex=42"))).toBe(42); + }); + + it("returns a 400 response when startIndex is not a number", () => { + const result = parseStreamStartIndex(url("?startIndex=abc")); + expect(result).toBeInstanceOf(NextResponse); + expect((result as NextResponse).status).toBe(400); + }); + + // The documented schema is `integer, minimum 0`. A negative value is + // meaningful to the underlying SDK (it counts back from the end of a live + // stream) but resolves differently on every call, so the contract excludes it. + it("returns a 400 response when startIndex is negative", () => { + const result = parseStreamStartIndex(url("?startIndex=-5")); + expect(result).toBeInstanceOf(NextResponse); + expect((result as NextResponse).status).toBe(400); + }); + + it("returns a 400 response when startIndex is fractional", () => { + const result = parseStreamStartIndex(url("?startIndex=1.5")); + expect(result).toBeInstanceOf(NextResponse); + expect((result as NextResponse).status).toBe(400); + }); + + it("returns a 400 response when startIndex is present but empty", () => { + const result = parseStreamStartIndex(url("?startIndex=")); + expect(result).toBeInstanceOf(NextResponse); + expect((result as NextResponse).status).toBe(400); + }); +}); diff --git a/lib/chat/__tests__/validateChatOwnership.test.ts b/lib/chat/__tests__/validateChatOwnership.test.ts new file mode 100644 index 000000000..152d6620d --- /dev/null +++ b/lib/chat/__tests__/validateChatOwnership.test.ts @@ -0,0 +1,79 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { NextRequest, NextResponse } from "next/server"; +import { validateChatOwnership } from "@/lib/chat/validateChatOwnership"; +import { validateAuthContext } from "@/lib/auth/validateAuthContext"; +import { selectChats } from "@/lib/supabase/chats/selectChats"; +import { selectSessions } from "@/lib/supabase/sessions/selectSessions"; + +vi.mock("@/lib/auth/validateAuthContext", () => ({ validateAuthContext: vi.fn() })); +vi.mock("@/lib/supabase/chats/selectChats", () => ({ selectChats: vi.fn() })); +vi.mock("@/lib/supabase/sessions/selectSessions", () => ({ selectSessions: vi.fn() })); + +const CHAT_ID = "11111111-2222-4333-8444-555555555555"; +const OWNER = "owner-account"; +const ADMIN_TARGET = "22222222-3333-4444-8555-666666666666"; + +const req = (qs = "") => + new NextRequest(`https://api.test/api/chat/${CHAT_ID}/stream${qs}`, { method: "GET" }); + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(validateAuthContext).mockResolvedValue({ accountId: OWNER, orgId: null } as never); + vi.mocked(selectChats).mockResolvedValue([{ id: CHAT_ID, session_id: "sess-1" }] as never); + vi.mocked(selectSessions).mockResolvedValue([{ id: "sess-1", account_id: OWNER }] as never); +}); + +describe("validateChatOwnership", () => { + it("resolves for the owning account", async () => { + const result = await validateChatOwnership(req(), CHAT_ID); + expect(result).not.toBeInstanceOf(NextResponse); + }); + + // Without this an org/admin key cannot reach a member's chat — the same gap + // DELETE /api/tasks has (chat#1918). validateAuthContext is what decides + // whether the caller may actually use the override. + it("forwards an account_id query override to validateAuthContext", async () => { + await validateChatOwnership(req(`?account_id=${ADMIN_TARGET}`), CHAT_ID); + + expect(validateAuthContext).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ accountId: ADMIN_TARGET }), + ); + }); + + it("lets an approved override through to a chat the key does not personally own", async () => { + // validateAuthContext approved the override, so the effective account is + // the target — which owns the session. + vi.mocked(validateAuthContext).mockResolvedValue({ + accountId: ADMIN_TARGET, + orgId: "org-1", + } as never); + vi.mocked(selectSessions).mockResolvedValue([ + { id: "sess-1", account_id: ADMIN_TARGET }, + ] as never); + + const result = await validateChatOwnership(req(`?account_id=${ADMIN_TARGET}`), CHAT_ID); + + expect(result).not.toBeInstanceOf(NextResponse); + }); + + it("still 403s when the resolved account does not own the session", async () => { + vi.mocked(selectSessions).mockResolvedValue([ + { id: "sess-1", account_id: "someone-else" }, + ] as never); + + const result = await validateChatOwnership(req(), CHAT_ID); + + expect(result).toBeInstanceOf(NextResponse); + expect((result as NextResponse).status).toBe(403); + }); + + it("omits the override key entirely when no account_id is supplied", async () => { + await validateChatOwnership(req(), CHAT_ID); + + expect(validateAuthContext).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ accountId: undefined }), + ); + }); +}); diff --git a/lib/chat/handleResumeChatStream.ts b/lib/chat/handleResumeChatStream.ts new file mode 100644 index 000000000..31edc942d --- /dev/null +++ b/lib/chat/handleResumeChatStream.ts @@ -0,0 +1,91 @@ +import { NextRequest, NextResponse } from "next/server"; +import { createUIMessageStreamResponse, type UIMessageChunk } from "ai"; +import { getRun } from "workflow/api"; +import { validateChatOwnership } from "@/lib/chat/validateChatOwnership"; +import { parseStreamStartIndex } from "@/lib/chat/parseStreamStartIndex"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; +import { wrapWorkflowStreamWatcher } from "@/lib/chat/wrapWorkflowStreamWatcher"; +import { errorResponse } from "@/lib/networking/errorResponse"; +import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; + +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set(["completed", "cancelled", "failed"]); + +/** + * Handles `GET /api/chat/{chatId}/stream` — reconnect to an in-progress + * chat response. + * + * A long turn's SSE stream can end before the run does, which leaves the + * client rendering a half-finished message with no way back in: `POST + * /api/chat` resumes only as a side effect of starting a turn, so today + * recovery needs a full page load (chat#1923). This is the resume path, + * and `startIndex` is what makes it gap-free. + * + * Contract (docs#286): 200 SSE + `x-workflow-run-id`, 204 when there is + * nothing to resume, 400 malformed `startIndex`, 401/403/404 from auth + * and ownership. + * + * @param request - The incoming request. + * @param chatId - Chat id from the route params. + * @returns The resumed stream, 204, or an error response. + */ +export async function handleResumeChatStream( + request: NextRequest, + chatId: string, +): Promise { + const validated = await validateChatOwnership(request, chatId); + if (validated instanceof NextResponse) return validated; + + const startIndex = parseStreamStartIndex(new URL(request.url)); + if (startIndex instanceof NextResponse) return startIndex; + + const activeStreamId = validated.chat.active_stream_id; + if (!activeStreamId) return new NextResponse(null, { status: 204, headers: getCorsHeaders() }); + + const run = getRun(activeStreamId); + + // A failed status read must not be reported as "nothing to resume" — that + // tells a client with a live run to stop reconnecting, which is the exact + // silent-truncation this endpoint exists to prevent. Surface it instead so + // the client retries. Mirrors reconcileExistingActiveStream, which treats a + // transient workflow-api failure as conflict rather than clearing the slot. + let status: string; + try { + status = await run.status; + } catch (error) { + console.error("[handleResumeChatStream] run status lookup failed:", error); + return errorResponse("Failed to read the workflow run", 502); + } + + if (TERMINAL_RUN_STATUSES.has(status)) { + // The run is done, so the slot is stale bookkeeping. Best-effort clear: + // a failed CAS just leaves it for the next request to heal. + const cleared = await compareAndSetChatActiveStreamId(chatId, activeStreamId, null); + if ("error" in cleared) { + console.error("[handleResumeChatStream] failed to clear stale active_stream_id:", cleared); + } + return new NextResponse(null, { status: 204, headers: getCorsHeaders() }); + } + + const readable = run.getReadable({ startIndex }); + + // Tell the client where this read ends so its next reconnect can resume + // exactly there instead of replaying from chunk zero. Upstream open-agents + // returns the same header, and the SDK's WorkflowChatTransport reads it to + // compute absolute chunk positions. Best-effort: if the runtime can't report + // a tail index we still stream — a replaying client beats no client. + let tailIndex: number | undefined; + try { + tailIndex = await readable.getTailIndex(); + } catch (error) { + console.error("[handleResumeChatStream] getTailIndex failed:", error); + } + + return createUIMessageStreamResponse({ + stream: wrapWorkflowStreamWatcher(activeStreamId, readable), + headers: { + ...getCorsHeaders(), + "x-workflow-run-id": activeStreamId, + ...(tailIndex === undefined ? {} : { "x-workflow-stream-tail-index": String(tailIndex) }), + }, + }); +} diff --git a/lib/chat/parseStreamStartIndex.ts b/lib/chat/parseStreamStartIndex.ts new file mode 100644 index 000000000..efb55ae5f --- /dev/null +++ b/lib/chat/parseStreamStartIndex.ts @@ -0,0 +1,32 @@ +import { NextResponse } from "next/server"; +import { validationErrorResponse } from "@/lib/zod/validationErrorResponse"; + +/** + * Parse the optional `startIndex` query param of + * `GET /api/chat/{chatId}/stream`. + * + * Documented as `integer, minimum 0` — the zero-based index of the first + * chunk to return, so a reconnecting client resumes where it left off + * instead of replaying the turn. Absent means "from the beginning", which + * is what a fresh reader (page load, or first watch of a headless run) + * wants. + * + * Negative values are rejected even though the SDK accepts them: it reads + * them relative to the end of a live stream, which resolves to a different + * absolute position on every call and so cannot give a client an exact, + * gap-free resume. + * + * @param url - The request URL carrying the query string. + * @returns The parsed index, `undefined` when absent, or a 400 response. + */ +export function parseStreamStartIndex(url: URL): number | undefined | NextResponse { + const raw = url.searchParams.get("startIndex"); + if (raw === null) return undefined; + + const parsed = Number(raw); + if (raw.trim() === "" || !Number.isInteger(parsed) || parsed < 0) { + return validationErrorResponse("startIndex must be a non-negative integer", ["startIndex"]); + } + + return parsed; +} diff --git a/lib/chat/runs/__tests__/handleStartChatRun.test.ts b/lib/chat/runs/__tests__/handleStartChatRun.test.ts index 3f1c8555b..999b12131 100644 --- a/lib/chat/runs/__tests__/handleStartChatRun.test.ts +++ b/lib/chat/runs/__tests__/handleStartChatRun.test.ts @@ -8,6 +8,7 @@ import { mintEphemeralAccountKey } from "@/lib/keys/mintEphemeralAccountKey"; import { deleteApiKey } from "@/lib/supabase/account_api_keys/deleteApiKey"; import { buildRunAgentInput } from "@/lib/chat/buildRunAgentInput"; import { start } from "workflow/api"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({ "Access-Control-Allow-Origin": "*" })), @@ -31,6 +32,9 @@ vi.mock("workflow/api", () => ({ start: vi.fn(), })); vi.mock("@/app/lib/workflows/runAgentWorkflow", () => ({ runAgentWorkflow: vi.fn() })); +vi.mock("@/lib/chat/compareAndSetChatActiveStreamId", () => ({ + compareAndSetChatActiveStreamId: vi.fn(async () => ({ ok: true, claimed: true })), +})); const req = () => new NextRequest("https://x.test/api/chat/generate", { @@ -121,4 +125,14 @@ describe("handleStartChatRun", () => { expect(mintEphemeralAccountKey).not.toHaveBeenCalled(); expect(deleteApiKey).not.toHaveBeenCalled(); }); + + // The published contract cross-references this: "start a headless run, then + // pass the returned chatId to GET /api/chat/{chatId}/stream to watch its + // output live". That route keys on chats.active_stream_id, so a headless run + // that never claims the slot is unresumable (chat#1923). + it("claims chats.active_stream_id with the run id so the run is resumable", async () => { + await handleStartChatRun({} as never); + + expect(compareAndSetChatActiveStreamId).toHaveBeenCalledWith("chat-1", null, "wrun_abc"); + }); }); diff --git a/lib/chat/runs/handleStartChatRun.ts b/lib/chat/runs/handleStartChatRun.ts index 41b68d1c7..4122834c8 100644 --- a/lib/chat/runs/handleStartChatRun.ts +++ b/lib/chat/runs/handleStartChatRun.ts @@ -8,6 +8,7 @@ import { mintEphemeralAccountKey } from "@/lib/keys/mintEphemeralAccountKey"; import { deleteApiKey } from "@/lib/supabase/account_api_keys/deleteApiKey"; import { buildRunAgentInput } from "@/lib/chat/buildRunAgentInput"; import { runAgentWorkflow } from "@/app/lib/workflows/runAgentWorkflow"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; /** Default title for the session a headless run provisions (no caller-supplied title). */ const DEFAULT_RUN_SESSION_TITLE = "Scheduled generation"; @@ -69,6 +70,21 @@ export async function handleStartChatRun(request: NextRequest): Promise; +} + +const chatIdSchema = z.string().uuid("chatId must be a valid UUID"); + +/** + * Authenticate the caller and confirm they own the chat behind a + * `/api/chat/{chatId}/…` route. + * + * Shared by `POST /api/chat/{chatId}/stop` and + * `GET /api/chat/{chatId}/stream` so both enforce identical auth, + * chat-id validation and ownership semantics. + * + * Honours the `account_id` **query** override, so an org/admin key can act + * on a member account's chat. Both routes carry their id in the path and + * parse no body, so a query param is the channel that works for `GET` and + * `POST` alike without consuming the request body. `validateAuthContext` is + * what decides whether the caller may actually use the override — passing it + * through does not weaken the check, and omitting it was why an admin key got + * a 403 on a chat it legitimately administers (same gap as `DELETE + * /api/tasks`, chat#1918). + * + * @param request - The incoming request, carrying the credentials. + * @param chatId - Chat id from the route params. + * @returns The auth context + chat row, or an error response + * (400 malformed id, 401 unauthenticated, 403 not owned, 404 missing). + */ +export async function validateChatOwnership( + request: NextRequest, + chatId: string, +): Promise { + const accountIdOverride = new URL(request.url).searchParams.get("account_id") ?? undefined; + const auth = await validateAuthContext(request, { accountId: accountIdOverride }); + if (auth instanceof NextResponse) return auth; + + const parsed = chatIdSchema.safeParse(chatId); + if (!parsed.success) { + const firstError = parsed.error.issues[0]; + return validationErrorResponse(firstError.message, firstError.path); + } + + const chats = await selectChats({ id: parsed.data }); + const chat = chats[0]; + if (!chat) return errorResponse("Chat not found", 404); + + const sessions = await selectSessions({ id: chat.session_id }); + if (sessions === null) return errorResponse("Internal server error", 500); + const session = sessions[0]; + if (!session) return errorResponse("Chat not found", 404); + if (session.account_id !== auth.accountId) return errorResponse("Forbidden", 403); + + return { auth, chat }; +} diff --git a/lib/chat/validateStopChatWorkflowRequest.ts b/lib/chat/validateStopChatWorkflowRequest.ts index 75bffe528..dc3d46a9d 100644 --- a/lib/chat/validateStopChatWorkflowRequest.ts +++ b/lib/chat/validateStopChatWorkflowRequest.ts @@ -1,43 +1,26 @@ import { NextRequest, NextResponse } from "next/server"; -import { z } from "zod"; -import { validateAuthContext } from "@/lib/auth/validateAuthContext"; -import type { AuthContext } from "@/lib/auth/validateAuthContext"; -import { selectChats } from "@/lib/supabase/chats/selectChats"; -import { selectSessions } from "@/lib/supabase/sessions/selectSessions"; -import { errorResponse } from "@/lib/networking/errorResponse"; -import { validationErrorResponse } from "@/lib/zod/validationErrorResponse"; -import type { Tables } from "@/types/database.types"; +import { + validateChatOwnership, + type ValidatedChatOwnership, +} from "@/lib/chat/validateChatOwnership"; -export interface ValidatedStopChatWorkflowRequest { - auth: AuthContext; - chat: Tables<"chats">; -} - -const chatIdSchema = z.string().uuid("chatId must be a valid UUID"); +export type ValidatedStopChatWorkflowRequest = ValidatedChatOwnership; -/** Validates POST /api/chat/{chatId}/stop: auth, chatId format, and chat + session-ownership lookup. */ +/** + * Validates POST /api/chat/{chatId}/stop: auth, chatId format, and chat + + * session-ownership lookup. + * + * Thin alias over `validateChatOwnership`, which `GET + * /api/chat/{chatId}/stream` shares — both routes must agree on who may + * touch a chat, so the rule lives in one place. + * + * @param request - The incoming request. + * @param chatId - Chat id from the route params. + * @returns The auth context + chat row, or an error response. + */ export async function validateStopChatWorkflowRequest( request: NextRequest, chatId: string, ): Promise { - const auth = await validateAuthContext(request); - if (auth instanceof NextResponse) return auth; - - const parsed = chatIdSchema.safeParse(chatId); - if (!parsed.success) { - const firstError = parsed.error.issues[0]; - return validationErrorResponse(firstError.message, firstError.path); - } - - const chats = await selectChats({ id: parsed.data }); - const chat = chats[0]; - if (!chat) return errorResponse("Chat not found", 404); - - const sessions = await selectSessions({ id: chat.session_id }); - if (sessions === null) return errorResponse("Internal server error", 500); - const session = sessions[0]; - if (!session) return errorResponse("Chat not found", 404); - if (session.account_id !== auth.accountId) return errorResponse("Forbidden", 403); - - return { auth, chat }; + return validateChatOwnership(request, chatId); }