feat(realtime): v3 client implementation

.superpowers/sdd/ie-flows-report.md

@@ -0,0 +1,101 @@
+# IE-3..6 E2E Integration Test Report
+
+## Summary
+
+All 13 integration tests (IE-1..6, 6 suites) pass GREEN against the live local Supabase instance (Docker, project `realtimev3`, `realtime:v2.107.5`).
+
+**Branch**: `claude/charming-euler-49c5a1`  
+**HEAD after commit**: see `git log --oneline -1`  
+**Test run**: `swift test --filter RealtimeV3IntegrationTests`  
+**Unit suite**: `swift test --filter RealtimeV3Tests` — 132 tests pass, 1 pre-existing known issue (unchanged)  
+**Supabase instance**: LEFT RUNNING (not stopped)
+
+---
+
+## IE-3 Broadcast (BroadcastE2ETests.swift) — GREEN
+
+### IE-3a: WS round-trip between two clients — PASS
+Two separate `Realtime` clients join the same topic. B opens a `broadcasts(of:event:)` stream before subscribing. A broadcasts a `ChatMsg`. B receives it within the timeout. Confirmed against live server.
+
+### IE-3b: HTTP broadcast received by WS subscriber — PASS (with SDK GAP noted)
+B subscribes via WebSocket. HTTP broadcast via `Realtime.httpBroadcastBatch` delivers the message. **SDK GAP discovered and documented** (see Concerns below). Test uses `httpBroadcastBatch` directly with the correct short topic as a workaround.
+
+### IE-3c: `acknowledge=true` returns without timeout — PASS
+Channel created with `broadcast.acknowledge = true`. `broadcast(...)` returns cleanly; server ACKs the push.
+
+---
+
+## IE-4 Presence (PresenceE2ETests.swift) — GREEN
+
+### IE-4a: presence sync between two clients — PASS
+A and B join same topic with `presence.enabled = true`. A tracks `UserPresence(userId: "user-a", status: "active")`. B's `presence.observe(UserPresence.self)` stream sees A appear in the active map. A cancels the track handle; B sees A's leave. Handle correctly cancelled to suppress leak warning.
+
+---
+
+## IE-5 Postgres Changes (PostgresChangesE2ETests.swift) — GREEN
+
+### IE-5a: INSERT delivers postgres change — PASS
+Registered `channel.inserts(schema:table:filter:)` with a unique `room_id` UUID before `subscribe()`. After join, inserted a row via `PostgrestClient`. The `postgresChanges(for:)` stream yielded the row within 15 seconds. `record["content"]` matched expected value. Server-id routing (`_buildServerIDRouting`) works correctly — join reply includes `postgres_changes: [{id: <int>, ...}]` which maps to the registration UUID.
+
+### IE-5b: UPDATE and DELETE deliver old_record — PASS (with server behavior note)
+UPDATE: `old_record` contains the full original row (REPLICA IDENTITY FULL working). New `record` contains updated values. DELETE: `old_record` contains **only the primary key**, not all columns. This is **not an SDK bug** — it is intentional Realtime server v2.x behavior: for DELETE, the deleted row no longer exists for RLS evaluation, so the server returns only the PK in `old_record`. Test updated to assert `old_record.id` matches the deleted row's UUID.
+
+---
+
+## IE-6 Reconnection (ReconnectionE2ETests.swift) — GREEN (with SDK GAP noted)
+
+### IE-6a: leave → disconnect → connect → subscribe — PASS
+Channel leaves cleanly, transitions to `.closed`. After `disconnect()` + `subscribe()` (which internally calls `connect()`), the channel rejoins and reaches `.joined`.
+
+### IE-6b: broadcast stream after leave → disconnect → connect cycle — PASS
+Pre-disconnect: receiver gets `Ping(seq: 1)`. After `leave()` + `disconnect()` + `subscribe()`: receiver re-joins, gets `Ping(seq: 2)` from the post-reconnect stream.
+
+**SDK GAP discovered**: After `disconnect()` WITHOUT a prior `leave()`, the channel state remains `.joined` in the SDK (transport severed, but logical state preserved for unclean-drop transparent rejoin). Calling `subscribe()` on a `.joined` channel is idempotent (returns immediately). Callers that want to reuse a channel after an intentional disconnect must call `leave()` before `disconnect()`. Forced-drop (unclean) reconnection is covered deterministically by `RealtimeV3Tests/RejoinTests` using `InMemoryTransport`.
+
+---
+
+## Real-Server Findings and SDK Gaps
+
+### SDK GAP 1 — `Channel.httpBroadcast` topic format
+
+**File**: `Sources/RealtimeV3/HTTP/HttpBroadcast.swift`, `Channel.httpBroadcast`  
+**Symptom**: `Channel.httpBroadcast(event:payload:)` sends HTTP 202 but the message is **not delivered** to WebSocket subscribers.  
+**Root cause**: `Channel.httpBroadcast` passes `topic` (the full `realtime:<short>` string) as the topic in the HTTP broadcast body. The Realtime server's `/api/broadcast` endpoint expects the **short topic without the `realtime:` prefix** to route to WS subscribers. Using the full prefixed topic causes a routing mismatch — accepted (202) but not delivered.  
+**Evidence**: `curl` tests confirmed that `"topic":"room:foo"` delivers; `"topic":"realtime:room:foo"` does not.  
+**Workaround in tests**: Uses `Realtime.httpBroadcastBatch` directly with the short topic.  
+**Fix needed**: Strip the `realtime:` prefix when building `HttpBroadcastMessage.topic` in `Channel.httpBroadcast`.
+
+### SDK GAP 2 — HTTP broadcast requires service-role JWT
+
+**File**: `Sources/RealtimeV3/HTTP/HttpBroadcast.swift`  
+**Symptom**: HTTP broadcast with `apikey: <anon_key>` returns HTTP 500 from the Realtime server.  
+**Root cause**: The Realtime `/api/broadcast` endpoint requires a JWT with `service_role` role (Bearer auth), not just the anon key header. The SDK sends `apikey: <key>` when no `accessToken` provider is configured, which is rejected.  
+**Evidence**: `curl` with `Authorization: Bearer <service_role_key>` → 202; with `apikey: <anon_key>` → 500.  
+**Fix needed**: Document that `httpBroadcast` requires the Realtime client to be initialized with a service-role `accessToken` provider, or add a specific error message when the server returns 500 for this call.
+
+### SDK GAP 3 — Intentional disconnect does not clear channel state
+
+**File**: `Sources/RealtimeV3/Realtime.swift`, `disconnect()`  
+**Symptom**: After `disconnect()`, channels remain in `.joined` state. Calling `subscribe()` is a no-op.  
+**Root cause**: `disconnect()` does not cascade state transitions to channels — this is by design (channels preserve state for unclean-drop transparent rejoin). But it creates a footgun for intentional disconnect+reconnect scenarios.  
+**Fix needed**: Either document explicitly that `leave()` must be called before `disconnect()` when the caller wants to resubscribe, or add a `disconnectMode` parameter to `disconnect()` that optionally leaves all channels first.
+
+### Server behavior note — DELETE old_record
+
+The Realtime server v2.107.5 returns only the PK columns in `old_record` for DELETE events, even with `REPLICA IDENTITY FULL` and permissive RLS. This is intentional security behavior (deleted row can't be evaluated against RLS). The SDK correctly surfaces what the server sends. No SDK fix needed; documentation note added in the test.
+
+---
+
+## Warnings Check
+
+Clean: `swift build --build-tests 2>&1 | grep "warning:" | grep -E "BroadcastE2E|PresenceE2E|PostgresChanges|Reconnection|IntegrationTests"` → no output.
+
+---
+
+## Files
+
+- `Tests/RealtimeV3IntegrationTests/BroadcastE2ETests.swift` (IE-3, 3 tests)
+- `Tests/RealtimeV3IntegrationTests/PresenceE2ETests.swift` (IE-4, 1 test)
+- `Tests/RealtimeV3IntegrationTests/PostgresChangesE2ETests.swift` (IE-5, 2 tests)
+- `Tests/RealtimeV3IntegrationTests/ReconnectionE2ETests.swift` (IE-6, 2 tests)
+- `Tests/RealtimeV3IntegrationTests/Support/IntegrationEnv.swift` (added `serviceRoleKey` + `makeRealtimeWithServiceRole()`)

Makefile

@@ -53,6 +53,15 @@ test-integration:
 	swift test --filter IntegrationTests
 	cd Tests/IntegrationTests && supabase stop
 
+# Run RealtimeV3 integration tests against a live local Supabase instance.
+# Starts the dedicated realtimev3 project, resets the DB (applies migrations + seed),
+# then runs the integration suite. The instance is intentionally left running after
+# the first invocation so subsequent runs skip the slow start step.
+# To stop: cd Tests/RealtimeV3IntegrationTests/supabase && supabase stop
+test-realtime-v3-integration:
+	cd Tests/RealtimeV3IntegrationTests/supabase && supabase start && supabase db reset --local
+	swift test --filter RealtimeV3IntegrationTests
+
 build-for-library-evolution:
 	swift build \
 		-q \
@@ -81,7 +90,7 @@ format:
 		-not -path '*/.*' -print0 \
 		| xargs -0 swift format --ignore-unparsable-files --in-place
 
-.PHONY: build-for-library-evolution format warm-simulator xcodebuild test-docs test-integration
+.PHONY: build-for-library-evolution format warm-simulator xcodebuild test-docs test-integration test-realtime-v3-integration
 
 .PHONY: coverage
 coverage:

Package.swift

@@ -18,6 +18,7 @@ let package = Package(
     .library(name: "Functions", targets: ["Functions"]),
     .library(name: "PostgREST", targets: ["PostgREST"]),
     .library(name: "Realtime", targets: ["Realtime"]),
+    .library(name: "RealtimeV3", targets: ["RealtimeV3"]),
     .library(name: "Storage", targets: ["Storage"]),
     .library(name: "Supabase", targets: ["Supabase"]),
   ],
@@ -159,6 +160,40 @@ let package = Package(
         "TestHelpers",
       ]
     ),
