Connection handler that keeps Supabase Realtime channels alive. It automatically reconnects after errors, handles reauthentication, and more.

README.md

I had some issues getting a reliable connection up. Made this after a lot of experimentation and it now works well.

Includes how to keep the connection alive when the document is not visible, handles reconnecting on errors, and also fixes certain authentication issues when coming back to a document that has not been visible while the token expired.

To use it, you pass in a factory function for your RealtimeChannel. Do not call .subscribe() on the channel, the handler will do this for you. See example.

PS: If you're not using supabase-js you'll need to change stuff a bit. Instead of passing in SupabaseClient switch it to use RealtimeClient, and update the refreshSessionIfNeeded method to your authentication implementation.

If you see any problems or have suggestions for improvements, please leave a comment.

Important for keeping connection alive when the document is not visible

When creating the client, make sure to enable web-workers for sending heartbeats. By default, browsers slow down or stop timeouts and intervals but web workers can overcome this.

Using realtime-js

const client = new RealtimeClient(REALTIME_URL, {
	...,
	worker: true
})

Using supabase-js

const client = createClient(SUPABASE_URL, SUPABASE_ANON_KEY, {
	realtime: {
		worker: true
	}
});

realtime-handler.ts

import type { RealtimeChannel, SupabaseClient } from '@supabase/supabase-js';

import { REALTIME_SUBSCRIBE_STATES } from '@supabase/realtime-js';

export type Topic = string;

export type ChannelFactory<T extends SupabaseClient = SupabaseClient> = (
	supabase: T
) => RealtimeChannel;

export type RealtimeChannelFactories<T extends SupabaseClient = SupabaseClient> = Map<
	Topic,
	ChannelFactory<T>
>;

export type RealtimeChannels = Map<Topic, RealtimeChannel>;

export type RealtimeHandlerConfig = {
	/** The number of milliseconds to wait before disconnecting from realtime when the document is not visible.
	 * Default is 10 minutes.
	 */
	inactiveTabTimeoutSeconds: number;
};

export type SubscriptionEventCallbacks = {
	onSubscribe?: (channel: RealtimeChannel) => void;
	onClose?: (channel: RealtimeChannel) => void;
	onTimeout?: (channel: RealtimeChannel) => void;
	onError?: (channel: RealtimeChannel, err: Error) => void;
};

export type SubscriptionEventCallbacksMap = Map<Topic, SubscriptionEventCallbacks>;

/**
 * Handles realtime subscriptions to multiple channels.
 *
 * Factories are used rather than channels themselves to allow for re-creation of channels when needed
 * to do a proper reconnection after an error or timeout.
 */
export class RealtimeHandler<T extends SupabaseClient> {
	private inactiveTabTimeoutSeconds = 10 * 60;

	private supabaseClient: T;

	private channelFactories: RealtimeChannelFactories<T> = new Map();
	private channels: RealtimeChannels = new Map();

	private subscriptionEventCallbacks: SubscriptionEventCallbacksMap = new Map();

	/** Timer reference used to disconnect when tab is inactive. */
	private inactiveTabTimer: ReturnType<typeof setTimeout> | undefined;

	/** Flag to indicate if the handler has been started. */
	private started = false;

	public constructor(supabaseClient: T, config?: RealtimeHandlerConfig) {
		this.supabaseClient = supabaseClient;
		if (config?.inactiveTabTimeoutSeconds) {
			this.inactiveTabTimeoutSeconds = config.inactiveTabTimeoutSeconds;
		}
	}

	/**
	 * Adds a new channel using the provided channel factory and, optionally, subscription event callbacks.
	 *
	 * @param channelFactory - A factory function responsible for creating the channel.
	 * @param subscriptionEventCallbacks - Optional callbacks for handling subscription-related events.
	 *
	 * @returns A function that, when executed, removes the channel. Use this for cleanup.
	 */
	public addChannel(
		channelFactory: ChannelFactory<T>,
		subscriptionEventCallbacks?: SubscriptionEventCallbacks
	) {
		const channel = this.createChannel(channelFactory);

		if (this.channelFactories.has(channel.topic)) {
			console.warn(`Overwriting existing channel factory for topic: ${channel.topic}`);
			this.unsubscribeFromChannel(channel.topic);
		}
		this.channelFactories.set(channel.topic, channelFactory);

		if (subscriptionEventCallbacks) {
			this.subscriptionEventCallbacks.set(channel.topic, subscriptionEventCallbacks);
		}

		if (this.started) {
			// No reason to await, as it's all event-driven.
			this.subscribeToChannel(channel);
		}

		return () => {
			this.removeChannel(channel.topic);
		};
	}

