//! Tests for `Archivist::list_sessions_paged` — pagination, filters, cursor stability.

use chrono::{Duration, Utc};
use dirigent_archivist::{
    backends::JsonlBackend, Archivist, RegisterConnectorRequest, RegisterSessionRequest,
    Result, SessionListQuery,
};
use std::sync::Arc;
use uuid::Uuid;

/// Scaffold: create a coordinator backed by a single `JsonlBackend` in a
/// unique temp dir, returning the backend alongside it so tests can probe
/// disk paths via `backend.paths()`.
async fn mk_archivist() -> Result<(Archivist, Arc<JsonlBackend>, std::path::PathBuf)> {
    let temp_dir = std::env::temp_dir().join(format!("paged_test_{}", Uuid::now_v7()));
    let backend = Arc::new(JsonlBackend::new(temp_dir.clone()).await?);
    let archivist =
        Archivist::from_single_backend("main".into(), backend.clone()).await?;
    Ok((archivist, backend, temp_dir))
}

/// Register a connector, return its UID.
async fn mk_connector(archivist: &Archivist, title: &str) -> Result<Uuid> {
    let resp = archivist
        .register_connector(
            RegisterConnectorRequest {
                r#type: "OpenCode".to_string(),
                title: title.to_string(),
                client_native_id: format!("{title}@local:{}", Uuid::now_v7()),
                custom_uid: None,
                metadata: serde_json::json!({}),
                fingerprint: None,
            },
            None,
        )
        .await?;
    Ok(resp.connector_uid)
}

/// Register a session and patch fields that `register_session` does not expose.
#[allow(clippy::too_many_arguments)]
async fn mk_session(
    archivist: &Archivist,
    backend: &JsonlBackend,
    connector_uid: Uuid,
    native_id: &str,
    title: Option<&str>,
    tags: Vec<String>,
    model: Option<&str>,
    project_id: Option<&str>,
    no_update: bool,
) -> Result<Uuid> {
    let mut metadata = serde_json::Map::new();
    if let Some(m) = model {
        metadata.insert("model".to_string(), serde_json::Value::String(m.to_string()));
    }
    if let Some(p) = project_id {
        metadata.insert(
            "project_id".to_string(),
            serde_json::Value::String(p.to_string()),
        );
    }

    let resp = archivist
        .register_session(
            RegisterSessionRequest {
                connector_uid,
                native_session_id: native_id.to_string(),
                title: title.map(String::from),
                custom_scroll_id: None,
                metadata: serde_json::Value::Object(metadata),
                completeness: Default::default(),
                parent_scroll_id: None,
                is_subagent: false,
                continuation: None,
                agent_id: None,
                subagent_type: None,
                spawning_tool_use_id: None,
            },
            None,
        )
        .await?;
    let scroll_id = resp.scroll_id;

    // Patch tags / no_update into session.json on disk.
    if !tags.is_empty() || no_update {
        let mut meta = archivist.get_session_metadata(scroll_id, None).await?;
        meta.tags = tags;
        meta.no_update = no_update;
        let path = backend.paths().session_json(scroll_id);
        dirigent_archivist::storage::json::write_json(&path, &meta)
            .await
            .map_err(dirigent_archivist::ArchivistError::Io)?;
    }

    Ok(scroll_id)
}

/// Overwrite a session's updated_at on disk for deterministic ordering.
async fn set_updated_at(
    archivist: &Archivist,
    backend: &JsonlBackend,
    scroll_id: Uuid,
    when: chrono::DateTime<chrono::Utc>,
) -> Result<()> {
    let mut meta = archivist.get_session_metadata(scroll_id, None).await?;
    meta.updated_at = when;
    let path = backend.paths().session_json(scroll_id);
    dirigent_archivist::storage::json::write_json(&path, &meta)
        .await
        .map_err(dirigent_archivist::ArchivistError::Io)?;
    Ok(())
}

fn cleanup(path: std::path::PathBuf) {
    let _ = std::fs::remove_dir_all(path);
}

