//! Unix-socket control plane: JSON-lines requests in, replies + pushed events out.
//!
//! Each connection gets a reader thread (parses requests, forwards them to the
//! SFU loop, relays the reply) and a writer thread (serializes replies AND
//! broadcast notices onto the stream, one JSON object per line).

use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::mpsc::{self, Sender};
use std::sync::{Arc, Mutex};
use std::thread;

use log::warn;

use crate::proto::{Notice, Reply, Request};
use crate::sfu::CtrlMsg;

/// Every connected control client's outbound line channel (for notices).
pub type Subscribers = Arc<Mutex<Vec<Sender<String>>>>;

/// Fan notices out to every connected control client.
pub fn broadcast_notices(rx: mpsc::Receiver<Notice>, subscribers: Subscribers) {
    for notice in rx {
        let json = serde_json::to_string(&notice).expect("notice to serialize");
        subscribers
            .lock()
            .unwrap()
            .retain(|tx| tx.send(json.clone()).is_ok());
    }
}

/// Accept control connections forever (runs on its own thread).
pub fn serve(listener: UnixListener, commands: Sender<CtrlMsg>, subscribers: Subscribers) {
    for stream in listener.incoming() {
        match stream {
            Ok(stream) => {
                let commands = commands.clone();
                let subscribers = subscribers.clone();
                thread::spawn(move || handle_connection(stream, commands, subscribers));
            }
            Err(e) => warn!("control accept failed: {e}"),
        }
    }
}

fn handle_connection(stream: UnixStream, commands: Sender<CtrlMsg>, subscribers: Subscribers) {
    let write_half = match stream.try_clone() {
        Ok(s) => s,
        Err(e) => {
            warn!("control connection clone failed: {e}");
            return;
        }
    };

    // One writer thread serializes replies + notices; dies when senders drop
    // or the peer closes (write error).
    let (out_tx, out_rx) = mpsc::channel::<String>();
    subscribers.lock().unwrap().push(out_tx.clone());

    let writer = thread::spawn(move || {
        let mut stream = write_half;
        for line in out_rx {
            if writeln!(stream, "{line}").and_then(|()| stream.flush()).is_err() {
                break;
            }
        }
    });

    let reader = BufReader::new(stream);
    for line in reader.lines() {
        let Ok(line) = line else { break };
        if line.trim().is_empty() {
            continue;
        }

        let request: Request = match serde_json::from_str(&line) {
            Ok(request) => request,
            Err(e) => {
                let reply = Reply::error(0, format!("bad request: {e}"));
                let _ = out_tx.send(serde_json::to_string(&reply).unwrap());
                continue;
            }
        };

        let (reply_tx, reply_rx) = mpsc::channel();
        if commands
            .send(CtrlMsg {
                request,
                reply: reply_tx,
            })
            .is_err()
        {
            break; // SFU loop gone (shutdown)
        }

        match reply_rx.recv() {
            Ok(reply) => {
                let _ = out_tx.send(serde_json::to_string(&reply).unwrap());
            }
            Err(_) => break,
        }
    }

    drop(out_tx);
    let _ = writer.join();
}