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>

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.

I am using the code from realtime-handler.ts, and my connection keeps getting loose as soon as it connects, giving a timeout error. Does anyone know why this happens, or has anyone come across the same problem?

I tried using this in a NextJS app, and found that even though I call supabaseClient.auth.setSession before passing the client to this class, the class throws an error on refreshSessionIfNeeded -- Unhandled Promise Rejection: Error: Session not found. My client is set up like this:

import { createClient } from '@supabase/supabase-js'

const supabaseClient = createClient<Database>(
  process.env.NEXT_PUBLIC_SUPABASE_URL!,
  process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!,
  {
    auth: {
      persistSession: false,
      autoRefreshToken: false
    },
    realtime: {
      worker: true
    }
  }
)

export function useSupabaseClient() {
  const { getToken } = useAuth()
  const [client, setClient] = useState(supabaseClient)
  const [isLoading, setIsLoading] = useState(true)

  useEffect(() => {
    const setupAuth = async () => {
      setIsLoading(true)
      try {
        const token = await getToken()
        if (token) {
          await supabaseClient.auth.setSession({
            access_token: token,
            refresh_token: ''
          })
          
          // Configure realtime to use Clerk's token refresh
          supabaseClient.realtime.accessToken = async () => {
            return await getToken()
          }
          
          setClient(supabaseClient)
          console.log('session set')
        } else {
          console.warn('No token found')
        }
      } catch (error) {
        console.error('Error setting up auth:', error)
      } finally {
        setIsLoading(false)
      }
    }
    setupAuth()
  }, [getToken])

  return { client, isLoading }
}

Using it as so:

export function ChatRealtime({ chatId, onUpdate }: ChatRealtimeProps) {
  const { client: supabase, isLoading } = useSupabaseClient()

  useEffect(() => {
    // Don't set up realtime until auth is ready
    if (isLoading) {
      return
    }

    console.log('setting up realtime')
    const realtimeHandler = new RealtimeHandler<SupabaseClient>(
      supabase
    )
    console.log('realtime handler created')
    realtimeHandler.addChannel(
      () => supabase.channel(`chat-${chatId}`).on('postgres_changes', {
        event: 'UPDATE',
        schema: 'public',
        table: 'chats',
        filter: `id=eq.${chatId}`
      }, (payload) => onUpdate(payload as unknown as Chat))
    )
    const realtimeCleanup = realtimeHandler.start()
    return () => {
      realtimeCleanup()
    }
  }, [chatId, onUpdate, supabase, isLoading])

  return null
}

@michaelaflores i think this stems from the fact that I didn't really take into account using external accessToken implementations. I'll need to look into it a bit more, but I think you'll need to reimplement refreshTokenIfNeeded by replacing supabaseClient.auth.getSession with Clerk's implementation of refreshing tokens.

That said, Supabase has had some updates recently that I have not tested. I'll try to get around to testing and updating with the changes this weekend.

I am using the code from realtime-handler.ts, and my connection keeps getting loose as soon as it connects, giving a timeout error. Does anyone know why this happens, or has anyone come across the same problem?

@Megha30501 Can you let me know the framework you're using and snippets from your code implementation?

@Cikmo, I'm using React 18 with Vite as the framework. I encountered an issue where addChannel() and start() from my RealtimeHandler class were being invoked twice in development, even though the handler was implemented as a singleton.

This behavior was caused by React 18's StrictMode, which intentionally runs lifecycle methods like useEffect twice to help detect side effects.

Originally, I had code like this in a useEffect

useEffect(() => {
    const handler = new RealtimeHandler(supabase);
    const removeChannel = handler.addChannel(
      (supabaseInstance) => supabaseInstance.channel('messages'),
      {
        onSubscribe: (channel) => {
          console.log('[Realtime] Channel connected:', channel.topic);
        },
        onClose: (channel) => {
          console.log('[Realtime] Channel closed:', channel.topic);
        },
        onTimeout: (channel) => {
          console.log('[Realtime] Channel timed out:', channel.topic);
        },
        onError: (channel, err) => {
          console.error('[Realtime] Channel error:', channel.topic, err);
        },
      }
    );
    const cleanupHandler = handler.start();
    return () => {
      removeChannel();
      if (cleanupHandler) cleanupHandler();
    };
  }, []);

To solve the issue, I moved the real-time logic out of React into a separate module.

@Cikmo what is the point of enabling web worker for the client if your code disconnects after inactiveTabTimeoutSeconds?

My understanding was that worker option was added if you want to keep the connection open while the tab is not visible, because when in background the browser stops the interval which triggers heartbeat, and this is avoided by moving heartbeat to worker. But in your example you are stopping the subscription when app goes to background so not sure I understand why would worker be needed.

@Cikmo what is the point of enabling web worker for the client if your code disconnects after inactiveTabTimeoutSeconds?

My understanding was that worker option was added if you want to keep the connection open while the tab is not visible, because when in background the browser stops the interval which triggers heartbeat, and this is avoided by moving heartbeat to worker. But in your example you are stopping the subscription when app goes to background so not sure I understand why would worker be needed.

If I remember correctly, I'm stopping the subscription mostly just as a resource saving measure, so that people who are inactive for substabtial amoun of time don't use up server resources. Lowers costs. Once they view the tab again, it reconnects. You can skip the disconnect if you wish, but keep the reconnect logic, cause browsers are weird sometimes.

I think disconnect and reconnect are safer than relying on web worker to keep the connection because web worker doesn't work on phones when entire browser goes to background or on desktop when machine goes to sleep.