diff --git a/backend/src/routes/v2.routes.ts b/backend/src/routes/v2.routes.ts index dfbd0ec..0afee1c 100644 --- a/backend/src/routes/v2.routes.ts +++ b/backend/src/routes/v2.routes.ts @@ -1,5 +1,6 @@ import { FastifyInstance } from 'fastify'; import { createClient, RedisClientType } from 'redis'; +import { randomUUID } from 'node:crypto'; import { DbService } from '../services/db.service.js'; import { SessionDirector } from '../services/session-director.service.js'; import { Candidate } from '../services/generators.service.js'; @@ -39,6 +40,34 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe await redisClient.del(planKey(userId)); } + // ponytail: per-user SET NX PX lock around the getActivePlan->mutate->setActivePlan + // span, released via a CAS Lua script so a slow holder can't delete a lock a later + // request already owns. Ceiling: this serializes /next and /feedback per-user only + // (fine — plans aren't shared across users); if plan mutation logic ever needs to + // span multiple keys/services atomically, swap this for a WATCH/MULTI transaction + // or move the plan into a single Lua script instead of an app-level lock. + const RELEASE_LOCK_LUA = `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`; + + async function withPlanLock(userId: string, fn: () => Promise): Promise { + const lockKey = `v2:planlock:${userId}`; + const token = randomUUID(); + const deadline = Date.now() + 5000; + let acquired = false; + while (Date.now() < deadline) { + const res = await redisClient.set(lockKey, token, { NX: true, PX: 3000 }); + if (res) { acquired = true; break; } + await new Promise((r) => setTimeout(r, 20 + Math.random() * 30)); + } + if (!acquired) { + throw new Error('Timed out waiting for active-plan lock'); + } + try { + return await fn(); + } finally { + await redisClient.eval(RELEASE_LOCK_LUA, { keys: [lockKey], arguments: [token] }); + } + } + /** * POST /api/v2/vibe/start — start a v2 session * Body: { seedTrackId? } @@ -61,25 +90,34 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe */ fastify.get('/v2/vibe/next', async (request, reply) => { const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const active = await getActivePlan(userId); - if (!active || active.plan.length === 0) { + const result = await withPlanLock(userId, async () => { + const active = await getActivePlan(userId); + + if (!active || active.plan.length === 0) { + return null; + } + + const next = active.plan.shift()!; + // Enrich with track details + const track = await dbService.getTrackById(next.trackId); + + // Replan if running low + if (active.plan.length < 5) { + const refill = await director.replan(userId, active.sessionId, active.plan, [next.trackId], active.seedTrackId ?? undefined); + active.plan = refill; + } + + await setActivePlan(userId, active); + + return { track, explanation: next.explanation, planRemaining: active.plan.length }; + }); + + if (!result) { return reply.code(404).send({ error: 'No active plan. POST /api/v2/vibe/start first.' }); } - const next = active.plan.shift()!; - // Enrich with track details - const track = await dbService.getTrackById(next.trackId); - - // Replan if running low - if (active.plan.length < 5) { - const refill = await director.replan(userId, active.sessionId, active.plan, [next.trackId], active.seedTrackId ?? undefined); - active.plan = refill; - } - - await setActivePlan(userId, active); - - return reply.send({ track, explanation: next.explanation, planRemaining: active.plan.length }); + return reply.send(result); }); /** @@ -107,15 +145,17 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe } // Replan the session - const active = await getActivePlan(userId); - if (active) { + const planRemaining = await withPlanLock(userId, async () => { + const active = await getActivePlan(userId); + if (!active) return 0; const playedTrackIds = [trackId]; const refill = await director.replan(userId, active.sessionId, active.plan, playedTrackIds, active.seedTrackId ?? undefined); active.plan = refill; await setActivePlan(userId, active); - } + return active.plan.length; + }); - return reply.send({ status: 'ok', planRemaining: active?.plan.length ?? 0 }); + return reply.send({ status: 'ok', planRemaining }); }); /**