mod common;

use std::sync::Arc;

use fafb_q::queue::{BackpressurePolicy, EngineConfig, InMemoryQueueEngine, QueueEngine};
use fafb_q::storage::mock::MockStorage;
use fafb_q::storage::{FsStorage, StorageBackend};
use fafb_q::types::{MessageEnvelope, MessageId, NamespaceId, TopicName, TopicOffset, now_millis};
use tempfile::TempDir;

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

async fn make_engine(tmp: &TempDir) -> InMemoryQueueEngine<FsStorage> {
    let storage = Arc::new(FsStorage::new(tmp.path().join("data")));
    storage.init().await.unwrap();
    let engine = InMemoryQueueEngine::new(
        storage,
        EngineConfig {
            max_depth: 100_000,
            backpressure: BackpressurePolicy::Reject,
            ..Default::default()
        },
    );
    engine.init().await.unwrap();
    engine
}

#[tokio::test]
async fn concurrent_publish_same_topic_sequential_offsets() {
    let tmp = TempDir::new().unwrap();
    let engine = Arc::new(make_engine(&tmp).await);

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    let tasks_count = 10;
    let msgs_per_task = 100u64;
    let mut handles = Vec::new();

    for task_id in 0..tasks_count {
        let e = Arc::clone(&engine);
        let ns = ns.clone();
        let topic = topic.clone();
        handles.push(tokio::spawn(async move {
            let mut offsets = Vec::new();
            for i in 0..msgs_per_task {
                let id = task_id * msgs_per_task + i;
                let result = e
                    .publish(&ns, &topic, make_envelope(id, "ns", "t", b"data"))
                    .await
                    .unwrap();
                offsets.push(result.offset.0);
            }
            offsets
        }));
    }

    let mut all_offsets = Vec::new();
    for h in handles {
        all_offsets.extend(h.await.unwrap());
    }

    all_offsets.sort();
    let total = (tasks_count * msgs_per_task) as u64;
    let expected: Vec<u64> = (0..total).collect();
    assert_eq!(all_offsets, expected);
}

#[tokio::test]
async fn concurrent_consume_consistent() {
    let tmp = TempDir::new().unwrap();
    let engine = Arc::new(make_engine(&tmp).await);

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Publish 100 messages
    for i in 0..100u64 {
        engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
    }

    // 5 concurrent readers each read all 100 messages
    let mut handles = Vec::new();
    for _ in 0..5 {
        let e = Arc::clone(&engine);
        let ns = ns.clone();
        let topic = topic.clone();
        handles.push(tokio::spawn(async move {
            let mut payloads = Vec::new();
            for i in 0..100u64 {
                let msg = e
                    .consume_from(&ns, &topic, TopicOffset(i))
                    .await
                    .unwrap()
                    .unwrap();
                payloads.push(msg.envelope.payload.clone());
            }
            payloads
        }));
    }

    let mut all_results = Vec::new();
    for h in handles {
        all_results.push(h.await.unwrap());
    }

    // All readers should see the same data
    for payloads in &all_results {
        for (i, payload) in payloads.iter().enumerate() {
            assert_eq!(payload, format!("msg-{i}").as_bytes());
        }
    }
}

#[tokio::test]
async fn publish_compact_concurrent() {
    let tmp = TempDir::new().unwrap();
    let engine = Arc::new(make_engine(&tmp).await);

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Publish 100 messages
    for i in 0..100u64 {
        engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
    }

    // Compact from another task while publishing more
    let compact_engine = Arc::clone(&engine);
    let compact_ns = ns.clone();
    let compact_topic = topic.clone();
    let compact_handle = tokio::spawn(async move {
        compact_engine
            .compact(&compact_ns, &compact_topic, TopicOffset(50))
            .await
            .unwrap();
    });

    // Continue publishing
    for i in 100..150u64 {
        engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
    }

    compact_handle.await.unwrap();

    // Offsets < 50 are compacted from memory but still readable from disk
    for i in 0..50u64 {
        let result = engine
            .consume_from(&ns, &topic, TopicOffset(i))
            .await
            .unwrap();
        assert!(
            result.is_some(),
            "offset {i} should be readable from disk after compaction"
        );
    }

    // Offsets >= 50 should be present
    for i in 50..150u64 {
        let msg = engine
            .consume_from(&ns, &topic, TopicOffset(i))
            .await
            .unwrap()
            .expect(&format!("offset {i} should be present"));
        assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
    }
}

