feat(realtime): Fase 2b.0 infra + first R-series tests green

Enable postgres_changes broadcasts on shopping_items / shopping_lists and prove
the path end-to-end with Vitest Realtime tests and a pgTAP configuration check.

Infra changes that were silently blocking events
- Migration 007: shopping_items + shopping_lists added to supabase_realtime
  publication with REPLICA IDENTITY FULL (UPDATE/DELETE payloads need the full
  row so list_id-based filters work and RLS can evaluate DELETE against OLD)
- Realtime service: DB_USER=supabase_admin (superuser; supabase_replication_admin
  lacks CREATE on the `realtime` schema that the service auto-creates per tenant)
- Realtime service: SELF_HOST_TENANT_NAME=realtime so the seed tenant name
  matches the default supabase-js resolves for localhost URLs (was realtime-dev
  → TenantNotFound)

Tests
- pgTAP 004_realtime_publication.sql — P-01..P-04: publication membership and
  REPLICA IDENTITY FULL for both shopping tables
- Vitest realtime-postgres-changes.test.ts — R-01 INSERT broadcast,
  R-02 UPDATE carries full row, R-03 list_id filter isolates events
- test-utils realtime-helpers.ts — subscribePostgresChanges() returns a
  waitFor(predicate, timeout) helper that captures the SUBSCRIBED handshake
  before returning so mutations issued right after are not lost
- createClientAs now calls realtime.setAuth(token) so the server evaluates
  RLS against the test user's JWT
- Justfile test-db now globs supabase/tests/*.sql so new pgTAP files are picked
  up automatically

Totals: 54→57 Vitest, 12→16 pgTAP, 15 Playwright = 88 tests 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/src/index.ts

@@ -1,3 +1,4 @@
 export * from './seed-constants.js';
 export * from './supabase-clients.js';
 export * from './db-helpers.js';
+export * from './realtime-helpers.js';

packages/test-utils/src/realtime-helpers.ts

@@ -0,0 +1,97 @@
+import type { RealtimeChannel, SupabaseClient } from '@supabase/supabase-js';
+
+/**
+ * Subscribe to `postgres_changes` on a table + optional filter for a given
+ * client, and return the channel plus a promise-based `waitFor` that resolves
+ * on the next event matching the given predicate. Always call `unsubscribe()`
+ * in a `finally` block — the WebSocket stays open otherwise and leaks across
+ * tests.
+ *
+ * The `SUBSCRIBED` handshake is awaited before the function resolves, so
+ * callers can start mutating immediately and trust that events will be caught.
+ */
+export async function subscribePostgresChanges<T = Record<string, unknown>>(
+	client: SupabaseClient,
+	params: {
+		table: string;
+		schema?: string;
+		filter?: string;
+		event?: 'INSERT' | 'UPDATE' | 'DELETE' | '*';
+		channelName?: string;
+	}
+): Promise<{
+	channel: RealtimeChannel;
+	events: Array<{ eventType: string; new: T; old: T | null }>;
+	waitFor: (
+		predicate: (evt: { eventType: string; new: T; old: T | null }) => boolean,
+		timeoutMs?: number
+	) => Promise<{ eventType: string; new: T; old: T | null }>;
+	unsubscribe: () => Promise<void>;
+}> {
+	const events: Array<{ eventType: string; new: T; old: T | null }> = [];
+	const waiters: Array<{
+		predicate: (e: { eventType: string; new: T; old: T | null }) => boolean;
+		resolve: (e: { eventType: string; new: T; old: T | null }) => void;
+	}> = [];
+
+	const channel = client.channel(params.channelName ?? `test:${params.table}:${Date.now()}`);
+
+	channel.on(
+		// @ts-expect-error — supabase-js types require a specific literal-union, filter
+		//                   is a runtime string so the narrow overload complains
+		'postgres_changes',
+		{
+			event: params.event ?? '*',
+			schema: params.schema ?? 'public',
+			table: params.table,
+			...(params.filter ? { filter: params.filter } : {})
+		},
+		(payload: { eventType: string; new: T; old: T | null }) => {
+			const evt = { eventType: payload.eventType, new: payload.new, old: payload.old };
+			events.push(evt);
+			for (let i = waiters.length - 1; i >= 0; i--) {
+				if (waiters[i].predicate(evt)) {
+					waiters[i].resolve(evt);
+					waiters.splice(i, 1);
+				}
+			}
+		}
+	);
+
+	await new Promise<void>((resolve, reject) => {
+		const timeout = setTimeout(() => reject(new Error('Realtime SUBSCRIBED timeout')), 10_000);
+		channel.subscribe((status) => {
+			if (status === 'SUBSCRIBED') {
+				clearTimeout(timeout);
+				resolve();
+			} else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT' || status === 'CLOSED') {
+				clearTimeout(timeout);
+				reject(new Error(`Realtime subscribe failed: ${status}`));
+			}
+		});
+	});
+
+	function waitFor(
+		predicate: (e: { eventType: string; new: T; old: T | null }) => boolean,
+		timeoutMs = 5_000
+	) {
+		const existing = events.find(predicate);
+		if (existing) return Promise.resolve(existing);
+		return new Promise<{ eventType: string; new: T; old: T | null }>((resolve, reject) => {
+			const t = setTimeout(() => reject(new Error('waitFor timeout')), timeoutMs);
+			waiters.push({
+				predicate,
+				resolve: (e) => {
+					clearTimeout(t);
+					resolve(e);
+				}
+			});
+		});
+	}
+
+	async function unsubscribe() {
+		await channel.unsubscribe();
+	}
+
+	return { channel, events, waitFor, unsubscribe };
+}

