fix(returns): keep the return-picture session binding on a revision conflict

src/__tests__/returns-guest-auth.test.ts

@@ -16,7 +16,7 @@
  * mocked — no live services required.
  */
 
-import { describe, it, expect, vi, beforeAll, beforeEach } from 'vitest';
+import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from 'vitest';
 import type { FastifyInstance } from 'fastify';
 import type { GuestInviteDoc, GuestSessionDoc, ProductionDoc } from '../db/types.js';
 
@@ -352,6 +352,84 @@ describe('Return-picture WHEP guest-token auth (issue #380)', () => {
   });
 });
 
+describe('Return-picture session binding under concurrent writes to the guest session', () => {
+  const bearer = (token: string) => ({ authorization: `Bearer ${token}` });
+  const revNumber = (rev: string | undefined) => Number(rev?.split('-')[0] ?? 0);
+  const defaultInsert = sessionsDb.insert.getMockImplementation()!;
+
+  // Enforce `_rev` like CouchDB: an update must carry the stored doc's current
+  // `_rev` or it gets a 409; each write bumps the revision.
+  beforeEach(() => {
+    sessionsDb.insert.mockImplementation(async (doc: GuestSessionDoc) => {
+      const stored = sessionsStore.get(doc._id);
+      if (stored && doc._rev !== stored._rev) throw Object.assign(new Error('conflict'), { statusCode: 409 });
+      sessionsStore.set(doc._id, { ...doc, _rev: `${revNumber(stored?._rev) + 1}-x` });
+      return { ok: true };
+    });
+  });
+  afterEach(() => {
+    sessionsDb.insert.mockImplementation(defaultInsert);
+  });
+
+  /** Answers the Strom WHEP POST after `during` has changed the guest session, as a concurrent request would. */
+  function stromAnswersAfter(during: () => void) {
+    vi.stubGlobal('fetch', vi.fn().mockImplementation(async () => {
+      during();
+      return {
+        ok: true,
+        status: 201,
+        text: async () => 'v=0 mock-answer-sdp',
+        headers: { get: (name: string) => (name === 'Location' ? '/whep/session-picture' : null) },
+      };
+    }));
+  }
+
+  it('keeps the binding, and the other write, when the session changed during the Strom round trip', async () => {
+    seedActiveProduction();
+    const invite = await createInvite('prod-1', { mixerInput: 'video_in_0' });
+    const { guestId } = await joinGuest(invite.id, invite.token);
+    // The guest page's talkback WHIP binding (guests.ts) lands while the
+    // return-picture offer is with Strom: the session doc this request read
+    // is no longer current when it writes its own binding.
+    stromAnswersAfter(() => {
+      const stored = sessionsStore.get(guestId)!;
+      sessionsStore.set(guestId, { ...stored, talkbackWhipSessionId: 'tb-1', _rev: `${revNumber(stored._rev) + 1}-x` });
+    });
+
+    const { res, sessionId } = await postPicture('video_in_0', bearer(invite.token));
+    expect(res.statusCode).toBe(201);
+    const session = sessionsStore.get(guestId)!;
+    expect(session.returnWhepSessionId).toBe(sessionId);
+    expect(session.talkbackWhipSessionId).toBe('tb-1');
+
+    // So the guest can still tear their own return session down.
+    const del = await app.inject({
+      method: 'DELETE',
+      url: `/api/v1/productions/prod-1/returns/video_in_0/picture/whep/${sessionId}`,
+      headers: bearer(invite.token),
+    });
+    expect(del.statusCode).toBe(204);
+  });
+
+  it('does not bind for a guest who left while the offer was with Strom', async () => {
+    seedActiveProduction();
+    const invite = await createInvite('prod-1', { mixerInput: 'video_in_0' });
+    const { guestId } = await joinGuest(invite.id, invite.token);
+    stromAnswersAfter(() => {
+      const stored = sessionsStore.get(guestId)!;
+      sessionsStore.set(guestId, { ...stored, state: 'left', _rev: `${revNumber(stored._rev) + 1}-x` });
+    });
+
+    const { res } = await postPicture('video_in_0', bearer(invite.token));
+    // The upstream session already exists, so the offer still succeeds…
+    expect(res.statusCode).toBe(201);
+    // …but the left session is not brought back to life with a binding.
+    const session = sessionsStore.get(guestId)!;
+    expect(session.state).toBe('left');
+    expect(session.returnWhepSessionId).toBeUndefined();
+  });
+});
+
 describe('Guest return mode: GET / PUT /api/v1/guests/:inviteId/session/return', () => {
   const returnUrl = (inviteId: string) => `/api/v1/guests/${inviteId}/session/return`;
   const bearer = (token: string) => ({ authorization: `Bearer ${token}` });

src/lib/db-errors.ts

@@ -13,3 +13,8 @@ export function isDocNotFoundError(err: unknown): boolean {
   if ((err as { statusCode?: number } | null)?.statusCode === 404) return true;
   return err instanceof Error && err.message.startsWith('Document type mismatch');
 }
+
+/** A CouchDB `.insert()` refused because the document's `_rev` is no longer current (409). */
+export function isDocConflictError(err: unknown): boolean {
+  return (err as { statusCode?: number } | null)?.statusCode === 409;
+}

src/routes/returns.ts

@@ -3,6 +3,7 @@ import { z } from 'zod';
 import { getDb, getGuestSessionsDb } from '../db/index.js';
 import type { GuestSessionDoc, ProductionDoc, ProductionSourceAssignment } from '../db/types.js';
 import { getStromToken } from '../lib/strom-token.js';
+import { isDocConflictError } from '../lib/db-errors.js';
 import { assertSameStromOrigin } from '../lib/url-validation.js';
 import { config, isGuestCallingEnabled } from '../config.js';
 import { resolveGuestSession } from '../lib/guest-scope.js';
@@ -116,23 +117,71 @@ async function proxyReturnPictureWhep(args: {
   }
   // Bind this WHEP session id to the guest's live session (issue #380) so the
   // matching DELETE can verify a guest caller only tears down their OWN return
-  // session — never another guest's :sessionId. Best-effort: a persist failure
-  // here does not fail the (already-established) upstream session; the guest
-  // just fails closed on their next DELETE instead of silently trusting an
-  // unbound id (see the DELETE handlers below).
+  // session — never another guest's :sessionId.
   if (guestSession && sessionId) {
+    await bindWhepSession(reply, 'return picture WHEP', {
+      session: guestSession,
+      apply: (doc, whepSessionId) => ({ ...doc, returnWhepSessionId: whepSessionId, updatedAt: new Date().toISOString() }),
+    }, sessionId);
+  }
+  reply.header('Content-Type', 'application/sdp');
+  return reply.status(201).send(answerSdp);
+}
+
+/**
+ * How a guest-scoped WHEP POST records its session id: the guest's session doc
+ * as read at the start of the request, and the one change to make to it.
+ * `apply` must only set its own key, because it may be re-applied to a newer
+ * copy of the doc (see `bindWhepSession`).
+ */
+interface GuestSessionBinding {
+  session: GuestSessionDoc;
+  apply: (doc: GuestSessionDoc, whepSessionId: string) => GuestSessionDoc;
+}
+
+/** Attempts at writing a WHEP session id into a guest session doc (one write plus retries on conflict). */
+const MAX_BIND_ATTEMPTS = 4;
+
+/**
+ * Writes the WHEP session id into the guest's session doc. The request read
+ * that doc at its start; while the Strom round trip ran, other requests of the
+ * same guest page may have written it (the talkback WHIP/WHEP bindings at join,
+ * a mute or a status change), so the insert can get a CouchDB 409. On a
+ * conflict the doc is read again, still has to be live (`state !== 'left'`),
+ * and only this request's key is applied to it, so neither this binding nor the
+ * other request's change is lost. Best-effort after that: a failure does not
+ * fail the (already-established) upstream session; the guest just fails closed
+ * on their next DELETE instead of trusting an unbound id.
+ */
+async function bindWhepSession(
+  reply: FastifyReply,
+  logName: string,
+  binding: GuestSessionBinding,
+  whepSessionId: string,
+): Promise<void> {
+  const db = getGuestSessionsDb();
+  let doc = binding.session;
+  for (let attempt = 1; ; attempt++) {
     try {
-      await getGuestSessionsDb().insert({
-        ...guestSession,
-        returnWhepSessionId: sessionId,
-        updatedAt: new Date().toISOString(),
-      });
+      await db.insert(binding.apply(doc, whepSessionId));
+      return;
+    } catch (err) {
+      if (!isDocConflictError(err) || attempt >= MAX_BIND_ATTEMPTS) {
+        reply.log.warn({ err, attempt }, `${logName}: guest session-id bind failed`);
+        return;
+      }
+    }
+    try {
+      doc = await db.get(doc._id);
     } catch (err) {
-      reply.log.warn({ err }, 'return picture WHEP: guest session-id bind failed');
+      reply.log.warn({ err }, `${logName}: guest session re-read failed, session-id not bound`);
+      return;
+    }
+    if (doc.state === 'left') {
+      reply.log.info(`${logName}: guest left during the request, session-id not bound`);
+      return;
     }
   }
-  reply.header('Content-Type', 'application/sdp');
-  return reply.status(201).send(answerSdp);
 }
 
 /**