//! `brokerd serve` answers `memoryd`'s session-state question on its third socket,
//! `run/memory-broker/state.sock` or `[sockets] state` (M5c spec, section 7). Do not edit.

#[path = "support/tmp.rs"]
mod tmp;

use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

use bxcore::{
    DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId,
    SessionState as StateQuestion, SessionStateReply,
};
use tmp::TempDir;

struct Home {
    dir: TempDir,
    config: PathBuf,
}

impl Home {
    /// A home whose config has `sockets` as the body of its `[sockets]` table.
    fn new(tag: &str, sockets: &str) -> Home {
        let dir = TempDir::new(tag);
        std::fs::create_dir_all(dir.path().join("grants")).unwrap();
        let text = format!(
            "[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n[sockets]\n{sockets}\n",
            home = dir.path().display()
        );
        let config = dir.write("brokerd.toml", &text);
        Home { dir, config }
    }

    fn path(&self, relative: &str) -> PathBuf {
        self.dir.path().join(relative)
    }

    fn state(&self) -> PathBuf {
        self.path("run/memory-broker/state.sock")
    }

    /// Writes the session's state file, as the ledger would.
    fn session(&self, id: &str, text: &str) {
        std::fs::create_dir_all(self.path("broker/sessions")).unwrap();
        std::fs::write(self.path(&format!("broker/sessions/{id}.json")), text).unwrap();
    }

    /// Starts `brokerd serve` and waits until the tool socket, the admin socket and `state`
    /// accept a connection.
    fn serve(&self, state: &Path) -> Running {
        let child = Command::new(env!("CARGO_BIN_EXE_brokerd"))
            .args(["serve", "--config"])
            .arg(&self.config)
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        let running = Running(Some(child));
        let until = Instant::now() + Duration::from_secs(10);
        for socket in [
            self.path("run/loop-broker/broker.sock"),
            self.path("run/owner-broker/admin.sock"),
            state.to_path_buf(),
        ] {
            while UnixStream::connect(&socket).is_err() {
                assert!(
                    Instant::now() < until,
                    "brokerd never listened on {}",
                    socket.display()
                );
                std::thread::sleep(Duration::from_millis(20));
            }
        }
        running
    }
}

/// Kills the daemon when dropped, and waits for it; `stop` returns what it printed.
struct Running(Option<Child>);

