fix(api): require owner auth for pipeline process (RL-002)

app/api/analysis/[id]/process/route.test.ts

@@ -0,0 +1,130 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextResponse } from "next/server";
+
+const mockRequireAuthUser = vi.fn();
+const mockFindUnique = vi.fn();
+const mockUpdateMany = vi.fn();
+const mockCheckRateLimit = vi.fn();
+const mockTriggerPipeline = vi.fn();
+
+vi.mock("@/lib/auth-helpers", () => ({
+  requireAuthUser: () => mockRequireAuthUser(),
+  getRequestId: () => "test-request-id",
+  unauthorizedResponse: () =>
+    NextResponse.json(
+      {
+        success: false,
+        error: "Sign in required",
+        code: "UNAUTHORIZED",
+      },
+      { status: 401 }
+    ),
+}));
+
+vi.mock("@/lib/prisma", () => ({
+  prisma: {
+    analysisSession: {
+      findUnique: (...args: unknown[]) => mockFindUnique(...args),
+      updateMany: (...args: unknown[]) => mockUpdateMany(...args),
+    },
+  },
+}));
+
+vi.mock("@/lib/rate-limit", () => ({
+  checkRateLimit: (...args: unknown[]) => mockCheckRateLimit(...args),
+  getClientIp: () => "127.0.0.1",
+  rateLimitResponseHeaders: () => ({}),
+}));
+
+vi.mock("@/lib/jobs/pipeline-trigger", () => ({
+  triggerAnalysisPipeline: (...args: unknown[]) => mockTriggerPipeline(...args),
+}));
+
+import { POST } from "./route";
+
+const SLUG = "process-slug-abc";
+const SESSION_ID = "sess_process_1";
+const OWNER_ID = "user_owner_1";
+const OTHER_ID = "user_other_1";
+
+function callPost(slug = SLUG) {
+  return POST(new Request(`http://localhost/api/analysis/${slug}/process`, {
+    method: "POST",
+  }), {
+    params: { id: slug },
+  });
+}
+
+function makeSession(overrides: Record<string, unknown> = {}) {
+  return {
+    id: SESSION_ID,
+    userId: OWNER_ID,
+    status: "PENDING",
+    updatedAt: new Date(),
+    ...overrides,
+  };
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  mockCheckRateLimit.mockResolvedValue({ ok: true });
+  mockRequireAuthUser.mockResolvedValue({ userId: OWNER_ID, session: {} });
+  mockFindUnique.mockResolvedValue(makeSession());
+  mockUpdateMany.mockResolvedValue({ count: 1 });
+  mockTriggerPipeline.mockResolvedValue("inline");
+});
+
+describe("POST /api/analysis/[id]/process", () => {
+  it("returns 401 when unauthenticated", async () => {
+    mockRequireAuthUser.mockResolvedValue(null);
+
+    const res = await callPost();
+    const body = await res.json();
+
+    expect(res.status).toBe(401);
+    expect(body.success).toBe(false);
+    expect(body.code).toBe("UNAUTHORIZED");
+    expect(mockTriggerPipeline).not.toHaveBeenCalled();
+  });
+
+  it("returns 403 when authenticated user is not the session owner", async () => {
+    mockRequireAuthUser.mockResolvedValue({ userId: OTHER_ID, session: {} });
+
+    const res = await callPost();
+    const body = await res.json();
+
+    expect(res.status).toBe(403);
+    expect(body.success).toBe(false);
+    expect(body.code).toBe("FORBIDDEN");
+    expect(mockTriggerPipeline).not.toHaveBeenCalled();
+  });
+
+  it("starts the pipeline for the session owner", async () => {
+    const res = await callPost();
+    const body = await res.json();
+
+    expect(res.status).toBe(200);
+    expect(body.success).toBe(true);
+    expect(body.data).toEqual({ started: true, mode: "inline" });
+    expect(mockTriggerPipeline).toHaveBeenCalledWith({
+      sessionId: SESSION_ID,
+      requestId: "test-request-id",
+    });
+  });
+
+  it("returns 429 when rate limited", async () => {
+    mockCheckRateLimit.mockResolvedValue({
+      ok: false,
+      retryAfterSec: 30,
+    });
+
+    const res = await callPost();
+    const body = await res.json();
+
+    expect(res.status).toBe(429);
+    expect(body.success).toBe(false);
+    expect(body.code).toBe("RATE_LIMITED");
+    expect(mockRequireAuthUser).not.toHaveBeenCalled();
+    expect(mockTriggerPipeline).not.toHaveBeenCalled();
+  });
+});