#[tokio::test]
async fn replay_from_storage_restores_state() {
    let tmp = TempDir::new().unwrap();

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Create engine, publish 50 messages, shutdown
    {
        let storage = Arc::new(FsStorage::new(tmp.path().join("data")));
        storage.init().await.unwrap();
        let engine = InMemoryQueueEngine::new(
            Arc::clone(&storage),
            EngineConfig {
                max_depth: 100_000,
                backpressure: BackpressurePolicy::Reject,
                ..Default::default()
            },
        );
        engine.init().await.unwrap();

        for i in 0..50u64 {
            engine
                .publish(
                    &ns,
                    &topic,
                    make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
                )
                .await
                .unwrap();
        }

        engine.shutdown().await.unwrap();
    }

    // New engine with same storage path, init should replay
    let storage = Arc::new(FsStorage::new(tmp.path().join("data")));
    storage.init().await.unwrap();
    let engine = InMemoryQueueEngine::new(
        storage,
        EngineConfig {
            max_depth: 100_000,
            backpressure: BackpressurePolicy::Reject,
            ..Default::default()
        },
    );
    engine.init().await.unwrap();

    // Verify head offset
    let head = engine.head_offset(&ns, &topic).await.unwrap();
    assert_eq!(head, TopicOffset(50));

    // Verify all messages are present
    for i in 0..50u64 {
        let msg = engine
            .consume_from(&ns, &topic, TopicOffset(i))
            .await
            .unwrap()
            .expect(&format!("offset {i} should be present after replay"));
        assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
    }

    engine.shutdown().await.unwrap();
}

#[tokio::test]
async fn cold_read_after_memory_eviction() {
    let storage = Arc::new(MockStorage::new());
    storage.init().await.unwrap();
    let engine = InMemoryQueueEngine::new(
        storage,
        EngineConfig {
            max_depth: 50,
            backpressure: BackpressurePolicy::DropOldest,
            ..Default::default()
        },
    );
    engine.init().await.unwrap();

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Publish 1000 messages; with max_depth=50 and DropOldest, older messages
    // are evicted from memory but remain on disk via MockStorage.
    for i in 0..1000u64 {
        engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
    }

    // Offset 0 was evicted from memory long ago but should be readable from disk.
    let cold_msg = engine
        .consume_from(&ns, &topic, TopicOffset(0))
        .await
        .unwrap();
    assert!(
        cold_msg.is_some(),
        "offset 0 should be readable from disk after memory eviction"
    );
    assert_eq!(cold_msg.unwrap().envelope.payload, b"msg-0");

    // Offset 999 should still be in memory.
    let hot_msg = engine
        .consume_from(&ns, &topic, TopicOffset(999))
        .await
        .unwrap();
    assert!(
        hot_msg.is_some(),
        "offset 999 should be readable from memory"
    );
    assert_eq!(hot_msg.unwrap().envelope.payload, b"msg-999");
}

#[tokio::test]
async fn disk_backed_publish_past_memory_budget() {
    let storage = Arc::new(MockStorage::new());
    storage.init().await.unwrap();
    let engine = InMemoryQueueEngine::new(
        storage,
        EngineConfig {
            max_depth: 50,
            backpressure: BackpressurePolicy::DropOldest,
            disk_backed: true,
            global_max_memory_bytes: 1024,
            ..Default::default()
        },
    );
    engine.init().await.unwrap();

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Publish 100 messages with 100-byte payloads each (total 10KB > 1KB budget).
    // All should succeed because disk_backed mode skips the memory budget check.
    let payload = vec![0xABu8; 100];
    for i in 0..100u64 {
        engine
            .publish(&ns, &topic, make_envelope(i, "ns", "t", &payload))
            .await
            .unwrap();
    }

    // All 100 messages should be consumable (from memory or disk).
    for i in 0..100u64 {
        let msg = engine
            .consume_from(&ns, &topic, TopicOffset(i))
            .await
            .unwrap();
        assert!(
            msg.is_some(),
            "offset {i} should be consumable in disk_backed mode"
        );
        assert_eq!(msg.unwrap().envelope.payload, payload);
    }
}

