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

//! A fake sync server for the activity tests: it answers the activity routes
//! of the server and the members list.
//!
//! It speaks the same JSON as `notescraft-sync-server`
//! (`/v1/vaults/{vault}/activity`, `/v1/vaults/{vault}/activity/read` and
//! `/v1/vaults/{vault}/members`) and keeps the events it was told about, so a
//! test can read the feed and see the events a client posted.

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

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde_json::{Value, json};

/// The token of the bound vault; it must never reach an answer or an error.
pub const TOKEN: &str = "secret-token";

/// The vault the tests bind to.
pub const VAULT: &str = "v-1";

/// How many events one page of the fake holds.
const MAX_PAGE: usize = 200;

/// One request the fake received.
#[derive(Debug, Clone)]
pub struct Request {
    /// The HTTP method.
    pub method: String,
    /// The path of the request line.
    pub path: String,
    /// The `Authorization` header, when there was one.
    pub auth: Option<String>,
    /// The body of the request.
    pub body: String,
}

/// One event of the feed of the fake.
#[derive(Debug, Clone)]
pub struct Event {
    /// The id of the event.
    pub id: u64,
    /// The kind of the event.
    pub kind: 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 mentions.
    pub mentions: Vec<String>,
    /// True once the member read the event.
    pub read: bool,
}

/// What the fake remembers between requests.
#[derive(Default)]
struct State {
    /// The events of the feed.
    events: Vec<Event>,
    /// The members of the vault.
    members: Vec<Value>,
    /// How many ids were handed out.
    next: u64,
}

/// A fake sync server: it answers the activity routes.
pub struct Fake {
    /// 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 Fake {
    /// Starts a fake on an ephemeral port.
    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::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 (status, body) = answer(&state, &request);
                                write_answer(&mut writer, status, &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()
    }

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

    /// Puts one event in the feed, as another member would have posted it.
    pub fn seed(&self, kind: &str, note: &str, mentions: &[&str]) -> u64 {
        let mut state = self.state.lock().expect("state");
        state.next += 1;
        let id = state.next;
        state.events.push(Event {
            id,
            kind: kind.to_owned(),
            note: Some(note.to_owned()),
            thread_id: Some(format!("c-{id}")),
            mentions: mentions.iter().map(|name| (*name).to_owned()).collect(),
            read: false,
        });
        id
    }

    /// Puts the members of the vault in the list of the fake.
    pub fn set_members(&self, members: Vec<Value>) {
        self.state.lock().expect("state").members = members;
    }
}

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

/// The answer of the fake for one request: `(status, body)`.
fn answer(state: &Arc<Mutex<State>>, request: &Request) -> (u16, String) {
    let method = request.method.as_str();
    let path = request.path.as_str();
    let feed = format!("/v1/vaults/{VAULT}/activity");
    let read = format!("/v1/vaults/{VAULT}/activity/read");
    let members = format!("/v1/vaults/{VAULT}/members");
    let mut state = state.lock().expect("state");
    if method == "POST" && path == feed {
        return post_event(&mut state, request);
    }
    if method == "GET" && path.starts_with(&feed) {
        return list_events(&state, path);
    }
    if method == "POST" && path == read {
        return mark_read(&mut state, request);
    }
    if method == "GET" && path == members {
        return (200, Value::Array(state.members.clone()).to_string());
    }
    not_found()
}

/// A new event of the feed of the fake.
fn post_event(state: &mut State, request: &Request) -> (u16, String) {
    let body: Value = serde_json::from_str(&request.body).unwrap_or(Value::Null);
    let kind = body.get("kind").and_then(Value::as_str).unwrap_or_default().to_owned();
    let note = body.get("note").and_then(Value::as_str).map(str::to_owned);
    let thread_id = body.get("thread_id").and_then(Value::as_str).map(str::to_owned);
    let mentions: Vec<String> =
        body.get("mentions").and_then(Value::as_array).map(|list| list.iter().filter_map(Value::as_str).map(str::to_owned).collect()).unwrap_or_default();
    state.next += 1;
    let id = state.next;
    state.events.push(Event { id, kind: kind.clone(), note, thread_id, mentions: mentions.clone(), read: false });
    let head = state.next;
    (200, json!({ "id": id, "kind": kind, "head": head, "recipients": mentions.len() }).to_string())
}

