339 lines
13 KiB
Swift
339 lines
13 KiB
Swift
import ClawdisProtocol
|
|
import Foundation
|
|
import OSLog
|
|
|
|
struct GatewayEvent: Codable {
|
|
let type: String
|
|
let event: String?
|
|
let payload: AnyCodable?
|
|
let seq: Int?
|
|
}
|
|
|
|
// Avoid ambiguity with the app's own AnyCodable type.
|
|
private typealias ProtoAnyCodable = ClawdisProtocol.AnyCodable
|
|
|
|
extension Notification.Name {
|
|
static let gatewaySnapshot = Notification.Name("clawdis.gateway.snapshot")
|
|
static let gatewayEvent = Notification.Name("clawdis.gateway.event")
|
|
static let gatewaySeqGap = Notification.Name("clawdis.gateway.seqgap")
|
|
}
|
|
|
|
private actor GatewayChannelActor {
|
|
private let logger = Logger(subsystem: "com.steipete.clawdis", category: "gateway")
|
|
private var task: URLSessionWebSocketTask?
|
|
private var pending: [String: CheckedContinuation<GatewayFrame, Error>] = [:]
|
|
private var connected = false
|
|
private var url: URL
|
|
private var token: String?
|
|
private let session = URLSession(configuration: .default)
|
|
private var backoffMs: Double = 500
|
|
private var shouldReconnect = true
|
|
private var lastSeq: Int?
|
|
private var lastTick: Date?
|
|
private var tickIntervalMs: Double = 30000
|
|
private let decoder = JSONDecoder()
|
|
private let encoder = JSONEncoder()
|
|
private var watchdogTask: Task<Void, Never>?
|
|
private let defaultRequestTimeoutMs: Double = 15_000
|
|
|
|
init(url: URL, token: String?) {
|
|
self.url = url
|
|
self.token = token
|
|
Task { [weak self] in
|
|
await self?.startWatchdog()
|
|
}
|
|
}
|
|
|
|
private func startWatchdog() {
|
|
self.watchdogTask?.cancel()
|
|
self.watchdogTask = Task { [weak self] in
|
|
guard let self else { return }
|
|
await self.watchdogLoop()
|
|
}
|
|
}
|
|
|
|
private func watchdogLoop() async {
|
|
// Keep nudging reconnect in case exponential backoff stalls.
|
|
while self.shouldReconnect {
|
|
try? await Task.sleep(nanoseconds: 30 * 1_000_000_000) // 30s cadence
|
|
if self.connected { continue }
|
|
do {
|
|
try await self.connect()
|
|
} catch {
|
|
let wrapped = self.wrap(error, context: "gateway watchdog reconnect")
|
|
self.logger.error("gateway watchdog reconnect failed \(wrapped.localizedDescription, privacy: .public)")
|
|
}
|
|
}
|
|
}
|
|
|
|
func connect() async throws {
|
|
if self.connected, self.task?.state == .running { return }
|
|
self.task?.cancel(with: .goingAway, reason: nil)
|
|
self.task = self.session.webSocketTask(with: self.url)
|
|
self.task?.resume()
|
|
do {
|
|
try await self.sendHello()
|
|
} catch {
|
|
let wrapped = self.wrap(error, context: "connect to gateway @ \(self.url.absoluteString)")
|
|
throw wrapped
|
|
}
|
|
self.listen()
|
|
self.connected = true
|
|
self.backoffMs = 500
|
|
self.lastSeq = nil
|
|
}
|
|
|
|
private func sendHello() async throws {
|
|
let osVersion = ProcessInfo.processInfo.operatingSystemVersion
|
|
let platform = "macos \(osVersion.majorVersion).\(osVersion.minorVersion).\(osVersion.patchVersion)"
|
|
let primaryLocale = Locale.preferredLanguages.first ?? Locale.current.identifier
|
|
|
|
let hello = Hello(
|
|
type: "hello",
|
|
minprotocol: GATEWAY_PROTOCOL_VERSION,
|
|
maxprotocol: GATEWAY_PROTOCOL_VERSION,
|
|
client: [
|
|
"name": ClawdisProtocol.AnyCodable("clawdis-mac"),
|
|
"version": ClawdisProtocol.AnyCodable(Bundle.main.infoDictionary?["CFBundleShortVersionString"] as? String ?? "dev"),
|
|
"platform": ClawdisProtocol.AnyCodable(platform),
|
|
"mode": ClawdisProtocol.AnyCodable("app"),
|
|
"instanceId": ClawdisProtocol.AnyCodable(Host.current().localizedName ?? UUID().uuidString),
|
|
],
|
|
caps: [],
|
|
auth: self.token.map { ["token": ClawdisProtocol.AnyCodable($0)] },
|
|
locale: primaryLocale,
|
|
useragent: ProcessInfo.processInfo.operatingSystemVersionString)
|
|
let data = try JSONEncoder().encode(hello)
|
|
try await self.task?.send(.data(data))
|
|
guard let msg = try await task?.receive() else {
|
|
throw NSError(domain: "Gateway", code: 1, userInfo: [NSLocalizedDescriptionKey: "hello failed (no response)"])
|
|
}
|
|
try await self.handleHelloResponse(msg)
|
|
}
|
|
|
|
private func handleHelloResponse(_ msg: URLSessionWebSocketTask.Message) async throws {
|
|
let data: Data? = switch msg {
|
|
case let .data(d): d
|
|
case let .string(s): s.data(using: .utf8)
|
|
@unknown default: nil
|
|
}
|
|
guard let data else {
|
|
throw NSError(domain: "Gateway", code: 1, userInfo: [NSLocalizedDescriptionKey: "hello failed (empty response)"])
|
|
}
|
|
let decoder = JSONDecoder()
|
|
if let ok = try? decoder.decode(HelloOk.self, from: data) {
|
|
if let tick = ok.policy["tickIntervalMs"]?.value as? Double {
|
|
self.tickIntervalMs = tick
|
|
} else if let tick = ok.policy["tickIntervalMs"]?.value as? Int {
|
|
self.tickIntervalMs = Double(tick)
|
|
}
|
|
self.lastTick = Date()
|
|
Task { await self.watchTicks() }
|
|
let frame = GatewayFrame.helloOk(ok)
|
|
NotificationCenter.default.post(name: .gatewaySnapshot, object: frame)
|
|
return
|
|
}
|
|
if let err = try? decoder.decode(HelloError.self, from: data) {
|
|
throw NSError(domain: "Gateway", code: 1, userInfo: [NSLocalizedDescriptionKey: "hello-error: \(err.reason)"])
|
|
}
|
|
throw NSError(domain: "Gateway", code: 1, userInfo: [NSLocalizedDescriptionKey: "hello failed (unexpected response)"])
|
|
}
|
|
|
|
private func listen() {
|
|
self.task?.receive { [weak self] result in
|
|
guard let self else { return }
|
|
switch result {
|
|
case let .failure(err):
|
|
Task { await self.handleReceiveFailure(err) }
|
|
case let .success(msg):
|
|
Task {
|
|
await self.handle(msg)
|
|
await self.listen()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private func handleReceiveFailure(_ err: Error) async {
|
|
let wrapped = self.wrap(err, context: "gateway receive")
|
|
self.logger.error("gateway ws receive failed \(wrapped.localizedDescription, privacy: .public)")
|
|
self.connected = false
|
|
await self.failPending(wrapped)
|
|
await self.scheduleReconnect()
|
|
}
|
|
|
|
private func handle(_ msg: URLSessionWebSocketTask.Message) async {
|
|
let data: Data? = switch msg {
|
|
case let .data(d): d
|
|
case let .string(s): s.data(using: .utf8)
|
|
@unknown default: nil
|
|
}
|
|
guard let data else { return }
|
|
guard let frame = try? self.decoder.decode(GatewayFrame.self, from: data) else {
|
|
self.logger.error("gateway decode failed")
|
|
return
|
|
}
|
|
switch frame {
|
|
case let .res(res):
|
|
let id = res.id
|
|
if let waiter = pending.removeValue(forKey: id) {
|
|
waiter.resume(returning: .res(res))
|
|
}
|
|
case let .event(evt):
|
|
if let seq = evt.seq {
|
|
if let last = lastSeq, seq > last + 1 {
|
|
NotificationCenter.default.post(
|
|
name: .gatewaySeqGap,
|
|
object: frame,
|
|
userInfo: ["expected": last + 1, "received": seq])
|
|
}
|
|
self.lastSeq = seq
|
|
}
|
|
if evt.event == "tick" { self.lastTick = Date() }
|
|
NotificationCenter.default.post(name: .gatewayEvent, object: frame)
|
|
case .helloOk:
|
|
self.lastTick = Date()
|
|
NotificationCenter.default.post(name: .gatewaySnapshot, object: frame)
|
|
default:
|
|
break
|
|
}
|
|
}
|
|
|
|
private func watchTicks() async {
|
|
let tolerance = self.tickIntervalMs * 2
|
|
while self.connected {
|
|
try? await Task.sleep(nanoseconds: UInt64(tolerance * 1_000_000))
|
|
guard self.connected else { return }
|
|
if let last = self.lastTick {
|
|
let delta = Date().timeIntervalSince(last) * 1000
|
|
if delta > tolerance {
|
|
self.logger.error("gateway tick missed; reconnecting")
|
|
self.connected = false
|
|
await self.failPending(
|
|
NSError(
|
|
domain: "Gateway",
|
|
code: 4,
|
|
userInfo: [NSLocalizedDescriptionKey: "gateway tick missed; reconnecting"]))
|
|
await self.scheduleReconnect()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private func scheduleReconnect() async {
|
|
guard self.shouldReconnect else { return }
|
|
let delay = self.backoffMs / 1000
|
|
self.backoffMs = min(self.backoffMs * 2, 30000)
|
|
try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
|
|
do {
|
|
try await self.connect()
|
|
} catch {
|
|
let wrapped = self.wrap(error, context: "gateway reconnect")
|
|
self.logger.error("gateway reconnect failed \(wrapped.localizedDescription, privacy: .public)")
|
|
await self.scheduleReconnect()
|
|
}
|
|
}
|
|
|
|
func request(method: String, params: [String: AnyCodable]?, timeoutMs: Double? = nil) async throws -> Data {
|
|
do {
|
|
try await self.connect()
|
|
} catch {
|
|
throw self.wrap(error, context: "gateway connect")
|
|
}
|
|
let id = UUID().uuidString
|
|
let effectiveTimeout = timeoutMs ?? self.defaultRequestTimeoutMs
|
|
// Encode request using the generated models to avoid JSONSerialization/ObjC bridging pitfalls.
|
|
let paramsObject: ProtoAnyCodable? = params.map { entries in
|
|
let dict = entries.reduce(into: [String: ProtoAnyCodable]()) { dict, entry in
|
|
dict[entry.key] = ProtoAnyCodable(entry.value.value)
|
|
}
|
|
return ProtoAnyCodable(dict)
|
|
}
|
|
let frame = RequestFrame(
|
|
type: "req",
|
|
id: id,
|
|
method: method,
|
|
params: paramsObject)
|
|
let data = try self.encoder.encode(frame)
|
|
let response = try await withCheckedThrowingContinuation { (cont: CheckedContinuation<GatewayFrame, Error>) in
|
|
self.pending[id] = cont
|
|
Task { [weak self] in
|
|
guard let self else { return }
|
|
try? await Task.sleep(nanoseconds: UInt64(effectiveTimeout * 1_000_000))
|
|
await self.timeoutRequest(id: id, timeoutMs: effectiveTimeout)
|
|
}
|
|
Task {
|
|
do {
|
|
try await self.task?.send(.data(data))
|
|
} catch {
|
|
self.pending.removeValue(forKey: id)
|
|
cont.resume(throwing: self.wrap(error, context: "gateway send \(method)"))
|
|
}
|
|
}
|
|
}
|
|
guard case let .res(res) = response else {
|
|
throw NSError(domain: "Gateway", code: 2, userInfo: [NSLocalizedDescriptionKey: "unexpected frame"])
|
|
}
|
|
if res.ok == false {
|
|
let msg = (res.error?["message"]?.value as? String) ?? "gateway error"
|
|
throw NSError(domain: "Gateway", code: 3, userInfo: [NSLocalizedDescriptionKey: msg])
|
|
}
|
|
if let payload = res.payload {
|
|
// Encode back to JSON with Swift's encoder to preserve types and avoid ObjC bridging exceptions.
|
|
return try self.encoder.encode(payload)
|
|
}
|
|
return Data() // Should not happen, but tolerate empty payloads.
|
|
}
|
|
|
|
// Wrap low-level URLSession/WebSocket errors with context so UI can surface them.
|
|
private func wrap(_ error: Error, context: String) -> Error {
|
|
if let urlError = error as? URLError {
|
|
let desc = urlError.localizedDescription.isEmpty ? "cancelled" : urlError.localizedDescription
|
|
return NSError(
|
|
domain: URLError.errorDomain,
|
|
code: urlError.errorCode,
|
|
userInfo: [NSLocalizedDescriptionKey: "\(context): \(desc)"])
|
|
}
|
|
let ns = error as NSError
|
|
let desc = ns.localizedDescription.isEmpty ? "unknown" : ns.localizedDescription
|
|
return NSError(domain: ns.domain, code: ns.code, userInfo: [NSLocalizedDescriptionKey: "\(context): \(desc)"])
|
|
}
|
|
|
|
private func failPending(_ error: Error) async {
|
|
let waiters = self.pending
|
|
self.pending.removeAll()
|
|
for (_, waiter) in waiters {
|
|
waiter.resume(throwing: error)
|
|
}
|
|
}
|
|
|
|
private func timeoutRequest(id: String, timeoutMs: Double) async {
|
|
guard let waiter = self.pending.removeValue(forKey: id) else { return }
|
|
let err = NSError(
|
|
domain: "Gateway",
|
|
code: 5,
|
|
userInfo: [NSLocalizedDescriptionKey: "gateway request timed out after \(Int(timeoutMs))ms"])
|
|
waiter.resume(throwing: err)
|
|
}
|
|
}
|
|
|
|
actor GatewayChannel {
|
|
private var inner: GatewayChannelActor?
|
|
|
|
func configure(url: URL, token: String?) {
|
|
self.inner = GatewayChannelActor(url: url, token: token)
|
|
}
|
|
|
|
func request(
|
|
method: String,
|
|
params: [String: AnyCodable]?,
|
|
timeoutMs: Double? = nil) async throws -> Data
|
|
{
|
|
guard let inner else {
|
|
throw NSError(domain: "Gateway", code: 0, userInfo: [NSLocalizedDescriptionKey: "not configured"])
|
|
}
|
|
return try await inner.request(method: method, params: params, timeoutMs: timeoutMs)
|
|
}
|
|
}
|