Blame for src/analysis.ts

import { createHash, timingSafeEqual } from "node:crypto";
import express from "express";
import type { SupabaseClient } from "@supabase/supabase-js";
import { z } from "zod";
import { env } from "./config.js";
import { activeSubscriptionFor } from "./subscriptions.js";

export function stableId(value: string) {
  const hex = createHash("sha256").update(value).digest("hex");
  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-5${hex.slice(13, 16)}-8${hex.slice(17, 20)}-${hex.slice(20, 32)}`;
}

const beginSchema = z.object({ clientId: z.uuid(), playerId: z.uuid(), name: z.string().trim().max(150) });
const reportSchema = z.object({
  index: z.number().int().nonnegative(), action: z.string().min(1).max(50),
  start_frame: z.number().int().nonnegative().optional(), end_frame: z.number().int().nonnegative().optional(),
  start_time_seconds: z.number().finite().nonnegative().optional(),
  end_time_seconds: z.number().finite().positive().optional(),
  error: z.string().max(2000).optional(),
});
export const completionSchema = z.discriminatedUnion("status", [
  z.object({ status: z.literal("completed"), path: z.string(), actions: z.array(reportSchema).max(5000) }),
  z.object({ status: z.enum(["failed", "cancelled"]) }),
]);

export function actionRows(sessionId: string, reports: z.infer<typeof reportSchema>[]) {
  if (new Set(reports.map((report) => report.index)).size !== reports.length) throw new Error("Duplicate action index");
  return reports.filter((report) => report.error === undefined).map((report) => {
    const { start_frame: startFrame, end_frame: endFrame,
      start_time_seconds: startTime, end_time_seconds: endTime } = report;
    if (startFrame === undefined || endFrame === undefined || endFrame < startFrame) throw new Error("Invalid action interval");
    if ((startTime === undefined) !== (endTime === undefined) ||
        (startTime !== undefined && endTime !== undefined && endTime <= startTime)) throw new Error("Invalid action playback times");
    return { action_id: stableId(`${sessionId}:${report.index}`), type: report.action,
      start_frame: startFrame, end_frame: endFrame,
      start_time_seconds: startTime ?? null, end_time_seconds: endTime ?? null };
  });
}

export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?: (sessionId: string, userId: string) => Promise<void>) {
  const router = express.Router();
  router.use((req, res, next) => {
    const supplied = Buffer.from(req.header("x-sporalize-proxy-secret") ?? "");
    const expected = Buffer.from(secret);
    if (supplied.length !== expected.length || !timingSafeEqual(supplied, expected)) return res.sendStatus(403);
    const user = z.uuid().safeParse(req.header("x-sporalize-user-id"));
    if (!user.success) return res.sendStatus(403);
    res.locals.userId = user.data;
    res.locals.isAdmin = req.header("x-sporalize-admin") === "true";
    next();
  });
  router.use(express.json({ limit: "2mb" }));
  router.post("/", async (req, res, next) => {
    try {
      const input = beginSchema.parse(req.body);
      const userId = res.locals.userId as string;
      const isAdmin = res.locals.isAdmin === true;
      const subscription = isAdmin ? null : await activeSubscriptionFor(userId, { refresh: true });
      if (!isAdmin && !subscription) {
        return res.status(402).json({ error: "An active subscription is required to start a new analysis.", code: "subscription_required" });
      }
      if (!isAdmin) {
        const { data: usage, error: usageError } = await db.from("users_subscriptions")
          .select("num_sessions").eq("user_id", userId).maybeSingle();
        if (usageError) return res.status(503).json({ error: "Unable to check your session quota." });
        const usedSessions = Number(usage?.num_sessions ?? 0);
        const sessionLimit = subscription?.session_limit == null ? null : Number(subscription.session_limit);
        if (sessionLimit !== null && usedSessions >= sessionLimit) {
          return res.status(403).json({
            error: "You have used all sessions included in this plan period. Renew or upgrade your plan to analyze another session.",
            code: "quota_exceeded", resource: "sessions", used: usedSessions, limit: sessionLimit,
          });
        }
      }
      let playerQuery = db.from("players").select("player_id,user_id").eq("player_id", input.playerId);
      if (!isAdmin) playerQuery = playerQuery.eq("user_id", userId);
      const { data: player, error: playerError } = await playerQuery.maybeSingle();
      if (playerError) return res.status(503).json({ error: "Unable to verify the selected player." });
      if (!player) return res.status(404).json({ error: isAdmin
        ? "The selected player was not found. Refresh the player list and try again."
        : "Choose a player from your own roster before starting analysis." });
      if (isAdmin) {
        const { error } = await db.rpc("admin_create_analysis_session", {
          p_session_id: input.clientId,
          p_user_id: player.user_id,
          p_player_id: input.playerId,
          p_name: input.name || `Session ${input.clientId.slice(0, 8)}`,
          p_path: `sessions/${input.clientId}`,
          p_by_admin: player.user_id !== userId,
        });
        if (error) return res.status(503).json({ error: "Unable to prepare analysis session." });
        return res.json({ session_id: input.clientId, player_id: input.playerId, user_id: player.user_id, by_admin: player.user_id !== userId, actor_is_admin: true });
      }
      const { data, error } = await db.rpc("begin_analysis", {
        p_session_id: input.clientId, p_user_id: userId,
        p_player_id: input.playerId,
        p_reference: null,
        p_name: input.name || `Session ${input.clientId.slice(0, 8)}`,
      });
      if (error) {
        console.error("Unable to prepare analysis session", { code: error.code, message: error.message });
        if (error.code === "P0001" && error.message.includes("subscription_quota_exceeded")) {
          return res.status(403).json({ error: "You have used all sessions included in this plan period. Renew or upgrade your plan to analyze another session.", code: "quota_exceeded", resource: "sessions" });
        }
        if (error.code === "P0001" && error.message.includes("subscription_required")) {
          return res.status(402).json({ error: "An active subscription is required to start an analysis.", code: "subscription_required" });
        }
        return res.status(error.code === "P0001" ? 409 : 503).json({ error: "Unable to prepare analysis session" });
      }
      return res.json({ ...data, user_id: userId, by_admin: false, actor_is_admin: false });
    } catch (error) { next(error); }
  });
  router.get("/:sessionId", async (req, res, next) => {
    try {
      const sessionId = z.uuid().parse(req.params.sessionId);
      let query = db.from("sessions").select("session_id,player_id,user_id,path,status,name,created_at,by_admin")
        .eq("session_id", sessionId);
      if (res.locals.isAdmin !== true) query = query.eq("user_id", res.locals.userId);
      const { data, error } = await query.maybeSingle();
      if (error) return res.status(503).json({ error: "Unable to load session" });
      if (!data) return res.sendStatus(404);
      if (req.query.playbackTimes === "false") return res.json(data);
      const { data: actionTimes, error: timesError } = await db.from("actions")
        .select("action_id,start_time_seconds,end_time_seconds").eq("session_id", sessionId);
      if (timesError) {
        console.error("Unable to load action playback times", { sessionId, code: timesError.code, message: timesError.message });
      }
      return res.json({ ...data, action_times: actionTimes ?? [] });
    } catch (error) { next(error); }
  });
  router.delete("/:sessionId/actions/:actionId", async (req, res, next) => {
    try {
      const sessionId = z.uuid().parse(req.params.sessionId);
      const actionId = z.uuid().parse(req.params.actionId);
      const userId = res.locals.userId as string;
      let sessionQuery = db.from("sessions").select("session_id,status").eq("session_id", sessionId);
      if (res.locals.isAdmin !== true) sessionQuery = sessionQuery.eq("user_id", userId);
      const { data: session, error: sessionError } = await sessionQuery.maybeSingle();
      if (sessionError) return res.status(503).json({ error: "Unable to verify session ownership." });
      if (!session) return res.status(404).json({ error: "Session not found." });
      if (["pending", "uploading", "queued", "running"].includes(session.status)) return res.status(409).json({ error: "Actions cannot be deleted while analysis is running." });

      const { data: action, error } = await db.from("actions").delete()
        .eq("action_id", actionId).eq("session_id", sessionId).select("action_id").maybeSingle();
      if (error) return res.status(503).json({ error: "Unable to delete action." });
      if (!action) return res.status(404).json({ error: "Action not found." });
      return res.sendStatus(204);
    } catch (error) { next(error); }
  });
  router.patch("/:sessionId/actions/:actionId", async (req, res, next) => {
    try {
      if (res.locals.isAdmin !== true) return res.sendStatus(403);
      const sessionId = z.uuid().parse(req.params.sessionId);
      const actionId = z.uuid().parse(req.params.actionId);
      const input = z.object({
        startFrame: z.number().int().nonnegative(),
        endFrame: z.number().int().nonnegative(),
        type: z.string().trim().min(1).max(50),
        startTimeSeconds: z.number().finite().nonnegative(),
        endTimeSeconds: z.number().finite().positive(),
      }).strict().parse(req.body);
      if (input.endFrame < input.startFrame) return res.status(400).json({ error: "Action end frame must be after its start frame." });
      if (input.endTimeSeconds <= input.startTimeSeconds) return res.status(400).json({ error: "Action playback interval is invalid." });
      const { data: session } = await db.from("sessions").select("status").eq("session_id", sessionId).maybeSingle();
      if (!session || ["pending", "uploading", "queued", "running"].includes(session.status)) return res.status(409).json({ error: "Actions cannot be edited while analysis is running." });
      const { data, error } = await db.from("actions").update({
        start_frame: input.startFrame, end_frame: input.endFrame, type: input.type,
        start_time_seconds: input.startTimeSeconds, end_time_seconds: input.endTimeSeconds,
      }).eq("action_id", actionId).eq("session_id", sessionId).select("action_id").maybeSingle();
      if (error && !["42703", "PGRST204"].includes(error.code)) return res.status(503).json({ error: "Unable to update action." });
      let updated = data;
      if (error) {
        const fallback = await db.from("actions").update({
          start_frame: input.startFrame, end_frame: input.endFrame, type: input.type,
        }).eq("action_id", actionId).eq("session_id", sessionId).select("action_id").maybeSingle();
        if (fallback.error) return res.status(503).json({ error: "Unable to update action." });
        updated = fallback.data;
      }
      if (!updated) return res.sendStatus(404);
      return res.json({ ok: true });
    } catch (error) { next(error); }
  });
  router.delete("/:sessionId", async (req, res, next) => {
    try {
      const sessionId = z.uuid().parse(req.params.sessionId);
      const userId = res.locals.userId as string;
      let sessionQuery = db.from("sessions").select("session_id").eq("session_id", sessionId);
      if (res.locals.isAdmin !== true) sessionQuery = sessionQuery.eq("user_id", userId);
      const { data: session, error: sessionError } = await sessionQuery.maybeSingle();
      if (sessionError) return res.status(503).json({ error: "Unable to verify session ownership." });
      if (!session) return res.status(404).json({ error: "Session not found." });

      // The schema keeps actions when a session row is removed, so explicitly
      // delete them first. Their annotations and feedback cascade from actions.
      const { error: actionsError } = await db.from("actions").delete().eq("session_id", sessionId);
      if (actionsError) return res.status(503).json({ error: "Unable to remove session actions." });
      let deleteQuery = db.from("sessions").delete().eq("session_id", sessionId);
      if (res.locals.isAdmin !== true) deleteQuery = deleteQuery.eq("user_id", userId);
      const { data: deleted, error } = await deleteQuery.select("session_id").maybeSingle();
      if (error) return res.status(503).json({ error: "Unable to delete session." });
      if (!deleted) return res.status(404).json({ error: "Session not found." });
      return res.sendStatus(204);
    } catch (error) { next(error); }
  });
  router.put("/:sessionId", async (req, res, next) => {
    try {
      const sessionId = z.uuid().parse(req.params.sessionId);
      const input = completionSchema.parse(req.body);
      if (input.status === "completed" && input.path !== `sessions/${sessionId}`) return res.sendStatus(400);
      let actions;
      try { actions = input.status === "completed" ? actionRows(sessionId, input.actions) : []; }
      catch { return res.status(400).json({ error: "Invalid action details" }); }
      if (res.locals.isAdmin === true) {
        const { data: session, error: sessionError } = await db.from("sessions").update({
          status: input.status,
          ...(input.status === "completed" ? { path: input.path } : {}),
        }).eq("session_id", sessionId).select("session_id,user_id").maybeSingle();
        if (sessionError) return res.status(503).json({ error: "Unable to save analysis session." });
        if (!session) return res.sendStatus(404);
        if (input.status === "completed") {
          const rows = actions.map((action) => ({ ...action, session_id: sessionId }));
          if (rows.length) {
            const { error } = await db.from("actions").upsert(rows, { onConflict: "action_id" });
            if (error && !["42703", "PGRST204"].includes(error.code)) return res.status(503).json({ error: "Unable to save analysis actions." });
            if (error) {
              const legacyRows = rows.map(({ start_time_seconds: _start, end_time_seconds: _end, ...row }) => row);
              const fallback = await db.from("actions").upsert(legacyRows, { onConflict: "action_id" });
              if (fallback.error) return res.status(503).json({ error: "Unable to save analysis actions." });
            }
          }
          if (onCompleted && session.user_id) void onCompleted(sessionId, session.user_id).catch((error) => console.error("Automatic Agent feedback generation failed", { sessionId, error }));
        }
        return res.json({ ok: true });
      }
      const { error } = await db.rpc("finish_analysis", {
        p_session_id: sessionId, p_user_id: res.locals.userId, p_status: input.status,
        p_path: input.status === "completed" ? input.path : null, p_actions: actions,
      });
      if (error) {
        console.error("Unable to save analysis results", { code: error.code, message: error.message });
        return res.status(error.code === "P0001" ? 409 : 503).json({ error: "Unable to save analysis" });
      }
      if (input.status === "completed" && onCompleted) {
        const { data: completedSession } = await db.from("sessions").select("user_id").eq("session_id", sessionId).maybeSingle();
        if (completedSession?.user_id) void onCompleted(sessionId, completedSession.user_id).catch((generationError) => console.error("Automatic Agent feedback generation failed", { sessionId, generationError }));
      }
      return res.json({ ok: true });
    } catch (error) { next(error); }
  });
  return router;
}