//! Two-archive fanout tests exercising `ArchiveFilter` semantics.
//!
//! The primary backend is unfiltered; the secondary backend carries a
//! restricted filter. Writes should always reach the primary but only
//! fan out to the secondary when the session passes the filter.

#![cfg(feature = "test-utils")]

use std::collections::HashSet;
use std::sync::Arc;

use chrono::Utc;
use uuid::Uuid;

use dirigent_archivist::backend::mock::MockBackend;
use dirigent_archivist::backend::{ArchiveBackend, HealthStatus};
use dirigent_archivist::coordinator::Archivist;
use dirigent_archivist::registry::{
    ArchiveFilter, ArchiveRegistration, FailureMode, WritePolicy,
};
use dirigent_archivist::types::{
    ConnectorRecord, MessageRecord, RegisterSessionRequest,
};

fn reg(
    name: &str,
    backend: Arc<MockBackend>,
    priority: u32,
    filter: ArchiveFilter,
) -> Arc<ArchiveRegistration> {
    Arc::new(
        ArchiveRegistration::new(
            name.into(),
            "mock",
            backend as Arc<dyn ArchiveBackend>,
            /* write_active */ true,
            FailureMode::Required,
            priority,
            /* enabled */ true,
            WritePolicy::Inline,
            /* writer */ None,
            HealthStatus::Healthy,
        )
        .with_filter(filter),
    )
}

/// Seed a connector into a MockBackend directly, bypassing the coordinator.
async fn seed_connector(backend: &MockBackend, connector_uid: Uuid, client_native_id: &str) {
    use dirigent_archivist::backend::ConnectorRegistryBackend;
    let rec = ConnectorRecord {
        version: 1,
        connector_uid,
        r#type: "Mock".into(),
        title: "Mock connector".into(),
        client_native_id: client_native_id.into(),
        alias_of: None,
        created_at: Utc::now(),
        metadata: serde_json::Value::Null,
        fingerprint: None,
    };
    backend
        .put_connector(rec)
        .await
        .expect("put_connector succeeds");
}

fn make_msg(session: Uuid, n: u32) -> MessageRecord {
    MessageRecord {
        version: 1,
        message_id: Uuid::now_v7(),
        session,
        parent_id: None,
        ts: Utc::now(),
        role: "user".into(),
        author: None,
        content_md: format!("msg {}", n),
        content_parts: None,
        attachments: vec![],
        metadata: serde_json::Value::Null,
    }
}

