Realtime guide

import { io } from "socket.io-client";

export function watchRun(eventId, { mintTicket, loadReport }) {
  const socket = io(SOCKET_ORIGIN, {
    transports: ["websocket"],
    auth: (cb) => mintTicket().then((ticket) => cb({ ticket })),
  });

  const finish = (outcome) => {
    socket.disconnect();
    return outcome === "failed" ? onFailed(eventId) : loadReport(eventId);
  };

  socket.on("connect", () => {
    socket.emit("join_workflow", eventId, (ack) => {
      if (ack.status === "error") return onJoinError(ack.error);   // not_found, wrong_channel
      if (ack.execution.status === "SUCCESSFUL") finish("done");    // finished before we joined
      if (ack.execution.status === "FAILED") finish("failed");
    });
  });

  socket.on("workflow_status_update", (payload) => finish(payload.error ? "failed" : "done"));
  socket.on("connect_error", (err) => console.warn("socket refused:", err.message));
  return socket;
}

/**
 * Watches one report run. The handshake auth can't be changed on the library's
 * own reconnect, so reconnection is done here: every attempt is a new socket
 * with a freshly minted ticket.
 */
class RunWatcher(
    private val eventId: String,
    private val scope: CoroutineScope,
    private val realtimeApi: RealtimeApi,            // POST /api/realtime/v1/ticket
    private val onFinished: suspend () -> Unit,      // then load the report via REST
) {
    private var socket: Socket? = null
    private var done = false

    fun start() = scope.launch { connect() }

    private suspend fun connect() {
        val ticket = realtimeApi.mintTicket().ticket
        val opts = IO.Options().apply {
            transports = arrayOf(WebSocket.NAME)
            reconnection = false
            auth = mapOf("ticket" to ticket)
        }
        socket = IO.socket(SOCKET_ORIGIN, opts).apply {
            on(Socket.EVENT_CONNECT) {
                emit("join_workflow", arrayOf(eventId), Ack { args ->
                    val ack = args.firstOrNull() as? JSONObject ?: return@Ack
                    if (ack.optString("status") != "ok") return@Ack
                    val status = ack.getJSONObject("execution").optString("status")
                    if (status == "SUCCESSFUL" || status == "FAILED") finish()   // finished before we joined
                })
            }
            on("workflow_status_update") { finish() }
            on(Socket.EVENT_CONNECT_ERROR) { retry() }
            on(Socket.EVENT_DISCONNECT) { retry() }
            connect()
        }
    }

    private fun retry() {
        if (done) return
        socket?.off()
        scope.launch { delay(3_000); if (!done) connect() }   // re-joins on the new connection
    }

    private fun finish() {
        done = true
        socket?.off()
        socket?.disconnect()
        scope.launch { onFinished() }                // a 404 from the report endpoint means it failed
    }
}

/// Watches one report run. Built-in reconnection would replay the old, already
/// used ticket, so it's turned off and each attempt connects with a fresh one.
final class RunWatcher {
    private let manager: SocketManager
    private var socket: SocketIOClient { manager.defaultSocket }
    private var done = false

    init(origin: URL) {
        manager = SocketManager(socketURL: origin, config: [.forceWebsockets(true), .reconnects(false)])
    }

    func watch(eventId: String, mintTicket: @escaping () async throws -> String, onFinished: @escaping () -> Void) {
        socket.on(clientEvent: .connect) { [weak self] _, _ in
            self?.socket.emitWithAck("join_workflow", eventId).timingOut(after: 10) { items in
                guard let ack = items.first as? [String: Any], ack["status"] as? String == "ok",
                      let execution = ack["execution"] as? [String: Any],
                      let status = execution["status"] as? String else { return }
                if status == "SUCCESSFUL" || status == "FAILED" { self?.finish(onFinished) }
            }
        }
        socket.on("workflow_status_update") { [weak self] _, _ in self?.finish(onFinished) }
        socket.on(clientEvent: .disconnect) { [weak self] _, _ in self?.retry(mintTicket) }
        socket.on(clientEvent: .error) { [weak self] _, _ in self?.retry(mintTicket) }
        connect(mintTicket)
    }

    private func connect(_ mintTicket: @escaping () async throws -> String) {
        Task {
            guard let ticket = try? await mintTicket() else { return retry(mintTicket) }
            socket.connect(withPayload: ["ticket": ticket])
        }
    }

    private func retry(_ mintTicket: @escaping () async throws -> String) {
        guard !done else { return }
        DispatchQueue.main.asyncAfter(deadline: .now() + 3) { [weak self] in
            guard let self, !self.done else { return }
            self.connect(mintTicket)                 // re-joins on the new connection
        }
    }

    private func finish(_ onFinished: () -> Void) {
        done = true
        socket.disconnect()
        onFinished()                                 // then load GET /api/workflows/reports/v1/:eventId
    }
}