//! Adaptador de transporte arje-bus → bus de agente de takana (último tramo de B.2).
//!
//! arje (el init/PID 1) supervisa cada Ente y, al morir, difunde `BusEvent` por su bus
//! (`arje-bus`: `BusRequest::Subscribe` + `BusPayload::Event`). Este módulo se **suscribe** a
//! ese stream y traduce cada evento al vocabulario de takana (`crashes::Lifecycle` →
//! `Event::Crashed`), publicándolo en el [`EventBus`] → `/run/agent.sock` → la capa de IA.
//!
//! ## Por qué un mirror y no una dependencia de `arje-bus`
//!
//! `arje-bus` arrastra el grafo de crates de arje (arje-card → card-core → …). Acoplarlo aquí
//! rompería el build hermético/standalone de takana. En su lugar **releemos el frame postcard**:
//! el bus usa frames `u32-BE-len + postcard(BusMessage)`, y reproducimos sólo el subconjunto que
//! un suscriptor necesita. El layout está verificado byte-a-byte contra `arje-bus` real (mismas
//! versiones de `ulid` 1.2 y `postcard` 1.1; `ulid` serializa como string). Si arje reordena las
//! variantes del enum, el contrato se rompe — por eso los discriminantes están documentados y
//! cubiertos por un test de round-trip local.
//!
//! Varios campos del mirror (`from`, `seq`, `id`) no se leen: existen para **consumir** los
//! bytes del wire al deserializar. Por eso el `allow(dead_code)` a nivel de módulo.
#![allow(dead_code)]

use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use std::path::PathBuf;
use std::thread;

use serde::Deserialize;
use ulid::Ulid;

use crate::crashes::{self, LifeStatus, Lifecycle};
use crate::events::EventBus;

/// Igual que `arje_bus::ENV_BUS_SOCK`: el env donde arje-zero publica la ruta del socket.
pub const ENV_BUS_SOCK: &str = "ENTE_BUS_SOCK";

/// Tope de frame, igual que `arje_bus::MAX_FRAME` (1 MiB) — protección contra OOM.
const MAX_FRAME: usize = 1 << 20;

// ── Mirror del wire de arje-bus (subconjunto de suscriptor) ──────────────────────────────
//
// Discriminantes (postcard = varint del índice de variante), verificados contra arje-bus:
//   BusPayload: Request=0, Response=1, Event=2
//   BusRequest::Subscribe = 14   (arje insertó StopCardFromDisk el 2026-06-26 y corrió
//                                 el índice 13→14; re-sincronizado 2026-07-16 — ver el
//                                 golden test de arje-bus `contrato_wire`)
//   BusEvent: EnteCrashed=0, EnteRestarting=1, EnteExited=2, EnteParked=3, EnteRefloored=4
//   LifecycleStatus: Exited=0, Killed=1

#[derive(Deserialize, Debug)]
struct BusMessage {
    from: Option<Ulid>,
    seq: u64,
    payload: BusPayload,
}

#[derive(Deserialize, Debug)]
enum BusPayload {
    /// Índice 0. Nunca lo recibe un suscriptor (los Invoke van a proveedores, no a subs);
    /// existe sólo para alinear el discriminante de `Response`/`Event`.
    Request,
    /// Índice 1. El único que recibimos por aquí es el `Ok` del ack de `Subscribe`.
    Response(MiniResponse),
    /// Índice 2. El stream de eventos de ciclo de vida.
    Event(BusEvent),
}

#[derive(Deserialize, Debug)]
enum MiniResponse {
    Ok,
    Error(String),
}

#[derive(Deserialize, Debug)]
enum BusEvent {
    EnteCrashed { id: Ulid, label: String, status: LifecycleStatus },
    EnteRestarting { id: Ulid, label: String, delay_ms: u64 },
    EnteExited { id: Ulid, label: String },
    /// Ente `Restart` aparcado: su "piso" (una capability) perdió proveedor.
    /// No es un crash — no hay señal para la capa de IA; se consume del wire
    /// para no tirar el frame como indescifrable.
    EnteParked { id: Ulid, label: String },
    /// El piso volvió y arje re-erigió al Ente aparcado. Ídem: sólo consumo.
    EnteRefloored { id: Ulid, label: String },
}

#[derive(Deserialize, Debug)]
enum LifecycleStatus {
    Exited(i32),
    Killed(i32),
}

impl BusEvent {
    /// Traduce el evento de arje a la señal normalizada de takana. El `label` del Ente es el
    /// `service`; una muerte por señal usa la convención de shell `128 + signum` para que el
    /// código sea siempre ≠ 0. `None` = evento consumido sin señal (Parked/Refloored: son
    /// vaivén del piso gráfico, no ciclo de vida que le importe a la capa de IA).
    fn to_lifecycle(self) -> Option<Lifecycle> {
        match self {
            BusEvent::EnteCrashed { label, status, .. } => {
                let code = match status {
                    LifecycleStatus::Exited(c) => c,
                    LifecycleStatus::Killed(sig) => 128 + sig,
                };
                Some(Lifecycle { service: label, status: LifeStatus::Crashed { code } })
            }
            BusEvent::EnteRestarting { label, delay_ms, .. } => {
                Some(Lifecycle { service: label, status: LifeStatus::Restarting { delay_ms } })
            }
            BusEvent::EnteExited { label, .. } => {
                Some(Lifecycle { service: label, status: LifeStatus::Exited })
            }
            BusEvent::EnteParked { .. } | BusEvent::EnteRefloored { .. } => None,
        }
    }
}