+    .target(
+      name: "RealtimeV3",
+      dependencies: [
+        .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+        .product(name: "HTTPTypes", package: "swift-http-types"),
+        .product(name: "Clocks", package: "swift-clocks"),
+        .product(name: "IssueReporting", package: "xctest-dynamic-overlay"),
+        "Helpers",
+      ]
+    ),
+    .testTarget(
+      name: "RealtimeV3Tests",
+      dependencies: [
+        .product(name: "CustomDump", package: "swift-custom-dump"),
+        .product(name: "Clocks", package: "swift-clocks"),
+        .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+        "Mocker",
+        "RealtimeV3",
+        "TestHelpers",
+      ]
+    ),
+    .testTarget(
+      name: "RealtimeV3IntegrationTests",
+      dependencies: [
+        .product(name: "CustomDump", package: "swift-custom-dump"),
+        .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+        "RealtimeV3",
+        "PostgREST",
+        "Helpers",
+      ],
+      exclude: [
+        "supabase",
+      ]
+    ),
     .target(
       name: "Storage",
       dependencies: [

Sources/RealtimeV3/Channel+Broadcast.swift

@@ -0,0 +1,160 @@
+//
+//  Channel+Broadcast.swift
+//  RealtimeV3
+//
+//  Created by Guilherme Souza on 29/06/26.
+//
+
+import Foundation
+import Helpers
+
+// MARK: - Channel + broadcast send
+
+extension Channel {
+
+  /// Sends a broadcast message with an `Encodable` payload.
+  ///
+  /// ## State gating
+  /// - `.unsubscribed` / `.joining` → throws `.notSubscribed`
+  /// - `.leaving` / `.closed` → throws `.channelClosed(reason)`
+  /// - `.joined` → encodes payload and sends the binary push frame
+  ///
+  /// ## Ack mode
+  /// When `options.broadcast.acknowledge == true`, the call suspends until the server
+  /// sends a `phx_reply` for this push, or `configuration.broadcastAckTimeout` elapses
+  /// (throws `.broadcastAckTimeout`). When `acknowledge == false` the frame is sent
+  /// fire-and-forget.
+  ///
+  /// ## Wire format
+  /// The binary frame is a Phoenix 2.0.0 broadcast push (kind byte `0x03`). The JSON
+  /// payload inside the frame is:
+  /// ```json
+  /// {"type": "broadcast", "event": "<event>", "payload": <encodedT>}
+  /// ```
+  /// This is symmetric with the receive side (`broadcasts(of:event:)`).
+  ///
+  /// - Parameters:
+  ///   - payload: The message payload. Encoded to JSON before sending.
+  ///   - event: The broadcast event name (e.g. `"chat"`).
+  /// - Throws: `RealtimeError`
+  public func broadcast<T: Encodable & Sendable>(_ payload: T, as event: String)
+    async throws(RealtimeError)
+  {
+    // Guard: ensure the owning Realtime is still alive (needed for the ack timeout below).
+    guard let realtime else { throw .channelClosed(.clientDisconnected) }
+    try _requireJoinedForSend()
+
+    // Build the inner broadcast envelope (user payload encoded via Configuration.encoder).
+    let envelope: JSONObject = [
+      "type": .string("broadcast"),
+      "event": .string(event),
+      "payload": try _encodeToJSON(payload),
+    ]
+
+    try await _push(
+      .broadcast, .broadcastJSON(envelope),
+      ack: options.broadcast.acknowledge
+        ? .require(timeout: realtime.configuration.broadcastAckTimeout, error: .broadcastAckTimeout)
+        : .none
+    )
+  }
+
+  /// Sends a broadcast message with a raw binary payload.
+  ///
+  /// The binary data is shipped as-is inside the Phoenix 2.0.0 broadcast push frame
+  /// (kind byte `0x03`, encoding byte `0x00`). Ack semantics are identical to the
+  /// `Encodable` overload.
+  ///
+  /// - Parameters:
+  ///   - data: The raw binary payload.
+  ///   - event: The broadcast event name.
+  /// - Throws: `RealtimeError`
+  public func broadcast(_ data: Data, as event: String) async throws(RealtimeError) {
+    // Guard: ensure the owning Realtime is still alive (needed for the ack timeout below).
+    guard let realtime else { throw .channelClosed(.clientDisconnected) }
+    try _requireJoinedForSend()
+
+    try await _push(
+      .broadcast, .broadcastData(data),
+      ack: options.broadcast.acknowledge
+        ? .require(timeout: realtime.configuration.broadcastAckTimeout, error: .broadcastAckTimeout)
+        : .none
+    )
+  }
+}
+
+// MARK: - Channel + broadcasts(of:event:)
+
+extension Channel {
+  /// Returns an `AsyncThrowingStream` that yields every broadcast message for the
+  /// given `event` name, decoded to `T`.
+  ///
+  /// ## Wire shape
+  /// A broadcast Phoenix frame has `event == "broadcast"` and a JSON payload of the form:
+  /// ```json
+  /// { "type": "broadcast", "event": "<name>", "payload": { ... } }
+  /// ```
+  /// This method filters frames whose inner `event` matches the requested name and
+  /// decodes the inner `payload` object to `T` using `Configuration.decoder`.
+  ///
+  /// ## Per-call fan-out (Decision 8)
+  /// Each call mints an independent stream. N concurrent calls each receive a copy
+  /// of every matching message. Streams created before `subscribe()` are valid —
+  /// they start producing once frames arrive after the join.
+  ///
+  /// ## Decode failure
+  /// If the inner `payload` cannot be decoded to `T`, the stream terminates by
+  /// throwing `RealtimeError.decoding(type:underlying:)`. Non-matching events are
+  /// silently ignored.
+  ///
+  /// ## Terminal close
+  /// When the channel transitions to `.closed(reason)` (e.g. via `leave()`), the
+  /// stream terminates by throwing `RealtimeError.channelClosed(reason)`.
+  ///
+  /// - Note: The thrown error is always a `RealtimeError`. Cast with `as? RealtimeError`
+  ///   or use `if case` matching on the caught `any Error`.
+  public func broadcasts<T: Decodable & Sendable>(
+    of type: T.Type,
+    event: String
+  ) -> AsyncThrowingStream<T, any Error> {
+    // Capture the decoder from realtime configuration, falling back to the default if
+    // realtime has already been deallocated (e.g. stream registered after client teardown).
+    let decoder = realtime?.configuration.decoder ?? .realtimeDefault
+
+    // The helper handles terminal close (→ `.channelClosed`) and task lifecycle; the body
+    // filters for the requested broadcast event and decodes, throwing on decode failure.
+    return _makeThrowingStream(initialState: ()) { _, message, continuation in
+      // Only handle broadcast Phoenix events matching the requested inner event.
+      guard message.event == .broadcast,
+        case .json(let jsonValue) = message.payload,
+        let obj = jsonValue.objectValue,
+        let innerEvent = obj["event"]?.stringValue, innerEvent == event
+      else { return }
+
+      // Extract the inner "payload" value.
+      guard let innerPayload = obj["payload"] else {
+        // No payload key — decode failure terminates the stream.
+        throw RealtimeError.decoding(
+          type: String(describing: T.self),
+          underlying: MissingPayloadError()
+        )
+      }
+
+      // Re-encode the JSONValue to Data, then decode T using the configured decoder.
+      // A decode failure throws and terminates the stream (spec: decode failure throws).
+      do {
+        let data = try JSONEncoder().encode(innerPayload)
+        continuation.yield(try decoder.decode(T.self, from: data))
+      } catch {
+        throw RealtimeError.decoding(type: String(describing: T.self), underlying: error)
+      }
+    }
+  }
+}
+
+// MARK: - MissingPayloadError
+
+/// Sentinel error used when a broadcast frame has no inner `payload` key.
+private struct MissingPayloadError: Error, Sendable {
+  var localizedDescription: String { "Broadcast frame is missing the inner 'payload' key." }
+}