feat(_Realtime): Realtime v3 — idiomatic Swift 6 actor-based API

.gitignore

@@ -91,7 +91,9 @@ iOSInjectionProject/
 
 .DS_Store
 /.build
-/Packages
+/Packages/*
+!/Packages/_Realtime
+!/Packages/_RealtimeTableMacros
 /*.xcodeproj
 /.swiftpm
 

Examples/Examples/Realtime/TodoRealtimeView.swift

@@ -105,14 +105,16 @@ struct TodoRealtimeView: View {
     // Handle insertions
     async let insertionObservation: () = { @MainActor in
       for await insertion in insertions {
-        try todos.insert(insertion.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder), at: 0)
+        try todos.insert(
+          insertion.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder), at: 0)
       }
     }()
 
     // Handle updates
     async let updatesObservation: () = { @MainActor in
       for await update in updates {
-        let record = try update.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder) as Todo
+        let record =
+          try update.decodeRecord(decoder: PostgrestClient.Configuration.jsonDecoder) as Todo
         todos[id: record.id] = record
       }
     }()

Package.swift

@@ -22,6 +22,7 @@ let package = Package(
     .library(name: "Supabase", targets: ["Supabase"]),
   ],
   dependencies: [
+    .package(path: "Packages/_Realtime"),
     .package(url: "https://github.com/apple/swift-crypto.git", "3.0.0"..<"5.0.0"),
     .package(url: "https://github.com/apple/swift-http-types.git", from: "1.3.0"),
     .package(url: "https://github.com/pointfreeco/swift-clocks", from: "1.0.0"),

Packages/_Realtime/Package.swift

@@ -0,0 +1,49 @@
+// swift-tools-version: 6.0
+import PackageDescription
+
+let package = Package(
+  name: "_Realtime",
+  platforms: [
+    .iOS(.v17),
+    .macOS(.v14),
+    .tvOS(.v17),
+    .watchOS(.v10),
+    .visionOS(.v1),
+  ],
+  products: [
+    .library(name: "_Realtime", targets: ["_Realtime"])
+  ],
+  dependencies: [
+    .package(path: "../_RealtimeTableMacros"),
+    .package(url: "https://github.com/pointfreeco/swift-clocks", from: "1.0.0"),
+    .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", from: "1.1.0"),
+    .package(url: "https://github.com/pointfreeco/xctest-dynamic-overlay", from: "1.2.2"),
+    .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.0"),
+  ],
+  targets: [
+    .target(
+      name: "_Realtime",
+      dependencies: [
+        .product(name: "_RealtimeTableMacros", package: "_RealtimeTableMacros"),
+        .product(name: "Clocks", package: "swift-clocks"),
+        .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+        .product(name: "IssueReporting", package: "xctest-dynamic-overlay"),
+      ],
+      swiftSettings: [
+        .swiftLanguageMode(.v6),
+        .enableUpcomingFeature("ExistentialAny"),
+      ]
+    ),
+    .testTarget(
+      name: "_RealtimeTests",
+      dependencies: [
+        .product(name: "Clocks", package: "swift-clocks"),
+        .product(name: "CustomDump", package: "swift-custom-dump"),
+        .product(name: "InlineSnapshotTesting", package: "swift-snapshot-testing"),
+        .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+        "_Realtime",
+      ]
+    ),
+  ]
+)

Packages/_Realtime/Sources/_Realtime/Broadcast/BroadcastMessage.swift

@@ -0,0 +1,24 @@
+//
+//  BroadcastMessage.swift
+//  _Realtime
+//
+//  Created by Guilherme Souza on 24/04/25.
+//
+
+import Foundation
+
+/// A broadcast message received from a Realtime channel.
+public struct BroadcastMessage: Sendable {
+  /// The event name carried by this broadcast.
+  public let event: String
+  /// The message payload as a `JSONValue`.
+  public let payload: JSONValue
+  /// The local time at which this message was received.
+  public let receivedAt: Date
+
+  public init(event: String, payload: JSONValue, receivedAt: Date = Date()) {
+    self.event = event
+    self.payload = payload
+    self.receivedAt = receivedAt
+  }
+}

Packages/_Realtime/Sources/_Realtime/Broadcast/Channel+Broadcast.swift

@@ -0,0 +1,148 @@
+//
+//  Channel+Broadcast.swift
+//  _Realtime
+//
+//  Created by Guilherme Souza on 24/04/25.
+//
+
+import Foundation
+
+extension Channel {
+  // MARK: - Receive API
+
+  /// Returns an `AsyncThrowingStream` that yields every broadcast message arriving on this channel.
+  ///
+  /// The channel is automatically joined on the first call if it has not been joined yet.
+  /// The stream finishes with a `RealtimeError` when the channel closes or an error occurs.
+  ///
+  /// Multiple calls create independent fan-out streams — all subscribers receive every message.
+  ///
+  /// - Returns: An async stream of `BroadcastMessage` values.
+  public func broadcasts() -> AsyncThrowingStream<BroadcastMessage, any Error> {
+    AsyncThrowingStream { continuation in
+      let id = UUID()
+      Task {
+        self.registerBroadcastContinuation(id: id, continuation: continuation)
+        continuation.onTermination = { [id] _ in
+          Task { await self.removeBroadcastContinuation(id: id) }
+        }
+        do {
+          try await self.joinIfNeeded()
+        } catch {
+          continuation.finish(throwing: error)
+        }
+      }
+    }
+  }
+
+  /// Returns an `AsyncThrowingStream` that yields broadcast messages matching `event`,
+  /// decoded as `T` using the channel's configured `JSONDecoder`.
+  ///
+  /// - Parameters:
+  ///   - event: Only messages whose `event` field equals this string are emitted.
+  ///   - decoder: A custom `JSONDecoder`. Defaults to `JSONDecoder()`.
+  ///   - type: The `Decodable` type to decode the payload into.
+  /// - Returns: An async stream of decoded values.
+  public func broadcasts<T: Decodable & Sendable>(
+    of _: T.Type = T.self,
+    event: String,
+    decoder: JSONDecoder = JSONDecoder()
+  ) -> AsyncThrowingStream<T, any Error> {
+    AsyncThrowingStream { continuation in
+      let base = self.broadcasts()
+      let task = Task {
+        do {
+          for try await msg in base {
+            guard msg.event == event else { continue }
+            do {
+              let data = try JSONEncoder().encode(msg.payload)
+              let value = try decoder.decode(T.self, from: data)
+              continuation.yield(value)
+            } catch {
+              continuation.finish(
+                throwing: RealtimeError.decoding(
+                  type: String(describing: T.self), underlying: error)
+              )
+              return
+            }
+          }
+          continuation.finish()
+        } catch {
+          continuation.finish(throwing: error)
+        }
+      }
+      continuation.onTermination = { _ in task.cancel() }
+    }
+  }
+
+  // MARK: - Send API
+
+  /// Broadcasts an `Encodable` value as a JSON payload on this channel.
+  ///
+  /// The channel must be in the `.joined` state. Use ``join()`` or rely on ``broadcasts()``
+  /// auto-join before calling this method.
+  ///
+  /// - Parameters:
+  ///   - value: The value to encode and broadcast.
+  ///   - event: The event name carried by the broadcast.
+  /// - Throws: `RealtimeError.channelNotJoined` if the channel has not joined,
+  ///           `RealtimeError.encoding` if the value cannot be encoded.
+  public func broadcast<T: Encodable & Sendable>(
+    _ value: T,
+    as event: String
+  ) async throws(RealtimeError) {
+    guard currentState == .joined else { throw .channelNotJoined }
+    guard let realtime else { throw .disconnected }
+
+    let payloadData: Data
+    do {
+      payloadData = try realtime.configuration.encoder.encode(value)
+    } catch {
+      throw .encoding(underlying: error)
+    }
+
+    let payloadDict: [String: JSONValue]
+    do {
+      payloadDict = try JSONDecoder().decode([String: JSONValue].self, from: payloadData)
+    } catch {
+      throw .encoding(underlying: error)
+    }
+
+    let msg = PhoenixMessage(
+      joinRef: nil, ref: nil,
+      topic: topic, event: "broadcast",
+      payload: [
+        "type": "broadcast",
+        "event": .string(event),
+        "payload": .object(payloadDict),
+      ]
+    )
+    try await realtime.send(msg)
+  }
+
+  /// Broadcasts raw binary data on this channel using the Realtime binary frame format.
+  ///
+  /// - Parameters:
+  ///   - data: The raw binary payload.
+  ///   - event: The event name carried by the broadcast.
+  /// - Throws: `RealtimeError.channelNotJoined` if not joined, or a transport error.
+  public func broadcast(_ data: Data, as event: String) async throws(RealtimeError) {
+    guard currentState == .joined else { throw .channelNotJoined }
+    guard let realtime else { throw .disconnected }
+
+    let frame: Data
+    do {
+      frame = try PhoenixSerializer.encodeBroadcastPush(
+        joinRef: nil, ref: nil,
+        topic: topic, event: event,
+        binaryPayload: data
+      )
+    } catch let e as RealtimeError {
+      throw e
+    } catch {
+      throw .encoding(underlying: error)
+    }
+
+    try await realtime.sendBinary(frame)
+  }
+}