haematite-cli 0.21.1

//! A fake store on a Unix socket for the HTTP MCP tests: it answers hello
//! as serve does, takes or refuses each connection's `session.pass`, and
//! records every request it reads, by connection, so a test can say what
//! reached the store and in what order.

use std::io;
use std::num::NonZeroU32;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread::JoinHandle;
use std::time::Duration;

use serde_json::{Value, json};

use super::event_fixtures::{Gate, Release};
use super::fixtures::TestResult;
use crate::serve::{frame, protocol};

const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(131_071);
/// A fixture connection no caller writes to for this long has been left open
/// by a failing test; it ends instead of holding the fixture's finish.
const IDLE: Duration = Duration::from_secs(30);

/// One request the store read: the connection it came on (counted from 0 in
/// the order the store accepted them), its method and its params.
#[derive(Clone, Debug)]
pub(super) struct Asked {
    pub(super) connection: usize,
    pub(super) method: String,
    pub(super) params: Value,
}

/// How the store answers each connection's `session.pass`.
#[derive(Clone, Copy)]
enum Verdict {
    Accept,
    Refuse(&'static str),
}

struct Shared {
    verdict: Verdict,
    hold: bool,
    asked: Mutex<Vec<Asked>>,
    connections: AtomicUsize,
    verbs: AtomicUsize,
    gate: Arc<Gate>,
    entered: Mutex<mpsc::Sender<()>>,
}

pub(super) struct Store {
    directory: tempfile::TempDir,
    pub(super) path: PathBuf,
    pub(super) entered: mpsc::Receiver<()>,
    shared: Arc<Shared>,
    stopping: Arc<AtomicBool>,
    task: Option<JoinHandle<TestResult>>,
}

fn request(
    socket: &mut UnixStream,
) -> Result<Option<Value>, Box<dyn std::error::Error + Send + Sync>> {
    match frame::read(socket, LIMIT) {
        Ok(Some(bytes)) => Ok(Some(serde_json::from_slice(&bytes)?)),
        Ok(None) => Ok(None),
        Err(error) => match error.kind() {
            io::ErrorKind::ConnectionReset | io::ErrorKind::BrokenPipe => Ok(None),
            _ => Err(error.into()),
        },
    }
}

fn respond(socket: &mut UnixStream, id: &Value, body: &Value) -> TestResult {
    let mut answer = json!({"id":id});
    answer
        .as_object_mut()
        .ok_or_else(|| io::Error::other("fixture answer is not an object"))?
        .extend(
            body.as_object()
                .ok_or_else(|| io::Error::other("fixture reply body is not an object"))?
                .clone(),
        );
    match frame::write(socket, &serde_json::to_vec(&answer)?, LIMIT) {
        Ok(()) => Ok(()),
        Err(error)
            if matches!(
                error.kind(),
                io::ErrorKind::ConnectionReset | io::ErrorKind::BrokenPipe
            ) =>
        {
            Ok(())
        }
        Err(error) => Err(error.into()),
    }
}

/// What the store answers a tool's verb: an empty page of collections on
/// main at version 1, so `describe_store` goes on to its next verb, and a
/// named refusal for every other verb.
fn verb_answer(method: &str) -> Value {
    if method == "collections.list" {
        json!({"result":{"selection":{"branch":"main","version":{"generation":"1","sequence":"1"}},"entries":[],"next_after":null}})
    } else {
        json!({"error":{"code":"fixture_service_refused","message":"The fixture store answers no further verb"}})
    }
}

fn serve(mut socket: UnixStream, shared: &Shared) -> TestResult {
    let connection = shared.connections.fetch_add(1, Ordering::SeqCst);
    socket.set_read_timeout(Some(IDLE))?;
    socket.set_write_timeout(Some(IDLE))?;
    while let Some(asked) = request(&mut socket)? {
        let id = asked
            .get("id")
            .ok_or_else(|| io::Error::other("fixture request id missing"))?
            .clone();
        let method = asked
            .pointer("/operation/method")
            .and_then(Value::as_str)
            .ok_or_else(|| io::Error::other("fixture request method missing"))?
            .to_owned();
        let params = asked
            .pointer("/operation/params")
            .cloned()
            .unwrap_or(Value::Null);
        shared
            .asked
            .lock()
            .map_err(|_| io::Error::other("fixture record poisoned"))?
            .push(Asked {
                connection,
                method: method.clone(),
                params,
            });
        let body = match method.as_str() {
            // The store's hello under BUSYSHARDNODEADLINE-001 Amendment 17:
            // its version, the typed receipt capability it agrees, its limits.
            "hello" => json!({"result":{
                "protocol":protocol::PROTOCOL,
                "capabilities":[protocol::RECEIPT_CAPABILITY],
                "max_frame":LIMIT.get(),"max_page":64,
            }}),
            "session.pass" => match shared.verdict {
                Verdict::Accept => {
                    json!({"result":{"holder":"fixture-holder","expires":4_102_444_800_u64}})
                }
                Verdict::Refuse(code) => {
                    json!({"error":{"code":code,"message":"The fixture store refused the pass"}})
                }
            },
            verb => {
                if shared.hold && shared.verbs.fetch_add(1, Ordering::SeqCst) == 0 {
                    shared
                        .entered
                        .lock()
                        .map_err(|_| io::Error::other("fixture entry poisoned"))?
                        .send(())?;
                    shared.gate.wait()?;
                }
                verb_answer(verb)
            }
        };
        respond(&mut socket, &id, &body)?;
    }
    Ok(())
}

impl Store {
    /// A store that takes every pass.
    pub(super) fn start() -> Result<Self, io::Error> {
        Self::with(Verdict::Accept, false)
    }

