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

/**
 * 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, afterEach } 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';
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);
	}
});

describe('Realtime postgres_changes — shopping_items', () => {
	it('R-01: INSERT made by Ana reaches Borja subscribed to the same list', async () => {
		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,
			{
				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 = 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
			.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 = 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 }>(
			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);
		}
	});
});