haematite-cli 0.21.1

use std::error::Error;
use std::io::{self, Write};
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::{fs::PermissionsExt, net::UnixStream};
use std::thread::JoinHandle;
use std::time::Duration;

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

use super::{Deadline, Limits, connection, deadline, ending};
use crate::serve::{frame, protocol::Version, service::Service, storage::Storage};

type TestResult = Result<(), Box<dyn Error>>;

struct Fixture {
    client: Option<UnixStream>,
    worker: Option<JoinHandle<io::Result<()>>>,
    directory: Option<tempfile::TempDir>,
    max_frame: NonZeroU32,
}

impl Fixture {
    fn new() -> Result<Self, Box<dyn Error>> {
        Self::with_frame_limit(65_536, None)
    }

    fn with_frame_limit(
        frame_limit: u32,
        lost_answer: Option<String>,
    ) -> Result<Self, Box<dyn Error>> {
        let directory = tempfile::tempdir()?;
        let storage = Storage::create(
            DatabaseConfig {
                data_dir: directory.path().join("database"),
                shard_count: 1,
                distributed: None,
                executor_threads: Some(1),
                node_cache_budget: Some(NodeCacheBudget::Unlimited),
                retention: haematite::Retention::KeepAll,
            },
            NodeCacheBudget::Unlimited,
            fixture_long_wait_interval(),
        )?;
        let service = Service::new(storage, NonZeroU32::new(64).ok_or("zero page")?)?;
        let max_frame = NonZeroU32::new(frame_limit).ok_or("zero frame")?;
        let limits = Limits {
            max_frame,
            max_clients: NonZeroUsize::new(1).ok_or("zero clients")?,
            frame_timeout: Duration::from_secs(60),
        };
        let (client, server) = UnixStream::pair()?;
        let worker = std::thread::Builder::new()
            .name("served-refusal".to_owned())
            .spawn(move || {
                if let Some(message) = lost_answer {
                    Storage::lose_next_mutation_answer_with_message(message);
                }
                connection(server, &service, limits)
            })?;
        Ok(Self {
            client: Some(client),
            worker: Some(worker),
            directory: Some(directory),
            max_frame,
        })
    }

    fn send(&mut self, request: &Value) -> Result<Value, Box<dyn Error>> {
        self.send_bytes(&serde_json::to_vec(request)?)
    }

    fn send_bytes(&mut self, bytes: &[u8]) -> Result<Value, Box<dyn Error>> {
        Ok(serde_json::from_str(&self.send_text(bytes)?)?)
    }

    /// The answer's frame as the text it was sent as.
    fn send_text(&mut self, bytes: &[u8]) -> Result<String, Box<dyn Error>> {
        let client = self.client.as_mut().ok_or("closed fixture")?;
        frame::write(client, bytes, self.max_frame)?;
        let bytes = frame::read(client, self.max_frame)?.ok_or("connection closed")?;
        Ok(String::from_utf8(bytes)?)
    }

    /// The hello of a peer that agrees the protocol and typed receipt
    /// fragments (BUSYSHARDNODEADLINE-001 Amendment 17): a hello without
    /// that agreement is refused by name and closes the connection.
    fn hello(&mut self) -> TestResult {
        let answer = self.send(
            &json!({"id":"hello", "operation":{"method":"hello", "params":{
                "protocol": crate::serve::protocol::PROTOCOL,
                "capabilities": [crate::serve::protocol::RECEIPT_CAPABILITY],
                "max_frame": self.max_frame,
            }}}),
        )?;
        assert_eq!(answer["id"], "hello");
        assert_eq!(answer["result"]["protocol"], "haem/1");
        assert!(answer.get("error").is_none());
        Ok(())
    }

    fn branches(&mut self) -> Result<Value, Box<dyn Error>> {
        let answer = self.send(&json!({
            "id":"read", "operation":{"method":"branches.list", "params":{"limit":1}},
        }))?;
        assert_eq!(answer["id"], "read");
        assert!(answer.get("error").is_none());
        answer
            .get("result")
            .cloned()
            .ok_or_else(|| "missing result".into())
    }