/// The page of the feed of the fake.
fn list_events(state: &State, path: &str) -> (u16, String) {
    let since = query_number(path, "since").unwrap_or(0);
    let cap = u64::try_from(MAX_PAGE).unwrap_or(u64::MAX);
    let limit = usize::try_from(query_number(path, "limit").unwrap_or(50).min(cap)).unwrap_or(MAX_PAGE);
    let visible: Vec<&Event> = state.events.iter().filter(|event| event.id > since).collect();
    let more = visible.len() > limit;
    let rows: Vec<Value> = visible.iter().take(limit).map(|event| event_json(event)).collect();
    let unread = state.events.iter().filter(|event| !event.read).count();
    (200, json!({ "events": rows, "head": state.next, "more": more, "unread": unread }).to_string())
}

/// The events of the fake as the rows of the feed.
fn event_json(event: &Event) -> Value {
    json!({
        "id": event.id,
        "ts_ms": 10,
        "kind": event.kind,
        "actor": "u-ana",
        "actor_name": "Ana",
        "note": event.note,
        "thread_id": event.thread_id,
        "line": Value::Null,
        "read": event.read,
    })
}

/// Marks the events of the fake read.
fn mark_read(state: &mut State, request: &Request) -> (u16, String) {
    let body: Value = serde_json::from_str(&request.body).unwrap_or(Value::Null);
    let ids = body.get("ids").and_then(Value::as_array).map(|list| list.iter().filter_map(Value::as_u64).collect::<Vec<u64>>()).unwrap_or_default();
    for event in &mut state.events {
        if ids.is_empty() || ids.contains(&event.id) {
            event.read = true;
        }
    }
    let unread = state.events.iter().filter(|event| !event.read).count();
    (200, json!({ "unread": unread }).to_string())
}

/// One number of the query of a request.
fn query_number(path: &str, name: &str) -> Option<u64> {
    let query = path.split_once('?').map(|(_, query)| query).unwrap_or_default();
    query.split('&').find_map(|pair| {
        let (key, value) = pair.split_once('=')?;
        (key == name).then(|| value.parse().ok())?
    })
}

/// The answer of a route the fake does not know.
fn not_found() -> (u16, String) {
    (404, "{\"error\":\"no such route\"}".to_owned())
}

/// Reads one request of a connection.
fn read_request(reader: &mut BufReader<TcpStream>) -> Option<Request> {
    let mut line = String::new();
    reader.read_line(&mut line).ok()?;
    let mut parts = line.split_whitespace();
    let method = parts.next()?.to_owned();
    let path = parts.next()?.to_owned();
    let mut auth = None;
    let mut length = 0_usize;
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header).ok()? == 0 {
            break;
        }
        let trimmed = header.trim_end();
        if trimmed.is_empty() {
            break;
        }
        if let Some((name, value)) = trimmed.split_once(':') {
            let name = name.trim().to_ascii_lowercase();
            let value = value.trim().to_owned();
            if name == "content-length" {
                length = value.parse().unwrap_or(0);
            }
            if name == "authorization" {
                auth = Some(value.clone());
            }
        }
    }
    let mut body = vec![0_u8; length];
    if length > 0 {
        reader.read_exact(&mut body).ok()?;
    }
    Some(Request { method, path, auth, body: String::from_utf8_lossy(&body).into_owned() })
}

/// Writes one answer of the fake.
fn write_answer(stream: &mut TcpStream, status: u16, body: &str) {
    let reason = match status {
        200 => "OK",
        400 => "Bad Request",
        404 => "Not Found",
        _ => "Status",
    };
    let head = format!("HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", body.len());
    let _ = stream.write_all(head.as_bytes());
    let _ = stream.write_all(body.as_bytes());
    let _ = stream.flush();
}