	/**
	 * Removes and unsubscribes the channel associated with the given topic.
	 */
	public removeChannel(topic: Topic) {
		if (!topic.startsWith('realtime:')) {
			// If not prefixed, the user passed in the `subTopic`.
			topic = `realtime:${topic}`;
		}
		this.channelFactories.delete(topic);
		this.unsubscribeFromChannel(topic);
	}

	/**
	 * Starts the realtime event handling process.
	 *
	 * @returns A cleanup function that stops realtime event handling by removing the visibility change listener
	 *          and unsubscribing from all channels.
	 */
	public start() {
		if (this.started) {
			console.warn('RealtimeHandler has already been started. Ignoring subsequent start call.');
			return () => {};
		}

		const removeVisibilityChangeListener = this.addOnVisibilityChangeListener();

		this.subscribeToAllCreatedChannels();

		this.started = true;

		return () => {
			// cleanup
			removeVisibilityChangeListener();
			this.unsubscribeFromAllChannels();
		};
	}

	/* -----------------------------------------------------------
	   Private / Internal Methods
	----------------------------------------------------------- */

	/**
	 * Recreates the channel for the specified topic.
	 */
	private createChannel(channelFactory: ChannelFactory<T>) {
		const channel = channelFactory(this.supabaseClient);
		this.channels.set(channel.topic, channel);
		return channel;
	}

	/**
	 * Subscribes to a single channel.
	 */
	private async subscribeToChannel(channel: RealtimeChannel) {
		if (channel.state === 'joined' || channel.state === 'joining') {
			console.debug(`Channel '${channel.topic}' is already joined or joining. Skipping subscribe.`);
			return;
		}

		await this.refreshSessionIfNeeded();

		channel.subscribe(async (status, err) => {
			await this.handleSubscriptionStateEvent(channel, status, err);
		});
	}

	private subscribeToAllCreatedChannels() {
		for (const channel of this.channels.values()) {
			if (channel) {
				this.subscribeToChannel(channel);
			}
		}
	}

	private resubscribeToAllChannels() {
		for (const topic of this.channelFactories.keys()) {
			if (!this.channels.get(topic)) {
				this.resubscribeToChannel(topic);
			}
		}
	}

	/**
	 * Recreates and subscribes to the realtime channel for the given topic.
	 */
	private resubscribeToChannel(topic: Topic) {
		const channelFactory = this.channelFactories.get(topic);
		if (!channelFactory) {
			throw new Error(`Channel factory not found for topic: ${topic}`);
		}
		const channel = this.createChannel(channelFactory);
		this.subscribeToChannel(channel);
	}

	private unsubscribeFromChannel(topic: Topic) {
		const channel = this.channels.get(topic);
		if (channel) {
			this.supabaseClient.removeChannel(channel);
			this.channels.delete(topic);
		}
	}

	private unsubscribeFromAllChannels() {
		for (const topic of this.channels.keys()) {
			this.unsubscribeFromChannel(topic);
		}
	}

	private async handleSubscriptionStateEvent(
		channel: RealtimeChannel,
		status: REALTIME_SUBSCRIBE_STATES,
		err: Error | undefined
	) {
		const { topic } = channel;

		switch (status) {
			case REALTIME_SUBSCRIBE_STATES.SUBSCRIBED: {
				console.debug(`Successfully subscribed to '${topic}'`);
				const subscriptionEventCallbacks = this.subscriptionEventCallbacks.get(topic);
				if (subscriptionEventCallbacks?.onSubscribe) {
					subscriptionEventCallbacks.onSubscribe(channel);
				}
				break;
			}
			case REALTIME_SUBSCRIBE_STATES.CLOSED: {
				console.debug(`Channel closed '${topic}'`);
				const subscriptionEventCallbacks = this.subscriptionEventCallbacks.get(topic);
				if (subscriptionEventCallbacks?.onClose) {
					subscriptionEventCallbacks.onClose(channel);
				}
				break;
			}
			case REALTIME_SUBSCRIBE_STATES.TIMED_OUT: {
				console.debug(`Channel timed out '${topic}'`);
				const subscriptionEventCallbacks = this.subscriptionEventCallbacks.get(topic);
				if (subscriptionEventCallbacks?.onTimeout) {
					subscriptionEventCallbacks.onTimeout(channel);
				}
				break;
			}
			case REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR: { // We'll just reconnect when the tab becomes visible again. // if the tab is hidden, we don't really care about reconnection
				if (document.hidden) {
					console.debug(`Channel error in '${topic}', but tab is hidden. Removing channel.`);
					await this.supabaseClient.removeChannel(channel);
					return;
				} else if (err && isTokenExpiredError(err)) {
					console.debug(`Token expired causing channel error in '${topic}'. Refreshing session.`);
					this.resubscribeToChannel(topic);
				} else {
					console.warn(`Channel error in '${topic}': `, err?.message);
				}
				const subscriptionEventCallbacks = this.subscriptionEventCallbacks.get(topic);
				if (subscriptionEventCallbacks?.onError) {
					subscriptionEventCallbacks.onError(channel, err!);
				}
				break;
			}
			default: {
				const exhaustiveCheck: never = status;
				throw new Error(`Unknown channel status: ${exhaustiveCheck}`);
			}
		}
	}

