Spaces:
Running
Running
| 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; | |
| } | |