//! The serve loop posts what `gateway.sock` brings, where it belongs, and marks it when it ends
//! or when a restart left it behind (M4b spec, sections 6 and 8). Do not edit.

#[path = "support/brokerd_client.rs"]
mod brokerd_client;
#[path = "support/fake_loop.rs"]
mod fake_loop;
#[path = "support/fake_mm.rs"]
mod fake_mm;
#[path = "support/gateway.rs"]
mod gateway;
#[path = "support/tmp.rs"]
mod tmp;

use std::path::PathBuf;
use std::sync::{Mutex, mpsc};
use std::time::Duration;

use brokerd_client::{ask, pending};
use fake_loop::{done, serve_loop};
use fake_mm::{DM, FakeMm, KYLE, id, post};
use gateway::{WAIT, config, loop_dir, read_state, start};
use tmp::TempDir;

fn socket(home: &TempDir) -> PathBuf {
    home.path().join("run/gateway/gateway.sock")
}

/// A turn that waits until the returned sender says so.
fn held_loop(home: &TempDir) -> mpsc::Sender<()> {
    loop_dir(home);
    let (release_tx, release_rx) = mpsc::channel::<()>();
    let release = Mutex::new(release_rx);
    let _turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |_, _| {
        let _ = release
            .lock()
            .unwrap()
            .recv_timeout(Duration::from_secs(20));
        vec![done("done")]
    });
    release_tx
}