    fn refusal(&mut self, request: &Value, code: &str) -> TestResult {
        let answer = self.send(request)?;
        assert_eq!(answer["id"], request["id"]);
        assert!(answer.get("result").is_none());
        let error = answer["error"].as_object().ok_or("missing refusal")?;
        if request["operation"]["method"] == "mutation.apply" {
            assert_eq!(error.len(), 6);
            assert_eq!(error["outcome"], "not_applied");
            assert_eq!(
                error["request_id"],
                request["operation"]["params"]["request_id"]
            );
            assert_eq!(
                error["receipt_branch"],
                request["operation"]["params"]["branch"]
            );
        } else {
            assert_eq!(error.len(), 3);
        }
        assert_eq!(error["code"], code);
        assert!(
            error["message"]
                .as_str()
                .is_some_and(|text| !text.is_empty())
        );
        assert_eq!(error["path"], Value::Null);
        Ok(())
    }

    fn receipt_absent(&mut self, request_id: &str) -> TestResult {
        let answer = self.send(&json!({
            "id":"receipt", "operation":{"method":"mutation.receipt", "params":{
                "branch":"main", "request_id":request_id,
            }},
        }))?;
        assert_eq!(answer["id"], "receipt");
        assert_eq!(answer.get("result"), Some(&Value::Null));
        assert!(answer.get("error").is_none());
        Ok(())
    }

    fn finish(&mut self) -> io::Result<()> {
        drop(self.client.take());
        let worker = self
            .worker
            .take()
            .ok_or_else(|| io::Error::other("missing worker"))?;
        worker
            .join()
            .map_err(|_| io::Error::other("connection worker panicked"))?
    }
}

impl Drop for Fixture {
    fn drop(&mut self) {
        drop(self.client.take());
        if let Some(worker) = self.worker.take() {
            match worker.join() {
                Ok(Ok(())) => {}
                Ok(Err(error)) => eprintln!("connection cleanup: {error}"),
                Err(_) => eprintln!("connection cleanup: worker panicked"),
            }
        }
        drop(self.directory.take());
    }
}

/// The connection loop as merged on 10 Oct: a hello carrying its agreement
/// sets the frame cap, then session.pass reaches the pass branch on the same
/// connection. This service is not served through Lys, so the pass is
/// refused by name, and the connection still answers after it.
#[test]
fn an_agreed_hello_then_session_pass_reaches_the_pass_on_one_connection() -> TestResult {
    let mut fixture = Fixture::new()?;
    let agreed = fixture.send(
        &json!({"id":"hello", "operation":{"method":"hello", "params":{
            "protocol": crate::serve::protocol::PROTOCOL,
            "capabilities": [crate::serve::protocol::RECEIPT_CAPABILITY],
            "max_frame": fixture.max_frame,
        }}}),
    )?;
    assert_eq!(agreed["id"], "hello");
    assert!(
        agreed.get("error").is_none(),
        "the agreed hello was refused: {agreed}"
    );
    fixture.refusal(
        &json!({"id":"pass", "operation":{"method":"session.pass", "params":{"pass":"a-pass"}}}),
        "invalid_operation",
    )?;
    fixture.branches()?;
    fixture.finish()?;
    Ok(())
}

#[test]
fn unknown_method_keeps_the_connection_and_branch_unchanged() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    let before = fixture.branches()?;
    fixture.refusal(
        &json!({"id":"unsupported", "operation":{"method":"modules.list", "params":{}}}),
        "unknown_method",
    )?;
    assert_eq!(fixture.branches()?, before);
    fixture.finish()?;
    Ok(())
}

#[test]
fn invalid_mutation_params_leave_no_write_or_receipt() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    let before = fixture.branches()?;
    fixture.refusal(
        &json!({"id":"bad-params", "operation":{"method":"mutation.apply", "params":{
            "request_id":"refused-params", "branch":"main", "expected":before["entries"][0]["version"],
            "change":{"kind":"collection.put", "value":{"collection":"records"}},
        }}}),
        "invalid_params",
    )?;
    assert_eq!(fixture.branches()?, before);
    fixture.receipt_absent("refused-params")?;
    fixture.finish()?;
    Ok(())
}

