feat(realtime): send broadcast events through HTTP (#476)

Author: grdsdev

.swiftpm/configuration/Package.resolved

@@ -35,6 +35,9 @@
          <TestPlanReference
             reference = "container:TestPlans/Integration.xctestplan">
          </TestPlanReference>
+         <TestPlanReference
+            reference = "container:TestPlans/AllTests.xctestplan">
+         </TestPlanReference>
       </TestPlans>
    </TestAction>
    <LaunchAction

.swiftpm/xcode/xcshareddata/xcschemes/Supabase.xcscheme

@@ -23,10 +23,10 @@ let package = Package(
   ],
   dependencies: [
     .package(url: "https://github.com/apple/swift-crypto.git", "1.0.0" ..< "4.0.0"),
-    .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", from: "1.0.0"),
-    .package(url: "https://github.com/pointfreeco/swift-custom-dump", from: "1.3.0"),
-    .package(url: "https://github.com/pointfreeco/swift-snapshot-testing", from: "1.8.1"),
-    .package(url: "https://github.com/pointfreeco/swift-issue-reporting", from: "1.2.0"),
+    .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", from: "1.1.0"),
+    .package(url: "https://github.com/pointfreeco/swift-custom-dump", from: "1.3.2"),
+    .package(url: "https://github.com/pointfreeco/swift-snapshot-testing", from: "1.17.2"),
+    .package(url: "https://github.com/pointfreeco/xctest-dynamic-overlay", from: "1.2.2"),
   ],
   targets: [
     .target(
@@ -55,7 +55,7 @@ let package = Package(
       dependencies: [
         .product(name: "CustomDump", package: "swift-custom-dump"),
         .product(name: "SnapshotTesting", package: "swift-snapshot-testing"),
-        .product(name: "IssueReporting", package: "swift-issue-reporting"),
+        .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Helpers",
         "Auth",
         "TestHelpers",
@@ -71,7 +71,7 @@ let package = Package(
       dependencies: [
         .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
         .product(name: "SnapshotTesting", package: "swift-snapshot-testing"),
-        .product(name: "IssueReporting", package: "swift-issue-reporting"),
+        .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Functions",
         "TestHelpers",
       ],
@@ -82,7 +82,7 @@ let package = Package(
       dependencies: [
         .product(name: "CustomDump", package: "swift-custom-dump"),
         .product(name: "InlineSnapshotTesting", package: "swift-snapshot-testing"),
-        .product(name: "IssueReporting", package: "swift-issue-reporting"),
+        .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Helpers",
         "Auth",
         "PostgREST",
@@ -129,7 +129,7 @@ let package = Package(
       name: "StorageTests",
       dependencies: [
         .product(name: "CustomDump", package: "swift-custom-dump"),
-        .product(name: "IssueReporting", package: "swift-issue-reporting"),
+        .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Storage",
       ]
     ),
@@ -155,7 +155,7 @@ let package = Package(
       name: "TestHelpers",
       dependencies: [
         .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
-        .product(name: "IssueReporting", package: "swift-issue-reporting"),
+        .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Auth",
       ]
     ),

Package.resolved

@@ -105,13 +105,6 @@ public struct RealtimeChannelOptions {
   }
 }
 
-/// Represents the different status of a push
-public enum PushStatus: String, Sendable {
-  case ok
-  case error
-  case timeout
-}
-
 public enum RealtimeSubscribeStates {
   case subscribed
   case timedOut

Package.swift

@@ -8,6 +8,13 @@
 import Foundation
 import Helpers
 
+/// Represents the different status of a push
+public enum PushStatus: String, Sendable {
+  case ok
+  case error
+  case timeout
+}
+
 actor PushV2 {
   private weak var channel: RealtimeChannelV2?
   let message: RealtimeMessageV2

Sources/Realtime/RealtimeChannel.swift

@@ -9,35 +9,56 @@ import ConcurrencyExtras
 import Foundation
 import Helpers
 
+#if canImport(FoundationNetworking)
+  import FoundationNetworking
+
+extension HTTPURLResponse {
+  convenience init() {
+    self.init(
+      url: URL(string: "http://127.0.0.1")!,
+      statusCode: 200,
+      httpVersion: nil,
+      headerFields: nil
+    )!
+  }
+}
+#endif
+
 public struct RealtimeChannelConfig: Sendable {
   public var broadcast: BroadcastJoinConfig
   public var presence: PresenceJoinConfig
   public var isPrivate: Bool
 }
 
 struct Socket: Sendable {
+  var broadcastURL: @Sendable () -> URL
   var status: @Sendable () -> RealtimeClientV2.Status
   var options: @Sendable () -> RealtimeClientOptions
   var accessToken: @Sendable () -> String?
+  var apiKey: @Sendable () -> String?
   var makeRef: @Sendable () -> Int
 
   var connect: @Sendable () async -> Void
   var addChannel: @Sendable (_ channel: RealtimeChannelV2) -> Void
   var removeChannel: @Sendable (_ channel: RealtimeChannelV2) async -> Void
   var push: @Sendable (_ message: RealtimeMessageV2) async -> Void
+  var httpSend: @Sendable (_ request: HTTPRequest) async throws -> HTTPResponse
 }
 
 extension Socket {
   init(client: RealtimeClientV2) {
     self.init(
+      broadcastURL: { [weak client] in client?.broadcastURL ?? URL(string: "http://localhost")! },
       status: { [weak client] in client?.status ?? .disconnected },
       options: { [weak client] in client?.options ?? .init() },
       accessToken: { [weak client] in client?.mutableState.accessToken },
+      apiKey: { [weak client] in client?.apikey },
       makeRef: { [weak client] in client?.makeRef() ?? 0 },
       connect: { [weak client] in await client?.connect() },
       addChannel: { [weak client] in client?.addChannel($0) },
       removeChannel: { [weak client] in await client?.removeChannel($0) },
-      push: { [weak client] in await client?.push($0) }
+      push: { [weak client] in await client?.push($0) },
+      httpSend: { [weak client] in try await client?.http.send($0) ?? .init(data: Data(), response: HTTPURLResponse()) }
     )
   }
 }
@@ -202,24 +223,64 @@ public final class RealtimeChannelV2: Sendable {
   ///   - event: Broadcast message event.
   ///   - message: Message payload.
   public func broadcast(event: String, message: JSONObject) async {
-    assert(
-      status == .subscribed,
-      "You can only broadcast after subscribing to the channel. Did you forget to call `channel.subscribe()`?"
-    )
+    if status != .subscribed {
+      struct Message: Encodable {
+        let topic: String
+        let event: String
+        let payload: JSONObject
+        let `private`: Bool
+      }
 
-    await push(
-      RealtimeMessageV2(
-        joinRef: mutableState.joinRef,
-        ref: socket.makeRef().description,
-        topic: topic,
-        event: ChannelEvent.broadcast,
-        payload: [
-          "type": "broadcast",
-          "event": .string(event),
-          "payload": .object(message),
-        ]
+      var headers = HTTPHeaders(["content-type": "application/json"])
+      if let apiKey = socket.apiKey() {
+        headers["apikey"] = apiKey
+      }
+      if let accessToken = socket.accessToken() {
+        headers["authorization"] = "Bearer \(accessToken)"
+      }
+
+      let task = Task { [headers] in
+        _ = try? await socket.httpSend(
+          HTTPRequest(
+            url: socket.broadcastURL(),
+            method: .post,
+            headers: headers,
+            body: JSONEncoder().encode(
+              [
+                "messages": [
+                  Message(
+                    topic: topic,
+                    event: event,
+                    payload: message,
+                    private: config.isPrivate
+                  ),
+                ],
+              ]
+            )
+          )
+        )
+      }
+
+      if config.broadcast.acknowledgeBroadcasts {
+        try? await withTimeout(interval: socket.options().timeoutInterval) {
+          await task.value
+        }
+      }
+    } else {
+      await push(
+        RealtimeMessageV2(
+          joinRef: mutableState.joinRef,
+          ref: socket.makeRef().description,
+          topic: topic,
+          event: ChannelEvent.broadcast,
+          payload: [
+            "type": "broadcast",
+            "event": .string(event),
+            "payload": .object(message),
+          ]
+        )
       )
-    )
+    }
   }
 
   public func track(_ state: some Codable) async throws {

Sources/Realtime/V2/PushV2.swift

@@ -79,6 +79,7 @@ public final class RealtimeClientV2: Sendable {
   let options: RealtimeClientOptions
   let ws: any WebSocketClient
   let mutableState = LockIsolated(MutableState())
+  let http: any HTTPClientType
   let apikey: String?
 
   public var subscriptions: [String: RealtimeChannelV2] {
@@ -128,6 +129,12 @@ public final class RealtimeClientV2: Sendable {
   }
 
   public convenience init(url: URL, options: RealtimeClientOptions) {
+    var interceptors: [any HTTPClientInterceptor] = []
+
+    if let logger = options.logger {
+      interceptors.append(LoggerInterceptor(logger: logger))
+    }
+
     self.init(
       url: url,
       options: options,
@@ -137,14 +144,24 @@ public final class RealtimeClientV2: Sendable {
           apikey: options.apikey
         ),
         options: options
+      ),
+      http: HTTPClient(
+        fetch: options.fetch ?? { try await URLSession.shared.data(for: $0) },
+        interceptors: interceptors
       )
     )
   }
 
-  init(url: URL, options: RealtimeClientOptions, ws: any WebSocketClient) {
+  init(
+    url: URL,
+    options: RealtimeClientOptions,
+    ws: any WebSocketClient,
+    http: any HTTPClientType
+  ) {
     self.url = url
     self.options = options
     self.ws = ws
+    self.http = http
     apikey = options.apikey
 
     mutableState.withValue {
@@ -471,7 +488,7 @@ public final class RealtimeClientV2: Sendable {
     return url
   }
 
-  private var broadcastURL: URL {
+  var broadcastURL: URL {
     url.appendingPathComponent("api/broadcast")
   }
 }