Mobile and Native Clients

import Foundation

/// A JSON value that can sit in an envelope payload.
enum JSON: Codable, Sendable, Equatable {
    case null, bool(Bool), number(Double), string(String)
    case array([JSON]), object([String: JSON])

    init(from decoder: Decoder) throws {
        let c = try decoder.singleValueContainer()
        if c.decodeNil() { self = .null }
        else if let v = try? c.decode(Bool.self) { self = .bool(v) }
        else if let v = try? c.decode(Double.self) { self = .number(v) }
        else if let v = try? c.decode(String.self) { self = .string(v) }
        else if let v = try? c.decode([JSON].self) { self = .array(v) }
        else { self = .object(try c.decode([String: JSON].self)) }
    }

    func encode(to encoder: Encoder) throws {
        var c = encoder.singleValueContainer()
        switch self {
        case .null: try c.encodeNil()
        case .bool(let v): try c.encode(v)
        case .number(let v): try c.encode(v)
        case .string(let v): try c.encode(v)
        case .array(let v): try c.encode(v)
        case .object(let v): try c.encode(v)
        }
    }

    subscript(key: String) -> JSON? {
        if case .object(let o) = self { return o[key] }
        return nil
    }
    var string: String? { if case .string(let s) = self { return s }; return nil }
    var int: Int? { if case .number(let n) = self { return Int(n) }; return nil }
    var bool: Bool? { if case .bool(let b) = self { return b }; return nil }
}

/// The Chatterbox v1 envelope.
struct Envelope: Codable, Sendable {
    var v = 1
    var type: String
    var id: String
    var room: String?
    var ts: Int = 0
    var payload: [String: JSON] = [:]

    // The protocol wants "room": null, not a missing key, so encode it explicitly.
    func encode(to encoder: Encoder) throws {
        var c = encoder.container(keyedBy: CodingKeys.self)
        try c.encode(v, forKey: .v)
        try c.encode(type, forKey: .type)
        try c.encode(id, forKey: .id)
        if let room { try c.encode(room, forKey: .room) } else { try c.encodeNil(forKey: .room) }
        try c.encode(ts, forKey: .ts)
        try c.encode(payload, forKey: .payload)
    }
}

enum ChatterboxError: Error { case ticket, closed(Int), server(code: String, retryable: Bool) }

