notescraft/crates/engine/tests/comments_e2e_stub/mod.rs

//! A fake sync server for the end-to-end proof of the comments: the members of
//! the vault, the activity feed and the files of the vault.
//!
//! It speaks the same JSON as `notescraft-sync-server`: `/v1/vaults/{vault}/members`,
//! `/v1/vaults/{vault}/activity`, `/v1/vaults/{vault}/activity/read`,
//! `/v1/vaults/{vault}/changes` and `/v1/vaults/{vault}/files/{path}`. The files go
//! through the ledger of `notescraft-sync`, so the decisions about a stale writer
//! are the ones the real server makes.
//!
//! The feed keeps the rules of the server: the actor of an event never sees it,
//! a `mention` reaches only the members it names and every other event reaches
//! the rest of the vault.
//!
//! The events no client posts yet - the `comment` and the `thread_resolved` -
//! are derived from the sidecar the fake serves, the way the server derives
//! `note_changed` from a file upload.

#![allow(dead_code)] // shared test helpers: each test file uses only some of them

mod feed;
mod files;
mod wire;

use std::collections::{BTreeMap, BTreeSet};
use std::io::BufReader;
use std::net::TcpListener;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use notescraft_sync::Ledger;
use serde_json::Value;

use wire::{Request, read_request, write_answer};

/// The vault the two devices share.
pub const VAULT: &str = "v-1";

/// Ana, the author of the comment.
pub const ANA: &str = "u-ana";

/// Bob, the member Ana mentions.
pub const BOB: &str = "u-bob";

/// The token of Ana's device.
pub const ANA_TOKEN: &str = "token-ana";

/// The token of Bob's device.
pub const BOB_TOKEN: &str = "token-bob";

/// The suffix of the sidecar of a note.
pub(crate) const SIDECAR_SUFFIX: &str = ".comments.json";

/// Most events one page of the feed holds.
pub(crate) const MAX_PAGE: u64 = 200;

/// The audience of an event the whole vault sees.
pub(crate) const AUDIENCE_VAULT: &str = "vault";

/// The audience of an event that names its members.
pub(crate) const AUDIENCE_MENTION: &str = "mention";

/// The kind of the event a mention posts.
pub(crate) const KIND_MENTION: &str = "mention";

/// The kind of the event a new thread posts.
pub(crate) const KIND_COMMENT: &str = "comment";

/// The kind of the event a closed thread posts.
pub(crate) const KIND_THREAD_RESOLVED: &str = "thread_resolved";

/// One event of the feed of the fake.
#[derive(Debug, Clone)]
pub struct Event {
    /// The id of the event; the cursor of the feed.
    pub id: u64,
    /// The kind of the event.
    pub kind: String,
    /// The user id of the actor.
    pub actor: String,
    /// The audience of the event: the whole vault or the members it names.
    pub audience: String,
    /// The note the event is about.
    pub note: Option<String>,
    /// The thread the event is about.
    pub thread_id: Option<String>,
    /// The members the event names.
    pub mentions: Vec<String>,
    /// True when a client posted the event, false when the fake derived it.
    pub posted: bool,
}

/// What the fake remembers between requests.
#[derive(Default)]
pub(crate) struct State {
    /// The files of the vault, as the ledger of the server sees them.
    pub(crate) ledger: Ledger,
    /// The bytes of every file the ledger holds.
    pub(crate) blobs: BTreeMap<String, Vec<u8>>,
    /// The members of the vault.
    pub(crate) members: Vec<Value>,
    /// The events of the feed, oldest first.
    pub(crate) events: Vec<Event>,
    /// The threads the fake has seen, with their resolved state.
    pub(crate) threads: BTreeMap<String, bool>,
    /// The events every member has read.
    pub(crate) read: BTreeSet<(String, u64)>,
    /// How many ids were handed out.
    pub(crate) next_event: u64,
}

impl State {
    /// The newest id of the feed.
    pub(crate) fn head(&self) -> u64 {
        self.next_event
    }

