fix(realtime): resolve race conditions and connection bugs

Examples/Examples.xcodeproj/project.pbxproj

@@ -682,7 +682,7 @@
 				SUPPORTS_MAC_DESIGNED_FOR_IPHONE_IPAD = NO;
 				SUPPORTS_XR_DESIGNED_FOR_IPHONE_IPAD = NO;
 				SWIFT_EMIT_LOC_STRINGS = YES;
-				TARGETED_DEVICE_FAMILY = 1;
+				TARGETED_DEVICE_FAMILY = "1,2";
 			};
 			name = Debug;
 		};
@@ -710,7 +710,7 @@
 				SUPPORTS_MAC_DESIGNED_FOR_IPHONE_IPAD = NO;
 				SUPPORTS_XR_DESIGNED_FOR_IPHONE_IPAD = NO;
 				SWIFT_EMIT_LOC_STRINGS = YES;
-				TARGETED_DEVICE_FAMILY = 1;
+				TARGETED_DEVICE_FAMILY = "1,2";
 			};
 			name = Release;
 		};

Examples/Examples/Realtime/BroadcastView.swift

@@ -56,7 +56,7 @@ struct BroadcastView: View {
     .navigationTitle("Broadcast")
     .gitHubSourceLink()
     .task {
-      subscribe()
+      await subscribe()
     }
     .onDisappear {
       Task {
@@ -67,24 +67,24 @@ struct BroadcastView: View {
     }
   }
 
-  func subscribe() {
-    let channel = supabase.channel("broadcast-example")
+  func subscribe() async {
+    let channel = supabase.channel("broadcast-example") {
+      $0.broadcast.receiveOwnBroadcasts = true
+    }
 
-    Task {
-      do {
-        let broadcast = channel.broadcastStream(event: "message")
+    do {
+      let broadcast = channel.broadcastStream(event: "message")
 
-        try await channel.subscribeWithError()
-        self.channel = channel
+      try await channel.subscribeWithError()
+      self.channel = channel
 
-        for await message in broadcast {
-          if let payload = try message["payload"]?.decode(as: BroadcastMessage.self) {
-            messages.append(payload)
-          }
+      for await message in broadcast {
+        if let payload = try message["payload"]?.decode(as: BroadcastMessage.self) {
+          messages.append(payload)
         }
-      } catch {
-        print(error)
       }
+    } catch {
+      print(error)
     }
   }
 

Examples/Examples/Realtime/PresenceView.swift

@@ -48,7 +48,7 @@ struct PresenceView: View {
     .navigationTitle("Presence")
     .gitHubSourceLink()
     .task {
-      try? await subscribe()
+      await subscribe()
     }
     .onDisappear {
       Task {
@@ -59,23 +59,25 @@ struct PresenceView: View {
     }
   }
 
-  func subscribe() async throws {
-    let channel = supabase.channel("presence-example")
+  func subscribe() async {
+    do {
+      let channel = supabase.channel("presence-example")
 
-    let presence = channel.presenceChange()
+      let presence = channel.presenceChange()
 
-    try await channel.subscribeWithError()
-    self.channel = channel
+      try await channel.subscribeWithError()
+      self.channel = channel
 
-    // Track current user
-    let userId = auth.currentUserID
-    try await channel.track([
-      "user_id": userId.uuidString,
-      "username": "User \(userId.uuidString.prefix(8))",
-    ])
+      // Track current user
+      let userId = auth.currentUserID
+      try await channel.track(
+        PresenceUser(
+          id: userId.uuidString,
+          username: "User \(userId.uuidString.prefix(8))"
+        )
+      )
 
-    // Listen to presence changes
-    Task {
+      // Listen to presence changes
       for await state in presence {
         // Convert presence state to array of users
         var users: [PresenceUser] = []
@@ -85,11 +87,13 @@ struct PresenceView: View {
         }
         onlineUsers = users
       }
+    } catch {
+      print("Error: \(error)")
     }
   }
 }
 
-struct PresenceUser: Identifiable, Decodable {
+struct PresenceUser: Identifiable, Codable {
   let id: String
   let username: String
 }

Examples/Examples/Realtime/TodoRealtimeView.swift

@@ -46,7 +46,7 @@ struct TodoRealtimeView: View {
     .gitHubSourceLink()
     .task {
       await loadInitialTodos()
-      subscribeToChanges()
+      await subscribeToChanges()
     }
     .onDisappear {
       Task {
@@ -73,57 +73,65 @@ struct TodoRealtimeView: View {
     }
   }
 
-  func subscribeToChanges() {
+  func subscribeToChanges() async {
     let channel = supabase.channel("live-todos")
 
-    Task {
-      let insertions = channel.postgresChange(
-        InsertAction.self,
-        schema: "public",
-        table: "todos"
-      )
+    let insertions = channel.postgresChange(
+      InsertAction.self,
+      schema: "public",
+      table: "todos"
+    )
 
-      let updates = channel.postgresChange(
-        UpdateAction.self,
-        schema: "public",
-        table: "todos"
-      )
+    let updates = channel.postgresChange(
+      UpdateAction.self,
+      schema: "public",
+      table: "todos"
+    )
 
-      let deletes = channel.postgresChange(
-        DeleteAction.self,
-        schema: "public",
-        table: "todos"
-      )
+    let deletes = channel.postgresChange(
+      DeleteAction.self,
+      schema: "public",
+      table: "todos"
+    )
 
+    do {
       try await channel.subscribeWithError()
-      self.channel = channel
+    } catch {
+      print("Error: \(error)")
+      return
+    }
+    self.channel = channel
 
-      // Handle insertions
-      Task {
-        for await insertion in insertions {
-          try todos.insert(insertion.decodeRecord(decoder: JSONDecoder()), at: 0)
-        }
+    // Handle insertions
+    async let insertionObservation: () = { @MainActor in
+      for await insertion in insertions {
+        try todos.insert(insertion.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder), at: 0)
       }
+    }()
 
-      // Handle updates
-      Task {
-        for await update in updates {
-          let record = try update.decodeRecord(decoder: JSONDecoder()) as Todo
-          todos[id: record.id] = record
-        }
+    // Handle updates
+    async let updatesObservation: () = { @MainActor in
+      for await update in updates {
+        let record = try update.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder) as Todo
+        todos[id: record.id] = record
       }
+    }()
 
-      // Handle deletes
-      Task {
-        for await delete in deletes {
-          await MainActor.run {
-            guard
-              let id = delete.oldRecord["id"].flatMap(\.stringValue).flatMap(UUID.init(uuidString:))
-            else { return }
-            todos.remove(id: id)
-          }
-        }
+    // Handle deletes
+    async let deletesObservation: () = { @MainActor in
+      for await delete in deletes {
+        guard
+          let id = delete.oldRecord["id"].flatMap(\.stringValue).flatMap(UUID.init(uuidString:))
+        else { return }
+        todos.remove(id: id)
       }
+    }()
+
+    do {
+      _ = try await (insertionObservation, updatesObservation, deletesObservation)
+    } catch {
+      print(error)
     }
   }
+
 }

Examples/Examples/Supabase.plist

@@ -5,6 +5,6 @@
 	<key>SUPABASE_URL</key>
 	<string>http://127.0.0.1:54321</string>
 	<key>SUPABASE_ANON_KEY</key>
-	<string>eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0</string>
+	<string>sb_publishable_ACJWlzQHlZjBrEguHvfOxg_3BJgxAaH</string>
 </dict>
 </plist>

Sources/Realtime/RealtimeChannelV2.swift

@@ -138,6 +138,14 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
 
           do {
             try await _clock.sleep(for: delay)
+
+            // Check if socket is still connected after delay
+            if socket.status != .connected {
+              logger?.debug(
+                "Socket disconnected during retry delay for channel '\(topic)', aborting subscription"
+              )
+              throw CancellationError()
+            }
           } catch {
             // If sleep is cancelled, break out of retry loop
             logger?.debug("Subscription retry cancelled for channel '\(topic)'")
@@ -196,6 +204,12 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
         return
       }
       await socket.connect()
+
+      // Verify connection succeeded after await
+      if socket.status != .connected {
+        logger?.debug("Socket failed to connect, cannot subscribe to channel \(topic)")
+        return
+      }
     }
 
     logger?.debug("Subscribing to channel \(topic)")
@@ -234,6 +248,9 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
     logger?.debug("Unsubscribing from channel \(topic)")
 
     await push(ChannelEvent.leave)
+
+    // Wait for server confirmation of unsubscription
+    _ = await statusChange.first { @Sendable in $0 == .unsubscribed }
   }
 
   @available(