mod common;

use std::sync::Arc;
use std::time::Duration;

use fafb_q::config::{
    CompiledInactiveOverride, RetentionPolicy, RuntimeConfig, ServerConfig, SharedConfig,
    TopicInactiveOverride,
};
use fafb_q::namespace::{DefaultNamespaceManager, ManagerConfig, NamespaceManager};
use fafb_q::queue::{EngineConfig, InMemoryQueueEngine, QueueEngine};
use fafb_q::retention::RetentionTask;
use fafb_q::server::state::ServerLifecycle;
use fafb_q::storage::mock::MockStorage;
use fafb_q::types::{
    ClientId, MessageEnvelope, MessageId, NamespaceId, SubscribeStart, TopicName, now_millis,
};
use fafb_q::wildcard::TopicPattern;

const NS: &str = "test-ns";

fn ns() -> NamespaceId {
    NamespaceId(NS.into())
}

fn make_envelope(id: u64, topic: &str, payload: &[u8]) -> MessageEnvelope {
    MessageEnvelope::new(
        MessageId(id + 1),
        NamespaceId(NS.into()),
        TopicName(topic.into()),
        payload.to_vec(),
        now_millis(),
    )
}

/// Build a SharedConfig with topic inactivity GC enabled.
///
/// - `topic_inactive_secs`: seconds of inactivity before a topic is eligible for reaping
/// - `grace_period_secs`: seconds to wait in grace period before hard delete
/// - `sweep_interval_secs`: how often the retention task runs
/// - `overrides`: optional per-topic pattern overrides
fn make_gc_shared_config(
    topic_inactive_secs: u64,
    grace_period_secs: u64,
    sweep_interval_secs: u64,
    overrides: Vec<TopicInactiveOverride>,
) -> SharedConfig {
    let mut rt = RuntimeConfig::from_server_config(&ServerConfig::default());

    let mut retention = RetentionPolicy::default();
    retention.topic_inactive_secs = Some(topic_inactive_secs);
    retention.topic_reap_grace_period_secs = Some(grace_period_secs);
    retention.topic_inactive_overrides = overrides.clone();

    rt.namespace_retention
        .insert(NamespaceId(NS.into()), retention);
    rt.retention_check_interval_secs = sweep_interval_secs;

    // Compile overrides for this namespace.
    let compiled: Vec<CompiledInactiveOverride> = overrides
        .iter()
        .map(|ov| CompiledInactiveOverride {
            pattern: TopicPattern::parse(&ov.pattern).expect("invalid pattern"),
            inactive_secs: ov.inactive_secs,
        })
        .collect();
    if !compiled.is_empty() {
        rt.compiled_inactive_overrides
            .insert(NamespaceId(NS.into()), compiled);
    }

    Arc::new(arc_swap::ArcSwap::from_pointee(rt))
}

/// Build engine + manager + storage for GC tests.
fn make_components(
    shared_cfg: SharedConfig,
) -> (
    Arc<InMemoryQueueEngine<MockStorage>>,
    Arc<DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>>,
    Arc<MockStorage>,
) {
    let storage = Arc::new(MockStorage::new());
    let engine = Arc::new(InMemoryQueueEngine::new(
        Arc::clone(&storage),
        EngineConfig {
            max_depth: 100_000,
            ..Default::default()
        },
    ));
    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![ns()],
        shared_cfg,
        ManagerConfig::default(),
    ));
    (engine, manager, storage)
}

/// Spawn a RetentionTask in the background and return a handle to shut it down.
fn spawn_retention_task(
    storage: Arc<MockStorage>,
    manager: Arc<DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>>,
    engine: Arc<InMemoryQueueEngine<MockStorage>>,
    shared_cfg: SharedConfig,
) -> Arc<ServerLifecycle> {
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);

    let shutdown_rx = lifecycle.subscribe();
    let task = RetentionTask::new(storage, manager, engine, shared_cfg);
    tokio::spawn(task.run(shutdown_rx));

    lifecycle
}

/// Publish a single message to a topic through the namespace manager.
async fn publish_one(
    manager: &DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>,
    topic: &str,
    msg_id: u64,
) {
    manager
        .publish(
            &ns(),
            &TopicName(topic.into()),
            make_envelope(msg_id, topic, b"data"),
        )
        .await
        .expect("publish should succeed");
}

/// Check whether a topic exists in the engine's in-memory map.
async fn topic_exists(engine: &InMemoryQueueEngine<MockStorage>, topic: &str) -> bool {
    let topics = engine.list_topics(&ns()).await;
    topics.contains(&TopicName(topic.into()))
}

// ---------------------------------------------------------------------------
// Test 1: Topic is reaped after inactivity threshold + grace period elapse
// ---------------------------------------------------------------------------

#[tokio::test]
async fn topic_reaped_after_inactivity_and_grace() {
    let shared_cfg = make_gc_shared_config(
        2, // topic_inactive_secs
        1, // grace_period_secs
        1, // sweep every second
        vec![],
    );

    let (engine, manager, storage) = make_components(Arc::clone(&shared_cfg));

    // Publish a message to create the topic.
    publish_one(&manager, "ephemeral.data", 1).await;
    assert!(
        topic_exists(&engine, "ephemeral.data").await,
        "topic should exist after publish"
    );

    // Start the retention task.
    let lifecycle = spawn_retention_task(
        Arc::clone(&storage),
        Arc::clone(&manager),
        Arc::clone(&engine),
        shared_cfg,
    );

    // Wait for inactivity (2s) + grace (1s) + sweep margin (~3s for sweep intervals).
    // The sweep runs every 1s but skips the first tick, so:
    //   t=1s: first sweep, topic is fresh (<2s inactive), no action
    //   t=2s: sweep, topic is ~2s inactive, enters grace period
    //   t=3s: sweep, grace elapsed (1s), topic is reaped
    // Use 6s total to be safe with timing jitter.
    tokio::time::sleep(Duration::from_secs(6)).await;

    assert!(
        !topic_exists(&engine, "ephemeral.data").await,
        "topic should have been reaped after inactivity + grace"
    );

    lifecycle.transition_to(fafb_q::server::state::ServerState::ShuttingDown);
}

