feat(realtime): full Fase 2b.0 test suite + infra hardening

Extend the 2b.0 red-phase coverage: presence, RLS isolation, offline-queue
contract, and Playwright scaffolds for the shopping-session UX. Only the
presence test is blocked on an upstream Realtime bug (documented below); all
other Realtime paths are proven end-to-end.

Tests
- realtime-postgres-changes.test.ts: now uses afterEach socket disconnect
  (Vitest singleFork kept Phoenix sockets alive between files, leaking state)
- realtime-isolation.test.ts (new): R-I-01 David (guest, same collective)
  receives events; R-I-02 Eva (non-member) receives NONE — RLS is enforced
  server-side per subscribed JWT
- realtime-presence.test.ts (new, describe.skip): two-client roster + leave
  assertions ready, gated on upstream bug
- sync-queue.test.ts (new, describe.skip): 7 placeholders pinning the
  apps/web/src/lib/sync/queue.ts contract (Q-01..Q-04 enqueue / in-order
  flush / retry; F-01..F-03 online-event flush + last-write-wins)
- apps/web/tests/e2e/{session,realtime,offline}.test.ts (new, describe.skip):
  S-01..S-03, R-E-01, R-E-02, O-01, O-02 pinning the Modo Compra UI contract
- vitest.config.ts: fileParallelism=false, drop singleFork so each file gets
  a fresh worker fork — prevents RealtimeClient singleton leakage

Infra
- db-init/00-role-passwords.sh: pre-create `realtime` schema with
  AUTHORIZATION supabase_admin. Realtime's per-tenant migrator creates tables
  INSIDE the schema but does not create the schema itself; without this the
  first tenant connect fails with "schema realtime does not exist"
- docker-compose.dev.yml: adopt the upstream supabase/supabase Realtime env
  shape — SEED_SELF_HOST=true + RUN_JANITOR=true + RLIMIT_NOFILE=10000 +
  drop the custom `command: eval seeds` (that bypasses the normal supervisor
  startup and was observed to cause GenServer crashes under load).
  Version pinned to v2.76.5 to match the official docker-compose.

