708 lines
30 KiB
Swift
708 lines
30 KiB
Swift
import Foundation
|
|
|
|
// On Linux, URLSession (incl. URLSessionWebSocketTask) lives in the separate
|
|
// FoundationNetworking module of swift-corelibs-foundation.
|
|
#if canImport(FoundationNetworking)
|
|
import FoundationNetworking
|
|
#endif
|
|
|
|
// The Covalence Relay data path (SYNC_PROTOCOL §3.2, mesh P2): a WebSocket per device
|
|
// into the host's relay room (`cloud/nucleic-edge`), carrying the same length-prefixed Noise
|
|
// frames as every other transport. The relay only ever sees ciphertext — admission is a
|
|
// two-tier token (long-lived membership → 2-minute connection token, `relayToken.ts`), never
|
|
// a content credential. This file is the platform-agnostic client leg shared by the phone
|
|
// and future mac-peer dialers; the host's room listener builds on `RelayWebSocket` from
|
|
// NucleicCore (it additionally speaks the `RelayEnvelope` demux).
|
|
|
|
/// Relay REST + WebSocket endpoints. One base URL serves both (the Worker routes by path),
|
|
/// so a dev override — a single `*.workers.dev` origin — swaps everything at once.
|
|
public enum RelayAPI {
|
|
/// The production relay every distributed build talks to.
|
|
public static let defaultBaseURL = URL(string: "https://relay.nucleic.blakeslee.xyz")!
|
|
|
|
/// Resolve a base URL: an explicit override (pairing payload / dev setting) or the default.
|
|
public static func baseURL(_ override: String?) -> URL {
|
|
override.flatMap(URL.init(string:)) ?? defaultBaseURL
|
|
}
|
|
|
|
/// The `/relay` WebSocket upgrade URL for a connection token (`?t=` — the Worker accepts
|
|
/// the token as a query param because `URLSessionWebSocketTask` can't set custom headers
|
|
/// portably across proxies).
|
|
public static func webSocketURL(base: URL, connectionToken: String) -> URL {
|
|
var components = URLComponents(url: base, resolvingAgainstBaseURL: false)!
|
|
components.scheme = components.scheme == "http" ? "ws" : "wss"
|
|
components.path = "/relay"
|
|
components.queryItems = [URLQueryItem(name: "t", value: connectionToken)]
|
|
return components.url!
|
|
}
|
|
|
|
public enum Error: Swift.Error, LocalizedError, Equatable {
|
|
case membershipRejected(status: Int)
|
|
case badResponse
|
|
case socketClosed
|
|
case connectTimedOut
|
|
|
|
public var errorDescription: String? {
|
|
switch self {
|
|
case .membershipRejected(let status):
|
|
status == 403
|
|
? "This device's relay access was revoked — pair again to restore it."
|
|
: "The relay didn't accept this device's credential (\(status)) — reconnect over LAN once to refresh it."
|
|
case .badResponse: "The relay returned an unexpected response."
|
|
case .socketClosed: "The relay connection closed."
|
|
case .connectTimedOut: "The relay didn't answer in time."
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A minted connection token and the instant it stops being accepted on the `/relay` upgrade.
|
|
/// The token is a stateless HMAC bearer (verified server-side with no KV read on the hot path)
|
|
/// and isn't enforced single-use, so a caller may cache it and reuse it on a reconnect until
|
|
/// `expiresAt` — skipping the `connectionToken` POST round-trip (a cold TLS handshake to the
|
|
/// relay after an iOS suspension) on the fast path back from a lock.
|
|
public struct MintedConnectionToken: Sendable, Equatable {
|
|
public let token: String
|
|
public let expiresAt: Date
|
|
public init(token: String, expiresAt: Date) {
|
|
self.token = token
|
|
self.expiresAt = expiresAt
|
|
}
|
|
}
|
|
|
|
/// Trade a long-lived membership token for the short-lived connection token the WebSocket
|
|
/// upgrade requires (`POST /v1/relay/connect`). Both tiers are HMAC-stateless server-side;
|
|
/// this is one small authenticated POST right before dialing. Returns the token with its
|
|
/// expiry so a caller can cache and reuse it within its TTL.
|
|
public static func mintConnectionToken(
|
|
base: URL, membershipToken: String, session: URLSession = .shared
|
|
) async throws -> MintedConnectionToken {
|
|
var request = URLRequest(url: base.appendingPathComponent("/v1/relay/connect"))
|
|
request.httpMethod = "POST"
|
|
request.timeoutInterval = 10
|
|
request.setValue("Bearer \(membershipToken)", forHTTPHeaderField: "Authorization")
|
|
let (data, response) = try await session.data(for: request)
|
|
guard let http = response as? HTTPURLResponse else { throw Error.badResponse }
|
|
guard http.statusCode == 200 else { throw Error.membershipRejected(status: http.statusCode) }
|
|
guard let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
|
|
let token = json["token"] as? String
|
|
else { throw Error.badResponse }
|
|
// `exp` is Unix-epoch seconds; fall back to a conservative near-term expiry if absent so a
|
|
// cache built on it never over-trusts an old token.
|
|
let exp = (json["exp"] as? Double).map { Date(timeIntervalSince1970: $0) }
|
|
?? Date().addingTimeInterval(60)
|
|
return MintedConnectionToken(token: token, expiresAt: exp)
|
|
}
|
|
|
|
/// Back-compat convenience: just the connection-token string (host/room callers that don't
|
|
/// cache). Prefer `mintConnectionToken` where the expiry matters.
|
|
public static func connectionToken(
|
|
base: URL, membershipToken: String, session: URLSession = .shared
|
|
) async throws -> String {
|
|
try await mintConnectionToken(base: base, membershipToken: membershipToken, session: session).token
|
|
}
|
|
}
|
|
|
|
/// The room's presence broadcast (`room.ts broadcastPresence`) — the relay's text control
|
|
/// plane, explicitly non-secret. The host uses it to reap virtual channels for departed
|
|
/// devices (the room socket outlives any one peer, so nothing else signals a disconnect);
|
|
/// the phone uses it to fail fast when no host holds the room (its frames would be dropped).
|
|
public struct RelayPresence: Sendable, Equatable, Decodable {
|
|
public struct Peer: Sendable, Equatable, Decodable {
|
|
public let deviceId: String
|
|
public let role: String
|
|
|
|
public init(deviceId: String, role: String) {
|
|
self.deviceId = deviceId
|
|
self.role = role
|
|
}
|
|
}
|
|
|
|
public let peers: [Peer]
|
|
|
|
public init(peers: [Peer]) {
|
|
self.peers = peers
|
|
}
|
|
|
|
public var hasHost: Bool { peers.contains { $0.role == "host" } }
|
|
|
|
/// Parse a relay text frame; nil for anything but a presence broadcast.
|
|
public static func parse(_ text: String) -> RelayPresence? {
|
|
struct Envelope: Decodable {
|
|
let t: String
|
|
let peers: [Peer]?
|
|
}
|
|
guard let envelope = try? JSONDecoder().decode(Envelope.self, from: Data(text.utf8)),
|
|
envelope.t == "presence"
|
|
else { return nil }
|
|
return RelayPresence(peers: envelope.peers ?? [])
|
|
}
|
|
}
|
|
|
|
/// Settles a `connect` exactly once. On Linux the upgrader's completion races the channel's
|
|
/// close (a rejected token never sends 101; the edge just closes) and completing an
|
|
/// `EventLoopPromise` twice traps; on Darwin the pong, the first inbound message, and the
|
|
/// connect deadline all race. Shared by both backends.
|
|
final class OnceResult: @unchecked Sendable {
|
|
private let lock = NSLock()
|
|
private var result: Result<Void, Error>?
|
|
private var waiter: CheckedContinuation<Void, Error>?
|
|
|
|
func succeed() { settle(.success(())) }
|
|
func fail(_ error: Error) { settle(.failure(error)) }
|
|
|
|
private func settle(_ outcome: Result<Void, Error>) {
|
|
lock.lock()
|
|
defer { lock.unlock() }
|
|
guard result == nil else { return }
|
|
result = outcome
|
|
if let waiter {
|
|
self.waiter = nil
|
|
waiter.resume(with: outcome)
|
|
}
|
|
}
|
|
|
|
func value() async throws {
|
|
try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, Error>) in
|
|
lock.lock()
|
|
if let result {
|
|
lock.unlock()
|
|
c.resume(with: result)
|
|
return
|
|
}
|
|
waiter = c
|
|
lock.unlock()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// What the host's room listener needs from its socket — seam for tests (an in-memory fake
|
|
/// drives the demux without a network). `RelayWebSocket` is the real implementation.
|
|
public protocol RelayRoomSocket: AnyObject, Sendable {
|
|
func messages() -> AsyncStream<RelayWebSocket.Message>
|
|
func send(_ data: Data)
|
|
func close()
|
|
}
|
|
|
|
/// A connected relay WebSocket: ordered sends, an inbound message stream, and a keepalive
|
|
/// ping loop. Shared by the phone's `RelayFrameChannel` (raw Noise frames) and the host's
|
|
/// room listener (enveloped frames + presence text). `connect` returns once admission is
|
|
/// confirmed — by a ping/pong round-trip *or* by the first inbound message, whichever lands
|
|
/// first. The room broadcasts presence to every socket the moment it accepts it, so the
|
|
/// first message is a deterministic application-level admission signal; requiring the pong
|
|
/// alone stranded healthy sockets whenever the network path ate protocol pings (observed in
|
|
/// production against Cloudflare: upgrade + data flowing, pong never delivered — every
|
|
/// connect "timed out" while the room saw us join).
|
|
///
|
|
/// Two backends, one surface: `URLSessionWebSocketTask` on Darwin, and SwiftNIO on Linux
|
|
/// (see the second `RelayWebSocket` below) — corelibs-foundation's WebSocket task rides
|
|
/// libcurl's experimental WS support, absent from distro builds.
|
|
#if canImport(Darwin)
|
|
public final class RelayWebSocket: RelayRoomSocket, @unchecked Sendable {
|
|
public enum Message: Sendable {
|
|
case binary(Data)
|
|
case text(String)
|
|
}
|
|
|
|
private let task: URLSessionWebSocketTask
|
|
private let inbound: AsyncStream<Message>
|
|
private let inboundContinuation: AsyncStream<Message>.Continuation
|
|
private let outboundContinuation: AsyncStream<Data>.Continuation
|
|
private var writerTask: Task<Void, Never>?
|
|
private var receiveTask: Task<Void, Never>?
|
|
private var pingTask: Task<Void, Never>?
|
|
|
|
/// Liveness bookkeeping for the keepalive loop (lock-guarded: receive and ping-completion
|
|
/// callbacks land on different queues). `pingSentAt` is the send time of the still-
|
|
/// unanswered keepalive ping, nil when none is outstanding.
|
|
private let liveLock = NSLock()
|
|
private var pingSentAt: Date?
|
|
private var lastInboundAt = Date()
|
|
/// Fires exactly once on the first inbound message — `connect`'s admission signal.
|
|
private var onFirstInbound: (() -> Void)?
|
|
|
|
/// How often to ping: keeps NAT bindings warm and detects a dead socket well inside the
|
|
/// reconnect budget. (Cloudflare's edge answers protocol pings without waking the DO —
|
|
/// though not on every path, so liveness also counts ordinary inbound traffic.)
|
|
private static let pingInterval: Duration = .seconds(25)
|
|
|
|
public static func connect(
|
|
url: URL, session: URLSession = .shared, timeout: Duration = .seconds(15)
|
|
) async throws -> RelayWebSocket {
|
|
let task = session.webSocketTask(with: url)
|
|
task.maximumMessageSize = WireFraming.maxFrameSize + 64
|
|
// A rejected upgrade (bad/expired token) surfaces as the ping failing; admission
|
|
// surfaces as the pong *or* the room's immediate presence broadcast (the receive
|
|
// pump is already running — the socket is built before resume). A blackholed socket
|
|
// (middlebox that accepts the upgrade but drops traffic) surfaces as *nothing* —
|
|
// URLSession imposes no post-handshake deadline — so the wait is raced against a
|
|
// hard timeout. Without it a single wedged connect suspends the caller forever: the
|
|
// sync-server start path, and every dial loop above this, have no deadline of their
|
|
// own until the socket is returned.
|
|
let settled = OnceResult()
|
|
let socket = RelayWebSocket(task: task, onFirstInbound: { settled.succeed() })
|
|
task.resume()
|
|
let deadline = Task {
|
|
try? await Task.sleep(for: timeout)
|
|
settled.fail(RelayAPI.Error.connectTimedOut)
|
|
}
|
|
task.sendPing { error in
|
|
if let error { settled.fail(error) } else { settled.succeed() }
|
|
}
|
|
do {
|
|
try await settled.value()
|
|
} catch {
|
|
deadline.cancel()
|
|
socket.close()
|
|
throw error
|
|
}
|
|
deadline.cancel()
|
|
return socket
|
|
}
|
|
|
|
private init(task: URLSessionWebSocketTask, onFirstInbound: (() -> Void)? = nil) {
|
|
self.task = task
|
|
self.onFirstInbound = onFirstInbound
|
|
var inCont: AsyncStream<Message>.Continuation!
|
|
inbound = AsyncStream(bufferingPolicy: .unbounded) { inCont = $0 }
|
|
inboundContinuation = inCont
|
|
|
|
var outCont: AsyncStream<Data>.Continuation!
|
|
let outbound = AsyncStream<Data>(bufferingPolicy: .unbounded) { outCont = $0 }
|
|
outboundContinuation = outCont
|
|
|
|
// Single writer so wire order matches send order (the Noise nonce sequence).
|
|
writerTask = Task { [task] in
|
|
for await data in outbound {
|
|
do { try await task.send(.data(data)) } catch { break }
|
|
}
|
|
}
|
|
receiveTask = Task { [weak self, task] in
|
|
while !Task.isCancelled {
|
|
do {
|
|
let message = try await task.receive()
|
|
guard let self else { break }
|
|
self.noteInbound()
|
|
switch message {
|
|
case .data(let data): self.inboundContinuation.yield(.binary(data))
|
|
case .string(let text): self.inboundContinuation.yield(.text(text))
|
|
@unknown default: break
|
|
}
|
|
} catch {
|
|
break
|
|
}
|
|
}
|
|
self?.inboundContinuation.finish()
|
|
}
|
|
pingTask = Task { [weak self, task] in
|
|
while !Task.isCancelled {
|
|
try? await Task.sleep(for: Self.pingInterval)
|
|
// Exit when the socket is gone, not just when cancelled: a RelayWebSocket
|
|
// dropped without close() (socket died, owner deallocated) would otherwise
|
|
// leave this loop pinging the dead URLSession task — which URLSession
|
|
// retains — every 25s forever, one leaked task set per redial.
|
|
guard !Task.isCancelled, let self else { break }
|
|
// The previous ping is still unanswered and nothing else arrived since it
|
|
// was sent: the socket is dead in the server→client direction. Without this,
|
|
// a blackholed socket whose pong handler simply never fires stays "connected"
|
|
// forever (URLSession puts no deadline on the pong) and the room quietly
|
|
// loses its host. Inbound traffic counts as life because some network paths
|
|
// deliver data fine while eating ping/pong frames.
|
|
if self.pingLooksDead() {
|
|
self.close()
|
|
break
|
|
}
|
|
self.notePingSent()
|
|
task.sendPing { [weak self] error in
|
|
self?.notePingAnswered()
|
|
if error != nil { self?.close() }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private func noteInbound() {
|
|
liveLock.lock()
|
|
lastInboundAt = Date()
|
|
let first = onFirstInbound
|
|
onFirstInbound = nil
|
|
liveLock.unlock()
|
|
first?()
|
|
}
|
|
|
|
private func notePingSent() {
|
|
liveLock.lock()
|
|
pingSentAt = Date()
|
|
liveLock.unlock()
|
|
}
|
|
|
|
private func notePingAnswered() {
|
|
liveLock.lock()
|
|
pingSentAt = nil
|
|
liveLock.unlock()
|
|
}
|
|
|
|
/// True when a keepalive ping has gone a full interval with neither a pong nor any other
|
|
/// inbound message — the socket is dead, not merely pong-deprived.
|
|
private func pingLooksDead() -> Bool {
|
|
liveLock.lock()
|
|
defer { liveLock.unlock() }
|
|
guard let sentAt = pingSentAt else { return false }
|
|
return lastInboundAt < sentAt
|
|
}
|
|
|
|
/// Ordered inbound messages; finishes when the socket dies.
|
|
public func messages() -> AsyncStream<Message> { inbound }
|
|
|
|
/// Enqueue one binary message (ordered, non-blocking).
|
|
public func send(_ data: Data) { outboundContinuation.yield(data) }
|
|
|
|
/// Enqueue one text message (relay control plane — presence/register/wake, never E2EE
|
|
/// payload). Ordering vs. binary sends is not guaranteed; control messages don't need it.
|
|
public func sendText(_ text: String) {
|
|
Task { [task] in try? await task.send(.string(text)) }
|
|
}
|
|
|
|
public func close() {
|
|
pingTask?.cancel()
|
|
receiveTask?.cancel()
|
|
writerTask?.cancel()
|
|
outboundContinuation.finish()
|
|
inboundContinuation.finish()
|
|
task.cancel(with: .goingAway, reason: nil)
|
|
}
|
|
}
|
|
#else
|
|
|
|
// Linux (the runner): the same `RelayWebSocket` surface on SwiftNIO — TLS via NIOSSL,
|
|
// HTTP/1.1 upgrade, RFC 6455 client framing (masked frames, ping/pong, fragment
|
|
// reassembly). The seam CLOUD_RUNTIME §3.1 names for the cloud host's outbound socket.
|
|
import NIOCore
|
|
import NIOFoundationCompat
|
|
import NIOHTTP1
|
|
import NIOPosix
|
|
import NIOSSL
|
|
import NIOWebSocket
|
|
|
|
public final class RelayWebSocket: RelayRoomSocket, @unchecked Sendable {
|
|
public enum Message: Sendable {
|
|
case binary(Data)
|
|
case text(String)
|
|
}
|
|
|
|
/// One small shared loop group for every relay socket this process opens (a runner holds
|
|
/// exactly one room socket plus reconnect attempts — a single thread is plenty).
|
|
private static let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
|
|
private static let sslContext: NIOSSLContext = {
|
|
// A client context with the platform trust roots; failure here is a build/deploy
|
|
// defect (BoringSSL always constructs a default client config).
|
|
try! NIOSSLContext(configuration: .makeClientConfiguration())
|
|
}()
|
|
|
|
/// How often to ping: keeps NAT bindings warm and detects a dead socket well inside the
|
|
/// reconnect budget. (Cloudflare's edge answers protocol pings without waking the DO.)
|
|
private static let pingInterval: Duration = .seconds(25)
|
|
|
|
private let channel: Channel
|
|
private let inbound: AsyncStream<Message>
|
|
private let inboundContinuation: AsyncStream<Message>.Continuation
|
|
private var pingTask: Task<Void, Never>?
|
|
|
|
public static func connect(
|
|
url: URL, session: URLSession = .shared, timeout: Duration = .seconds(15)
|
|
) async throws -> RelayWebSocket {
|
|
guard let host = url.host else { throw RelayAPI.Error.badResponse }
|
|
let useTLS = url.scheme == "wss" || url.scheme == "https"
|
|
let port = url.port ?? (useTLS ? 443 : 80)
|
|
|
|
var inCont: AsyncStream<Message>.Continuation!
|
|
let inbound = AsyncStream<Message>(bufferingPolicy: .unbounded) { inCont = $0 }
|
|
let continuation = inCont!
|
|
|
|
let components = URLComponents(url: url, resolvingAgainstBaseURL: false)
|
|
var uri = components?.percentEncodedPath ?? "/relay"
|
|
if uri.isEmpty { uri = "/relay" }
|
|
if let query = components?.percentEncodedQuery { uri += "?" + query }
|
|
|
|
// The upgrade settles exactly once: success from the upgrader's completion, or failure
|
|
// from the channel dying first (a rejected token closes the connection without a 101).
|
|
let upgraded = OnceResult()
|
|
let messageHandler = RelayWebSocketFrameHandler(continuation: continuation)
|
|
let upgrader = NIOWebSocketClientUpgrader(
|
|
requestKey: Data((0..<16).map { _ in UInt8.random(in: .min ... .max) }).base64EncodedString(),
|
|
maxFrameSize: WireFraming.maxFrameSize
|
|
) { channel, _ in
|
|
channel.pipeline.addHandler(messageHandler)
|
|
}
|
|
|
|
let bootstrap = ClientBootstrap(group: group)
|
|
.connectTimeout(.seconds(10))
|
|
.channelInitializer { channel in
|
|
let requestSender = RelayUpgradeRequestSender(host: host, uri: uri)
|
|
let tls: EventLoopFuture<Void>
|
|
if useTLS {
|
|
do {
|
|
let ssl = try NIOSSLClientHandler(context: sslContext, serverHostname: host)
|
|
tls = channel.pipeline.addHandler(ssl)
|
|
} catch {
|
|
return channel.eventLoop.makeFailedFuture(error)
|
|
}
|
|
} else {
|
|
tls = channel.eventLoop.makeSucceededVoidFuture()
|
|
}
|
|
return tls.flatMap {
|
|
channel.pipeline.addHTTPClientHandlers(withClientUpgrade: (
|
|
upgraders: [upgrader],
|
|
completionHandler: { _ in upgraded.succeed() }
|
|
))
|
|
}.flatMap {
|
|
channel.pipeline.addHandler(requestSender)
|
|
}
|
|
}
|
|
|
|
let channel = try await bootstrap.connect(host: host, port: port).get()
|
|
channel.closeFuture.whenComplete { _ in
|
|
upgraded.fail(RelayAPI.Error.socketClosed)
|
|
continuation.finish()
|
|
}
|
|
// The close future only fires if the server closes: a non-101 answer over HTTP
|
|
// keep-alive (NIO's client upgrader forwards it and removes itself without closing),
|
|
// or a server that accepts TCP and then goes silent, would otherwise park this await
|
|
// forever — `connectTimeout` above covers only the TCP leg. Race a hard deadline.
|
|
let deadline = Task {
|
|
try? await Task.sleep(for: timeout)
|
|
upgraded.fail(RelayAPI.Error.connectTimedOut)
|
|
}
|
|
do {
|
|
try await upgraded.value()
|
|
} catch {
|
|
deadline.cancel()
|
|
try? await channel.close()
|
|
throw error
|
|
}
|
|
deadline.cancel()
|
|
return RelayWebSocket(channel: channel, inbound: inbound, continuation: continuation)
|
|
}
|
|
|
|
private init(
|
|
channel: Channel, inbound: AsyncStream<Message>,
|
|
continuation: AsyncStream<Message>.Continuation
|
|
) {
|
|
self.channel = channel
|
|
self.inbound = inbound
|
|
self.inboundContinuation = continuation
|
|
pingTask = Task { [weak self, channel] in
|
|
while !Task.isCancelled {
|
|
try? await Task.sleep(for: Self.pingInterval)
|
|
guard !Task.isCancelled else { break }
|
|
guard channel.isActive else {
|
|
self?.close()
|
|
break
|
|
}
|
|
let frame = WebSocketFrame(
|
|
fin: true, opcode: .ping, maskKey: .random(),
|
|
data: channel.allocator.buffer(capacity: 0))
|
|
channel.writeAndFlush(frame, promise: nil)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Ordered inbound messages; finishes when the socket dies.
|
|
public func messages() -> AsyncStream<Message> { inbound }
|
|
|
|
/// Enqueue one binary message (ordered, non-blocking — NIO serializes channel writes).
|
|
public func send(_ data: Data) {
|
|
let frame = WebSocketFrame(
|
|
fin: true, opcode: .binary, maskKey: .random(),
|
|
data: channel.allocator.buffer(bytes: data))
|
|
channel.writeAndFlush(frame, promise: nil)
|
|
}
|
|
|
|
/// Enqueue one text message (relay control plane — presence/register/wake, never E2EE
|
|
/// payload).
|
|
public func sendText(_ text: String) {
|
|
let frame = WebSocketFrame(
|
|
fin: true, opcode: .text, maskKey: .random(),
|
|
data: channel.allocator.buffer(string: text))
|
|
channel.writeAndFlush(frame, promise: nil)
|
|
}
|
|
|
|
public func close() {
|
|
pingTask?.cancel()
|
|
let goAway = WebSocketFrame(
|
|
fin: true, opcode: .connectionClose, maskKey: .random(),
|
|
data: channel.allocator.buffer(capacity: 0))
|
|
channel.writeAndFlush(goAway, promise: nil)
|
|
channel.close(promise: nil)
|
|
inboundContinuation.finish()
|
|
}
|
|
}
|
|
|
|
/// Fires the HTTP upgrade GET the moment the (TLS) channel is up; the
|
|
/// `NIOHTTPClientUpgradeHandler` ahead of it stamps the Upgrade/Sec-WebSocket-* headers on.
|
|
private final class RelayUpgradeRequestSender: ChannelInboundHandler, RemovableChannelHandler,
|
|
@unchecked Sendable
|
|
{
|
|
typealias InboundIn = HTTPClientResponsePart
|
|
typealias OutboundOut = HTTPClientRequestPart
|
|
|
|
private let host: String
|
|
private let uri: String
|
|
|
|
init(host: String, uri: String) {
|
|
self.host = host
|
|
self.uri = uri
|
|
}
|
|
|
|
func channelActive(context: ChannelHandlerContext) {
|
|
var head = HTTPRequestHead(version: .http1_1, method: .GET, uri: uri)
|
|
head.headers.add(name: "Host", value: host)
|
|
context.write(wrapOutboundOut(.head(head)), promise: nil)
|
|
context.writeAndFlush(wrapOutboundOut(.end(nil)), promise: nil)
|
|
context.fireChannelActive()
|
|
}
|
|
}
|
|
|
|
/// The post-upgrade WebSocket leg: reassembles fragmented messages, answers pings, and
|
|
/// yields whole binary/text messages to the inbound stream.
|
|
private final class RelayWebSocketFrameHandler: ChannelInboundHandler, @unchecked Sendable {
|
|
typealias InboundIn = WebSocketFrame
|
|
typealias OutboundOut = WebSocketFrame
|
|
|
|
private let continuation: AsyncStream<RelayWebSocket.Message>.Continuation
|
|
/// A fragmented message in progress: its initial opcode + accumulated bytes.
|
|
private var pending: (opcode: WebSocketOpcode, buffer: ByteBuffer)?
|
|
|
|
init(continuation: AsyncStream<RelayWebSocket.Message>.Continuation) {
|
|
self.continuation = continuation
|
|
}
|
|
|
|
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
|
|
var frame = unwrapInboundIn(data)
|
|
switch frame.opcode {
|
|
case .binary, .text:
|
|
let payload = frame.unmaskedData
|
|
if frame.fin {
|
|
emit(opcode: frame.opcode, buffer: payload)
|
|
} else {
|
|
pending = (frame.opcode, payload)
|
|
}
|
|
case .continuation:
|
|
guard var partial = pending else { break }
|
|
var payload = frame.unmaskedData
|
|
partial.buffer.writeBuffer(&payload)
|
|
if frame.fin {
|
|
pending = nil
|
|
emit(opcode: partial.opcode, buffer: partial.buffer)
|
|
} else {
|
|
pending = partial
|
|
}
|
|
case .ping:
|
|
let pong = WebSocketFrame(
|
|
fin: true, opcode: .pong, maskKey: .random(), data: frame.unmaskedData)
|
|
context.writeAndFlush(wrapOutboundOut(pong), promise: nil)
|
|
case .connectionClose:
|
|
let close = WebSocketFrame(
|
|
fin: true, opcode: .connectionClose, maskKey: .random(),
|
|
data: context.channel.allocator.buffer(capacity: 0))
|
|
context.writeAndFlush(wrapOutboundOut(close), promise: nil)
|
|
context.close(promise: nil)
|
|
default:
|
|
break
|
|
}
|
|
}
|
|
|
|
private func emit(opcode: WebSocketOpcode, buffer: ByteBuffer) {
|
|
var buffer = buffer
|
|
if opcode == .text {
|
|
let text = buffer.readString(length: buffer.readableBytes) ?? ""
|
|
continuation.yield(.text(text))
|
|
} else {
|
|
let data = buffer.readData(length: buffer.readableBytes) ?? Data()
|
|
continuation.yield(.binary(data))
|
|
}
|
|
}
|
|
|
|
func channelInactive(context: ChannelHandlerContext) {
|
|
continuation.finish()
|
|
context.fireChannelInactive()
|
|
}
|
|
|
|
func errorCaught(context: ChannelHandlerContext, error: Error) {
|
|
continuation.finish()
|
|
context.close(promise: nil)
|
|
}
|
|
}
|
|
#endif
|
|
|
|
/// The phone-side relay `FrameChannel` (the client leg — raw Noise frames, no routing
|
|
/// envelope; the Room DO addresses clients itself). Frames keep the same `WireFraming`
|
|
/// length prefix as TCP so every layer above sees identical bytes on every transport.
|
|
public final class RelayFrameChannel: FrameChannel, @unchecked Sendable {
|
|
public let peerDescription: String
|
|
private let socket: RelayWebSocket
|
|
private let accumulator = FrameAccumulator()
|
|
private let stream: AsyncStream<Data>
|
|
private let continuation: AsyncStream<Data>.Continuation
|
|
private var pumpTask: Task<Void, Never>?
|
|
|
|
/// Trade the membership token for a connection token and open the room WebSocket.
|
|
public static func dial(
|
|
base: URL, membershipToken: String, session: URLSession = .shared
|
|
) async throws -> RelayFrameChannel {
|
|
let token = try await RelayAPI.connectionToken(
|
|
base: base, membershipToken: membershipToken, session: session)
|
|
return try await dial(base: base, connectionToken: token, session: session)
|
|
}
|
|
|
|
/// Open the room WebSocket with an already-obtained connection token — the fast reconnect path,
|
|
/// where the caller reuses a still-valid cached token and skips the `connectionToken` POST. A
|
|
/// stale/expired token surfaces as a rejected upgrade (`RelayWebSocket.connect` throws), which
|
|
/// the caller catches to re-mint and retry.
|
|
public static func dial(
|
|
base: URL, connectionToken: String, session: URLSession = .shared
|
|
) async throws -> RelayFrameChannel {
|
|
let socket = try await RelayWebSocket.connect(
|
|
url: RelayAPI.webSocketURL(base: base, connectionToken: connectionToken), session: session)
|
|
return RelayFrameChannel(socket: socket, host: base.host ?? "relay")
|
|
}
|
|
|
|
init(socket: RelayWebSocket, host: String) {
|
|
self.socket = socket
|
|
self.peerDescription = "relay:\(host)"
|
|
var cont: AsyncStream<Data>.Continuation!
|
|
stream = AsyncStream(bufferingPolicy: .unbounded) { cont = $0 }
|
|
continuation = cont
|
|
pumpTask = Task { [weak self] in
|
|
for await message in socket.messages() {
|
|
guard let self else { break }
|
|
switch message {
|
|
case .binary(let data):
|
|
guard let frames = try? self.accumulator.push(data) else {
|
|
self.close()
|
|
return
|
|
}
|
|
for frame in frames { self.continuation.yield(frame) }
|
|
case .text(let text):
|
|
// No host in the room means our frames are being dropped (`routeFromClient`
|
|
// has no host socket) — fail fast so the dial chain / retry loop moves on.
|
|
if let presence = RelayPresence.parse(text), !presence.hasHost {
|
|
self.close()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
self?.continuation.finish()
|
|
}
|
|
}
|
|
|
|
public func frames() -> AsyncStream<Data> { stream }
|
|
|
|
public func send(_ frame: Data) { socket.send(WireFraming.frame(frame)) }
|
|
|
|
public func close() {
|
|
pumpTask?.cancel()
|
|
socket.close()
|
|
continuation.finish()
|
|
}
|
|
}
|