impl Running {
    fn stop(mut self) -> String {
        let mut child = self.0.take().unwrap();
        child.kill().unwrap();
        let output = child.wait_with_output().unwrap();
        String::from_utf8_lossy(&output.stderr).to_string()
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if let Some(child) = &mut self.0 {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

fn mode(path: &Path) -> u32 {
    std::fs::metadata(path).unwrap().permissions().mode() & 0o777
}

/// A connection with a ten-second read limit, so a daemon that never answers fails the test.
fn connect(socket: &Path) -> UnixStream {
    let stream = UnixStream::connect(socket).unwrap();
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .unwrap();
    stream
}

/// Sends one message under `id` and returns the one final frame that answers it.
fn exchange(socket: &Path, id: u64, msg: Message) -> Envelope {
    let mut stream = connect(socket);
    bxcore::write_frame(
        &mut stream,
        &Envelope {
            v: PROTOCOL_VERSION,
            id,
            r#final: true,
            msg,
        },
    )
    .unwrap();
    let env = bxcore::read_frame(&mut stream).unwrap();
    assert_eq!((env.id, env.r#final), (id, true), "{env:?}");
    env
}

fn question(session: &str) -> Message {
    Message::SessionState(StateQuestion {
        session: SessionId::new(session).unwrap(),
    })
}

fn reply(env: &Envelope) -> SessionStateReply {
    match &env.msg {
        Message::SessionStateReply(reply) => reply.clone(),
        other => panic!("wanted a session_state_reply, got {other:?}"),
    }
}

#[test]
fn it_answers_on_its_default_path_from_the_state_files() {
    let home = Home::new("serve-state-answers", "");
    home.session("s1", "{\"taint\":\"secret\",\"untrusted\":true}\n");
    home.session("s2", "{\"taint\":\"private\",\"untrusted\":true}\n");
    let running = home.serve(&home.state());
    for (id, session, class, untrusted) in [
        (1, "s1", DataClass::Secret, true),
        (2, "s2", DataClass::Private, true),
        (3, "unseen", DataClass::Private, false),
    ] {
        assert_eq!(
            reply(&exchange(&home.state(), id, question(session))),
            SessionStateReply { class, untrusted },
            "{session}"
        );
    }
    assert!(!home.path("broker/sessions/unseen.json").exists());
    let printed = running.stop();
    assert!(
        printed.contains(&format!(
            "brokerd: serving session state on {}\n",
            home.state().display()
        )),
        "{printed}"
    );
}

/// As for the other two sockets: the directory is made private even when it was there at 0755,
/// the socket is 0600, and a stale socket left by a killed `brokerd` is replaced.
#[test]
fn its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replaced() {
    let home = Home::new("serve-state-modes", "");
    std::fs::create_dir_all(home.path("run/memory-broker")).unwrap();
    std::fs::set_permissions(
        home.path("run/memory-broker"),
        std::fs::Permissions::from_mode(0o755),
    )
    .unwrap();
    drop(UnixListener::bind(home.state()).unwrap());
    assert!(home.state().exists(), "the stale socket file is there");
    let _running = home.serve(&home.state());
    assert_eq!(mode(&home.path("run/memory-broker")), 0o700);
    assert_eq!(mode(&home.state()), 0o600);
    assert_eq!(
        reply(&exchange(&home.state(), 9, question("s9"))),
        SessionStateReply {
            class: DataClass::Private,
            untrusted: false,
        }
    );
}

#[test]
fn sockets_state_in_the_config_moves_it() {
    let dir = TempDir::new("serve-state-moved-sock");
    let moved = dir.path().join("st/state.sock");
    let home = Home::new(
        "serve-state-moved",
        &format!("state = \"{}\"", moved.display()),
    );
    home.session("s1", "{\"taint\":\"secret\",\"untrusted\":false}\n");
    let running = home.serve(&moved);
    assert_eq!(
        reply(&exchange(&moved, 1, question("s1"))),
        SessionStateReply {
            class: DataClass::Secret,
            untrusted: false,
        }
    );
    assert_eq!(mode(&dir.path().join("st")), 0o700);
    assert!(
        !home.path("run/memory-broker").exists(),
        "nothing at the default"
    );
    let printed = running.stop();
    assert!(
        printed.contains(&format!(
            "brokerd: serving session state on {}\n",
            moved.display()
        )),
        "{printed}"
    );
}

/// A connection that sends nothing holds only its own thread: the next question is answered.
#[test]
fn a_silent_connection_does_not_hold_up_the_next() {
    let home = Home::new("serve-state-silent", "");
    let _running = home.serve(&home.state());
    let _silent = connect(&home.state());
    assert_eq!(
        reply(&exchange(&home.state(), 2, question("s2"))),
        SessionStateReply {
            class: DataClass::Private,
            untrusted: false,
        }
    );
}

/// The question belongs on `state.sock` only: the tool and admin sockets refuse it.
#[test]
fn the_other_sockets_refuse_the_question() {
    let home = Home::new("serve-state-others", "");
    let running = home.serve(&home.state());
    for (id, socket) in [
        (4, home.path("run/loop-broker/broker.sock")),
        (5, home.path("run/owner-broker/admin.sock")),
    ] {
        match exchange(&socket, id, question("s1")).msg {
            Message::Error(e) => assert_eq!(e.code, ErrorCode::Forbidden, "{e:?}"),
            other => panic!("{other:?}"),
        }
    }
    let printed = running.stop();
    assert!(
        printed.contains("session_state on broker.sock\nsee docs/runbook.md#socket-forbidden"),
        "{printed}"
    );
    assert!(
        printed.contains("session_state on admin.sock\nsee docs/runbook.md#socket-forbidden"),
        "{printed}"
    );
}