	/**
	 * Refreshes the session token if needed and sets the token for Supabase Realtime.
	 */
	private async refreshSessionIfNeeded() {
		const { data, error } = await this.supabaseClient.auth.getSession();
		if (error) {
			throw error;
		}
		if (!data.session) {
			throw new Error('Session not found');
		}
		if (this.supabaseClient.realtime.accessTokenValue !== data.session.access_token) {
			await this.supabaseClient.realtime.setAuth(data.session.access_token);
		}
	}

	private addOnVisibilityChangeListener() {
		const handler = () => this.handleVisibilityChange();
		document.addEventListener('visibilitychange', handler);

		return () => {
			document.removeEventListener('visibilitychange', handler);
		};
	}

	private handleVisibilityChange() {
		if (document.hidden) {
			if (!this.inactiveTabTimer) {
				this.inactiveTabTimer = setTimeout(async () => {
					console.log(
						`Tab inactive for ${this.inactiveTabTimeoutSeconds} seconds. Disconnecting from realtime.`
					);
					this.unsubscribeFromAllChannels();
				}, this.inactiveTabTimeoutSeconds * 1000);
			}
		} else {
			if (this.inactiveTabTimer) {
				clearTimeout(this.inactiveTabTimer);
				this.inactiveTabTimer = undefined;
			}

			this.resubscribeToAllChannels();
		}
	}
}

/**
 * Determines if the provided error relates to an expired token.
 */
const isTokenExpiredError = (err: Error) => {
	// For some reason, message has sometimes been undefined. Adding a ? just in case.
	return err.message?.startsWith('"Token has expired');
};

usage-example.svelte

<!--
	Using Svelte as an example, but will work in anything.
-->

<script lang="ts">
	import { page } from '$app/state';
	import { RealtimeHandler } from '$lib/realtime-handler';

	const realtimeHandler = new RealtimeHandler(page.data.supabase);

	const unsubscribeChatMessages = realtimeHandler.addChannel((supabase) => {
		return supabase
			.channel('chat_messages')
			.on('broadcast', { event: 'new' }, (event) => {
				console.log('New message:', event);
			});
	});

	onMount(() => {
		const realtimeCleanup = realtimeHandler.start();

		return () => {
			realtimeCleanup();
		};
	});
</script>

Thanks so much for this! It seems to be working fine however I seem to keep getting an exception after a tab is restored after a timeout:

realtime-handler.ts:163 Uncaught (in promise) tried to subscribe multiple times. 'subscribe' can only be called a single time per channel instance
subscribeToChannel @ realtime-handler.ts:163
await in subscribeToChannel
resubscribeToChannel @ realtime-handler.ts:190
resubscribeToAllChannels @ realtime-handler.ts:176
handleVisibilityChange @ realtime-handler.ts:305
handler @ realtime-handler.ts:281

Thanks so much for this! It seems to be working fine however I seem to keep getting an exception after a tab is restored after a timeout:

realtime-handler.ts:163 Uncaught (in promise) tried to subscribe multiple times. 'subscribe' can only be called a single time per channel instance
subscribeToChannel @ realtime-handler.ts:163
await in subscribeToChannel
resubscribeToChannel @ realtime-handler.ts:190
resubscribeToAllChannels @ realtime-handler.ts:176
handleVisibilityChange @ realtime-handler.ts:305
handler @ realtime-handler.ts:281