#[tokio::test]
async fn cold_batch_read_from_fsstorage() {
    let tmp = TempDir::new().unwrap();

    let ns = NamespaceId("ns".into());
    let topic = TopicName("t".into());

    // Use small segment_max_bytes to force multiple segments
    let storage = Arc::new(FsStorage::with_segment_max_bytes(
        tmp.path().join("data"),
        256,
    ));
    storage.init().await.unwrap();
    let engine = InMemoryQueueEngine::new(
        storage,
        EngineConfig {
            max_depth: 100_000,
            backpressure: BackpressurePolicy::DropOldest,
            ..Default::default()
        },
    );
    engine.init().await.unwrap();

    // Publish 100 messages (will span multiple segments due to small segment size)
    for i in 0..100u64 {
        engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
    }

    // Compact first 50 from in-memory buffer
    engine.compact(&ns, &topic, TopicOffset(50)).await.unwrap();

    // read_cold_batch should read from disk across multiple segments
    let msgs = engine
        .read_cold_batch(&ns, &topic, TopicOffset(0), 20)
        .await
        .unwrap();
    assert_eq!(msgs.len(), 20, "should return 20 messages from disk");

    // Verify they are sequential starting from offset 0
    for (i, msg) in msgs.iter().enumerate() {
        assert_eq!(msg.offset, TopicOffset(i as u64));
        assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
    }

    // Also test reading from middle of the range
    let msgs = engine
        .read_cold_batch(&ns, &topic, TopicOffset(10), 15)
        .await
        .unwrap();
    assert_eq!(
        msgs.len(),
        15,
        "should return 15 messages starting from offset 10"
    );
    for (i, msg) in msgs.iter().enumerate() {
        let expected_offset = 10 + i as u64;
        assert_eq!(msg.offset, TopicOffset(expected_offset));
        assert_eq!(
            msg.envelope.payload,
            format!("msg-{expected_offset}").as_bytes()
        );
    }

    engine.shutdown().await.unwrap();
}

/// Full-stack integration test: FsStorage + engine + manager with disk quota.
/// Verifies that publish is rejected when the namespace exceeds its disk quota.
#[tokio::test]
async fn disk_quota_rejection_with_fsstorage() {
    use fafb_q::error::Error;
    use fafb_q::namespace::manager::{ManagerConfig, NamespaceQuota};
    use fafb_q::namespace::{DefaultNamespaceManager, NamespaceManager};

    let tmp = TempDir::new().unwrap();
    let storage = Arc::new(FsStorage::new(tmp.path().join("data")));
    storage.init().await.unwrap();

    let engine = Arc::new(InMemoryQueueEngine::new(
        Arc::clone(&storage),
        EngineConfig {
            max_depth: 100_000,
            backpressure: BackpressurePolicy::Reject,
            ..Default::default()
        },
    ));
    engine.init().await.unwrap();

    let ns = NamespaceId("quota-ns".into());
    let topic = TopicName("t".into());

    // Compute per-record size to set quota precisely
    let sample = fafb_q::types::PublishedMessage {
        envelope: Arc::new(make_envelope(0, "quota-ns", "t", b"test-payload")),
        offset: TopicOffset(0),
    };
    let record_size = fafb_q::storage::fs::encode_record(&sample).unwrap().len() as u64;

    // Allow 5 records worth of disk space
    let quota_bytes = record_size * 5;

    // Build a SharedConfig with the disk quota for this namespace.
    let shared_cfg = {
        let mut rt = fafb_q::config::RuntimeConfig::from_server_config(
            &fafb_q::config::ServerConfig::default(),
        );
        rt.namespace_quotas.insert(
            ns.clone(),
            NamespaceQuota {
                max_disk_bytes: Some(quota_bytes),
                ..Default::default()
            },
        );
        std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
    };
    let mgr = DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![ns.clone()],
        shared_cfg,
        ManagerConfig::default(),
    );

    // Publish 5 messages — should all succeed
    for i in 0..5u64 {
        mgr.publish(
            &ns,
            &topic,
            make_envelope(i, "quota-ns", "t", b"test-payload"),
        )
        .await
        .unwrap();
    }

    // Verify disk bytes are tracked through the full stack
    let disk_bytes = engine.namespace_disk_bytes(&ns).await;
    assert!(
        disk_bytes >= quota_bytes,
        "disk bytes ({disk_bytes}) should be >= quota ({quota_bytes})"
    );

    // 6th message should be rejected
    let err = mgr
        .publish(
            &ns,
            &topic,
            make_envelope(5, "quota-ns", "t", b"test-payload"),
        )
        .await
        .unwrap_err();
    assert!(
        matches!(err, Error::ResourceExhausted(_)),
        "expected ResourceExhausted, got: {err}"
    );

    // Verify the error message is descriptive
    let msg = err.to_string();
    assert!(
        msg.contains("namespace disk quota exceeded"),
        "error should mention disk quota: {msg}"
    );

    engine.shutdown().await.unwrap();
}