#[test]
fn an_approval_goes_into_the_asking_thread_and_is_marked_when_it_ends() {
    let home = TempDir::new("sa-thread");
    let release = held_loop(&home);
    let mm = FakeMm::start();
    let _running = start(config(&home, &mm.url(), ""));
    let mut ws = mm.next_ws(WAIT);
    let p1 = id('p', 1);
    ws.posted(&post(&p1, KYLE, DM, "", "read my notes", 5), "D");
    let until = std::time::Instant::now() + WAIT;
    let path = home.path().join("gateway/state.json");
    let in_flight = || -> bool {
        let Ok(text) = std::fs::read_to_string(&path) else {
            return false;
        };
        let v: serde_json::Value = serde_json::from_str(&text).unwrap();
        !v["in_flight"].as_array().unwrap().is_empty()
    };
    while !in_flight() {
        assert!(std::time::Instant::now() < until, "the turn never started");
        std::thread::sleep(Duration::from_millis(20));
    }
    let asking = ask(
        &socket(&home),
        pending(42, &format!("mm-{p1}"), r#"{"path":"/n/a"}"#),
    )
    .unwrap();
    let posts = mm.posts();
    assert_eq!(posts.len(), 1, "{posts:?}");
    let (channel, root, text) = &posts[0];
    assert_eq!(
        (channel.as_str(), root.as_str()),
        (DM, p1.as_str()),
        "in the asking thread"
    );
    assert!(text.starts_with("**Approval 42** · tool `read_file` · grant `notes` · taint private · expires 2026-09-24 08:20 UTC\n```\n{\"path\":\"/n/a\"}\n```\n"), "{text}");
    assert_eq!(asking.posts, [FakeMm::post_id(0)]);
    assert_eq!(
        read_state(&home)["approvals"],
        serde_json::json!([{"approval": 42, "channel": DM, "posts": [FakeMm::post_id(0)]}])
    );
    asking.close();
    let edits = mm.wait_edits(1, WAIT);
    assert_eq!(edits.len(), 1, "{edits:?}");
    assert_eq!(edits[0].0, FakeMm::post_id(0));
    assert_eq!(
        edits[0].1,
        format!("{text}\n\nNo longer pending (answered elsewhere, expired, or its turn ended).")
    );
    let until = std::time::Instant::now() + WAIT;
    while !read_state(&home)["approvals"]
        .as_array()
        .unwrap()
        .is_empty()
    {
        assert!(std::time::Instant::now() < until, "still recorded");
        std::thread::sleep(Duration::from_millis(20));
    }
    release.send(()).unwrap();
}

#[test]
fn without_a_turn_in_flight_it_goes_to_the_first_approvers_direct_messages_in_parts() {
    let home = TempDir::new("sa-dm");
    loop_dir(&home);
    let mm = FakeMm::start();
    let _running = start(config(&home, &mm.url(), ""));
    let _ws = mm.next_ws(WAIT);
    let long = format!("{{\"content\":\"{}\"}}", "x".repeat(20_000));
    let asking = ask(&socket(&home), pending(7, "chat-1789700000-1", &long)).unwrap();
    let posts = mm.posts();
    assert_eq!(posts.len(), 3, "a first post, then two parts");
    assert_eq!(
        (posts[0].0.as_str(), posts[0].1.as_str()),
        (DM, ""),
        "a new thread in the direct channel"
    );
    assert!(
        posts[0]
            .2
            .starts_with("**Approval 7** was asked in session `chat-1789700000-1`"),
        "{}",
        posts[0].2
    );
    for (n, part) in posts[1..].iter().enumerate() {
        assert_eq!(
            (part.0.as_str(), part.1.clone()),
            (DM, FakeMm::post_id(0)),
            "part {n} in its thread"
        );
    }
    assert!(posts[1].2.starts_with("**Approval 7**"), "{}", posts[1].2);
    assert!(
        posts[2]
            .2
            .contains("Reply in this thread with `!approve 7`")
    );
    assert_eq!(
        asking.posts,
        [FakeMm::post_id(0), FakeMm::post_id(1), FakeMm::post_id(2)]
    );
    asking.close();
    let edits = mm.wait_edits(1, WAIT);
    assert_eq!(
        edits[0].0,
        FakeMm::post_id(2),
        "only the last part is edited"
    );
}

#[test]
fn an_approval_that_cannot_be_posted_is_refused_to_brokerd() {
    let home = TempDir::new("sa-refused");
    loop_dir(&home);
    let mm = FakeMm::start();
    let _running = start(config(&home, &mm.url(), ""));
    let _ws = mm.next_ws(WAIT);
    mm.refuse_posts(true);
    let why = ask(&socket(&home), pending(8, "chat-1789700000-1", "{}"))
        .err()
        .expect("refused");
    assert!(
        why.starts_with("cannot post approval 8: status 400"),
        "{why}"
    );
    assert!(
        read_state(&home)["approvals"]
            .as_array()
            .unwrap()
            .is_empty()
    );
}

#[test]
fn a_restart_marks_the_posts_it_left_behind() {
    let home = TempDir::new("sa-restart");
    let (a, b) = (id('q', 1), id('q', 2));
    let state = serde_json::json!({
        "channels": {}, "recent": [], "threads": [], "in_flight": [],
        "approvals": [{"approval": 5, "channel": DM, "posts": [a, b]}]
    });
    home.write("gateway/state.json", &state.to_string());
    loop_dir(&home);
    let mm = FakeMm::start();
    let _running = start(config(&home, &mm.url(), ""));
    let _ws = mm.next_ws(WAIT);
    let edits = mm.wait_edits(1, WAIT);
    assert_eq!(edits, [(b, "**Approval 5**\n\nNo longer answerable here: gatewayd restarted. If it is still pending it is posted again.".to_string())]);
    assert!(
        read_state(&home)["approvals"]
            .as_array()
            .unwrap()
            .is_empty()
    );
}

#[test]
fn the_socket_is_there_while_gatewayd_runs() {
    let home = TempDir::new("sa-socket");
    loop_dir(&home);
    let mm = FakeMm::start();
    let running = start(config(&home, &mm.url(), ""));
    let _ws = mm.next_ws(WAIT);
    use std::os::unix::fs::{FileTypeExt, PermissionsExt};
    let meta = std::fs::metadata(socket(&home)).unwrap();
    assert!(meta.file_type().is_socket());
    assert_eq!(meta.permissions().mode() & 0o777, 0o600);
    drop(running);
}