Hm. Are you sure you're not calling .subscribe in your channel construction? Cause if you do, it will end up doing it twice, first you, then internally.

Also, unrelated, but I just fixed isTokenExpiredError to update it to the new error message returned by realtime.

Apologies I replied via email but it didn't seem to come through.
Definitely not calling it anywhere else. If I look at the logs it initially creates each channel correctly, closes them correctly when the tab is inactive but it looks like it somehow gets called twice at https://gist.github.com/Cikmo/bcba91318ba19dae1f914b32bf2b94b2#file-realtime-handler-ts-L158 in quick succession with a channel state of "closed" in both cases, for each channel

Hm, I'll try to replicate it

@surrealroad which browser are you testing it with?

Chrome Version 134.0.6998.46 (Official Build) (arm64)

Sorry for taking a while to respond. I tried replicating it and was unable to do so. Could there be some client code gets called twice for some reason, such as by react in dev mode?

I'll try a bit more tomorrow, I'd love to track this down.

Hey thanks for this. That's kind of what I'm wondering too. FWIW I am using SvelteKit. I thought it might be a hydration thing but I wrapped the calls to this in a browser ? check, so I don't think that's it. I also checked it is happening on both my prod and local site:

From this screenshot it almost looks like there's two instances of RealtimeHandler, as everything is logged twice. But because RealtimeHandler uses the same supabase client, and supabase keeps an internal record of active channels, it would cause a conflict like you're seeing. But maybe I'm not entirely understanding the logs.

But if this is correct, I'd start with checking that the instance of RealtimeHandler is not created in a place that will ever be called more than once on the browser.

My suggestion is to have a central instance for RealtimeHandler, like you would for supabase client itself. You can then just access it anywhere to add channels. You can add channels either before or after .start is called.

You could, for example use context, or simply attach it to page data at the root layout during load and checking for browser.

Yeah you might be onto something. I have an effect which does this:

$effect(() => {
		// reload data on new/changed route
		if (data.exploration?.id !== $explorationId) {
			removeSubscriptions().then(() => { // realtimeCleanup
				startSubscriptions(); // instantiate realtimehandler
			});
		}
	});

I think what might be happening is that startSubscriptions is creating the realtimehandler but maybe it is not fully destroyed by realtimecleanup. I will try some things and see.

I believe you have a typo in line 312 of realtime-handler.ts; the error message has a space between 'has' and 'expired'. In general this seems quite fragile, depending on string comparison; wondering if there's another way to determine the token expiry in that case.

I believe you have a typo in line 312 of realtime-handler.ts; the error message has a space between 'has' and 'expired'. In general this seems quite fragile, depending on string comparison; wondering if there's another way to determine the token expiry in that case.

You’d think it was a typo, but it is indeed the correct error message, you can confirm by checking the realtime source code. This was even their "fix" for a previous error message which said "token as expired".

as far as I’ve been able to figure out this is the only way to find the correct error cause. But if you have suggestions for other ways to handle this, please come with them.

Pretty sure I got a "Token has expired [n] seconds ago", but in any case I'm indeed delving into this issue myself and will report my findings if they could be of benefit. (Where did you find that message in the source code? For the life of me I don't find it)

Pretty sure I got a "Token has expired [n] seconds ago", but in any case I'm indeed delving into this issue myself and will report my findings if they could be of benefit. (Where did you find that message in the source code? For the life of me I don't find it)

oh you are indeed correct. They seem to have fixed the typo once again. It is now with a space. Thanks for pointing it out!

They fixed it in this commit. Just ctrl f for "has expired", and you'll see it.

But as you can see from their source, the only thing the client ever gets is a string that we need to just directly match. I find it weird that supabase client doesn't handle this error internally to refresh the token. The way it refreshes currently, I'm pretty sure, is just keeping an internal timer and refreshing before it expires, but not error handling for this edge case. It's weird.

I think what might be happening is that startSubscriptions is creating the realtimehandler but maybe it is not fully destroyed by realtimecleanup. I will try some things and see.

Just wanted to come back to this. I refactored a bunch of code today to not rely on $effect and the problem went away. Thanks again.

nevermind this seems to work great on nuxt, just my code i messed up creating and destroying channels incorrectly

That's great! I'll spend some time one day to document what needs to be done to make stuff work as intended, to make it easier to adapt to different frameworks.