//! Producer threads. Each owns one input, reconnects on its own, and never blocks
//! on the writer: CAN frames that do not fit the queue are counted, not waited for.

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixListener;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{self, SyncSender, TrySendError};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

use crate::can::{CanSocket, Frame, Recv};
use crate::clock;

pub enum Event {
    Can(Frame),
    /// Frames the kernel dropped before we read them (SO_RXQ_OVFL).
    KernelDrops(u32),
    CanLink(bool, String),
    Gnss(f64, String),
    GnssLink(bool, String),
    /// An envelope on yukart/vehicle/<signal>: receive time, signal, payload, retained.
    Vehicle(f64, String, String, bool),
    BusLink(bool, String),
    Ctl(String, mpsc::Sender<String>),
}

pub fn spawn_can(iface: String, tx: SyncSender<Event>, queue_drops: Arc<AtomicU64>, stop: Arc<AtomicBool>) {
    thread::Builder::new().name("can".into()).spawn(move || {
        let mut reported: Option<String> = None;
        while !stop.load(Ordering::Relaxed) {
            let mut sock = match CanSocket::open(&iface, Duration::from_millis(200)) {
                Ok(s) => s,
                Err(e) => {
                    // An unplugged adapter is a normal state; say so once, then poll quietly.
                    let msg = format!("{iface}: {e}");
                    if reported.as_deref() != Some(&msg) {
                        let _ = tx.send(Event::CanLink(false, msg.clone()));
                        reported = Some(msg);
                    }
                    thread::sleep(Duration::from_secs(1));
                    continue;
                }
            };
            let _ = tx.send(Event::CanLink(true, format!("{iface}: socket open")));
            reported = None;
            while !stop.load(Ordering::Relaxed) {
                match sock.recv() {
                    Ok(Recv::Timeout) => {}
                    Ok(Recv::Frame(f, drops)) => {
                        if drops > 0 {
                            let _ = tx.try_send(Event::KernelDrops(drops));
                        }
                        match tx.try_send(Event::Can(f)) {
                            Ok(()) => {}
                            Err(TrySendError::Full(_)) => {
                                queue_drops.fetch_add(1, Ordering::Relaxed);
                            }
                            Err(TrySendError::Disconnected(_)) => return,
                        }
                    }
                    Err(e) => {
                        // ENETDOWN / ENODEV: the link went down or the adapter left.
                        let _ = tx.send(Event::CanLink(false, format!("{iface}: {e}")));
                        reported = Some(String::new());
                        thread::sleep(Duration::from_millis(500));
                        break;
                    }
                }
            }
        }
    }).expect("spawn can thread");
}

pub fn spawn_gnss(addr: String, tx: SyncSender<Event>, stop: Arc<AtomicBool>) {
    thread::Builder::new().name("gnss".into()).spawn(move || {
        // Report a failing gpsd once, not every retry.
        let mut reported_down = false;
        while !stop.load(Ordering::Relaxed) {
            let conn = addr
                .to_socket_addrs()
                .ok()
                .and_then(|mut a| a.next())
                .ok_or_else(|| std::io::Error::other("unresolvable"))
                .and_then(|a| TcpStream::connect_timeout(&a, Duration::from_secs(2)));
            let mut stream = match conn {
                Ok(s) => s,
                Err(e) => {
                    if !reported_down {
                        let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: {e}")));
                        reported_down = true;
                    }
                    thread::sleep(Duration::from_secs(3));
                    continue;
                }
            };
            let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
            if stream.write_all(b"?WATCH={\"enable\":true,\"json\":true};\n").is_err() {
                continue;
            }
            let _ = tx.send(Event::GnssLink(true, format!("gpsd {addr}: watching")));
            let mut reader = BufReader::new(stream);
            let mut line = String::new();
            loop {
                if stop.load(Ordering::Relaxed) {
                    return;
                }
                match reader.read_line(&mut line) {
                    Ok(0) => break,
                    Ok(_) => {
                        if line.ends_with('\n') {
                            let rx = clock::realtime();
                            let _ = tx.send(Event::Gnss(rx, line.trim_end().to_string()));
                            line.clear();
                        }
                    }
                    Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {}
                    Err(_) => break,
                }
            }
            let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: connection lost")));
            reported_down = true;
            thread::sleep(Duration::from_secs(1));
        }
    }).expect("spawn gnss thread");
}

/// Envelopes from the broker, for the record only: raw capture never waits on this.
pub fn spawn_bus(addr: String, tx: SyncSender<Event>, stop: Arc<AtomicBool>) {
    let Some((host, port)) = addr.rsplit_once(':').and_then(|(h, p)| Some((h.to_string(), p.parse::<u16>().ok()?))) else {
        let _ = tx.send(Event::BusLink(false, format!("bad broker address {addr}")));
        return;
    };
    thread::Builder::new().name("bus".into()).spawn(move || {
        let mut opts = rumqttc::MqttOptions::new("recorder", host, port);
        opts.set_keep_alive(Duration::from_secs(5));
        let (client, mut connection) = rumqttc::Client::new(opts, 64);
        let mut net = rumqttc::NetworkOptions::new();
        net.set_tcp_nodelay(true);
        connection.eventloop.set_network_options(net);
        let mut up = false;
        for ev in connection.iter() {
            if stop.load(Ordering::Relaxed) {
                return;
            }
            match ev {
                Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(_))) => {
                    up = true;
                    let _ = client.try_subscribe("yukart/vehicle/+", rumqttc::QoS::AtMostOnce);
                    let _ = tx.send(Event::BusLink(true, format!("{addr}: subscribed to yukart/vehicle/+")));
                }
                Ok(rumqttc::Event::Incoming(rumqttc::Packet::Publish(p))) => {
                    let signal = p.topic.rsplit('/').next().unwrap_or_default().to_string();
                    let payload = String::from_utf8_lossy(&p.payload).into_owned();
                    // Never block the event loop on the writer; CAN frames have priority in the queue.
                    let _ = tx.try_send(Event::Vehicle(clock::realtime(), signal, payload, p.retain));
                }
                Ok(_) => {}
                Err(e) => {
                    if std::mem::take(&mut up) {
                        let _ = tx.send(Event::BusLink(false, format!("{addr}: {e}")));
                    }
                    thread::sleep(Duration::from_secs(1));
                }
            }
        }
    }).expect("spawn bus thread");
}

/// One command per connection, one reply line back.
pub fn spawn_ctl(path: String, tx: SyncSender<Event>) -> std::io::Result<()> {
    let _ = std::fs::remove_file(&path);
    let listener = UnixListener::bind(&path)?;
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660))?;
    thread::Builder::new().name("ctl".into()).spawn(move || {
        for conn in listener.incoming() {
            let Ok(mut conn) = conn else { continue };
            let _ = conn.set_read_timeout(Some(Duration::from_secs(2)));
            let mut buf = Vec::new();
            let _ = (&mut conn).take(4096).read_to_end(&mut buf);
            let cmd = String::from_utf8_lossy(&buf).trim().to_string();
            let (rtx, rrx) = mpsc::channel();
            if tx.send(Event::Ctl(cmd, rtx)).is_err() {
                return;
            }
            let reply = rrx.recv_timeout(Duration::from_secs(5)).unwrap_or_else(|_| "error: no reply".into());
            let _ = conn.write_all(reply.as_bytes());
            let _ = conn.write_all(b"\n");
        }
    })?;
    Ok(())
}