packages/test-utils/src/supabase-clients.ts

@@ -26,10 +26,14 @@ async function signToken(userId: string): Promise<string> {
 /**
  * Create a Supabase client that authenticates as the given seed user.
  * RLS policies see auth.uid() = userId.
+ *
+ * Also calls `realtime.setAuth(token)` so Realtime subscriptions carry the
+ * same identity — the server evaluates RLS against this JWT when deciding
+ * which Postgres-changes events to forward.
  */
 export async function createClientAs(userId: string) {
 	const token = await signToken(userId);
-	return createClient<Database>(
+	const client = createClient<Database>(
 		requiredEnv('PUBLIC_SUPABASE_URL'),
 		requiredEnv('PUBLIC_SUPABASE_ANON_KEY'),
 		{
@@ -37,6 +41,9 @@ export async function createClientAs(userId: string) {
 			auth: { persistSession: false, autoRefreshToken: false }
 		}
 	);
+	// @ts-expect-error — realtime.setAuth exists at runtime on supabase-js v2
+	client.realtime.setAuth(token);
+	return client;
 }
 
 /**

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

@@ -0,0 +1,157 @@
+/**
+ * R-series: Supabase Realtime `postgres_changes` over `shopping_items`.
+ *
+ * Verifies the end-to-end path: mutation in Postgres → WAL → supabase_realtime
+ * publication → Realtime service → WebSocket → subscribed client.
+ *
+ * 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 { createClientAs, createAdminClient } from '../src/supabase-clients.js';
+import { subscribePostgresChanges } from '../src/realtime-helpers.js';
+import {
+	ANA_ID,
+	BORJA_ID,
+	SEED_LIST_ID,
+	COLLECTIVE_ID
+} from '../src/seed-constants.js';
+
+const admin = createAdminClient();
+const createdItemIds: string[] = [];
+
+afterAll(async () => {
+	if (createdItemIds.length > 0) {
+		await admin.from('shopping_items').delete().in('id', createdItemIds);
+	}
+});
+
+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 sub = await subscribePostgresChanges<{ id: string; list_id: string; name: string }>(
+			borja,
+			{
+				table: 'shopping_items',
+				event: 'INSERT',
+				filter: `list_id=eq.${SEED_LIST_ID}`
+			}
+		);
+
+		try {
+			const itemName = `R-01-item-${Date.now()}`;
+			const { data, error } = await ana
+				.from('shopping_items')
+				.insert({ list_id: SEED_LIST_ID, name: itemName, sort_order: 900, created_by: ANA_ID })
+				.select('id')
+				.single();
+			expect(error).toBeNull();
+			if (data?.id) createdItemIds.push(data.id);
+
+			const evt = await sub.waitFor((e) => e.eventType === 'INSERT' && e.new.name === itemName);
+			expect(evt.new.list_id).toBe(SEED_LIST_ID);
+			expect(evt.new.name).toBe(itemName);
+		} finally {
+			await sub.unsubscribe();
+			await borja.removeAllChannels();
+			await ana.removeAllChannels();
+		}
+	});
+
+	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);
+
+		// Seed an item owned by Ana to mutate
+		const { data: item } = await admin
+			.from('shopping_items')
+			.insert({ list_id: SEED_LIST_ID, name: 'R-02-seed', sort_order: 901, created_by: ANA_ID })
+			.select('id')
+			.single();
+		createdItemIds.push(item!.id);
+
+		const sub = await subscribePostgresChanges<{
+			id: string;
+			list_id: string;
+			name: string;
+			is_checked: boolean;
+		}>(borja, {
+			table: 'shopping_items',
+			event: 'UPDATE',
+			filter: `list_id=eq.${SEED_LIST_ID}`
+		});
+
+		try {
+			await ana
+				.from('shopping_items')
+				.update({ is_checked: true, checked_by: ANA_ID, checked_at: new Date().toISOString() })
+				.eq('id', item!.id);
+
+			const evt = await sub.waitFor((e) => e.eventType === 'UPDATE' && e.new.id === item!.id);
+			// With REPLICA IDENTITY FULL, `new` carries the full row including list_id
+			expect(evt.new.list_id).toBe(SEED_LIST_ID);
+			expect(evt.new.is_checked).toBe(true);
+			expect(evt.new.name).toBe('R-02-seed');
+		} finally {
+			await sub.unsubscribe();
+			await ana.removeAllChannels();
+			await borja.removeAllChannels();
+		}
+	});
+
+	it('R-03: filter on list_id excludes events from other lists', async () => {
+		// Create a second list in the same collective
+		const { data: otherList } = await admin
+			.from('shopping_lists')
+			.insert({ collective_id: COLLECTIVE_ID, name: 'R-03 other list', created_by: ANA_ID })
+			.select('id')
+			.single();
+		const otherListId = otherList!.id;
+
+		const borja = await createClientAs(BORJA_ID);
+		const ana = await createClientAs(ANA_ID);
+
+		// Borja subscribes to SEED_LIST_ID only
+		const sub = await subscribePostgresChanges<{ id: string; list_id: string; name: string }>(
+			borja,
+			{
+				table: 'shopping_items',
+				event: 'INSERT',
+				filter: `list_id=eq.${SEED_LIST_ID}`
+			}
+		);
+
+		try {
+			// Insert into the OTHER list — Borja should NOT receive this event
+			const noiseName = `R-03-noise-${Date.now()}`;
+			const { data: noise } = await ana
+				.from('shopping_items')
+				.insert({ list_id: otherListId, name: noiseName, sort_order: 902, created_by: ANA_ID })
+				.select('id')
+				.single();
+			createdItemIds.push(noise!.id);
+
+			// Insert into SEED_LIST_ID — Borja SHOULD receive this
+			const signalName = `R-03-signal-${Date.now()}`;
+			const { data: signal } = await ana
+				.from('shopping_items')
+				.insert({ list_id: SEED_LIST_ID, name: signalName, sort_order: 903, created_by: ANA_ID })
+				.select('id')
+				.single();
+			createdItemIds.push(signal!.id);
+
+			// Wait for the signal event
+			await sub.waitFor((e) => e.new.name === signalName);
+
+			// The noise event should not be present
+			const noiseEvt = sub.events.find((e) => (e.new as { name: string }).name === noiseName);
+			expect(noiseEvt).toBeUndefined();
+		} finally {
+			await sub.unsubscribe();
+			await ana.removeAllChannels();
+			await borja.removeAllChannels();
+			await admin.from('shopping_lists').delete().eq('id', otherListId);
+		}
+	});
+});