/// One Chatterbox session over URLSessionWebSocketTask: ticket login, hello, resume, outbox and acks.
actor ChatterboxClient {
    private let baseURL: URL
    private let user: String
    private let session = URLSession(configuration: .default)
    private var task: URLSessionWebSocketTask?
    private var receiver: Task<Void, Never>?

    private(set) var sessionId: String?
    private(set) var lastSeq: [String: Int] = [:]
    private var outbox: [String: Envelope] = [:]      // unacknowledged envelopes, by id
    private var order: [String] = []                  // original send order
    private var waiting: [String: CheckedContinuation<JSON, Error>] = [:]
    private var helloWaiter: CheckedContinuation<JSON, Error>?

    var onBroadcast: (@Sendable (Envelope) -> Void)?

    init(baseURL: URL, user: String) { self.baseURL = baseURL; self.user = user }

    func setBroadcastHandler(_ h: @escaping @Sendable (Envelope) -> Void) { onBroadcast = h }

    /// Opens a socket, says hello (with resume data when there is a previous session) and returns the hello reply payload.
    func connect() async throws -> JSON {
        var ticketRequest = URLRequest(url: baseURL.appending(path: "ticket").appending(queryItems: [.init(name: "user", value: user)]))
        ticketRequest.httpMethod = "POST"
        let (data, response) = try await session.data(for: ticketRequest)
        guard (response as? HTTPURLResponse)?.statusCode == 200,
              let ticket = try JSONDecoder().decode([String: String].self, from: data)["ticket"]
        else { throw ChatterboxError.ticket }

        var wsURL = URLComponents(url: baseURL, resolvingAgainstBaseURL: false)!
        wsURL.scheme = wsURL.scheme == "https" ? "wss" : "ws"
        let task = session.webSocketTask(with: wsURL.url!, protocols: ["chatterbox.v1"])
        task.maximumMessageSize = 1_048_576          // history pages may be large (server frames up to 262144 bytes)
        self.task = task
        task.resume()
        receiver = Task { await self.receiveLoop(task) }

        var hello = Envelope(type: "hello", id: Self.newId())
        hello.payload["token"] = .string(ticket)
        hello.payload["client"] = .string("ios/1.0")
        if let sid = sessionId {
            hello.payload["resume"] = .object([
                "sessionId": .string(sid),
                "rooms": .object(lastSeq.mapValues { .number(Double($0)) }),
            ])
        }
        let reply: JSON = try await withCheckedThrowingContinuation { cont in
            helloWaiter = cont
            Task { try? await self.write(hello) }
        }
        sessionId = reply["userId"] != nil ? reply["sessionId"]?.string : sessionId
        // Anything still unacknowledged goes out again with the SAME id and in the original order.
        for id in order { if let e = outbox[id] { try await write(e) } }
        return reply
    }

    /// Sends an acked request and returns the ack payload; the id stays in the outbox until the ack arrives.
    func request(type: String, room: String?, payload: [String: JSON], id: String = newId()) async throws -> JSON {
        let env = Envelope(type: type, id: id, room: room, payload: payload)
        outbox[id] = env; order.append(id)
        return try await withCheckedThrowingContinuation { cont in
            waiting[id] = cont
            Task { try? await self.write(env) }
        }
    }

    /// Simulates a lost connection (what Wi-Fi to cellular handover looks like to the app).
    func dropConnection() {
        task?.cancel(with: .goingAway, reason: nil)
    }

    func close() { task?.cancel(with: .normalClosure, reason: nil); receiver?.cancel() }

    private func write(_ env: Envelope) async throws {
        let text = String(decoding: try JSONEncoder().encode(env), as: UTF8.self)
        try await task?.send(.string(text))
    }

    private func receiveLoop(_ task: URLSessionWebSocketTask) async {
        do {
            while true {
                let message = try await task.receive()
                guard case .string(let text) = message else { continue }
                let env = try JSONDecoder().decode(Envelope.self, from: Data(text.utf8))
                handle(env)
            }
        } catch {
            let code = Int(task.closeCode.rawValue)
            let failure = ChatterboxError.closed(code)
            helloWaiter?.resume(throwing: failure); helloWaiter = nil
            for (_, c) in waiting { c.resume(throwing: failure) }
            waiting.removeAll()
        }
    }

    private func handle(_ env: Envelope) {
        let payload = JSON.object(env.payload)
        switch env.type {
        case "hello":
            helloWaiter?.resume(returning: payload); helloWaiter = nil
        case "ack", "error":
            guard let ref = payload["ref"]?.string else { return }
            outbox[ref] = nil; order.removeAll { $0 == ref }
            if env.type == "error" {
                let failure = ChatterboxError.server(code: payload["code"]?.string ?? "internal", retryable: payload["retryable"]?.bool ?? false)
                if let w = helloWaiter { helloWaiter = nil; w.resume(throwing: failure) }
                waiting.removeValue(forKey: ref)?.resume(throwing: failure)
            } else {
                waiting.removeValue(forKey: ref)?.resume(returning: payload)
            }
        default:
            if let room = env.room, let seq = payload["seq"]?.int, seq > (lastSeq[room] ?? 0) {
                lastSeq[room] = seq
            }
            onBroadcast?(env)
        }
    }

    static func newId() -> String { UUID().uuidString.lowercased() }
}

import Foundation
setvbuf(stdout, nil, _IOLBF, 0)