#[test]
fn unknown_change_kind_leaves_identity_available_for_a_valid_mutation() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    let before = fixture.branches()?;
    let version = &before["entries"][0]["version"];
    fixture.refusal(
        &json!({"id":"bad-kind", "operation":{"method":"mutation.apply", "params":{
            "request_id":"refused-kind", "branch":"main", "expected":version,
            "change":{"kind":"branch.rebase", "value":{}},
        }}}),
        "unknown_change_kind",
    )?;
    assert_eq!(fixture.branches()?, before);
    fixture.receipt_absent("refused-kind")?;
    let applied = fixture.send(&json!({
        "id":"valid-write", "operation":{"method":"mutation.apply", "params":{
            "request_id":"refused-kind", "branch":"main", "expected":version,
            "change":{"kind":"collection.put", "value":{"collection":"records", "schema":null}},
        }},
    }))?;
    assert_eq!(applied["id"], "valid-write");
    assert!(applied.get("error").is_none());
    assert_eq!(applied["result"]["request_id"], "refused-kind");
    assert_eq!(applied["result"]["version"]["sequence"], "1");
    assert_eq!(
        fixture.branches()?["entries"][0]["version"],
        applied["result"]["version"]
    );
    fixture.finish()?;
    Ok(())
}

#[test]
fn a_decode_refusal_does_not_satisfy_hello_first() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.refusal(
        &json!({"id":"unsupported", "operation":{"method":"modules.list"}}),
        "unknown_method",
    )?;
    fixture.refusal(
        &json!({"id":"early-read", "operation":{"method":"branches.list", "params":{"limit":1}}}),
        "hello_required",
    )?;
    fixture.hello()?;
    assert_eq!(
        fixture.branches()?["entries"]
            .as_array()
            .ok_or("missing branches")?
            .len(),
        1
    );
    fixture.finish()?;
    Ok(())
}

#[test]
fn malformed_known_operations_keep_the_connection_usable() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    let before = fixture.branches()?;
    for operation in [
        json!({"method":"branches.list", "params":{"limit":0}}),
        json!({"method":"branches.list", "params":{"limit":"1"}}),
        json!({"method":"branches.list", "params":{"limit":1, "extra":true}}),
        json!({"method":"branches.list"}),
    ] {
        fixture.refusal(
            &json!({"id":"bad-shape", "operation":operation}),
            "invalid_params",
        )?;
        assert_eq!(fixture.branches()?, before);
    }
    let duplicate = fixture.send_bytes(
        br#"{"id":"duplicate-operation","operation":{"method":"hello"},"operation":{"method":"hello"}}"#,
    )?;
    assert_eq!(duplicate["id"], "duplicate-operation");
    assert_eq!(duplicate["error"]["code"], "invalid_params");
    assert_eq!(fixture.branches()?, before);
    fixture.finish()?;
    Ok(())
}

#[test]
fn uncorrelatable_frames_still_close_the_connection() -> TestResult {
    for bytes in [
        b"{".to_vec(),
        br#"{"operation":{"method":"hello"}}"#.to_vec(),
        br#"{"id":7,"operation":{"method":"hello"}}"#.to_vec(),
        br#"{"id":"","operation":{"method":"hello"}}"#.to_vec(),
        br#"{"id":"one","id":"two","operation":{"method":"hello"}}"#.to_vec(),
        serde_json::to_vec(&json!({"id":"x".repeat(513), "operation":{"method":"hello"}}))?,
    ] {
        let mut fixture = Fixture::new()?;
        let client = fixture.client.as_mut().ok_or("closed fixture")?;
        frame::write(client, &bytes, fixture.max_frame)?;
        assert!(frame::read(client, fixture.max_frame)?.is_none());
        assert_eq!(
            fixture.finish().err().ok_or("frame was accepted")?.kind(),
            io::ErrorKind::InvalidData
        );
    }
    Ok(())
}