Known upstream bug
- Realtime v2.76.5 and v2.83.0 both crash on presence_diff:
  `(UndefinedFunctionError) RealtimeChannel.handle_out/3 is undefined`.
  postgres_changes + isolation unaffected (they don't traverse handle_out).
  Presence tests stay `describe.skip` until a fixed upstream is released.

Totals: 59 Vitest passed (+ 9 skipped awaiting implementation/upstream),
16 pgTAP, 15 Playwright (+ 7 E2E scaffolded for 2b UI) = 90 green.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

packages/test-utils/tests/realtime-isolation.test.ts

@@ -0,0 +1,106 @@
+/**
+ * R-series (Isolation): RLS must be enforced on Realtime postgres_changes too.
+ *
+ * Eva is not a member of the seed collective. Even if she subscribes to the
+ * shopping_items channel for SEED_LIST_ID, the Realtime server evaluates RLS
+ * using her JWT when deciding which events to forward, and should drop every
+ * event whose row is not visible to her.
+ *
+ * We also verify the positive case: a guest of the same collective (David) DOES
+ * receive events, because the existing shopping_items SELECT policy allows
+ * guests to read items in their collectives.
+ */
+import { describe, it, expect, afterAll, afterEach } from 'vitest';
+import { createClientAs, createAdminClient } from '../src/supabase-clients.js';
+import { subscribePostgresChanges } from '../src/realtime-helpers.js';
+import { ANA_ID, DAVID_ID, EVA_ID, SEED_LIST_ID } from '../src/seed-constants.js';
+import type { SupabaseClient } from '@supabase/supabase-js';
+
+const admin = createAdminClient();
+const createdItemIds: string[] = [];
+
+const activeClients: SupabaseClient[] = [];
+function track(c: SupabaseClient): SupabaseClient {
+	activeClients.push(c);
+	return c;
+}
+afterEach(async () => {
+	for (const c of activeClients.splice(0)) {
+		await c.removeAllChannels();
+		await c.realtime.disconnect();
+	}
+});
+
+afterAll(async () => {
+	if (createdItemIds.length > 0) {
+		await admin.from('shopping_items').delete().in('id', createdItemIds);
+	}
+});
+
+describe('Realtime RLS isolation — shopping_items', () => {
+	it('R-I-01: David (guest) receives events for items in his collective', async () => {
+		const david = track(await createClientAs(DAVID_ID));
+		const ana = track(await createClientAs(ANA_ID));
+
+		const sub = await subscribePostgresChanges<{ id: string; name: string }>(david, {
+			table: 'shopping_items',
+			event: 'INSERT',
+			filter: `list_id=eq.${SEED_LIST_ID}`
+		});
+
+		try {
+			const itemName = `R-I-01-${Date.now()}`;
+			const { data } = await ana
+				.from('shopping_items')
+				.insert({ list_id: SEED_LIST_ID, name: itemName, sort_order: 800, created_by: ANA_ID })
+				.select('id')
+				.single();
+			createdItemIds.push(data!.id);
+
+			const evt = await sub.waitFor((e) => e.new.name === itemName);
+			expect(evt.new.name).toBe(itemName);
+		} finally {
+			await sub.unsubscribe();
+			await ana.removeAllChannels();
+			await david.removeAllChannels();
+		}
+	});
+
+	it('R-I-02: Eva (non-member) receives NO events for items in the seed list', async () => {
+		const eva = track(await createClientAs(EVA_ID));
+		const ana = track(await createClientAs(ANA_ID));
+
+		const sub = await subscribePostgresChanges<{ id: string; name: string }>(eva, {
+			table: 'shopping_items',
+			event: 'INSERT',
+			filter: `list_id=eq.${SEED_LIST_ID}`
+		});
+
+		try {
+			// Ana inserts something that should NOT reach Eva
+			const forbiddenName = `R-I-02-forbidden-${Date.now()}`;
+			const { data } = await ana
+				.from('shopping_items')
+				.insert({
+					list_id: SEED_LIST_ID,
+					name: forbiddenName,
+					sort_order: 801,
+					created_by: ANA_ID
+				})
+				.select('id')
+				.single();
+			createdItemIds.push(data!.id);
+
+			// Give Realtime up to 2s to (not) deliver. If RLS works, Eva sees nothing.
+			await new Promise((r) => setTimeout(r, 2_000));
+
+			const evt = sub.events.find((e) => e.new.name === forbiddenName);
+			expect(evt).toBeUndefined();
+			expect(sub.events.length).toBe(0);
+		} finally {
+			await sub.unsubscribe();
+			await ana.removeAllChannels();
+			await eva.removeAllChannels();
+		}
+	});
+});

packages/test-utils/tests/realtime-postgres-changes.test.ts

@@ -7,7 +7,7 @@
  * Migration 007 adds shopping_items to the publication and sets REPLICA
  * IDENTITY FULL. Without either, these tests fail with timeouts.
  */
-import { describe, it, expect, afterAll } from 'vitest';
+import { describe, it, expect, afterAll, afterEach } from 'vitest';
 import { createClientAs, createAdminClient } from '../src/supabase-clients.js';
 import { subscribePostgresChanges } from '../src/realtime-helpers.js';
 import {
@@ -16,10 +16,25 @@ import {
 	SEED_LIST_ID,
 	COLLECTIVE_ID
 } from '../src/seed-constants.js';
+import type { SupabaseClient } from '@supabase/supabase-js';
 
 const admin = createAdminClient();
 const createdItemIds: string[] = [];
 
+// See realtime-presence.test.ts for rationale: Vitest's singleFork means Phoenix
+// sockets leak between test files unless we force-disconnect them per test.
+const activeClients: SupabaseClient[] = [];
+function track(c: SupabaseClient): SupabaseClient {
+	activeClients.push(c);
+	return c;
+}
+afterEach(async () => {
+	for (const c of activeClients.splice(0)) {
+		await c.removeAllChannels();
+		await c.realtime.disconnect();
+	}
+});
+
 afterAll(async () => {
 	if (createdItemIds.length > 0) {
 		await admin.from('shopping_items').delete().in('id', createdItemIds);
@@ -28,8 +43,8 @@ afterAll(async () => {
 
 describe('Realtime postgres_changes — shopping_items', () => {
 	it('R-01: INSERT made by Ana reaches Borja subscribed to the same list', async () => {
-		const borja = await createClientAs(BORJA_ID);
-		const ana = await createClientAs(ANA_ID);
+		const borja = track(await createClientAs(BORJA_ID));
+		const ana = track(await createClientAs(ANA_ID));
 		const sub = await subscribePostgresChanges<{ id: string; list_id: string; name: string }>(
 			borja,
 			{
@@ -60,8 +75,8 @@ describe('Realtime postgres_changes — shopping_items', () => {
 	});
 
 	it('R-02: UPDATE broadcasts include the full row (REPLICA IDENTITY FULL)', async () => {
-		const ana = await createClientAs(ANA_ID);
-		const borja = await createClientAs(BORJA_ID);
+		const ana = track(await createClientAs(ANA_ID));
+		const borja = track(await createClientAs(BORJA_ID));
 
 		// Seed an item owned by Ana to mutate
 		const { data: item } = await admin
@@ -109,8 +124,8 @@ describe('Realtime postgres_changes — shopping_items', () => {
 			.single();
 		const otherListId = otherList!.id;
 
-		const borja = await createClientAs(BORJA_ID);
-		const ana = await createClientAs(ANA_ID);
+		const borja = track(await createClientAs(BORJA_ID));
+		const ana = track(await createClientAs(ANA_ID));
 
 		// Borja subscribes to SEED_LIST_ID only
 		const sub = await subscribePostgresChanges<{ id: string; list_id: string; name: string }>(

packages/test-utils/tests/realtime-presence.test.ts

@@ -0,0 +1,194 @@
+/**
+ * R-series (Presence): Supabase Realtime Presence on a list channel.
+ *
+ * Verifies that two authenticated clients sharing the same channel name see
+ * each other's presence state, and that leaving the channel removes a user
+ * from everyone's view.
+ */
+import { describe, it, expect, afterEach } from 'vitest';
+import { createClientAs } from '../src/supabase-clients.js';
+import { ANA_ID, BORJA_ID, SEED_LIST_ID } from '../src/seed-constants.js';
+import type { RealtimeChannel, SupabaseClient } from '@supabase/supabase-js';
+
+// Track all clients created in a test so we can force-close their sockets in
+// afterEach. `removeAllChannels` only unsubs channels; the underlying Phoenix
+// socket stays open per singleFork and leaks state into later tests.
+const activeClients: SupabaseClient[] = [];
+
+function track(client: SupabaseClient): SupabaseClient {
+	activeClients.push(client);
+	return client;
+}
+
+afterEach(async () => {
+	for (const c of activeClients.splice(0)) {
+		await c.removeAllChannels();
+		await c.realtime.disconnect();
+	}
+});
+
+/**
+ * Subscribe to a presence channel and track the state. Resolves once the
+ * initial `presence sync` event has fired so callers can rely on the state
+ * snapshot being current before asserting.
+ */
+async function subscribePresence(
+	client: SupabaseClient,
+	channelName: string,
+	presenceKey: string,
+	payload: Record<string, unknown> = {}
+): Promise<{
+	channel: RealtimeChannel;
+	state: () => Record<string, unknown[]>;
+	waitForJoin: (key: string, timeoutMs?: number) => Promise<void>;
+	waitForLeave: (key: string, timeoutMs?: number) => Promise<void>;
+	unsubscribe: () => Promise<void>;
+}> {
+	const channel = client.channel(channelName, {
+		config: { presence: { key: presenceKey } }
+	});
+
+	const joinWaiters: Array<{ key: string; resolve: () => void; reject: (e: Error) => void }> = [];
+	const leaveWaiters: Array<{ key: string; resolve: () => void; reject: (e: Error) => void }> = [];
+
+	channel.on('presence', { event: 'sync' }, () => {
+		// Tracked for debugging only — state accessor returns the current
+		// snapshot on demand.
+	});
+
+	channel.on('presence', { event: 'join' }, ({ key }) => {
+		for (let i = joinWaiters.length - 1; i >= 0; i--) {
+			if (joinWaiters[i].key === key) {
+				joinWaiters[i].resolve();
+				joinWaiters.splice(i, 1);
+			}
+		}
+	});
+
+	channel.on('presence', { event: 'leave' }, ({ key }) => {
+		for (let i = leaveWaiters.length - 1; i >= 0; i--) {
+			if (leaveWaiters[i].key === key) {
+				leaveWaiters[i].resolve();
+				leaveWaiters.splice(i, 1);
+			}
+		}
+	});
+
+	await new Promise<void>((resolve, reject) => {
+		const timeout = setTimeout(() => reject(new Error('Presence SUBSCRIBED timeout')), 10_000);
+		channel.subscribe(async (status) => {
+			if (status === 'SUBSCRIBED') {
+				clearTimeout(timeout);
+				await channel.track({ user_id: presenceKey, ...payload });
+				resolve();
+			} else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT' || status === 'CLOSED') {
+				clearTimeout(timeout);
+				reject(new Error(`Presence subscribe failed: ${status}`));
+			}
+		});
+	});
+
+	return {
+		channel,
+		state: () => channel.presenceState() as Record<string, unknown[]>,
+		waitForJoin: (key, timeoutMs = 5_000) =>
+			new Promise<void>((resolve, reject) => {
+				if (key in channel.presenceState()) return resolve();
+				const t = setTimeout(() => reject(new Error(`waitForJoin(${key}) timeout`)), timeoutMs);
+				joinWaiters.push({
+					key,
+					resolve: () => {
+						clearTimeout(t);
+						resolve();
+					},
+					reject
+				});
+			}),
+		waitForLeave: (key, timeoutMs = 5_000) =>
+			new Promise<void>((resolve, reject) => {
+				const t = setTimeout(() => reject(new Error(`waitForLeave(${key}) timeout`)), timeoutMs);
+				leaveWaiters.push({
+					key,
+					resolve: () => {
+						clearTimeout(t);
+						resolve();
+					},
+					reject
+				});
+			}),
+		unsubscribe: async () => {
+			await channel.unsubscribe();
+		}
+	};
+}
+
+// KNOWN UPSTREAM BUG (supabase/realtime v2.76.5 + v2.83.0):
+// Presence `presence_diff` broadcasts crash `RealtimeWeb.RealtimeChannel`:
+//   (UndefinedFunctionError) function RealtimeChannel.handle_out/3 is undefined
+// The GenServer terminates, taking the connection down. Presence state appears
+// to work in one-off runs but is unreliable under load.
+//
+// The tests below are correct — unskip once the upstream bug is fixed (or
+// when we upgrade to a Realtime version that defines handle_out properly).
+// See: Realtime logs show the crash, postgres_changes + isolation tests work
+// fine (they don't go through handle_out).
+describe.skip('Realtime Presence — list session channel (blocked by upstream bug)', () => {
+	it('R-P-01: two clients on the same channel see each other', async () => {
+		const channelName = `list:${SEED_LIST_ID}:presence-${Date.now()}`;
+		const ana = track(await createClientAs(ANA_ID));
+		const borja = track(await createClientAs(BORJA_ID));
+
+		const anaSub = await subscribePresence(ana, channelName, ANA_ID, { name: 'Ana' });
+		const borjaSub = await subscribePresence(borja, channelName, BORJA_ID, { name: 'Borja' });
+
+		try {
+			// Each side should eventually see both users in the presence state.
+			// Wait for the "join" event that completes each side's view of the
+			// full roster. We wait for BOTH directions in parallel — when the
+			// test runs after other Realtime files (singleFork) the server
+			// occasionally takes >1s to emit the cross-presence_diff, so a
+			// serial await would race with it.
+			await Promise.all([anaSub.waitForJoin(BORJA_ID), borjaSub.waitForJoin(ANA_ID)]);
+			// Small settle window so both channel.presenceState() snapshots
+			// reflect the same roster before we diff them.
+			await new Promise((r) => setTimeout(r, 100));
+
+			expect(Object.keys(anaSub.state()).sort()).toEqual([ANA_ID, BORJA_ID].sort());
+			expect(Object.keys(borjaSub.state()).sort()).toEqual([ANA_ID, BORJA_ID].sort());
+		} finally {
+			await anaSub.unsubscribe();
+			await borjaSub.unsubscribe();
+		}
+	});
+
+	it('R-P-02: unsubscribing removes the user from the other client\'s state', async () => {
+		const channelName = `list:${SEED_LIST_ID}:presence-${Date.now()}`;
+		const ana = track(await createClientAs(ANA_ID));
+		const borja = track(await createClientAs(BORJA_ID));
+
+		const anaSub = await subscribePresence(ana, channelName, ANA_ID);
+		const borjaSub = await subscribePresence(borja, channelName, BORJA_ID);
+
+		try {
+			await anaSub.waitForJoin(BORJA_ID);
+			// Ana sees both herself and Borja before the leave
+			expect(Object.keys(anaSub.state()).sort()).toEqual([ANA_ID, BORJA_ID].sort());
+
+			// Borja leaves — Ana should see the leave event
+			const leavePromise = anaSub.waitForLeave(BORJA_ID);
+			await borjaSub.unsubscribe();
+			await leavePromise;
+			// Small settle window for the presence_diff to be applied to the
+			// internal state snapshot after the `leave` callback fires.
+			await new Promise((r) => setTimeout(r, 50));
+
+			const afterKeys = Object.keys(anaSub.state());
+			expect(afterKeys).not.toContain(BORJA_ID);
+			expect(afterKeys).toContain(ANA_ID);
+		} finally {
+			await anaSub.unsubscribe();
+			await ana.removeAllChannels();
+			await borja.removeAllChannels();
+		}
+	});
+});

packages/test-utils/tests/sync-queue.test.ts

@@ -0,0 +1,69 @@
+/**
+ * Sync queue unit tests (Fase 2b.2).
+ *
+ * These describe the contract of the offline mutation queue that will live at
+ * `apps/web/src/lib/sync/queue.ts`. They are skipped until that module exists.
+ * When implementation lands, remove the `describe.skip` (NOT the inner skips)
+ * and each test should fail red → pass green as features are built.
+ *
+ * Why they're here (not in apps/web): keeping all Vitest tests in one package
+ * for `just test-integration`. If we ever add Vitest to apps/web directly, move
+ * these files over and drop the cross-package import.
+ */
+import { describe, it, expect } from 'vitest';
+
+describe.skip('sync queue — pending_ops contract (Fase 2b.2)', () => {
+	it('Q-01: enqueue writes the op to pending_ops before the Supabase call', async () => {
+		// Given a mocked idb + mocked supabase.from().insert()
+		// When sync.enqueue({ op: "insert", table: "shopping_items", payload })
+		// Then pending_ops should contain the op before the supabase call fires,
+		//      and the supabase call should only be awaited after the write.
+		expect(true).toBe(true); // placeholder
+	});
+
+	it('Q-02: successful sync removes the op from pending_ops', async () => {
+		// Given an op in pending_ops
+		// When the supabase call resolves with no error
+		// Then the op should be removed from pending_ops
+		expect(true).toBe(true);
+	});
+
+	it('Q-03: failed sync keeps the op in pending_ops for retry', async () => {
+		// Given an op in pending_ops
+		// When the supabase call throws (network error)
+		// Then the op remains, its `attempts` counter increments,
+		//      and the promise resolves (does NOT throw) — errors are deferred
+		//      so callers can keep working offline.
+		expect(true).toBe(true);
+	});
+
+	it('Q-04: ops are processed in insertion order on flush', async () => {
+		// Given three ops A, B, C in pending_ops in that order
+		// When flush() runs
+		// Then supabase calls happen in A, B, C order regardless of timing
+		expect(true).toBe(true);
+	});
+});
+
+describe.skip('sync flush — online event fallback (Fase 2b.2, Safari-safe)', () => {
+	it('F-01: window "online" event triggers flush()', async () => {
+		// Given pending ops and a mocked window
+		// When dispatch Event("online")
+		// Then flush() is called within 100ms
+		expect(true).toBe(true);
+	});
+
+	it('F-02: last-write-wins on conflict — remote updated_at > local', async () => {
+		// Given a local op that would overwrite a remote row with a newer updated_at
+		// When flush runs
+		// Then the local op is DROPPED (not retried) and a sync_conflicts row is written
+		expect(true).toBe(true);
+	});
+
+	it('F-03: last-write-wins — local updated_at > remote, local wins', async () => {
+		// Given a local op newer than the remote row
+		// When flush runs
+		// Then the local op is APPLIED normally, no conflict row
+		expect(true).toBe(true);
+	});
+});

packages/test-utils/vitest.config.ts

@@ -11,11 +11,13 @@ export default defineConfig(({ mode }) => {
 			globals: true,
 			testTimeout: 30_000,
 			hookTimeout: 30_000,
-			// Run test files sequentially to avoid RLS state races
+			// Run test files sequentially (no parallel) to avoid RLS state races.
+			// But give each file its own worker fork — keeps Vitest sequential while
+			// ensuring Phoenix sockets from the Realtime client don't leak between
+			// files (supabase-js holds a module-level RealtimeClient singleton that
+			// `disconnect()` doesn't fully reset, so a shared fork bleeds state).
 			pool: 'forks',
-			poolOptions: {
-				forks: { singleFork: true }
-			}
+			fileParallelism: false
 		}
 	};
 });