//! Tests for the chat message store: backlog replay, the ban purge, and both
//! halves of retention.
//!
//! These drive `mt_db` directly rather than going through routes, because the
//! routes do not exist yet and because what is worth proving here is the SQL:
//! the cursor semantics the reconnect path depends on, and the two sweep
//! statements, which are the ones a unit test over a mock cannot check.

use mt_db::{mutations, queries};
use uuid::Uuid;

use crate::harness::TestHarness;

/// Default retention, matching `livechat::Retention::forum_default()` and the
/// column default in migration 039.
const FORUM_HOURS: i32 = 168;

struct Room {
    id: Uuid,
    alice: Uuid,
    bob: Uuid,
}

async fn room(h: &mut TestHarness) -> Room {
    let alice = h.login_as("alice").await;
    let id = h.create_community("Test", "test").await;
    h.add_membership(alice, id, "owner").await;
    let bob = h.login_as("bob").await;
    h.add_membership(bob, id, "member").await;
    Room { id, alice, bob }
}

async fn say(h: &TestHarness, room: &Room, author: Uuid, body: &str) -> i64 {
    mutations::insert_chat_message(&h.db, room.id, author, body, FORUM_HOURS)
        .await
        .expect("insert")
        .id
}

/// Everything currently in the room, oldest first.
async fn all(h: &TestHarness, room: &Room) -> Vec<String> {
    queries::recent_backlog(&h.db, room.id, 1_000)
        .await
        .expect("backlog")
        .into_iter()
        .map(|m| m.body_html)
        .collect()
}

#[sqlx::test]
async fn ids_are_monotonic_and_the_backlog_reads_oldest_first(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let first = say(&h, &r, r.alice, "one").await;
    let second = say(&h, &r, r.bob, "two").await;
    let third = say(&h, &r, r.alice, "three").await;

    assert!(first < second && second < third, "ids order the room");
    assert_eq!(all(&h, &r).await, vec!["one", "two", "three"]);
}

#[sqlx::test]
async fn a_cursor_replays_only_what_came_after_it(_pool: sqlx::PgPool) {
    // The reconnect path: every deploy drops every connection, and the client
    // resumes from the last id it saw.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    say(&h, &r, r.alice, "before").await;
    let cursor = say(&h, &r, r.alice, "last seen").await;
    say(&h, &r, r.bob, "missed").await;
    say(&h, &r, r.bob, "also missed").await;

    let replay: Vec<_> = queries::backlog_after(&h.db, r.id, cursor, 100)
        .await
        .unwrap()
        .into_iter()
        .map(|m| m.body_html)
        .collect();

    assert_eq!(replay, vec!["missed", "also missed"]);
}

#[sqlx::test]
async fn a_long_absence_is_capped_and_resumable(_pool: sqlx::PgPool) {
    // A client away longer than the window must not be able to make the server
    // materialize the whole room in one reply. It gets the oldest slice past
    // its cursor and walks forward.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    for i in 0..10 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    let first = queries::backlog_after(&h.db, r.id, 0, 4).await.unwrap();
    assert_eq!(first.len(), 4);
    assert_eq!(first[0].body_html, "m0", "oldest first, not newest");

    let next = queries::backlog_after(&h.db, r.id, first[3].id, 4)
        .await
        .unwrap();
    assert_eq!(next[0].body_html, "m4", "the cursor advances without a gap");
}

#[sqlx::test]
async fn a_fresh_connection_gets_the_newest_window_in_reading_order(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    for i in 0..10 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    let window: Vec<_> = queries::recent_backlog(&h.db, r.id, 3)
        .await
        .unwrap()
        .into_iter()
        .map(|m| m.body_html)
        .collect();

    assert_eq!(window, vec!["m7", "m8", "m9"], "newest three, ascending");
}

#[sqlx::test]
async fn rooms_do_not_leak_into_each_other(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let other = h.create_community("Other", "other").await;
    h.add_membership(r.alice, other, "owner").await;

    say(&h, &r, r.alice, "in test").await;
    mutations::insert_chat_message(&h.db, other, r.alice, "in other", FORUM_HOURS)
        .await
        .unwrap();

    assert_eq!(all(&h, &r).await, vec!["in test"]);
}

#[sqlx::test]
async fn the_author_lookup_is_scoped_to_the_room(_pool: sqlx::PgPool) {
    // A message id is a guessable integer. Asking who wrote one without saying
    // which room it is in would confirm authorship across a community the
    // caller cannot read.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let other = h.create_community("Other", "other").await;
    h.add_membership(r.alice, other, "owner").await;
    let elsewhere = mutations::insert_chat_message(&h.db, other, r.alice, "secret", FORUM_HOURS)
        .await
        .unwrap()
        .id;

    let mine = say(&h, &r, r.bob, "mine").await;

    assert_eq!(
        queries::chat_message_author(&h.db, r.id, mine)
            .await
            .unwrap(),
        Some(r.bob)
    );
    assert_eq!(
        queries::chat_message_author(&h.db, r.id, elsewhere)
            .await
            .unwrap(),
        None,
        "a message in another room is indistinguishable from absent"
    );
}