/// macOS refuses a socket timeout once the peer has closed; the bytes the peer
/// sent before closing are still read, and then its close is a clean end.
#[test]
fn a_whole_frame_then_a_close_reads_as_that_frame_then_the_end() -> TestResult {
    let limit = NonZeroU32::new(64).ok_or("zero frame")?;
    let (mut client, server) = UnixStream::pair()?;
    frame::write(&mut client, b"last", limit)?;
    drop(client);
    let mut stream = Deadline {
        stream: server,
        end: deadline(Duration::from_secs(60))?,
    };
    assert_eq!(frame::read(&mut stream, limit)?, Some(b"last".to_vec()));
    assert_eq!(frame::read(&mut stream, limit)?, None);
    Ok(())
}

#[test]
fn an_answer_to_a_client_that_has_closed_is_a_broken_pipe() -> TestResult {
    let limit = NonZeroU32::new(64).ok_or("zero frame")?;
    let (client, server) = UnixStream::pair()?;
    drop(client);
    let mut stream = Deadline {
        stream: server,
        end: deadline(Duration::from_secs(60))?,
    };
    let error = frame::write(&mut stream, b"answer", limit)
        .err()
        .ok_or("an answer was written to a closed client")?;
    assert_eq!(error.kind(), io::ErrorKind::BrokenPipe, "{error}");
    Ok(())
}

#[test]
fn a_client_that_closes_after_its_answer_ends_the_connection_cleanly() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    fixture.finish()?;
    Ok(())
}

#[test]
fn a_client_that_closes_inside_a_frame_is_named_a_cut_frame() -> TestResult {
    for (bytes, part) in [
        (&[0, 0][..], "header"),
        (&[0, 0, 0, 10, b'{', b'"', b'i'][..], "body"),
    ] {
        let mut fixture = Fixture::new()?;
        fixture.hello()?;
        fixture
            .client
            .as_mut()
            .ok_or("closed fixture")?
            .write_all(bytes)?;
        let error = fixture.finish().err().ok_or("a cut frame ended cleanly")?;
        let cut = format!("frame cut: the input ended inside its {part}");
        assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof, "{error}");
        assert_eq!(error.to_string(), cut);
        assert_eq!(
            ending(&error),
            format!("haematite client closed inside a frame: {cut}")
        );
    }
    Ok(())
}

#[test]
fn each_ending_is_logged_by_what_it_was() {
    for (kind, words) in [
        (io::ErrorKind::UnexpectedEof, "closed inside a frame"),
        (
            io::ErrorKind::TimedOut,
            "sent no whole frame within the frame deadline",
        ),
        (
            io::ErrorKind::WouldBlock,
            "sent no whole frame within the frame deadline",
        ),
        (io::ErrorKind::InvalidData, "sent a frame that was refused"),
        (
            io::ErrorKind::BrokenPipe,
            "closed before its answer was written",
        ),
        (io::ErrorKind::Other, "connection failed"),
    ] {
        let line = ending(&io::Error::new(kind, "the cause"));
        assert_eq!(line, format!("haematite client {words}: the cause"));
    }
}

#[test]
fn late_frame_refusal_keeps_committed_mutation_identity() -> TestResult {
    // The cap is above the derived receipt floor (Amendment 19 refuses a
    // hello below it); the lost reply is longer than the cap.
    let mut fixture = Fixture::with_frame_limit(4_096, Some("lost committed reply ".repeat(300)))?;
    fixture.hello()?;
    let branches = fixture.branches()?;
    let expected = &branches["entries"][0]["version"];
    let answer = fixture.send(&json!({
        "id":"apply", "operation":{"method":"mutation.apply", "params":{
            "request_id":"frame-lost", "branch":"main", "expected":expected,
            "change":{"kind":"collection.put", "value":{"collection":"records", "schema":null}},
        }},
    }))?;
    assert_eq!(answer["error"]["code"], "response_limit");
    assert_eq!(answer["error"]["outcome"], "unknown");
    assert_eq!(answer["error"]["request_id"], "frame-lost");
    assert_eq!(answer["error"]["receipt_branch"], "main");
    let receipt = fixture.send(&json!({
        "id":"receipt", "operation":{"method":"mutation.receipt", "params":{
            "branch":"main", "request_id":"frame-lost",
        }},
    }))?;
    assert_eq!(receipt["result"]["request_id"], "frame-lost");
    fixture.finish()?;
    Ok(())
}

