//! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the
//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path (a batch of
//! messages or conclusions answered with as many as it asked for, up to the two recorded), and
//! records every request, body included. It can be set to answer every request with a status
//! instead, to answer nothing, to answer as a workspace that does not exist, or to answer a chat
//! with a given text, for every workspace or for one. Stopped, and its held connections released, 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::collections::{HashMap, HashSet};
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use serde_json::Value;

/// One request as the fake read it.
#[derive(Debug, Clone, PartialEq)]
pub struct Request {
    pub method: String,
    pub path: String,
    /// Header names in lower case.
    pub headers: Vec<(String, String)>,
    pub body: Option<Value>,
}

impl Request {
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|(k, _)| k == name)
            .map(|(_, v)| v.as_str())
    }
}

/// How the fake answers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
    /// From the fixtures.
    Fixtures,
    /// Every request with this status. A 422 echoes the request's body, as Honcho does.
    Status(u16),
    /// Read the request, then answer nothing until the fake is dropped (at most 30 s).
    Hang,
    /// As a workspace that does not exist: a search or a conclusion query answers 200 with no
    /// hits (fixtures 34, 35), chat 404 (36).
    Missing,
    /// A chat answers 200 with this text as its `content`; anything else as `Fixtures`.
    Chat(&'static str),
}

struct Shared {
    mode: Mutex<Mode>,
    /// A mode for one workspace's requests, by its name, instead of `mode`.
    workspaces: Mutex<HashMap<String, Mode>>,
    requests: Mutex<Vec<Request>>,
    /// What was made, by method and path and id: a second create answers 200, a second delete
    /// 404.
    made: Mutex<HashSet<String>>,
    stop: AtomicBool,
}

pub struct FakeHoncho {
    pub url: String,
    port: u16,
    shared: Arc<Shared>,
    accept: Option<JoinHandle<()>>,
}

/// A recorded fixture: its response's status and body.
fn fixture(name: &str) -> (u16, String) {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures/honcho")
        .join(name);
    let text = std::fs::read_to_string(&path).unwrap();
    let value: Value = serde_json::from_str(&text).unwrap();
    let status = u16::try_from(value["response"]["status"].as_u64().unwrap()).unwrap();
    let body = match &value["response"]["body"] {
        Value::String(s) if s.is_empty() => String::new(),
        other => other.to_string(),
    };
    (status, body)
}

impl FakeHoncho {
    pub fn start() -> FakeHoncho {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let shared = Arc::new(Shared {
            mode: Mutex::new(Mode::Fixtures),
            workspaces: Mutex::new(HashMap::new()),
            requests: Mutex::new(Vec::new()),
            made: Mutex::new(HashSet::new()),
            stop: AtomicBool::new(false),
        });
        let for_accept = Arc::clone(&shared);
        let accept = std::thread::spawn(move || {
            for stream in listener.incoming() {
                if for_accept.stop.load(Ordering::SeqCst) {
                    return;
                }
                let Ok(stream) = stream else { return };
                let shared = Arc::clone(&for_accept);
                std::thread::spawn(move || serve(stream, &shared));
            }
        });
        FakeHoncho {
            url: format!("http://127.0.0.1:{port}"),
            port,
            shared,
            accept: Some(accept),
        }
    }

    pub fn set(&self, mode: Mode) {
        *self.shared.mode.lock().unwrap() = mode;
    }

    /// Answer `workspace`'s requests in `mode`, whatever `set` says for the others.
    pub fn set_workspace(&self, workspace: &str, mode: Mode) {
        self.shared
            .workspaces
            .lock()
            .unwrap()
            .insert(workspace.to_string(), mode);
    }

    /// Every request read so far, in order.
    pub fn requests(&self) -> Vec<Request> {
        self.shared.requests.lock().unwrap().clone()
    }

    /// Forget the requests read so far (not what was made).
    pub fn clear(&self) {
        self.shared.requests.lock().unwrap().clear();
    }

    /// Waits up to 10 s until at least `n` requests 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.requests().len() >= n {
                return true;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        false
    }
}

impl Drop for FakeHoncho {
    fn drop(&mut self) {
        self.shared.stop.store(true, Ordering::SeqCst);
        let _ = TcpStream::connect(("127.0.0.1", self.port));
        if let Some(accept) = self.accept.take() {
            let _ = accept.join();
        }
    }
}

/// Reads one request: the head up to its blank line, then `Content-Length` bytes of body.
fn read_request(stream: &mut TcpStream) -> Option<Request> {
    let mut buf = Vec::new();
    let mut byte = [0u8; 1];
    while !buf.ends_with(b"\r\n\r\n") {
        if stream.read(&mut byte).ok()? == 0 {
            return None;
        }
        buf.push(byte[0]);
    }
    let head = String::from_utf8(buf).ok()?;
    let mut lines = head.split("\r\n");
    let mut first = lines.next()?.split(' ');
    let (method, path) = (first.next()?.to_string(), first.next()?.to_string());
    let headers: Vec<(String, String)> = lines
        .filter_map(|l| l.split_once(": "))
        .map(|(k, v)| (k.to_ascii_lowercase(), v.to_string()))
        .collect();
    let length: usize = headers
        .iter()
        .find(|(k, _)| k == "content-length")
        .map_or(Some(0), |(_, v)| v.parse().ok())?;
    let mut body = vec![0u8; length];
    stream.read_exact(&mut body).ok()?;
    let body = if length == 0 {
        None
    } else {
        serde_json::from_slice(&body).ok()
    };
    Some(Request {
        method,
        path,
        headers,
        body,
    })
}

