feat: [US-045] - Supabase Realtime channel and connection management

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-23 15:28:23 -03:00
parent dfbaa0066d
commit 2e313a0264
5 changed files with 221 additions and 2 deletions
@@ -0,0 +1,167 @@
+import { createClient } from '@/lib/supabase/client'
+import type { RealtimeChannel } from '@supabase/supabase-js'
+
+export type ConnectionState = 'connecting' | 'connected' | 'disconnected' | 'reconnecting'
+
+type RealtimeCallbacks = {
+  onConnectionStateChange: (state: ConnectionState) => void
+  onPresenceSync?: (presences: Record<string, unknown[]>) => void
+}
+
+const HEARTBEAT_INTERVAL_MS = 30_000
+const RECONNECT_BASE_DELAY_MS = 1000
+const RECONNECT_MAX_DELAY_MS = 30_000
+
+export class RealtimeConnection {
+  private channel: RealtimeChannel | null = null
+  private heartbeatTimer: ReturnType<typeof setInterval> | null = null
+  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
+  private reconnectAttempts = 0
+  private projectId: string
+  private userId: string
+  private callbacks: RealtimeCallbacks
+  private isDestroyed = false
+  private supabase = createClient()
+
+  constructor(projectId: string, userId: string, callbacks: RealtimeCallbacks) {
+    this.projectId = projectId
+    this.userId = userId
+    this.callbacks = callbacks
+  }
+
+  async connect(): Promise<void> {
+    if (this.isDestroyed) return
+    this.callbacks.onConnectionStateChange('connecting')
+
+    this.channel = this.supabase.channel(`project:${this.projectId}`, {
+      config: { presence: { key: this.userId } },
+    })
+
+    this.channel
+      .on('presence', { event: 'sync' }, () => {
+        if (this.channel) {
+          const state = this.channel.presenceState()
+          this.callbacks.onPresenceSync?.(state)
+        }
+      })
+      .subscribe(async (status) => {
+        if (this.isDestroyed) return
+
+        if (status === 'SUBSCRIBED') {
+          this.reconnectAttempts = 0
+          this.callbacks.onConnectionStateChange('connected')
+          this.startHeartbeat()
+          await this.createSession()
+        } else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT') {
+          this.callbacks.onConnectionStateChange('reconnecting')
+          this.scheduleReconnect()
+        } else if (status === 'CLOSED') {
+          this.callbacks.onConnectionStateChange('disconnected')
+        }
+      })
+  }
+
+  async disconnect(): Promise<void> {
+    this.isDestroyed = true
+    this.stopHeartbeat()
+    this.clearReconnectTimer()
+
+    if (this.channel) {
+      await this.deleteSession()
+      this.supabase.removeChannel(this.channel)
+      this.channel = null
+    }
+
+    this.callbacks.onConnectionStateChange('disconnected')
+  }
+
+  getChannel(): RealtimeChannel | null {
+    return this.channel
+  }
+
+  private async createSession(): Promise<void> {
+    try {
+      await this.supabase.from('collaboration_sessions').upsert(
+        {
+          project_id: this.projectId,
+          user_id: this.userId,
+          cursor_position: null,
+          selected_node_id: null,
+          connected_at: new Date().toISOString(),
+          last_heartbeat: new Date().toISOString(),
+        },
+        { onConflict: 'project_id,user_id' }
+      )
+    } catch {
+      // Fire-and-forget: session creation failure shouldn't block editing
+    }
+  }
+
+  private async deleteSession(): Promise<void> {
+    try {
+      await this.supabase
+        .from('collaboration_sessions')
+        .delete()
+        .eq('project_id', this.projectId)
+        .eq('user_id', this.userId)
+    } catch {
+      // Fire-and-forget: cleanup failure is non-critical
+    }
+  }
+
+  private startHeartbeat(): void {
+    this.stopHeartbeat()
+    this.heartbeatTimer = setInterval(async () => {
+      if (this.isDestroyed) {
+        this.stopHeartbeat()
+        return
+      }
+      try {
+        await this.supabase
+          .from('collaboration_sessions')
+          .update({ last_heartbeat: new Date().toISOString() })
+          .eq('project_id', this.projectId)
+          .eq('user_id', this.userId)
+      } catch {
+        // Heartbeat failure is non-critical
+      }
+    }, HEARTBEAT_INTERVAL_MS)
+  }
+
+  private stopHeartbeat(): void {
+    if (this.heartbeatTimer) {
+      clearInterval(this.heartbeatTimer)
+      this.heartbeatTimer = null
+    }
+  }
+
+  private scheduleReconnect(): void {
+    if (this.isDestroyed) return
+    this.clearReconnectTimer()
+
+    const delay = Math.min(
+      RECONNECT_BASE_DELAY_MS * Math.pow(2, this.reconnectAttempts),
+      RECONNECT_MAX_DELAY_MS
+    )
+    this.reconnectAttempts++
+
+    this.reconnectTimer = setTimeout(async () => {
+      if (this.isDestroyed) return
+      this.callbacks.onConnectionStateChange('reconnecting')
+
+      if (this.channel) {
+        this.supabase.removeChannel(this.channel)
+        this.channel = null
+      }
+
+      await this.connect()
+    }, delay)
+  }
+
+  private clearReconnectTimer(): void {
+    if (this.reconnectTimer) {
+      clearTimeout(this.reconnectTimer)
+      this.reconnectTimer = null
+    }
+  }
+}