fix: realtime safeguards and logging

rockfactory · rockfactory · commit a7ef86d91426 · 2026-04-16T00:19:20.000+02:00
diff --git a/src/games/sync/peersSlice.ts b/src/games/sync/peersSlice.ts
@@ -1,7 +1,8 @@
+import type { RealtimeChannel } from '@supabase/supabase-js';
 import { useMemo } from 'react';
 import { useStore } from '@/core/zustand';
 import { createSlice } from '@/core/zustand-helpers/slices';
-import { SENDER_ID } from './realtimeSyncTypes';
+import { type PresencePayload, SENDER_ID } from './realtimeSyncTypes';
 
 export interface PeerInfo {
   senderId: string;
@@ -53,6 +54,22 @@ export function countOtherPeers(peers: Record<string, PeerInfo>): number {
   return count;
 }
 
-export function hasOtherPeersConnected(): boolean {
-  return countOtherPeers(useStore.getState().peers.peers) > 0;
+/**
+ * Reads peers directly from the realtime channel's presenceState instead of
+ * the zustand slice. The slice is updated only on `presence.sync` events
+ * (handled by `computeLeaderAndPeers`), so it lags behind the channel by up
+ * to a few hundred ms after subscribe — long enough that the first patches
+ * after joining could be wrongly skipped as "no peers". The channel itself
+ * always knows the current presence list.
+ */
+export function hasOtherPeersConnectedOnChannel(
+  channel: RealtimeChannel,
+): boolean {
+  const state = channel.presenceState<PresencePayload>();
+  for (const presences of Object.values(state)) {
+    for (const p of presences) {
+      if (p.senderId && p.senderId !== SENDER_ID) return true;
+    }
+  }
+  return false;
 }
diff --git a/src/games/sync/realtimeSyncHandlers.ts b/src/games/sync/realtimeSyncHandlers.ts
@@ -18,6 +18,7 @@ import {
   SENDER_ID,
   withSuppressedBroadcast,
 } from './realtimeSyncTypes';
+import { safeChannelSend } from './safeChannelSend';
 
 const logger = loglev.getLogger('games:realtime-sync');
 
@@ -99,15 +100,19 @@ export function handleFullStateRequest(
       share_token: latestGame.shareToken,
     };
 
-    channel.send({
-      type: 'broadcast',
-      event: BROADCAST_FULL_RESPONSE,
-      payload: {
-        senderId: SENDER_ID,
-        seq: refs.seq.current,
-        serialized,
-        remoteData,
-      } satisfies FullStateResponsePayload,
+    safeChannelSend({
+      channel,
+      message: {
+        type: 'broadcast',
+        event: BROADCAST_FULL_RESPONSE,
+        payload: {
+          senderId: SENDER_ID,
+          seq: refs.seq.current,
+          serialized,
+          remoteData,
+        } satisfies FullStateResponsePayload,
+      },
+      context: 'full state response',
     });
   } catch (err) {
     logger.error('Failed to send full state response', err);
@@ -145,10 +150,14 @@ export function requestFullStateWithFallback(
   refs: SyncRefs,
   timers: SyncTimers,
 ) {
-  channel.send({
-    type: 'broadcast',
-    event: BROADCAST_FULL_REQUEST,
-    payload: { senderId: SENDER_ID } satisfies FullStateRequestPayload,
+  safeChannelSend({
+    channel,
+    message: {
+      type: 'broadcast',
+      event: BROADCAST_FULL_REQUEST,
+      payload: { senderId: SENDER_ID } satisfies FullStateRequestPayload,
+    },
+    context: 'full state request',
   });
 
   if (timers.dbFallback !== null) clearTimeout(timers.dbFallback);
diff --git a/src/games/sync/safeChannelSend.ts b/src/games/sync/safeChannelSend.ts
@@ -0,0 +1,33 @@
+import type { RealtimeChannel } from '@supabase/supabase-js';
+import { loglev } from '@/core/logger/log';
+
+const logger = loglev.getLogger('games:realtime-sync');
+
+/**
+ * Wraps `channel.send` to log non-ok responses and rejections. The Supabase
+ * client returns a Promise<'ok' | 'timed out' | 'error'> that we otherwise
+ * ignore — failures (e.g. RLS denies, channel closed, payload too big) would
+ * be silent and very hard to diagnose at runtime.
+ */
+export interface SafeChannelSendArgs {
+  channel: RealtimeChannel;
+  message: Parameters<RealtimeChannel['send']>[0];
+  context: string;
+}
+
+export function safeChannelSend({
+  channel,
+  message,
+  context,
+}: SafeChannelSendArgs): void {
+  channel
+    .send(message)
+    .then(status => {
+      if (status !== 'ok') {
+        logger.warn(`channel.send (${context}) returned: ${status}`);
+      }
+    })
+    .catch(err => {
+      logger.error(`channel.send (${context}) failed`, err);
+    });
+}
diff --git a/src/games/sync/useRealtimeGameSync.ts b/src/games/sync/useRealtimeGameSync.ts
@@ -1,14 +1,14 @@
 import type { RealtimeChannel } from '@supabase/supabase-js';
 import type { Patch } from 'immer';
-import { useEffect, useRef } from 'react';
+import { useEffect, useRef, useState } from 'react';
 import { loglev } from '@/core/logger/log';
 import { supabaseClient } from '@/core/supabase';
 import { useStore } from '@/core/zustand';
 import { onStorePatches } from '@/core/zustand-helpers/immer';
 import { saveRemoteGame } from '@/games/save/saveRemoteGame';
 import { useCurrentFactoryId } from '@/notes/useNotesContext';
 import { flushRemoteGameOnUnload } from './flushRemoteGameOnUnload';
-import { hasOtherPeersConnected } from './peersSlice';
+import { hasOtherPeersConnectedOnChannel } from './peersSlice';
 import {
   computeLeaderAndPeers,
   handleFullStateRequest,
@@ -32,6 +32,7 @@ import {
   type PresencePayload,
   SENDER_ID,
 } from './realtimeSyncTypes';
+import { safeChannelSend } from './safeChannelSend';
 
 const logger = loglev.getLogger('games:realtime-sync');
 
@@ -50,6 +51,10 @@ export function useRealtimeGameSync() {
   const isLeaderRef = useRef(false);
   const factoryIdRef = useRef(factoryId);
   factoryIdRef.current = factoryId;
+  // Bumped to force the subscribe effect to re-run and recreate the channel
+  // after CHANNEL_ERROR / CLOSED / TIMED_OUT. Reset on successful SUBSCRIBED.
+  const [reconnectEpoch, setReconnectEpoch] = useState(0);
+  const reconnectAttemptsRef = useRef(0);
 
   useEffect(() => {
     if (!channelRef.current || !session) return;
@@ -76,7 +81,9 @@ export function useRealtimeGameSync() {
     }
 
     const channelName = `game:${savedId}`;
-    logger.info(`Joining realtime channel: ${channelName}`);
+    logger.info(
+      `Joining realtime channel: ${channelName} (epoch=${reconnectEpoch})`,
+    );
 
     const channel = supabaseClient.channel(channelName, {
       config: { private: true },
@@ -93,7 +100,29 @@ export function useRealtimeGameSync() {
     let pendingPatches: Patch[] = [];
     let flushTimer: ReturnType<typeof setTimeout> | null = null;
     let autoSaveTimer: ReturnType<typeof setTimeout> | null = null;
+    let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
     let hasDirtySinceLastSave = false;
+    // Set to true in the effect cleanup so the subscribe callback (which fires
+    // with CLOSED when we voluntarily remove the channel) does not schedule a
+    // reconnect loop. Only genuine errors while the effect is still active
+    // should trigger a retry.
+    let isCleaningUp = false;
+
+    function scheduleReconnect() {
+      if (isCleaningUp) return;
+      if (reconnectTimer !== null) return;
+      const attempt = reconnectAttemptsRef.current;
+      // Exponential backoff capped at 30s: 1s, 2s, 4s, 8s, 16s, 30s, 30s, ...
+      const delay = Math.min(1000 * 2 ** attempt, 30_000);
+      reconnectAttemptsRef.current = attempt + 1;
+      logger.warn(
+        `Scheduling realtime reconnect in ${delay}ms (attempt=${attempt + 1})`,
+      );
+      reconnectTimer = setTimeout(() => {
+        reconnectTimer = null;
+        setReconnectEpoch(e => e + 1);
+      }, delay);
+    }
 
     const doRequestFullState = () =>
       requestFullStateWithFallback(channel, gameId, refs, timers);
@@ -117,9 +146,8 @@ export function useRealtimeGameSync() {
       flushTimer = null;
       if (!channelRef.current || pendingPatches.length === 0) return;
 
-      if (!hasOtherPeersConnected()) {
+      if (!hasOtherPeersConnectedOnChannel(channelRef.current)) {
         pendingPatches = [];
-        logger.debug('Skipping broadcast: no other peers connected');
         return;
       }
 
@@ -128,20 +156,20 @@ export function useRealtimeGameSync() {
       const batch = pendingPatches;
       pendingPatches = [];
 
-      try {
-        channelRef.current.send({
+      safeChannelSend({
+        channel: channelRef.current,
+        message: {
           type: 'broadcast',
           event: BROADCAST_EVENT,
           payload: {
             senderId: SENDER_ID,
             seq,
             patches: batch,
           } satisfies PatchBroadcastPayload,
-        });
-        logger.debug(`Broadcasted ${batch.length} patches (seq=${seq})`);
-      } catch (err) {
-        logger.error('Failed to broadcast patches', err);
-      }
+        },
+        context: `patch batch seq=${seq}`,
+      });
+      logger.debug(`Broadcasted ${batch.length} patches (seq=${seq})`);
     }
 
     channel
@@ -177,6 +205,7 @@ export function useRealtimeGameSync() {
         useStore.getState().setRealtimeSyncConnected(status === 'SUBSCRIBED');
 
         if (status === 'SUBSCRIBED') {
+          reconnectAttemptsRef.current = 0;
           const user = session.user;
           await channel.track({
             senderId: SENDER_ID,
@@ -189,6 +218,12 @@ export function useRealtimeGameSync() {
             factoryId: factoryIdRef.current,
           } satisfies PresencePayload);
           doRequestFullState();
+        } else if (
+          status === 'CHANNEL_ERROR' ||
+          status === 'CLOSED' ||
+          status === 'TIMED_OUT'
+        ) {
+          scheduleReconnect();
         }
       });
 
@@ -222,6 +257,7 @@ export function useRealtimeGameSync() {
     });
 
     return () => {
+      isCleaningUp = true;
       window.removeEventListener('beforeunload', onBeforeUnload);
       unsubscribePatches();
       if (flushTimer !== null) {
@@ -245,6 +281,10 @@ export function useRealtimeGameSync() {
         clearTimeout(timers.dbFallback);
         timers.dbFallback = null;
       }
+      if (reconnectTimer !== null) {
+        clearTimeout(reconnectTimer);
+        reconnectTimer = null;
+      }
 
       if (channelRef.current) {
         logger.info(`Leaving realtime channel: ${channelName}`);
@@ -255,5 +295,5 @@ export function useRealtimeGameSync() {
       useStore.getState().setRealtimeSyncConnected(false);
       useStore.getState().clearPeers();
     };
-  }, [session, savedId, selectedGameId]);
+  }, [session, savedId, selectedGameId, reconnectEpoch]);
 }