app/api/analysis/[id]/process/route.ts

@@ -8,7 +8,11 @@ import {
 } from "@/lib/rate-limit";
 import { triggerAnalysisPipeline } from "@/lib/jobs/pipeline-trigger";
 import { createLogger } from "@/lib/logger";
-import { getRequestId } from "@/lib/auth-helpers";
+import {
+  getRequestId,
+  requireAuthUser,
+  unauthorizedResponse,
+} from "@/lib/auth-helpers";
 import type { ApiResponse } from "@/types";
 
 export const maxDuration = 60;
@@ -46,16 +50,28 @@ export async function POST(
     );
   }
 
+  const authUser = await requireAuthUser();
+  if (!authUser) {
+    return unauthorizedResponse();
+  }
+
   try {
     const session = await prisma.analysisSession.findUnique({
       where: { shareableSlug: slug },
-      select: { id: true, status: true, updatedAt: true },
+      select: { id: true, userId: true, status: true, updatedAt: true },
     });
 
     if (!session) {
       return NextResponse.json(
-        { success: false as const, error: "Session not found" },
-        { status: 404 }
+        { success: false as const, error: "Forbidden", code: "FORBIDDEN" },
+        { status: 403 }
+      );
+    }
+
+    if (session.userId !== authUser.userId) {
+      return NextResponse.json(
+        { success: false as const, error: "Forbidden", code: "FORBIDDEN" },
+        { status: 403 }
       );
     }
 

app/dashboard/[id]/dashboard-client.tsx

@@ -103,6 +103,8 @@ export function DashboardClient({
   }, [slug]);
 
   const startPipeline = useCallback(async () => {
+    if (!isOwner) return;
+
     try {
       await apiPost(`/api/analysis/${slug}/process`, {});
     } catch (err) {
@@ -111,7 +113,7 @@ export function DashboardClient({
       }
       // Non-fatal — pipeline may already be running
     }
-  }, [slug]);
+  }, [slug, isOwner]);
 
   const retryAnalysis = useCallback(async () => {
     setPageError(null);
@@ -131,7 +133,10 @@ export function DashboardClient({
     }, 1_000);
 
     async function init() {
-      if (initialStatus === "PENDING" || initialStatus === "PROCESSING") {
+      if (
+        isOwner &&
+        (initialStatus === "PENDING" || initialStatus === "PROCESSING")
+      ) {
         await startPipeline();
       }
 
@@ -173,6 +178,7 @@ export function DashboardClient({
     return () => stopPolling();
   }, [
     slug,
+    isOwner,
     initialStatus,
     fetchStatus,
     startPipeline,

e2e/golden-path.spec.ts

@@ -31,6 +31,22 @@ test.describe("golden path: upload → preview → submit", () => {
       });
     });
 
+    // Process is owner-only; mock it so the dashboard handoff stays deterministic.
+    await page.route("**/api/analysis/*/process", async (route) => {
+      if (route.request().method() !== "POST") {
+        await route.fallback();
+        return;
+      }
+      await route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({
+          success: true,
+          data: { started: true, mode: "inline" },
+        }),
+      });
+    });
+
     await page.goto("/analyze");
 
     // Upload the CSV via the (possibly hidden) file input.

middleware.ts

@@ -33,6 +33,6 @@ export default auth((req) => {
 
 export const config = {
   matcher: [
-    "/((?!_next/static|_next/image|favicon.ico|opengraph-image|api/inngest|api/health|api/auth|api/analysis/.+/status|api/analysis/.+/process).*)",
+    "/((?!_next/static|_next/image|favicon.ico|opengraph-image|api/inngest|api/health|api/auth|api/analysis/.+/status).*)",
   ],
 };