#[tokio::test]
async fn secondary_archive_filters_by_exclude_connector() {
    let primary_backend = Arc::new(MockBackend::new());
    let secondary_backend = Arc::new(MockBackend::new());

    let connector_a = Uuid::now_v7();
    let connector_b = Uuid::now_v7();

    // Connector A is excluded from the secondary.
    let mut excluded = HashSet::new();
    excluded.insert(connector_a);
    let secondary_filter = ArchiveFilter {
        exclude_connectors: excluded,
        ..Default::default()
    };

    // Seed connectors on primary (and on secondary so mapping writes that DO
    // pass the filter don't fail for unrelated reasons).
    seed_connector(&primary_backend, connector_a, "native/a").await;
    seed_connector(&primary_backend, connector_b, "native/b").await;

    let archivist = Archivist::from_registrations(vec![
        reg("primary", primary_backend.clone(), 0, ArchiveFilter::default()),
        reg("secondary", secondary_backend.clone(), 10, secondary_filter),
    ]);

    // Register a session for each connector.
    let resp_a = archivist
        .register_session(
            RegisterSessionRequest {
                connector_uid: connector_a,
                native_session_id: "sess-a".into(),
                title: None,
                custom_scroll_id: None,
                metadata: serde_json::Value::Null,
                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
        .expect("register session a");
    let scroll_a = resp_a.scroll_id;

    let resp_b = archivist
        .register_session(
            RegisterSessionRequest {
                connector_uid: connector_b,
                native_session_id: "sess-b".into(),
                title: None,
                custom_scroll_id: None,
                metadata: serde_json::Value::Null,
                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
        .expect("register session b");
    let scroll_b = resp_b.scroll_id;

    // Append 3 messages to each session.
    archivist
        .append_messages(
            scroll_a,
            vec![make_msg(scroll_a, 1), make_msg(scroll_a, 2), make_msg(scroll_a, 3)],
            None,
        )
        .await
        .expect("append to a");
    archivist
        .append_messages(
            scroll_b,
            vec![make_msg(scroll_b, 1), make_msg(scroll_b, 2), make_msg(scroll_b, 3)],
            None,
        )
        .await
        .expect("append to b");

    // Primary sees every message.
    assert_eq!(primary_backend.appended_count(scroll_a), 3);
    assert_eq!(primary_backend.appended_count(scroll_b), 3);

    // Secondary excludes connector_a: scroll_a is filtered out,
    // scroll_b is replicated.
    assert_eq!(
        secondary_backend.appended_count(scroll_a),
        0,
        "secondary should NOT receive messages for the excluded connector"
    );
    assert_eq!(
        secondary_backend.appended_count(scroll_b),
        3,
        "secondary should receive messages for the allowed connector"
    );

    // Session metadata fanout follows the same rule.
    assert!(
        primary_backend
            .get_session(scroll_a)
            .await
            .unwrap()
            .is_some(),
        "primary has scroll_a"
    );
    assert!(
        secondary_backend
            .get_session(scroll_a)
            .await
            .unwrap()
            .is_none(),
        "secondary should NOT have scroll_a (excluded connector)"
    );
    assert!(
        secondary_backend
            .get_session(scroll_b)
            .await
            .unwrap()
            .is_some(),
        "secondary should have scroll_b (allowed connector)"
    );
}

#[tokio::test]
async fn secondary_archive_filters_by_include_tag() {
    let primary_backend = Arc::new(MockBackend::new());
    let secondary_backend = Arc::new(MockBackend::new());

    let connector = Uuid::now_v7();
    seed_connector(&primary_backend, connector, "native/tagged").await;

    let mut include = HashSet::new();
    include.insert("prod".to_string());
    let secondary_filter = ArchiveFilter {
        include_tags: include,
        ..Default::default()
    };

    let archivist = Archivist::from_registrations(vec![
        reg("primary", primary_backend.clone(), 0, ArchiveFilter::default()),
        reg("secondary", secondary_backend.clone(), 10, secondary_filter),
    ]);

    // Register two sessions on the same connector.
    let prod_resp = archivist
        .register_session(
            RegisterSessionRequest {
                connector_uid: connector,
                native_session_id: "sess-prod".into(),
                title: None,
                custom_scroll_id: None,
                metadata: serde_json::Value::Null,
                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
        .expect("register prod session");
    let scroll_prod = prod_resp.scroll_id;

    let dev_resp = archivist
        .register_session(
            RegisterSessionRequest {
                connector_uid: connector,
                native_session_id: "sess-dev".into(),
                title: None,
                custom_scroll_id: None,
                metadata: serde_json::Value::Null,
                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
        .expect("register dev session");
    let scroll_dev = dev_resp.scroll_id;

    // Tag the prod session directly on the primary so the coordinator can
    // see it on the next fanout metadata lookup. We mutate via the primary
    // backend to avoid going through update_session_metadata (which doesn't
    // expose a tag API).
    {
        use dirigent_archivist::backend::ArchiveBackend as _;
        let mut md = primary_backend
            .get_session(scroll_prod)
            .await
            .unwrap()
            .expect("prod session on primary");
        md.tags.push("prod".into());
        primary_backend.put_session(md).await.unwrap();
    }

    // Append messages AFTER tagging — now the filter consults the tagged metadata.
    archivist
        .append_messages(
            scroll_prod,
            vec![
                make_msg(scroll_prod, 1),
                make_msg(scroll_prod, 2),
                make_msg(scroll_prod, 3),
            ],
            None,
        )
        .await
        .expect("append prod");
    archivist
        .append_messages(
            scroll_dev,
            vec![make_msg(scroll_dev, 1), make_msg(scroll_dev, 2), make_msg(scroll_dev, 3)],
            None,
        )
        .await
        .expect("append dev");

    // Primary keeps both.
    assert_eq!(primary_backend.appended_count(scroll_prod), 3);
    assert_eq!(primary_backend.appended_count(scroll_dev), 3);

    // Secondary only keeps the tagged session.
    assert_eq!(
        secondary_backend.appended_count(scroll_prod),
        3,
        "secondary receives messages for the `prod`-tagged session"
    );
    assert_eq!(
        secondary_backend.appended_count(scroll_dev),
        0,
        "secondary rejects the untagged session"
    );
}