/// RC2: a stored number goes out over the socket as the text it was written
/// with. A change row is written member by member in its own order (kind,
/// origin, before, after, `value_before`, `value_after`). The transport decoding
/// the answer into a `Value` and encoding it again would sort those members,
/// so the row below would not be found. `serde_json` is built here with
/// `arbitrary_precision`, so a decode alone keeps `1e2`: the order is what a
/// second decode changes.
#[test]
fn changes_list_answers_a_stored_number_as_its_written_text() -> TestResult {
    let mut fixture = Fixture::new()?;
    fixture.hello()?;
    let head = fixture.branches()?["entries"][0]["version"].clone();
    let declared = fixture.send(&json!({
        "id":"declare", "operation":{"method":"mutation.apply", "params":{
            "request_id":"declare", "branch":"main", "expected":head,
            "change":{"kind":"collection.put", "value":{"collection":"records", "schema":null}},
        }},
    }))?;
    let from = declared["result"]["version"].clone();
    let write = json!({
        "id":"write", "operation":{"method":"mutation.apply", "params":{
            "request_id":"write", "branch":"main", "expected":&from,
            "change":{"kind":"document.put", "value":{
                "collection":"records", "document_id":"a", "document":"@body",
            }},
        }},
    })
    .to_string()
    .replace(r#""@body""#, r#"{"n":1e2}"#);
    let written = fixture.send_bytes(write.as_bytes())?;
    assert!(written.get("error").is_none(), "{written}");
    let to = written["result"]["version"].clone();
    let asked = json!({
        "id":"changes", "operation":{"method":"changes.list", "params":{
            "from":{"branch":"main", "version":&from}, "to":{"branch":"main", "version":&to},
            "rows":"values", "page":{"rows":10, "bytes":65_536},
        }},
    });
    let answer = fixture.send_text(&serde_json::to_vec(&asked)?)?;
    let after = serde_json::to_string(&serde_json::from_value::<Version>(to)?)?;
    let row = [
        r#"{"kind":"added","origin":{"collection":"records","id":"a"},"before":null,"after":"#,
        &after,
        r#","value_before":null,"value_after":{"n":1e2}}"#,
    ]
    .concat();
    assert!(answer.contains(&format!(r#""rows":[{row}]"#)), "{answer}");
    fixture.finish()?;
    Ok(())
}

#[test]
fn socket_parent_preflight_leaves_a_private_directory_empty() -> TestResult {
    let directory = tempfile::tempdir()?;
    std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o700))?;
    let path = directory.path().join("haematite.sock");
    crate::serve::socket_owner::check_parent(&path)?;
    let entries = std::fs::read_dir(directory.path())?.collect::<Result<Vec<_>, io::Error>>()?;
    assert!(entries.is_empty());
    directory.close()?;
    Ok(())
}

#[test]
fn socket_bind_rechecks_directory_mode_after_preflight() -> TestResult {
    let directory = tempfile::tempdir()?;
    std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o700))?;
    let path = directory.path().join("haematite.sock");
    crate::serve::socket_owner::check_parent(&path)?;
    std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o755))?;
    let error = super::Listener::bind(&path)
        .err()
        .ok_or("socket bound after parent became non-private")?;
    assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
    assert_eq!(
        error.to_string(),
        "socket directory must be private (mode 0700)"
    );
    let entries = std::fs::read_dir(directory.path())?.collect::<Result<Vec<_>, io::Error>>()?;
    assert!(entries.is_empty());
    directory.close()?;
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
    LONG_WAIT_INTERVAL
}