//! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running.

use std::fs::{DirBuilder, Permissions};
use std::io;
use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::mpsc;
use std::time::Duration;

use crate::admin;
use crate::audit::{AuditError, Writer};
use crate::broker::{self, Broker};
use crate::config::Config;
use crate::ledger::Ledger;
use crate::runner::Runtime;
use crate::state::StateStore;
use proto::{ChainFailure, Timestamp};

/// A failure while `brokerd serve` is starting up.
#[derive(Debug)]
pub enum ServeError {
    /// The audit chain would not open. Displayed as the audit error's own text.
    Audit(AuditError),
    /// A directory could not be prepared, or made private.
    Dir(PathBuf, io::Error),
    /// A socket could not be bound, or made private.
    Socket(PathBuf, io::Error),
}

impl std::fmt::Display for ServeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ServeError::Audit(e) => write!(f, "{e}"),
            ServeError::Dir(path, e) => write!(f, "cannot prepare {}: {e}", path.display()),
            ServeError::Socket(path, e) => write!(f, "cannot listen on {}: {e}", path.display()),
        }
    }
}

impl std::error::Error for ServeError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            ServeError::Audit(e) => Some(e),
            ServeError::Dir(_, e) | ServeError::Socket(_, e) => Some(e),
        }
    }
}

/// The result of `start`: a running broker with both sockets open.
pub struct Started {
    pub broker: Arc<Broker>,
    pub recovered: bool,
    pub accepted: Option<ChainFailure>,
    tools: UnixListener,
    admin: UnixListener,
}

/// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker.
pub fn start(
    cfg: Config,
    accept_break: bool,
    runtime: Box<dyn Runtime>,
    log: Arc<dyn Fn(&str) + Send + Sync>,
) -> Result<Started, ServeError> {
    // The audit lock comes first: if the chain will not open, no directory or socket is made.
    let opened = Writer::open(&cfg.audit_dir(), accept_break).map_err(ServeError::Audit)?;

    // Both sockets, in the order the task gives.
    let tools = listen(&cfg.broker_socket())?;
    let admin = listen(&cfg.admin_socket())?;

    // The ledger and broker share the one log, each through its own box.
    let log1: Box<dyn Fn(&str) + Send + Sync> = {
        let log = Arc::clone(&log);
        Box::new(move |line: &str| log(line))
    };
    let ledger = Ledger::new(
        Box::new(opened.writer),
        StateStore::new(&cfg.state_dir()),
        log1,
    );
    let log2: Box<dyn Fn(&str) + Send + Sync> = {
        let log = Arc::clone(&log);
        Box::new(move |line: &str| log(line))
    };
    let broker = Broker::new(cfg, ledger, runtime, log2);

    Ok(Started {
        broker: Arc::new(broker),
        recovered: opened.recovered,
        accepted: opened.accepted.map(|b| *b),
        tools,
        admin,
    })
}

/// Bind one socket: prepare its directory at 0700, make it private, remove any stale
/// socket, bind, then make the socket itself private at 0600.
fn listen(socket: &Path) -> Result<UnixListener, ServeError> {
    // A socket needs a directory of its own: step 2 makes that directory 0700, which must never
    // be `/` or a shared directory reached through a symbolic link.
    let dir = match socket.parent() {
        Some(parent) if parent.parent().is_some() && !parent.as_os_str().is_empty() => {
            parent.to_path_buf()
        }
        _ => {
            return Err(ServeError::Dir(
                PathBuf::from(socket),
                io::Error::other("a socket needs a directory of its own, not / or none"),
            ));
        }
    };

    // 1. The directory, at 0700.
    DirBuilder::new()
        .recursive(true)
        .mode(0o700)
        .create(&dir)
        .map_err(|e| ServeError::Dir(dir.clone(), e))?;
    // `set_permissions` follows a symbolic link, so a link would make the directory it points to
    // private instead: refuse it.
    let kind = std::fs::symlink_metadata(&dir)
        .map_err(|e| ServeError::Dir(dir.clone(), e))?
        .file_type();
    if !kind.is_dir() {
        return Err(ServeError::Dir(
            dir,
            io::Error::other(
                "is a symbolic link or not a directory; brokerd will not change its mode",
            ),
        ));
    }
    // 2. Make it private, always, even when it was already there at 0755.
    std::fs::set_permissions(&dir, Permissions::from_mode(0o700))
        .map_err(|e| ServeError::Dir(dir.clone(), e))?;

    // 3. A stale socket is removed; a missing one is fine.
    match std::fs::remove_file(socket) {
        Ok(()) => {}
        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
        Err(e) => return Err(ServeError::Socket(socket.to_path_buf(), e)),
    }

    // 4. Bind, then 5. make the socket private.
    let listener =
        UnixListener::bind(socket).map_err(|e| ServeError::Socket(PathBuf::from(socket), e))?;
    std::fs::set_permissions(socket, Permissions::from_mode(0o600))
        .map_err(|e| ServeError::Socket(PathBuf::from(socket), e))?;
    Ok(listener)
}