    /// A store that refuses every pass with `code`.
    pub(super) fn refusing(code: &'static str) -> Result<Self, io::Error> {
        Self::with(Verdict::Refuse(code), false)
    }

    /// A store that takes every pass and holds the first verb it reads
    /// until the test opens it.
    pub(super) fn held() -> Result<Self, io::Error> {
        Self::with(Verdict::Accept, true)
    }

    fn with(verdict: Verdict, hold: bool) -> Result<Self, io::Error> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("service.sock");
        let listener = UnixListener::bind(&path)?;
        let (entered, received) = mpsc::channel();
        let shared = Arc::new(Shared {
            verdict,
            hold,
            asked: Mutex::new(Vec::new()),
            connections: AtomicUsize::new(0),
            verbs: AtomicUsize::new(0),
            gate: Arc::<Gate>::default(),
            entered: Mutex::new(entered),
        });
        let stopping = Arc::new(AtomicBool::new(false));
        let task = {
            let shared = Arc::clone(&shared);
            let stopping = Arc::clone(&stopping);
            std::thread::spawn(move || -> TestResult {
                let mut tasks = Vec::new();
                for socket in listener.incoming() {
                    if stopping.load(Ordering::SeqCst) {
                        break;
                    }
                    let socket = socket?;
                    let shared = Arc::clone(&shared);
                    tasks.push(std::thread::spawn(move || serve(socket, &shared)));
                }
                for task in tasks {
                    task.join()
                        .map_err(|_| io::Error::other("fixture store connection panicked"))??;
                }
                Ok(())
            })
        };
        Ok(Self {
            directory,
            path,
            entered: received,
            shared,
            stopping,
            task: Some(task),
        })
    }

    /// Every request the store read so far, in the order it read them.
    pub(super) fn asked(&self) -> Result<Vec<Asked>, io::Error> {
        Ok(self
            .shared
            .asked
            .lock()
            .map_err(|_| io::Error::other("fixture record poisoned"))?
            .clone())
    }

    /// How many times a pass was presented to the store.
    pub(super) fn presented(&self) -> Result<usize, io::Error> {
        Ok(self
            .asked()?
            .iter()
            .filter(|asked| asked.method == "session.pass")
            .count())
    }

    pub(super) fn open(&self) -> io::Result<()> {
        self.shared.gate.open()
    }

    pub(super) fn release_on_exit(&self) -> Release {
        Release(Arc::clone(&self.shared.gate))
    }

    pub(super) fn finish(mut self) -> TestResult {
        self.close()
    }

    fn close(&mut self) -> TestResult {
        self.shared.gate.open()?;
        if let Some(task) = self.task.take() {
            self.stopping.store(true, Ordering::SeqCst);
            // The accept loop reads the stop only when a connection arrives.
            match UnixStream::connect(self.directory.path().join("service.sock")) {
                Ok(wake) => drop(wake),
                Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => {}
                Err(error) => return Err(error.into()),
            }
            task.join()
                .map_err(|_| io::Error::other("fixture store task panicked"))??;
        }
        Ok(())
    }
}

impl Drop for Store {
    fn drop(&mut self) {
        if let Err(error) = self.close() {
            eprintln!("store_fixture_cleanup_failed: {error}");
        }
    }
}