// ---------------------------------------------------------------------------
// Test 2: Publishing resets the inactivity clock
// ---------------------------------------------------------------------------

#[tokio::test]
async fn publish_resets_inactivity_clock() {
    let shared_cfg = make_gc_shared_config(
        2, // topic_inactive_secs
        1, // grace_period_secs
        1, // sweep every second
        vec![],
    );

    let (engine, manager, storage) = make_components(Arc::clone(&shared_cfg));

    // Publish to create the topic.
    publish_one(&manager, "active.topic", 1).await;

    let lifecycle = spawn_retention_task(
        Arc::clone(&storage),
        Arc::clone(&manager),
        Arc::clone(&engine),
        shared_cfg,
    );

    // Wait 1.5s (less than the 2s threshold), then publish again to reset clock.
    tokio::time::sleep(Duration::from_millis(1500)).await;
    publish_one(&manager, "active.topic", 2).await;

    // Wait another 1.5s — total elapsed since last publish is only 1.5s (< 2s threshold).
    tokio::time::sleep(Duration::from_millis(1500)).await;

    assert!(
        topic_exists(&engine, "active.topic").await,
        "topic should still exist because publish reset the inactivity clock"
    );

    // Now stop publishing and wait for full inactivity + grace + sweep margin.
    tokio::time::sleep(Duration::from_secs(6)).await;

    assert!(
        !topic_exists(&engine, "active.topic").await,
        "topic should be reaped after inactivity resumed"
    );

    lifecycle.transition_to(fafb_q::server::state::ServerState::ShuttingDown);
}

// ---------------------------------------------------------------------------
// Test 3: Active subscriber blocks reaping
// ---------------------------------------------------------------------------

#[tokio::test]
async fn active_subscriber_blocks_reap() {
    let shared_cfg = make_gc_shared_config(
        2, // topic_inactive_secs
        1, // grace_period_secs
        1, // sweep every second
        vec![],
    );

    let (engine, manager, storage) = make_components(Arc::clone(&shared_cfg));

    // Publish to create the topic.
    publish_one(&manager, "subscribed.topic", 1).await;

    // Subscribe a client to the topic (starts at head, which is fine for this test).
    let client_id = ClientId(42);
    let sub_result = manager
        .subscribe(
            &ns(),
            &TopicName("subscribed.topic".into()),
            client_id,
            None,
            SubscribeStart::Head,
        )
        .await
        .expect("subscribe should succeed");
    // Keep the receiver alive — dropping it would remove the subscriber.
    let _rx = sub_result.rx;

    let lifecycle = spawn_retention_task(
        Arc::clone(&storage),
        Arc::clone(&manager),
        Arc::clone(&engine),
        shared_cfg,
    );

    // Wait well past the inactivity threshold + grace period.
    tokio::time::sleep(Duration::from_secs(6)).await;

    assert!(
        topic_exists(&engine, "subscribed.topic").await,
        "topic should NOT be reaped while a subscriber is active"
    );

    // Unsubscribe and verify the topic eventually gets reaped.
    manager
        .unsubscribe(
            &ns(),
            &TopicName("subscribed.topic".into()),
            client_id,
            None,
        )
        .await
        .expect("unsubscribe should succeed");

    // Wait for inactivity + grace + sweep to catch up.
    tokio::time::sleep(Duration::from_secs(6)).await;

    assert!(
        !topic_exists(&engine, "subscribed.topic").await,
        "topic should be reaped after subscriber disconnected and inactivity elapsed"
    );

    lifecycle.transition_to(fafb_q::server::state::ServerState::ShuttingDown);
}

// ---------------------------------------------------------------------------
// Test 4: Override with inactive_secs=0 exempts topic from reaping
// ---------------------------------------------------------------------------

#[tokio::test]
async fn override_exempts_topic_from_reaping() {
    let shared_cfg = make_gc_shared_config(
        2, // topic_inactive_secs (namespace default)
        1, // grace_period_secs
        1, // sweep every second
        vec![TopicInactiveOverride {
            pattern: "system.>".to_string(),
            inactive_secs: 0, // 0 = never reap
        }],
    );

    let (engine, manager, storage) = make_components(Arc::clone(&shared_cfg));

    // Publish to both a system topic (exempt) and a regular topic (should be reaped).
    publish_one(&manager, "system.logs", 1).await;
    publish_one(&manager, "regular.events", 2).await;

    assert!(topic_exists(&engine, "system.logs").await);
    assert!(topic_exists(&engine, "regular.events").await);

    let lifecycle = spawn_retention_task(
        Arc::clone(&storage),
        Arc::clone(&manager),
        Arc::clone(&engine),
        shared_cfg,
    );

    // Wait for inactivity + grace + sweep margin.
    tokio::time::sleep(Duration::from_secs(6)).await;

    assert!(
        topic_exists(&engine, "system.logs").await,
        "system.logs should NOT be reaped because the override sets inactive_secs=0"
    );
    assert!(
        !topic_exists(&engine, "regular.events").await,
        "regular.events should be reaped (no override applies)"
    );

    lifecycle.transition_to(fafb_q::server::state::ServerState::ShuttingDown);
}
