//! A fake `brokerd` state socket for tests: it records each question and answers with a fixed
//! state, an error frame, or nothing at all (it holds the connection open without a byte). It
//! stops, and releases every held connection, when dropped; every wait has a deadline. Do not
//! edit.

#![allow(dead_code)] // each test file uses a different part of this module

use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use bxcore::{
    Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionStateReply, WireError, read_frame,
    write_frame,
};

/// How the fake answers.
#[derive(Debug, Clone)]
pub enum Answer {
    State(SessionStateReply),
    Error(String),
    /// Read the question, then send nothing until the fake is dropped (at most 30 s).
    Hang,
}

pub struct FakeBrokerd {
    path: PathBuf,
    stop: Arc<AtomicBool>,
    seen: Arc<Mutex<Vec<Envelope>>>,
    accept: Option<JoinHandle<()>>,
}

impl FakeBrokerd {
    /// Listen on `path` (its directory is made) and answer every connection with `answer`.
    pub fn start(path: &Path, answer: Answer) -> FakeBrokerd {
        if let Some(dir) = path.parent() {
            std::fs::create_dir_all(dir).unwrap();
        }
        let _ = std::fs::remove_file(path);
        let listener = UnixListener::bind(path).unwrap();
        let stop = Arc::new(AtomicBool::new(false));
        let seen = Arc::new(Mutex::new(Vec::new()));
        let (stop_accept, seen_accept) = (Arc::clone(&stop), Arc::clone(&seen));
        let accept = std::thread::spawn(move || {
            for stream in listener.incoming() {
                if stop_accept.load(Ordering::SeqCst) {
                    return;
                }
                let Ok(stream) = stream else { return };
                let (stop, seen, answer) = (
                    Arc::clone(&stop_accept),
                    Arc::clone(&seen_accept),
                    answer.clone(),
                );
                std::thread::spawn(move || serve(stream, &answer, &seen, &stop));
            }
        });
        FakeBrokerd {
            path: path.to_path_buf(),
            stop,
            seen,
            accept: Some(accept),
        }
    }

    /// Every question read so far, in order.
    pub fn seen(&self) -> Vec<Envelope> {
        self.seen.lock().unwrap().clone()
    }

    /// Waits up to 10 s until at least `n` questions were read; true if they were.
    pub fn wait_for(&self, n: usize) -> bool {
        let deadline = Instant::now() + Duration::from_secs(10);
        while Instant::now() < deadline {
            if self.seen().len() >= n {
                return true;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        false
    }
}

fn serve(mut stream: UnixStream, answer: &Answer, seen: &Mutex<Vec<Envelope>>, stop: &AtomicBool) {
    let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
    let Ok(question) = read_frame(&mut stream) else {
        return;
    };
    let id = question.id;
    seen.lock().unwrap().push(question);
    let msg = match answer {
        Answer::State(state) => Message::SessionStateReply(state.clone()),
        Answer::Error(detail) => Message::Error(WireError {
            code: ErrorCode::Internal,
            detail: detail.clone(),
        }),
        Answer::Hang => {
            let deadline = Instant::now() + Duration::from_secs(30);
            while !stop.load(Ordering::SeqCst) && Instant::now() < deadline {
                std::thread::sleep(Duration::from_millis(10));
            }
            return;
        }
    };
    let env = Envelope {
        v: PROTOCOL_VERSION,
        id,
        r#final: true,
        msg,
    };
    let _ = write_frame(&mut stream, &env);
}

impl Drop for FakeBrokerd {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::SeqCst);
        // Wakes the accept loop, which then sees `stop`.
        let _ = UnixStream::connect(&self.path);
        if let Some(accept) = self.accept.take() {
            let _ = accept.join();
        }
        let _ = std::fs::remove_file(&self.path);
    }
}