let client = ChatterboxClient(baseURL: URL(string: "http://127.0.0.1:18900/")!, user: "alice")
await client.setBroadcastHandler { env in
    print("broadcast \(env.type) room=\(env.room ?? "-") seq=\(env.payload["seq"]?.int ?? -1)")
}

let hello = try await client.connect()
print("hello userId=\(hello["userId"]?.string ?? "?") resumed=\(hello["resumed"]?.bool ?? false)")

let joined = try await client.request(type: "join", room: "general", payload: ["since": .null])
print("joined lastSeq=\(joined["result"]?["lastSeq"]?.int ?? -1)")

let fixedId = ChatterboxClient.newId()
let first = try await client.request(type: "message", room: "general", payload: ["text": .string("Hello from Swift"), "replyTo": .null, "attachments": .array([])], id: fixedId)
print("message ack seq=\(first["seq"]?.int ?? -1) dup=\(first["dup"]?.bool ?? false)")

try await Task.sleep(for: .milliseconds(200))   // let the broadcast copy of our message arrive
await client.dropConnection()
try await Task.sleep(for: .milliseconds(300))

let again = try await client.connect()
print("reconnect resumed=\(again["resumed"]?.bool ?? false)")

let retry = try await client.request(type: "message", room: "general", payload: ["text": .string("Hello from Swift"), "replyTo": .null, "attachments": .array([])], id: fixedId)
print("retry ack seq=\(retry["seq"]?.int ?? -1) dup=\(retry["dup"]?.bool ?? false)")
await client.close()

hello userId=alice resumed=false
joined lastSeq=1
message ack seq=2 dup=false
broadcast message room=general seq=2
reconnect resumed=true
retry ack seq=2 dup=true

import java.util.UUID
import java.util.concurrent.CompletableFuture
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import okhttp3.RequestBody.Companion.toRequestBody
import org.json.JSONObject

/** One Chatterbox session over OkHttp: ticket login, hello with resume, acks and an outbox keyed by envelope id. */
class ChatterboxClient(private val baseUrl: String, private val user: String) {
    private val http = OkHttpClient.Builder()
        .pingInterval(20, TimeUnit.SECONDS)          // WebSocket ping frames: detects a dead network path
        .readTimeout(0, TimeUnit.MILLISECONDS)       // a socket is allowed to be silent between pings
        .build()

    private var socket: WebSocket? = null
    private val acks = ConcurrentHashMap<String, CompletableFuture<JSONObject>>()
    private val outbox = LinkedHashMap<String, JSONObject>()      // unacknowledged envelopes in send order
    private val lastSeq = HashMap<String, Int>()
    private var sessionId: String? = null
    var onBroadcast: (JSONObject) -> Unit = {}

    private fun envelope(type: String, room: String?, payload: JSONObject, id: String = UUID.randomUUID().toString()) =
        JSONObject().put("v", 1).put("type", type).put("id", id)
            .put("room", room ?: JSONObject.NULL).put("ts", 0).put("payload", payload)