#[tokio::test]
async fn list_sessions_paged_respects_limit() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-a").await?;

    let base = Utc::now();
    for i in 0..30 {
        let scroll = mk_session(
            &archivist,
            &backend,
            uid,
            &format!("native-{i}"),
            Some(&format!("title-{i}")),
            Vec::new(),
            None,
            None,
            false,
        )
        .await?;
        set_updated_at(&archivist, &backend, scroll, base - Duration::seconds(i)).await?;
    }

    let page = archivist
        .list_sessions_paged(SessionListQuery::default().with_connector(uid).with_limit(10))
        .await?;

    assert_eq!(page.items.len(), 10);
    assert!(page.next_cursor.is_some());

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_end_of_list() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-b").await?;

    let base = Utc::now();
    for i in 0..5 {
        let scroll = mk_session(
            &archivist,
            &backend,
            uid,
            &format!("native-{i}"),
            Some(&format!("title-{i}")),
            Vec::new(),
            None,
            None,
            false,
        )
        .await?;
        set_updated_at(&archivist, &backend, scroll, base - Duration::seconds(i)).await?;
    }

    let page = archivist
        .list_sessions_paged(SessionListQuery::default().with_connector(uid).with_limit(100))
        .await?;

    assert_eq!(page.items.len(), 5);
    assert!(page.next_cursor.is_none());

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_cursor_stability() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-c").await?;

    let fixed = Utc::now();
    for i in 0..6 {
        let scroll = mk_session(
            &archivist,
            &backend,
            uid,
            &format!("native-{i}"),
            Some(&format!("title-{i}")),
            Vec::new(),
            None,
            None,
            false,
        )
        .await?;
        set_updated_at(&archivist, &backend, scroll, fixed).await?;
    }

    let p1 = archivist
        .list_sessions_paged(SessionListQuery::default().with_connector(uid).with_limit(3))
        .await?;
    assert_eq!(p1.items.len(), 3);
    assert!(p1.next_cursor.is_some());

    let p2 = archivist
        .list_sessions_paged(
            SessionListQuery::default()
                .with_connector(uid)
                .with_limit(3)
                .with_cursor(p1.next_cursor.clone()),
        )
        .await?;
    assert_eq!(p2.items.len(), 3);

    let ids1: std::collections::HashSet<_> = p1.items.iter().map(|s| s.scroll_id).collect();
    let ids2: std::collections::HashSet<_> = p2.items.iter().map(|s| s.scroll_id).collect();
    assert!(ids1.is_disjoint(&ids2), "page 1 and page 2 must not overlap");
    assert!(p2.next_cursor.is_none());

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_title_filter() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-d").await?;

    mk_session(&archivist, &backend, uid, "n1", Some("Alpha beta"), vec![], None, None, false).await?;
    mk_session(&archivist, &backend, uid, "n2", Some("BETA only"), vec![], None, None, false).await?;
    mk_session(&archivist, &backend, uid, "n3", Some("gamma"), vec![], None, None, false).await?;
    mk_session(&archivist, &backend, uid, "n4", None, vec![], None, None, false).await?;

    let page = archivist
        .list_sessions_paged(
            SessionListQuery::default()
                .with_connector(uid)
                .with_limit(50)
                .with_title_query("beta"),
        )
        .await?;

    let titles: Vec<_> = page.items.iter().filter_map(|s| s.title.clone()).collect();
    assert_eq!(titles.len(), 2, "got {titles:?}");
    assert!(titles.iter().any(|t| t == "Alpha beta"));
    assert!(titles.iter().any(|t| t == "BETA only"));

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_tags_and() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-e").await?;

    mk_session(
        &archivist, &backend, uid, "n1", Some("s1"),
        vec!["red".into(), "blue".into()], None, None, false,
    ).await?;
    mk_session(
        &archivist, &backend, uid, "n2", Some("s2"),
        vec!["red".into()], None, None, false,
    ).await?;
    mk_session(
        &archivist, &backend, uid, "n3", Some("s3"),
        vec!["blue".into()], None, None, false,
    ).await?;

    let mut q = SessionListQuery::default().with_connector(uid).with_limit(50);
    q.tags = vec!["red".into(), "blue".into()];

    let page = archivist.list_sessions_paged(q).await?;

    assert_eq!(page.items.len(), 1);
    assert_eq!(page.items[0].title.as_deref(), Some("s1"));

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_model_filter() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-f").await?;

    mk_session(&archivist, &backend, uid, "n1", Some("s1"), vec![], Some("claude-3-5-sonnet"), None, false).await?;
    mk_session(&archivist, &backend, uid, "n2", Some("s2"), vec![], Some("gpt-4o"), None, false).await?;
    mk_session(&archivist, &backend, uid, "n3", Some("s3"), vec![], None, None, false).await?;

    let mut q = SessionListQuery::default().with_connector(uid).with_limit(50);
    q.model_filter = Some("sonnet".into());

    let page = archivist.list_sessions_paged(q).await?;

    assert_eq!(page.items.len(), 1);
    assert_eq!(page.items[0].title.as_deref(), Some("s1"));

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_include_hidden() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let uid = mk_connector(&archivist, "connector-g").await?;

    mk_session(&archivist, &backend, uid, "n1", Some("visible"), vec![], None, None, false).await?;
    mk_session(&archivist, &backend, uid, "n2", Some("hidden"),  vec![], None, None, true).await?;

    let visible_only = archivist
        .list_sessions_paged(SessionListQuery::default().with_connector(uid).with_limit(50))
        .await?;
    assert_eq!(visible_only.items.len(), 1);
    assert_eq!(visible_only.items[0].title.as_deref(), Some("visible"));

    let all = archivist
        .list_sessions_paged(
            SessionListQuery::default()
                .with_connector(uid)
                .with_limit(50)
                .with_include_hidden(true),
        )
        .await?;
    assert_eq!(all.items.len(), 2);

    cleanup(temp);
    Ok(())
}

#[tokio::test]
async fn list_sessions_paged_project_scope() -> Result<()> {
    let (archivist, backend, temp) = mk_archivist().await?;
    let c1 = mk_connector(&archivist, "connector-h1").await?;
    let c2 = mk_connector(&archivist, "connector-h2").await?;

    mk_session(&archivist, &backend, c1, "n1", Some("proj-a-1"), vec![], None, Some("proj-a"), false).await?;
    mk_session(&archivist, &backend, c1, "n2", Some("proj-b-1"), vec![], None, Some("proj-b"), false).await?;
    mk_session(&archivist, &backend, c2, "n3", Some("proj-a-2"), vec![], None, Some("proj-a"), false).await?;

    let page = archivist
        .list_sessions_paged(
            SessionListQuery::default()
                .with_project("proj-a")
                .with_limit(50),
        )
        .await?;

    assert_eq!(page.items.len(), 2);
    for s in &page.items {
        assert_eq!(s.metadata.get("project_id").and_then(|v| v.as_str()), Some("proj-a"));
    }

    cleanup(temp);
    Ok(())
}
