sergio/humanoid

1package com.rizoma.consola.data
2
3import android.util.Log
4import com.rizoma.consola.DebugLog
5import kotlinx.coroutines.Dispatchers
6import kotlinx.coroutines.withContext
7import okhttp3.MediaType.Companion.toMediaType
8import okhttp3.OkHttpClient
9import okhttp3.Request
10import okhttp3.RequestBody.Companion.toRequestBody
11import okhttp3.Response
12import okhttp3.WebSocket
13import okhttp3.WebSocketListener
14import okio.ByteString
15import okio.ByteString.Companion.toByteString
16import org.json.JSONObject
17import java.net.URLEncoder
18import java.util.concurrent.TimeUnit
19
20/**
21 * Cliente del shuma-gateway: `POST /rpc` (JSON 1:1) y `GET /ws/pty`
22 * (WebSocket full-duplex hacia una sesión PTY). Sin estado de UI; lo
23 * orquesta el ViewModel.
24 */
25class GatewayClient(@Volatile var config: GatewayConfig) {
26
27    private val json = "application/json; charset=utf-8".toMediaType()
28
29    // El WS de un terminal puede estar minutos en silencio; sin timeout de
30    // lectura pero con ping para mantener vivo el NAT/proxy.
31    private val client = OkHttpClient.Builder()
32        .connectTimeout(10, TimeUnit.SECONDS)
33        .readTimeout(0, TimeUnit.MILLISECONDS)
34        .pingInterval(20, TimeUnit.SECONDS)
35        .build()
36
37    /** Round-trip de una Request JSON (cuerpos de [Rpc]) → [RpcResponse]. */
38    suspend fun rpc(body: String): RpcResponse = withContext(Dispatchers.IO) {
39        val rpcUrl = config.baseUrl.trimEnd('/') + "/rpc"
40        DebugLog.log("RPC", "POST $rpcUrl  (token ${if (config.token.isBlank()) "ausente" else "presente"})  ← ${body.take(80)}")
41        val req = Request.Builder()
42            .url(rpcUrl)
43            .post(body.toRequestBody(json))
44            .apply { if (config.token.isNotBlank()) header("Authorization", "Bearer ${config.token}") }
45            .build()
46        runCatching {
47            client.newCall(req).execute().use { resp: Response ->
48                val text = resp.body?.string().orEmpty()
49                if (text.isEmpty()) {
50                    DebugLog.log("RPC", "HTTP ${resp.code} con cuerpo vacío")
51                    RpcResponse.Failure("respuesta vacía (HTTP ${resp.code})")
52                } else {
53                    DebugLog.log("RPC", "HTTP ${resp.code} ← ${text.take(120)}")
54                    RpcParser.parse(text)
55                }
56            }
57        }.getOrElse { e ->
58            DebugLog.log("RPC", "FALLO ${e.javaClass.simpleName}: ${e.message ?: "(sin mensaje)"}")
59            RpcResponse.Failure(e.message ?: "fallo de red")
60        }
61    }
62
63    /**
64     * Abre el WS de un PTY. Si [session] es null, **crea** una sesión nueva
65     * (spawn) con [program]/[args]/[cwd]/[label]; si no, se **adjunta** a la
66     * existente. Devuelve el handle para mandar teclas/resize y cerrar.
67     */
68    fun openPty(
69        session: String?,
70        program: String,
71        args: List<String>,
72        cwd: String,
73        label: String,
74        rows: Int,
75        cols: Int,
76        listener: PtyListener,
77    ): PtyConnection {
78        var url = config.baseUrl.trimEnd('/') + "/ws/pty"
79        url = when {
80            url.startsWith("https://") -> "wss://" + url.removePrefix("https://")
81            url.startsWith("http://") -> "ws://" + url.removePrefix("http://")
82            else -> url
83        }
84        // El token también va en el query: muchos gateways no pueden leer la
85        // cabecera Authorization en el handshake WS (los WebSocket de navegador
86        // no la permiten). DEBE ir URL-encoded: un token con '+', '/' o '='
87        // (base64/JWT) crudo rompe el query y el server contesta 401/400 al
88        // upgrade aunque el mismo token funcione por header en /rpc.
89        if (config.token.isNotBlank()) {
90            url += "?token=" + URLEncoder.encode(config.token, "UTF-8")
91        }
92        // URL sin query: para logs y para el mensaje de error sin filtrar el token.
93        val safeUrl = url.substringBefore('?')
94        DebugLog.log(
95            "WS",
96            "abriendo $safeUrl  (${if (session != null) "attach $session" else "spawn $program"}, ${rows}x$cols, " +
97                "token ${if (config.token.isBlank()) "ausente" else "presente"})",
98        )
99        val req = Request.Builder()
100            .url(url)
101            .apply { if (config.token.isNotBlank()) header("Authorization", "Bearer ${config.token}") }
102            .build()
103
104        // Primer mensaje (texto): adjuntar o crear.
105        val open = JSONObject().put("rows", rows).put("cols", cols)
106        if (session != null) {
107            open.put("session", session)
108        } else {
109            open.put("program", program)
110            open.put("args", org.json.JSONArray(args))
111            open.put("cwd", cwd)
112            open.put("label", label)
113        }
114
115        // El open DEBE ser el primer frame de texto del WS: el gateway lee el
116        // primer texto como spec de apertura. Pero el layout del terminal puede
117        // disparar resize() antes de que llegue onOpen, y OkHttp encola los
118        // envíos en orden de llamada → el resize se colaría delante del open y
119        // el gateway lo tomaría como apertura ("falta 'session' o 'program'").
120        // [gate] difiere cualquier resize hasta haber mandado el open.
121        val gate = OpenGate()
122        val ws = client.newWebSocket(req, object : WebSocketListener() {
123            override fun onOpen(webSocket: WebSocket, response: Response) {
124                DebugLog.log("WS", "handshake OK (HTTP ${response.code}) → enviando open: ${open.toString().take(120)}")
125                webSocket.send(open.toString())
126                gate.markOpened(webSocket)
127                listener.onOpen()
128            }
129
130            override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
131                DebugLog.log("WS", "← ${bytes.size} bytes")
132                listener.onBytes(bytes.toByteArray())
133            }
134
135            override fun onMessage(webSocket: WebSocket, text: String) {
136                DebugLog.log("WS", "← texto: ${text.take(160)}")
137                val o = runCatching { JSONObject(text) }.getOrNull() ?: return
138                when (o.optString("t")) {
139                    "session" -> listener.onSessionId(o.optString("id"))
140                    "exit" -> listener.onExit(o.optInt("code", -1))
141                    "error" -> listener.onError(o.optString("msg", "error"))
142                }
143            }
144
145            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
146                DebugLog.log("WS", "cerrando: code=$code reason=${reason.ifBlank { "—" }}")
147                webSocket.close(1000, null)
148            }
149
150            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
151                DebugLog.log("WS", "cerrado: code=$code reason=${reason.ifBlank { "—" }}")
152                listener.onClosed(reason.ifBlank { null })
153            }
154
155            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
156                // El cuerpo de la respuesta de error (si el server lo mandó) suele
157                // explicar el rechazo del upgrade; léelo ANTES de armar el detalle.
158                val httpBody = response?.let { runCatching { it.body?.string() }.getOrNull() }
159
160                // Stack completo a Logcat: la única forma fiable de ver la causa
161                // raíz cuando en pantalla solo cabe una línea. Captúralo con:
162                //   adb logcat -s ConsolaWS
163                Log.w(TAG, "WS handshake/stream falló contra $safeUrl (HTTP ${response?.code ?: "—"})", t)
164                if (!httpBody.isNullOrBlank()) Log.w(TAG, "cuerpo de respuesta: ${httpBody.take(500)}")
165
166                // Detalle diagnóstico para la UI: el handshake suele fallar ANTES
167                // del primer mensaje. Sin esto, una caída llega como "fallo del
168                // WebSocket" a secas (t.message null, response null) y no se sabe
169                // si es 401, conexión rechazada, proxy sin upgrade, o TLS. Por eso
170                // siempre incluimos algo: clase, HTTP, cuerpo, o al menos la URL.
171                val detail = buildString {
172                    append(t.javaClass.simpleName)
173                    t.message?.let { append(": "); append(it) }
174                    response?.let { append(" · HTTP "); append(it.code) }
175                    if (!httpBody.isNullOrBlank()) { append(" · "); append(httpBody.trim().take(120)) }
176                    if (t.message == null && response == null) {
177                        append(" · sin respuesta al abrir "); append(safeUrl)
178                    }
179                }
180
181                // Pista interpretada: un EOF/conexión cerrada sin respuesta HTTP
182                // casi siempre es que el server cerró el TCP sin completar el
183                // upgrade. Las causas típicas, para no adivinar a ciegas.
184                val hint = when {
185                    response == null && (t is java.io.EOFException ||
186                        t.javaClass.simpleName == "EOFException") ->
187                        "el server cerró la conexión sin responder al upgrade. Revisa: " +
188                            "(1) ¿es ws:// vs wss:// correcto para el puerto? " +
189                            "(2) ¿la ruta /ws/pty existe en el gateway? " +
190                            "(3) ¿hay un proxy/reverse-proxy que no pasa el Upgrade? " +
191                            "(4) ¿el token es válido?"
192                    response?.code == 401 || response?.code == 403 ->
193                        "rechazo de auth: el token del query/header no es válido para el WS"
194                    t is java.net.ConnectException ->
195                        "no se pudo conectar al host:puerto — ¿gateway caído o URL/puerto mal?"
196                    else -> null
197                }
198
199                DebugLog.log("WS", "FALLO contra $safeUrl → $detail")
200                if (!httpBody.isNullOrBlank()) DebugLog.log("WS", "cuerpo: ${httpBody.take(300)}")
201                hint?.let { DebugLog.log("WS", "pista: $it") }
202
203                listener.onError(detail)
204                listener.onClosed(detail)
205            }
206        })
207        return PtyConnection(ws, gate)
208    }
209
210    private companion object {
211        const val TAG = "ConsolaWS"
212    }
213}
214
215/**
216 * Compuerta que garantiza que el `open` salga antes que cualquier `resize`.
217 * Hasta que [markOpened] confirme que se envió el open, los resize se guardan
218 * (solo el último importa) y se sueltan después. Thread-safe: `markOpened`
219 * corre en el hilo de OkHttp y `gateResize` en el de la UI.
220 */
221class OpenGate {
222    @Volatile private var opened = false
223    private var pending: Pair<Int, Int>? = null
224
225    /** Llamado tras enviar el open: marca abierto y suelta el resize pendiente. */
226    @Synchronized fun markOpened(ws: WebSocket) {
227        opened = true
228        pending?.let { (r, c) -> ws.send(resizeFrame(r, c)) }
229        pending = null
230    }
231
232    /**
233     * Devuelve true si el resize quedó diferido (aún no hay open); false si ya
234     * se puede mandar directo.
235     */
236    @Synchronized fun gateResize(rows: Int, cols: Int): Boolean {
237        if (opened) return false
238        pending = rows to cols
239        return true
240    }
241
242    companion object {
243        fun resizeFrame(rows: Int, cols: Int): String =
244            JSONObject().put("t", "resize").put("rows", rows).put("cols", cols).toString()
245    }
246}
247
248/** Eventos de una conexión PTY (callbacks desde el hilo de OkHttp). */
249interface PtyListener {
250    fun onOpen() {}
251    fun onSessionId(id: String) {}
252    fun onBytes(data: ByteArray) {}
253    fun onExit(code: Int) {}
254    fun onError(message: String) {}
255    fun onClosed(reason: String?) {}
256}
257
258/** Handle de un PTY adjunto: manda teclas/resize y cierra (DETACH). */
259class PtyConnection(private val ws: WebSocket, private val gate: OpenGate) {
260    fun sendInput(bytes: ByteArray) {
261        ws.send(bytes.toByteString())
262    }
263
264    fun sendInput(text: String) {
265        ws.send(text.toByteArray(Charsets.UTF_8).toByteString())
266    }
267
268    fun resize(rows: Int, cols: Int) {
269        // Si el open todavía no salió, se difiere (lo manda la compuerta luego).
270        if (gate.gateResize(rows, cols)) return
271        ws.send(OpenGate.resizeFrame(rows, cols))
272    }
273
274    /** Cierra el WS → DETACH en el daemon (la sesión sigue viva). */
275    fun detach() {
276        ws.close(1000, null)
277    }
278}