/// The fixture that answers `request`, by its method and the shape of its path; the second of a
/// pair when the same thing was made (or deleted) before.
fn answer(request: &Request, made: &Mutex<HashSet<String>>) -> (u16, String) {
    let parts: Vec<&str> = request.path.trim_start_matches('/').split('/').collect();
    let id = request
        .body
        .as_ref()
        .and_then(|b| b["id"].as_str())
        .unwrap_or_default()
        .to_string();
    let again = !made
        .lock()
        .unwrap()
        .insert(format!("{} {} {id}", request.method, request.path));
    let configured = request
        .body
        .as_ref()
        .is_some_and(|b| b.get("configuration").is_some());
    match (request.method.as_str(), parts.as_slice()) {
        ("POST", ["v3", "workspaces"]) if again => fixture("02-workspace-create-again.json"),
        ("POST", ["v3", "workspaces"]) => fixture("01-workspace-create.json"),
        ("POST", ["v3", "workspaces", _, "peers"]) if again => {
            fixture("05-peer-create-boxmaker0-again.json")
        }
        ("POST", ["v3", "workspaces", _, "peers"]) if configured => {
            fixture("04-peer-create-boxmaker0.json")
        }
        ("POST", ["v3", "workspaces", _, "peers"]) => fixture("03-peer-create-kyle.json"),
        ("POST", ["v3", "workspaces", _, "sessions"]) if again => {
            fixture("07-session-create-again.json")
        }
        ("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"),
        ("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => {
            first(fixture("08-messages-add-batch.json"), request, "messages")
        }
        ("POST", ["v3", "workspaces", _, "conclusions"]) => first(
            fixture("11-conclusions-create-batch.json"),
            request,
            "conclusions",
        ),
        // A workspace chat answers the recorded one at `minimal` (fixture 19), whatever it asked.
        ("POST", ["v3", "workspaces", _, "chat"]) => fixture("19-workspace-chat.json"),
        // A search answers the three recorded messages about tea (fixture 30), whatever it asked.
        ("POST", ["v3", "workspaces", _, "search"]) => fixture("30-workspace-search-limit.json"),
        // A conclusion query answers the recorded one about the owner (32) when it asks about
        // `kyle`, else the one about an agent (33).
        ("POST", ["v3", "workspaces", _, "conclusions", "query"]) => {
            let about = request
                .body
                .as_ref()
                .and_then(|b| b["filters"]["observed_id"].as_str().map(str::to_string));
            if about.as_deref() == Some("kyle") {
                fixture("32-conclusions-query-owner.json")
            } else {
                fixture("33-conclusions-query-agent.json")
            }
        }
        ("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => {
            fixture("16-conclusion-delete-again.json")
        }
        ("DELETE", ["v3", "workspaces", _, "conclusions", _]) => {
            fixture("15-conclusion-delete.json")
        }
        _ => (404, r#"{"detail":"Not Found"}"#.to_string()),
    }
}

/// A recorded batch answer cut to as many items as the request's `field` holds (the recorded ones
/// hold two), so that one message or conclusion gets one id.
fn first(recorded: (u16, String), request: &Request, field: &str) -> (u16, String) {
    let asked = request
        .body
        .as_ref()
        .and_then(|b| b[field].as_array())
        .map_or(0, Vec::len);
    let mut items: Vec<Value> = serde_json::from_str(&recorded.1).unwrap();
    items.truncate(asked);
    (recorded.0, Value::Array(items).to_string())
}

fn serve(mut stream: TcpStream, shared: &Shared) {
    let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
    let Some(request) = read_request(&mut stream) else {
        return;
    };
    shared.requests.lock().unwrap().push(request.clone());
    let workspace = request.path.split('/').nth(3).unwrap_or("").to_string();
    let mode = match shared.workspaces.lock().unwrap().get(&workspace) {
        Some(mode) => *mode,
        None => *shared.mode.lock().unwrap(),
    };
    let (status, body) = match mode {
        Mode::Fixtures => answer(&request, &shared.made),
        Mode::Missing if request.path.ends_with("/search") => {
            fixture("34-workspace-search-missing-workspace.json")
        }
        Mode::Missing if request.path.ends_with("/conclusions/query") => {
            fixture("35-conclusions-query-missing-workspace.json")
        }
        Mode::Missing => fixture("36-workspace-chat-missing-workspace.json"),
        Mode::Chat(content) if request.path.ends_with("/chat") => (
            200,
            serde_json::json!({ "content": content, "evidence": null }).to_string(),
        ),
        Mode::Chat(_) => answer(&request, &shared.made),
        Mode::Status(422) => {
            let input = request.body.clone().unwrap_or(Value::Null);
            (
                422,
                serde_json::json!({ "detail": [{ "input": input }] }).to_string(),
            )
        }
        Mode::Status(status) => (status, r#"{"detail":"refused"}"#.to_string()),
        Mode::Hang => {
            let deadline = Instant::now() + Duration::from_secs(30);
            while !shared.stop.load(Ordering::SeqCst) && Instant::now() < deadline {
                std::thread::sleep(Duration::from_millis(10));
            }
            return;
        }
    };
    let head = format!(
        "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    let _ = stream.write_all(head.as_bytes());
    let _ = stream.write_all(body.as_bytes());
}