fix: realtime reconnection (#261)

* wip

* refactor: EventEmitter

* fix: client reconnect

* chore: add existential any to missing places

* chore: remove unused SharedStream type

* Fix tests

* Add Sendable requirement to Event generic type

* chore: remove TestLogger

* test: use withMainSerialExecutor

* chore(deps): update all deps

* chore: fix build on Windows

* fix: do try reconnecting when connection error

* chore(examples): track presence on channel subscription

* test: add tests

* Fix build for Windows and Linux

* Add Sendable to SupabaseLogMessage

Author: grdsdev

Examples/SlackClone/AppView.swift

@@ -14,6 +14,8 @@ final class AppViewModel {
   var session: Session?
   var selectedChannel: Channel?
 
+  var realtimeConnectionStatus: RealtimeClientV2.Status?
+
   init() {
     Task { [weak self] in
       for await (event, session) in await supabase.auth.authStateChanges {
@@ -27,23 +29,46 @@ final class AppViewModel {
         }
       }
     }
+
+    Task {
+      for await status in await supabase.realtimeV2.statusChange {
+        realtimeConnectionStatus = status
+      }
+    }
   }
 }
 
 @MainActor
 struct AppView: View {
   @Bindable var model: AppViewModel
+  let log = LogStore.shared
+
+  @State var logPresented = false
 
   @ViewBuilder
   var body: some View {
     if model.session != nil {
       NavigationSplitView {
         ChannelListView(channel: $model.selectedChannel)
+          .toolbar {
+            ToolbarItem {
+              Button("Log") {
+                logPresented = true
+              }
+            }
+          }
       } detail: {
         if let channel = model.selectedChannel {
           MessagesView(channel: channel).id(channel.id)
         }
       }
+      .sheet(isPresented: $logPresented) {
+        List {
+          ForEach(0 ..< log.messages.count, id: \.self) { i in
+            Text(log.messages[i].description)
+          }
+        }
+      }
     } else {
       AuthView()
     }

Examples/SlackClone/Logger.swift

@@ -13,11 +13,21 @@ extension Logger {
   static let main = Self(subsystem: "com.supabase.SlackClone", category: "app")
 }
 
-final class SupabaseLoggerImpl: SupabaseLogger, @unchecked Sendable {
+@Observable
+final class LogStore: SupabaseLogger {
   private let lock = NSLock()
   private var loggers: [String: Logger] = [:]
 
+  static let shared = LogStore()
+
+  @MainActor
+  var messages: [SupabaseLogMessage] = []
+
   func log(message: SupabaseLogMessage) {
+    Task {
+      await add(message: message)
+    }
+
     lock.withLock {
       if loggers[message.system] == nil {
         loggers[message.system] = Logger(
@@ -29,11 +39,16 @@ final class SupabaseLoggerImpl: SupabaseLogger, @unchecked Sendable {
       let logger = loggers[message.system]!
 
       switch message.level {
-      case .debug: logger.debug("\(message)")
-      case .error: logger.error("\(message)")
-      case .verbose: logger.info("\(message)")
-      case .warning: logger.notice("\(message)")
+      case .debug: logger.debug("\(message, privacy: .public)")
+      case .error: logger.error("\(message, privacy: .public)")
+      case .verbose: logger.info("\(message, privacy: .public)")
+      case .warning: logger.notice("\(message, privacy: .public)")
       }
     }
   }
+
+  @MainActor
+  private func add(message: SupabaseLogMessage) {
+    messages.insert(message, at: 0)
+  }
 }

Examples/SlackClone/Supabase.swift

@@ -21,10 +21,10 @@ let decoder: JSONDecoder = {
 }()
 
 let supabase = SupabaseClient(
-  supabaseURL: URL(string: "http://127.0.0.1:54321")!,
+  supabaseURL: URL(string: "http://localhost:54321")!,
   supabaseKey: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0",
   options: SupabaseClientOptions(
     db: .init(encoder: encoder, decoder: decoder),
-    global: SupabaseClientOptions.GlobalOptions(logger: SupabaseLoggerImpl())
+    global: SupabaseClientOptions.GlobalOptions(logger: LogStore.shared)
   )
 )

Examples/SlackClone/UserStore.swift

@@ -22,12 +22,17 @@ final class UserStore {
       let channel = await supabase.realtimeV2.channel("public:users")
       let changes = await channel.postgresChange(AnyAction.self, table: "users")
 
-      let prenseces = await channel.presenceChange()
+      let presences = await channel.presenceChange()
 
       await channel.subscribe()
 
-      let userId = try await supabase.auth.session.user.id
-      try await channel.track(UserPresence(userId: userId, onlineAt: Date()))
+      Task {
+        let statusChange = await channel.statusChange
+        for await _ in statusChange.filter({ $0 == .subscribed }) {
+          let userId = try await supabase.auth.session.user.id
+          try await channel.track(UserPresence(userId: userId, onlineAt: Date()))
+        }
+      }
 
       Task {
         for await change in changes {
@@ -36,7 +41,7 @@ final class UserStore {
       }
 
       Task {
-        for await presence in prenseces {
+        for await presence in presences {
           let joins = try presence.decodeJoins(as: UserPresence.self)
           let leaves = try presence.decodeLeaves(as: UserPresence.self)
 

Package.swift

@@ -79,6 +79,7 @@ let package = Package(
       dependencies: [
         "Auth",
         "_Helpers",
+        "TestHelpers",
         .product(name: "SnapshotTesting", package: "swift-snapshot-testing"),
         .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
       ],
@@ -115,6 +116,7 @@ let package = Package(
       name: "RealtimeTests",
       dependencies: [
         "Realtime",
+        "TestHelpers",
         .product(name: "CustomDump", package: "swift-custom-dump"),
       ]
     ),
@@ -132,6 +134,7 @@ let package = Package(
       ]
     ),
     .testTarget(name: "SupabaseTests", dependencies: ["Supabase"]),
+    .target(name: "TestHelpers"),
   ]
 )
 

Sources/Auth/AuthClient.swift

@@ -199,9 +199,9 @@ public actor AuthClient {
   public func onAuthStateChange(
     _ listener: @escaping AuthStateChangeListener
   ) async -> some AuthStateChangeListenerRegistration {
-    let handle = eventEmitter.attachListener(listener)
-    await emitInitialSession(forHandle: handle)
-    return handle
+    let token = eventEmitter.attachListener(listener)
+    await emitInitialSession(forToken: token)
+    return token
   }
 
   /// Listen for auth state changes.
@@ -908,9 +908,9 @@ public actor AuthClient {
     return session
   }
 
-  private func emitInitialSession(forHandle handle: AuthStateChangeListenerHandle) async {
+  private func emitInitialSession(forToken token: ObservationToken) async {
     let session = try? await session
-    eventEmitter.emit(.initialSession, session: session, handle: handle)
+    eventEmitter.emit(.initialSession, session: session, token: token)
   }
 
   private func prepareForPKCE() -> (codeChallenge: String?, codeChallengeMethod: String?) {

Sources/Auth/AuthStateChangeListener.swift

@@ -5,35 +5,19 @@
 //  Created by Guilherme Souza on 17/02/24.
 //
 
+import _Helpers
 import ConcurrencyExtras
 import Foundation
 
 /// A listener that can be removed by calling ``AuthStateChangeListenerRegistration/remove()``.
 ///
 /// - Note: Listener is automatically removed on deinit.
-public protocol AuthStateChangeListenerRegistration: Sendable, AnyObject {
+public protocol AuthStateChangeListenerRegistration: Sendable {
   /// Removes the listener. After the initial call, subsequent calls have no effect.
   func remove()
 }
 
-final class AuthStateChangeListenerHandle: AuthStateChangeListenerRegistration {
-  let _onRemove = LockIsolated((@Sendable () -> Void)?.none)
-
-  public func remove() {
-    _onRemove.withValue {
-      if $0 == nil {
-        return
-      }
-
-      $0?()
-      $0 = nil
-    }
-  }
-
-  deinit {
-    remove()
-  }
-}
+extension ObservationToken: AuthStateChangeListenerRegistration {}
 
 public typealias AuthStateChangeListener = @Sendable (
   _ event: AuthChangeEvent,

Sources/Auth/Internal/EventEmitter.swift

@@ -1,15 +1,16 @@
 import ConcurrencyExtras
 import Foundation
+@_spi(Internal) import _Helpers
 
 protocol EventEmitter: Sendable {
   func attachListener(
     _ listener: @escaping AuthStateChangeListener
-  ) -> AuthStateChangeListenerHandle
+  ) -> ObservationToken
 
   func emit(
     _ event: AuthChangeEvent,
     session: Session?,
-    handle: AuthStateChangeListenerHandle?
+    token: ObservationToken?
   )
 }
 
@@ -18,7 +19,7 @@ extension EventEmitter {
     _ event: AuthChangeEvent,
     session: Session?
   ) {
-    emit(event, session: session, handle: nil)
+    emit(event, session: session, token: nil)
   }
 }
 
@@ -27,31 +28,24 @@ final class DefaultEventEmitter: EventEmitter {
 
   private init() {}
 
-  let listeners = LockIsolated<[ObjectIdentifier: AuthStateChangeListener]>([:])
+  let emitter = _Helpers.EventEmitter<(AuthChangeEvent, Session?)?>(
+    initialEvent: nil,
+    emitsLastEventWhenAttaching: false
+  )
 
   func attachListener(
     _ listener: @escaping AuthStateChangeListener
-  ) -> AuthStateChangeListenerHandle {
-    let handle = AuthStateChangeListenerHandle()
-    let key = ObjectIdentifier(handle)
-
-    handle._onRemove.setValue { [weak self] in
-      self?.listeners.withValue {
-        $0[key] = nil
-      }
-    }
-
-    listeners.withValue {
-      $0[key] = listener
+  ) -> ObservationToken {
+    emitter.attach { event in
+      guard let event else { return }
+      listener(event.0, event.1)
     }
-
-    return handle
   }
 
   func emit(
     _ event: AuthChangeEvent,
     session: Session?,
-    handle: AuthStateChangeListenerHandle? = nil
+    token: ObservationToken? = nil
   ) {
     NotificationCenter.default.post(
       name: AuthClient.didChangeAuthStateNotification,
@@ -62,14 +56,6 @@ final class DefaultEventEmitter: EventEmitter {
       ]
     )
 
-    let listeners = listeners.value
-
-    if let handle {
-      listeners[ObjectIdentifier(handle)]?(event, session)
-    } else {
-      for listener in listeners.values {
-        listener(event, session)
-      }
-    }
+    emitter.emit((event, session), to: token)
   }
 }

Sources/Realtime/V2/PushV2.swift

@@ -21,7 +21,7 @@ actor PushV2 {
 
   func send() async -> PushStatus {
     do {
-      try await channel?.socket?.ws?.send(message)
+      try await channel?.socket?.ws.send(message)
 
       if channel?.config.broadcast.acknowledgeBroadcasts == true {
         return await withCheckedContinuation {

Sources/Realtime/V2/RealtimeChannelV2.swift

@@ -33,19 +33,20 @@ public actor RealtimeChannelV2 {
   let logger: (any SupabaseLogger)?
 
   private let callbackManager = CallbackManager()
-  private let statusStream = SharedStream<Status>(initialElement: .unsubscribed)
+
+  private let statusEventEmitter = EventEmitter<Status>(initialEvent: .unsubscribed)
 
   private var clientChanges: [PostgresJoinConfig] = []
   private var joinRef: String?
   private var pushes: [String: PushV2] = [:]
 
   public private(set) var status: Status {
-    get { statusStream.lastElement }
-    set { statusStream.yield(newValue) }
+    get { statusEventEmitter.lastEvent.value }
+    set { statusEventEmitter.emit(newValue) }
   }
 
   public var statusChange: AsyncStream<Status> {
-    statusStream.makeStream()
+    statusEventEmitter.stream()
   }
 
   init(