/// Cuerpo del frame `Subscribe` (sin el prefijo de longitud), hand-encodeado: `from=None` +
/// `seq` (varint) + `BusPayload::Request`(0) + `BusRequest::Subscribe`(14). Verificado
/// byte-a-byte contra `postcard::to_stdvec` de arje-bus (golden test `contrato_wire`).
fn subscribe_body(seq: u64) -> Vec<u8> {
    let mut v = vec![0x00u8]; // Option::None
    let mut s = seq; // u64 como varint LEB128
    loop {
        let b = (s & 0x7f) as u8;
        s >>= 7;
        if s == 0 {
            v.push(b);
            break;
        } else {
            v.push(b | 0x80);
        }
    }
    v.push(0x00); // BusPayload::Request
    v.push(0x0E); // BusRequest::Subscribe (índice 14 desde 2026-06-26)
    v
}

fn write_frame(w: &mut impl Write, body: &[u8]) -> std::io::Result<()> {
    w.write_all(&(body.len() as u32).to_be_bytes())?;
    w.write_all(body)?;
    w.flush()
}

fn read_frame(r: &mut impl Read) -> std::io::Result<Vec<u8>> {
    let mut len = [0u8; 4];
    r.read_exact(&mut len)?;
    let n = u32::from_be_bytes(len) as usize;
    if n > MAX_FRAME {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("frame oversize: {n} > {MAX_FRAME}"),
        ));
    }
    let mut buf = vec![0u8; n];
    r.read_exact(&mut buf)?;
    Ok(buf)
}

/// Conecta al bus de arje, se suscribe y bombea eventos de ciclo de vida al `EventBus` hasta
/// que la conexión se cierra. Bloqueante — pensado para correr en su propio thread.
/// `pub(crate)` para que el e2e (`tests/arje_link_e2e.rs`) lo maneje contra un bus falso.
pub(crate) fn run(sock: &PathBuf, bus: &EventBus) -> std::io::Result<()> {
    let mut stream = UnixStream::connect(sock)?;
    write_frame(&mut stream, &subscribe_body(1))?;
    tracing::info!(sock = %sock.display(), "arje-link: suscrito al bus de init");
    loop {
        let body = read_frame(&mut stream)?;
        let msg: BusMessage = match postcard::from_bytes(&body) {
            Ok(m) => m,
            Err(e) => {
                // Un frame que no entendemos no debe matar el puente: lo saltamos.
                tracing::warn!(error = %e, "arje-link: frame indescifrable, ignoro");
                continue;
            }
        };
        match msg.payload {
            BusPayload::Event(ev) => {
                if let Some(life) = ev.to_lifecycle() {
                    if let Some(event) = crashes::to_event(&life) {
                        bus.publish(&event);
                    }
                }
            }
            // El ack del Subscribe y cualquier otra respuesta: nada que hacer.
            BusPayload::Response(_) | BusPayload::Request => {}
        }
    }
}

/// Lanza el puente en un thread si `ENTE_BUS_SOCK` está definido (arje corriendo). Si no, es un
/// no-op: hammerd sigue sin la fuente de crashes (los tests del bus no la necesitan). El thread
/// es resiliente: si el bus muere, loguea y termina sin tumbar al daemon.
pub fn spawn_if_configured(bus: EventBus) {
    let Ok(path) = std::env::var(ENV_BUS_SOCK) else {
        tracing::debug!("arje-link: {ENV_BUS_SOCK} no definido; sin fuente de crashes");
        return;
    };
    let sock = PathBuf::from(path);
    let _ = thread::Builder::new().name("arje-link".into()).spawn(move || {
        if let Err(e) = run(&sock, &bus) {
            tracing::warn!(error = %e, "arje-link: puente terminó (¿bus de init caído?)");
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn subscribe_body_es_el_frame_esperado() {
        // Verificado contra arje-bus real: [from=None, seq=1, Request, Subscribe=14].
        assert_eq!(subscribe_body(1), vec![0x00, 0x01, 0x00, 0x0E]);
        // seq grande → varint multibyte (300 = 0xAC 0x02).
        assert_eq!(subscribe_body(300), vec![0x00, 0xAC, 0x02, 0x00, 0x0E]);
    }

    /// Round-trip local con el MISMO `postcard`/`ulid` que arje-bus: serializamos un
    /// `BusMessage::Event` con tipos espejo `Serialize` y lo deserializamos con los `Deserialize`
    /// reales del módulo. Si los discriminantes/orden de campos se desalinean, esto rompe.
    #[test]
    fn evento_crashed_roundtrips_por_el_wire() {
        use serde::Serialize;
        #[derive(Serialize)]
        enum TxPayload {
            #[allow(dead_code)]
            Request,
            #[allow(dead_code)]
            Response,
            Event(TxEvent),
        }
        #[derive(Serialize)]
        enum TxEvent {
            EnteCrashed { id: Ulid, label: String, status: TxStatus },
        }
        #[derive(Serialize)]
        enum TxStatus {
            #[allow(dead_code)]
            Exited(i32),
            Killed(i32),
        }
        #[derive(Serialize)]
        struct TxMsg {
            from: Option<Ulid>,
            seq: u64,
            payload: TxPayload,
        }

        let id = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").unwrap();
        let bytes = postcard::to_stdvec(&TxMsg {
            from: Some(id),
            seq: 7,
            payload: TxPayload::Event(TxEvent::EnteCrashed {
                id,
                label: "demonio".into(),
                status: TxStatus::Killed(11),
            }),
        })
        .unwrap();

        let msg: BusMessage = postcard::from_bytes(&bytes).unwrap();
        match msg.payload {
            BusPayload::Event(ev) => {
                let life = ev.to_lifecycle().expect("un crash sí produce señal");
                assert_eq!(life.service, "demonio");
                // Killed(11) → 128 + 11 = 139.
                assert_eq!(life.status, LifeStatus::Crashed { code: 139 });
            }
            other => panic!("esperaba Event, fue {other:?}"),
        }
    }
}