    /// Stores one event and answers its id.
    pub(crate) fn push_event(&mut self, event: Event) -> u64 {
        self.next_event += 1;
        let id = self.next_event;
        self.events.push(Event { id, ..event });
        id
    }

    /// The events the member sees, oldest first.
    pub(crate) fn visible(&self, user: &str) -> Vec<&Event> {
        self.events.iter().filter(|event| event.actor != user && (event.audience == AUDIENCE_VAULT || event.mentions.iter().any(|name| name == user))).collect()
    }

    /// True when the member has read the event.
    pub(crate) fn is_read(&self, user: &str, id: u64) -> bool {
        self.read.contains(&(user.to_owned(), id))
    }

    /// The events the member sees and has not read.
    pub(crate) fn unread(&self, user: &str) -> usize {
        self.visible(user).iter().filter(|event| !self.is_read(user, event.id)).count()
    }
}

/// A fake sync server: it answers the members, the feed and the files.
pub struct Stub {
    /// The base URL of the fake.
    pub base: String,
    /// Every request the fake received.
    seen: Arc<Mutex<Vec<Request>>>,
    /// What the fake remembers.
    state: Arc<Mutex<State>>,
    /// Whether the listener thread should stop.
    stop: Arc<AtomicBool>,
}

impl Stub {
    /// Starts a fake on an ephemeral port, with Ana and Bob as members.
    pub fn spawn() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind the fake");
        listener.set_nonblocking(true).expect("nonblocking");
        let port = listener.local_addr().expect("addr").port();
        let seen: Arc<Mutex<Vec<Request>>> = Arc::new(Mutex::new(Vec::new()));
        let state: Arc<Mutex<State>> = Arc::new(Mutex::new(State { members: feed::members(), ..State::default() }));
        let stop = Arc::new(AtomicBool::new(false));
        let thread_seen = Arc::clone(&seen);
        let thread_state = Arc::clone(&state);
        let thread_stop = Arc::clone(&stop);
        std::thread::spawn(move || {
            while !thread_stop.load(Ordering::Relaxed) {
                match listener.accept() {
                    Ok((stream, _)) => {
                        let _ = stream.set_nonblocking(false);
                        let Ok(mut writer) = stream.try_clone() else { continue };
                        let mut reader = BufReader::new(stream);
                        let _ = reader.get_ref().set_read_timeout(Some(Duration::from_secs(30)));
                        let seen = Arc::clone(&thread_seen);
                        let state = Arc::clone(&thread_state);
                        std::thread::spawn(move || {
                            while let Some(request) = read_request(&mut reader) {
                                seen.lock().expect("seen").push(request.clone());
                                let reply = answer(&state, &request);
                                write_answer(&mut writer, reply.status, reply.content_type, &reply.body);
                            }
                        });
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(Duration::from_millis(2)),
                    Err(_) => break,
                }
            }
        });
        Self { base: format!("http://127.0.0.1:{port}"), seen, state, stop }
    }

    /// Every request the fake received.
    pub fn requests(&self) -> Vec<Request> {
        self.seen.lock().expect("seen").clone()
    }

    /// Every event of the feed, oldest first.
    pub fn events(&self) -> Vec<Event> {
        self.state.lock().expect("state").events.clone()
    }

    /// The events a client posted, oldest first.
    pub fn posted(&self) -> Vec<Event> {
        self.events().into_iter().filter(|event| event.posted).collect()
    }

    /// The bytes of a file of the vault.
    pub fn file(&self, path: &str) -> Option<Vec<u8>> {
        self.state.lock().expect("state").blobs.get(path).cloned()
    }
}

impl Drop for Stub {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

/// What the fake answers: the status, the content type and the body.
pub(crate) struct Reply {
    /// The HTTP status of the answer.
    pub(crate) status: u16,
    /// The content type of the body.
    pub(crate) content_type: &'static str,
    /// The body of the answer.
    pub(crate) body: Vec<u8>,
}

impl Reply {
    /// A JSON answer.
    pub(crate) fn json(status: u16, value: &Value) -> Self {
        Self { status, content_type: "application/json", body: serde_json::to_vec(value).unwrap_or_default() }
    }