    /** Opens a socket, says hello (resuming the previous session when there is one) and returns the hello reply payload. */
    fun connect(): JSONObject {
        val ticketRequest = Request.Builder().url("${baseUrl}ticket?user=$user")
            .post("".toRequestBody()).build()
        val ticket = http.newCall(ticketRequest).execute().use { JSONObject(it.body.string()).getString("ticket") }

        val helloReply = CompletableFuture<JSONObject>()
        val request = Request.Builder().url(baseUrl)
            .header("Sec-WebSocket-Protocol", "chatterbox.v1")
            .build()
        socket = http.newWebSocket(request, object : WebSocketListener() {
            override fun onMessage(webSocket: WebSocket, text: String) = handle(JSONObject(text), helloReply)
            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                helloReply.completeExceptionally(t)
                acks.values.forEach { it.completeExceptionally(t) }
                acks.clear()
            }
            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                val closed = IllegalStateException("closed $code $reason")
                helloReply.completeExceptionally(closed)
                acks.values.forEach { it.completeExceptionally(closed) }
                acks.clear()
            }
        })

        val hello = JSONObject().put("token", ticket).put("client", "android/1.0")
        sessionId?.let {
            hello.put("resume", JSONObject().put("sessionId", it).put("rooms", JSONObject(lastSeq as Map<*, *>)))
        }
        socket!!.send(envelope("hello", null, hello).toString())
        val reply = helloReply.get(10, TimeUnit.SECONDS)
        sessionId = reply.getString("sessionId")
        // Unacknowledged envelopes go out again with the SAME id, in the original order.
        synchronized(outbox) { outbox.values.forEach { socket!!.send(it.toString()) } }
        return reply
    }

    /** Sends an acked request and waits for the ack; the id stays in the outbox until the ack arrives. */
    fun request(type: String, room: String?, payload: JSONObject, id: String = UUID.randomUUID().toString()): JSONObject {
        val env = envelope(type, room, payload, id)
        val future = CompletableFuture<JSONObject>()
        acks[id] = future
        synchronized(outbox) { outbox[id] = env }
        socket!!.send(env.toString())
        return future.get(10, TimeUnit.SECONDS)
    }

    /** What a Wi-Fi to cellular handover looks like to the app: the socket dies without a close handshake. */
    fun dropConnection() { socket?.cancel() }

    fun close() { socket?.close(1000, "bye"); http.dispatcher.executorService.shutdown(); http.connectionPool.evictAll() }

    private fun handle(env: JSONObject, helloReply: CompletableFuture<JSONObject>) {
        val payload = env.getJSONObject("payload")
        when (env.getString("type")) {
            "hello" -> helloReply.complete(payload)
            "ack", "error" -> {
                val ref = payload.optString("ref")
                synchronized(outbox) { outbox.remove(ref) }
                val future = acks.remove(ref)
                if (env.getString("type") == "ack") {
                    future?.complete(payload)
                } else {
                    val failure = IllegalStateException("${payload.getString("code")} retryable=${payload.getBoolean("retryable")}")
                    helloReply.completeExceptionally(failure)
                    future?.completeExceptionally(failure)
                }
            }
            else -> {
                val room = env.optString("room")
                val seq = payload.optInt("seq", 0)
                if (room.isNotEmpty() && seq > (lastSeq[room] ?: 0)) lastSeq[room] = seq
                onBroadcast(env)
            }
        }
    }
}

import java.util.UUID
import org.json.JSONObject

fun main() {
    val client = ChatterboxClient("http://127.0.0.1:18900/", "alice")
    client.onBroadcast = { env -> println("broadcast ${env.getString("type")} room=${env.optString("room")} seq=${env.getJSONObject("payload").optInt("seq", -1)}") }

    val hello = client.connect()
    println("hello userId=${hello.getString("userId")} resumed=${hello.getBoolean("resumed")}")

    val joined = client.request("join", "general", JSONObject().put("since", JSONObject.NULL))
    println("joined lastSeq=${joined.getJSONObject("result").getInt("lastSeq")}")

    val id = UUID.randomUUID().toString()
    fun send() = client.request("message", "general", JSONObject().put("text", "Hello from Kotlin").put("replyTo", JSONObject.NULL).put("attachments", org.json.JSONArray()), id)
    val first = send()
    println("message ack seq=${first.getInt("seq")} dup=${first.getBoolean("dup")}")

    Thread.sleep(200)
    client.dropConnection()
    Thread.sleep(300)

    val again = client.connect()
    println("reconnect resumed=${again.getBoolean("resumed")}")
    val retry = send()
    println("retry ack seq=${retry.getInt("seq")} dup=${retry.getBoolean("dup")}")
    client.close()
}

hello userId=alice resumed=false
joined lastSeq=3
message ack seq=4 dup=false
broadcast message room=general seq=4
reconnect resumed=true
retry ack seq=4 dup=true