#[sqlx::test]
async fn a_purge_removes_one_author_and_leaves_the_room(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    say(&h, &r, r.alice, "alice one").await;
    say(&h, &r, r.bob, "bob spam").await;
    say(&h, &r, r.alice, "alice two").await;
    say(&h, &r, r.bob, "bob spam again").await;

    let removed = mutations::purge_author_messages(&h.db, r.id, r.bob)
        .await
        .unwrap();

    assert_eq!(removed, 2);
    assert_eq!(all(&h, &r).await, vec!["alice one", "alice two"]);
}

#[sqlx::test]
async fn a_purge_does_not_reach_the_same_author_in_another_room(_pool: sqlx::PgPool) {
    // Banning someone from one community must not erase them from another.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let other = h.create_community("Other", "other").await;
    h.add_membership(r.bob, other, "member").await;
    mutations::insert_chat_message(&h.db, other, r.bob, "innocent", FORUM_HOURS)
        .await
        .unwrap();

    say(&h, &r, r.bob, "spam").await;
    mutations::purge_author_messages(&h.db, r.id, r.bob)
        .await
        .unwrap();

    let survivors = queries::recent_backlog(&h.db, other, 100).await.unwrap();
    assert_eq!(survivors.len(), 1, "the other room is untouched");
}

#[sqlx::test]
async fn deleting_one_message_reports_whether_it_was_there(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let id = say(&h, &r, r.alice, "oops").await;

    assert_eq!(
        mutations::delete_chat_message(&h.db, r.id, id)
            .await
            .unwrap(),
        1
    );
    assert_eq!(
        mutations::delete_chat_message(&h.db, r.id, id)
            .await
            .unwrap(),
        0,
        "a second delete is a no-op, not an error"
    );
    assert!(all(&h, &r).await.is_empty());
}

#[sqlx::test]
async fn the_sweep_takes_expired_messages_and_spares_live_ones(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    say(&h, &r, r.alice, "fresh").await;
    let stale = say(&h, &r, r.alice, "stale").await;

    // Expire one row directly. Reaching past the insert path is the point:
    // waiting out a real retention window is not a test.
    sqlx::query("UPDATE chat_messages SET expires_at = now() - interval '1 hour' WHERE id = $1")
        .bind(stale)
        .execute(&h.db)
        .await
        .unwrap();

    let swept = mutations::sweep_expired_chat_messages(&h.db).await.unwrap();

    assert_eq!(swept, 1);
    assert_eq!(all(&h, &r).await, vec!["fresh"]);
}

#[sqlx::test]
async fn shortening_retention_applies_to_messages_already_sent(_pool: sqlx::PgPool) {
    // The cost of stamping expiry at insert: a shortened window has to reach
    // back, or an owner who just cut retention from 30 days to 1 would still be
    // holding 30 days of chat.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    // Sent under a long policy, and old enough that a short one expires it.
    let old = say(&h, &r, r.alice, "old").await;
    sqlx::query(
        "UPDATE chat_messages
         SET created_at = now() - interval '48 hours',
             expires_at = now() + interval '600 hours'
         WHERE id = $1",
    )
    .bind(old)
    .execute(&h.db)
    .await
    .unwrap();

    say(&h, &r, r.alice, "recent").await;

    // Owner cuts the window to 24 hours.
    let restamped = mutations::recompute_chat_expiry(&h.db, r.id, 24)
        .await
        .unwrap();
    assert_eq!(restamped, 2, "every message in the room is restamped");

    let swept = mutations::sweep_expired_chat_messages(&h.db).await.unwrap();
    assert_eq!(swept, 1, "the 48-hour-old message is now past a 24h window");
    assert_eq!(all(&h, &r).await, vec!["recent"]);
}

#[sqlx::test]
async fn recompute_is_measured_from_send_time_not_from_now(_pool: sqlx::PgPool) {
    // Restamping from now() would silently extend every old message's life by
    // the full window each time an owner touched the setting.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let old = say(&h, &r, r.alice, "old").await;
    sqlx::query("UPDATE chat_messages SET created_at = now() - interval '10 hours' WHERE id = $1")
        .bind(old)
        .execute(&h.db)
        .await
        .unwrap();

    mutations::recompute_chat_expiry(&h.db, r.id, 6)
        .await
        .unwrap();

    assert_eq!(
        mutations::sweep_expired_chat_messages(&h.db).await.unwrap(),
        1,
        "a 10-hour-old message under a 6-hour window is already expired"
    );
}