    /// An answer with raw bytes.
    pub(crate) fn bytes(status: u16, body: Vec<u8>) -> Self {
        Self { status, content_type: "application/octet-stream", body }
    }

    /// The answer of a route the fake does not know.
    pub(crate) fn not_found() -> Self {
        Self::json(404, &serde_json::json!({ "error": "no such route" }))
    }
}

/// The answer of the fake for one request.
fn answer(state: &Arc<Mutex<State>>, request: &Request) -> Reply {
    let (path, query) = split_query(&request.path);
    let prefix = format!("/v1/vaults/{VAULT}");
    let Some(rest) = path.strip_prefix(&prefix) else { return Reply::not_found() };
    match (request.method.as_str(), rest) {
        ("GET", "/members") => {
            let state = state.lock().expect("state");
            Reply::json(200, &Value::Array(state.members.clone()))
        }
        ("POST", "/activity") => {
            let mut state = state.lock().expect("state");
            feed::post_event(&mut state, request)
        }
        ("GET", "/activity") => {
            let state = state.lock().expect("state");
            feed::list_events(&state, request, query)
        }
        ("POST", "/activity/read") => {
            let mut state = state.lock().expect("state");
            feed::mark_read(&mut state, request)
        }
        ("GET", "/changes") => {
            let state = state.lock().expect("state");
            files::changes(&state, query)
        }
        (_, path) if path.starts_with("/files/") => {
            let mut state = state.lock().expect("state");
            files::file_route(&mut state, request, &decode(path.trim_start_matches("/files/")))
        }
        _ => Reply::not_found(),
    }
}

/// The user behind the token of a request.
pub(crate) fn user_of(request: &Request) -> String {
    let token = request.auth.as_deref().and_then(|auth| auth.strip_prefix("Bearer ")).unwrap_or_default();
    match token {
        ANA_TOKEN => ANA.to_owned(),
        BOB_TOKEN => BOB.to_owned(),
        _ => "u-stranger".to_owned(),
    }
}

/// The path of a request without its query.
pub(crate) fn split_query(path: &str) -> (&str, &str) {
    match path.split_once('?') {
        Some((path, query)) => (path, query),
        None => (path, ""),
    }
}

/// One number of the query of a request.
pub(crate) fn query_number(query: &str, name: &str) -> Option<u64> {
    query.split('&').find_map(|pair| {
        let (key, value) = pair.split_once('=')?;
        if key != name {
            return None;
        }
        value.parse().ok()
    })
}

/// One number of a header of a request.
pub(crate) fn header_number(request: &Request, name: &str) -> u64 {
    request.header(name).and_then(|value| value.parse().ok()).unwrap_or(0)
}

/// The list of strings a field of a body holds.
pub(crate) fn strings(body: &Value, name: &str) -> Vec<String> {
    body.get(name).and_then(Value::as_array).map(|list| list.iter().filter_map(Value::as_str).map(str::to_owned).collect()).unwrap_or_default()
}

/// Decodes the percent escapes of a path.
pub(crate) fn decode(path: &str) -> String {
    let mut out: Vec<u8> = Vec::with_capacity(path.len());
    let mut rest = path.as_bytes().iter().copied();
    while let Some(byte) = rest.next() {
        if byte != b'%' {
            out.push(byte);
            continue;
        }
        let pair = [rest.next().unwrap_or(b' '), rest.next().unwrap_or(b' ')];
        match std::str::from_utf8(&pair).ok().and_then(|text| u8::from_str_radix(text, 16).ok()) {
            Some(decoded) => out.push(decoded),
            None => out.push(b'%'),
        }
    }
    String::from_utf8_lossy(&out).into_owned()
}