/// Which handler the accept thread runs for each accepted stream.
#[derive(Copy, Clone)]
enum Which {
    Tools,
    Admin,
}

/// Run the broker: start the expiry thread, one accept thread per socket, then wait for the
/// first listener error.
impl Started {
    pub fn run(self) -> io::Result<()> {
        let Started {
            broker,
            tools,
            admin,
            ..
        } = self;

        // The expiry thread: once a second, expire the approvals whose time has run out.
        // `Builder` rather than `spawn`, which panics when the system refuses a thread.
        {
            let broker = Arc::clone(&broker);
            std::thread::Builder::new()
                .name("expiry".to_string())
                .spawn(move || {
                    loop {
                        std::thread::sleep(Duration::from_secs(1));
                        admin::expire_due(&broker, Timestamp::now());
                    }
                })?;
        }

        // One accept thread per socket, each with its own clone of the broker.
        let (tx, rx) = mpsc::channel::<io::Error>();

        accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools)?;
        accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?;

        // The first listener error stops the daemon.
        match rx.recv() {
            Ok(e) => Err(e),
            Err(_) => Err(io::Error::other("a listener thread ended")),
        }
    }
}

/// Accept streams from one listener forever, starting a handler thread per stream. When the
/// system is out of file descriptors or memory, the listener pauses and tries again instead of
/// stopping: otherwise a peer holding many idle connections could stop the daemon, and the other
/// socket with it. Any other error stops the daemon through `tx`.
fn accept_loop(
    listener: UnixListener,
    broker: Arc<Broker>,
    tx: mpsc::Sender<io::Error>,
    which: Which,
) -> io::Result<()> {
    let name = match which {
        Which::Tools => "accept-broker",
        Which::Admin => "accept-admin",
    };
    std::thread::Builder::new()
        .name(name.to_string())
        .spawn(move || {
            let mut reported = false;
            for stream in listener.incoming() {
                match stream {
                    Ok(stream) => {
                        reported = false;
                        serve_one(stream, &broker, which);
                    }
                    Err(e) if out_of_resources(&e) => {
                        if !reported {
                            broker.log(&format!(
                                "brokerd: cannot accept on {name} for now, retrying: {e}\nsee docs/runbook.md#brokerd-listener-lost"
                            ));
                            reported = true;
                        }
                        std::thread::sleep(Duration::from_millis(200));
                    }
                    Err(e) => {
                        // Once `run` has returned, nobody receives; dropping this is correct.
                        let _ = tx.send(e);
                        return;
                    }
                }
            }
        })?;
    Ok(())
}

/// Hand one stream to its own thread. If the system refuses the thread, the stream is closed, which
/// the peer sees as the broker being unavailable, and the refusal is printed.
fn serve_one(stream: UnixStream, broker: &Arc<Broker>, which: Which) {
    let for_thread = Arc::clone(broker);
    let started = std::thread::Builder::new().spawn(move || match which {
        Which::Tools => broker::handle(stream, &for_thread),
        Which::Admin => admin::handle(stream, &for_thread),
    });
    if let Err(e) = started {
        broker.log(&format!(
            "brokerd: cannot start a thread for a connection, so it was closed: {e}\nsee docs/runbook.md#brokerd-listener-lost"
        ));
    }
}

/// Errors from `accept` that mean "not now" rather than "never": out of file descriptors
/// (`EMFILE`, `ENFILE`, the same numbers on Linux and macOS) or out of memory. `accept` already
/// retries `EINTR` itself.
fn out_of_resources(e: &io::Error) -> bool {
    const ENFILE: i32 = 23;
    const EMFILE: i32 = 24;
    matches!(e.raw_os_error(), Some(ENFILE | EMFILE)) || e.kind() == io::ErrorKind::OutOfMemory
}