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

/**
 * 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();
		}
	});
});