#[sqlx::test]
async fn the_count_cap_trims_the_oldest_and_keeps_the_newest(_pool: sqlx::PgPool) {
    // The other half of retention: age expires a quiet room, the cap bounds a
    // busy one that would reach its age limit holding far more.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    sqlx::query("UPDATE communities SET chat_max_messages = 3 WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .unwrap();

    for i in 0..10 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    let trimmed = mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap();

    assert_eq!(trimmed, 7);
    assert_eq!(all(&h, &r).await, vec!["m7", "m8", "m9"]);
}

#[sqlx::test]
async fn the_cap_is_per_room_not_global(_pool: sqlx::PgPool) {
    // One statement trims the whole table, so the partition boundary is the
    // thing that can be wrong: a room under its own cap must not lose messages
    // because another room is over.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    let quiet = h.create_community("Quiet", "quiet").await;
    h.add_membership(r.alice, quiet, "owner").await;

    sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .unwrap();

    for i in 0..5 {
        say(&h, &r, r.alice, &format!("loud{i}")).await;
    }
    for i in 0..2 {
        mutations::insert_chat_message(&h.db, quiet, r.alice, &format!("q{i}"), FORUM_HOURS)
            .await
            .unwrap();
    }

    assert_eq!(mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap(), 3);
    assert_eq!(all(&h, &r).await, vec!["loud3", "loud4"]);
    assert_eq!(
        queries::recent_backlog(&h.db, quiet, 100)
            .await
            .unwrap()
            .len(),
        2,
        "the room under its cap kept everything"
    );
}

#[sqlx::test]
async fn a_room_exactly_at_its_cap_loses_nothing(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    sqlx::query("UPDATE communities SET chat_max_messages = 3 WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .unwrap();

    for i in 0..3 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    assert_eq!(
        mutations::trim_chat_rooms_to_cap(&h.db).await.unwrap(),
        0,
        "the cap is inclusive"
    );
    assert_eq!(all(&h, &r).await.len(), 3);
}

#[sqlx::test]
async fn deleting_a_community_takes_its_chat_with_it(_pool: sqlx::PgPool) {
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;
    say(&h, &r, r.alice, "hello").await;

    sqlx::query("DELETE FROM communities WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .expect("the cascade must not be blocked by a chat row");

    let left: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM chat_messages WHERE community_id = $1")
            .bind(r.id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(left, 0);
}

#[sqlx::test]
async fn retention_columns_refuse_values_past_the_crate_ceilings(_pool: sqlx::PgPool) {
    // The schema restates `livechat`'s ceilings so a bad UPDATE fails at the
    // database rather than at whichever call site forgot to validate.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    for (column, value) in [
        ("chat_retention_hours", 721),
        ("chat_retention_hours", 0),
        ("chat_max_messages", 20_001),
        ("chat_max_messages", 0),
    ] {
        let result = sqlx::query(&format!(
            "UPDATE communities SET {column} = $1 WHERE id = $2"
        ))
        .bind(value)
        .bind(r.id)
        .execute(&h.db)
        .await;
        assert!(result.is_err(), "{column} = {value} must be refused");
    }
}

// The scheduled sweep

#[sqlx::test]
async fn one_sweep_round_enforces_both_halves_of_retention(_pool: sqlx::PgPool) {
    // The two statements answer different questions and both must run every
    // round. A room busy enough to be over its cap is usually one whose
    // messages are all too new to have expired, so gating the trim on the
    // expiry sweep finding something would skip it exactly when it is needed.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .unwrap();

    // Five fresh messages: nothing is expired, and three are over the cap.
    for i in 0..5 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    let (expired, trimmed) = multithreaded::maintenance::sweep_chat_once(&h.db).await;

    assert_eq!(expired, 0, "nothing was old enough to expire");
    assert_eq!(trimmed, 3, "the cap still bit");
    assert_eq!(all(&h, &r).await, vec!["m3", "m4"]);
}

#[sqlx::test]
async fn a_sweep_round_over_an_empty_table_is_a_no_op(_pool: sqlx::PgPool) {
    let h = TestHarness::new().await;
    assert_eq!(
        multithreaded::maintenance::sweep_chat_once(&h.db).await,
        (0, 0)
    );
}

#[sqlx::test]
async fn the_sweep_is_convergent(_pool: sqlx::PgPool) {
    // Steady-state work is zero: a second round immediately after the first
    // must find nothing, or the sweep would churn the same rows every interval.
    let mut h = TestHarness::new().await;
    let r = room(&mut h).await;

    sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
        .bind(r.id)
        .execute(&h.db)
        .await
        .unwrap();
    for i in 0..5 {
        say(&h, &r, r.alice, &format!("m{i}")).await;
    }

    multithreaded::maintenance::sweep_chat_once(&h.db).await;
    assert_eq!(
        multithreaded::maintenance::sweep_chat_once(&h.db).await,
        (0, 0),
        "the second round must find nothing left to do"
    );
}