mod common;

use std::sync::Arc;

use fafb_q::storage::{FsStorage, StorageBackend};
use fafb_q::types::{
    MessageEnvelope, MessageId, NamespaceId, PublishedMessage, TopicName, TopicOffset, now_millis,
};
use tempfile::TempDir;

fn make_msg(offset: u64, payload: &[u8]) -> PublishedMessage {
    PublishedMessage {
        envelope: Arc::new(MessageEnvelope::new(
            MessageId(offset + 1),
            NamespaceId("test-ns".into()),
            TopicName("test-topic".into()),
            payload.to_vec(),
            now_millis(),
        )),
        offset: TopicOffset(offset),
    }
}

#[tokio::test]
async fn append_and_replay_returns_messages_in_order() {
    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

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

    // Append 100 messages
    for i in 0..100u64 {
        let msg = make_msg(i, format!("payload-{i}").as_bytes());
        storage.append(&ns, &topic, &msg).await.unwrap();
    }

    // Replay and verify
    let replayed = storage.replay(&ns, &topic).await.unwrap();
    assert_eq!(replayed.len(), 100);
    for (i, msg) in replayed.iter().enumerate() {
        assert_eq!(msg.offset, TopicOffset(i as u64));
        assert_eq!(msg.envelope.payload, format!("payload-{i}").as_bytes());
    }
}

#[tokio::test]
async fn replay_recovers_from_truncation() {
    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

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

    // Append 10 messages
    for i in 0..10u64 {
        let msg = make_msg(i, format!("payload-{i}").as_bytes());
        storage.append(&ns, &topic, &msg).await.unwrap();
    }

    // Flush to ensure data is durable before we externally truncate
    storage.flush(&ns, &topic).await.unwrap();

    // Find the segment file and truncate it
    let seg_dir = tmp
        .path()
        .join("data")
        .join("test-ns")
        .join("test-topic")
        .join("segments");
    let mut entries: Vec<_> = std::fs::read_dir(&seg_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .map(|e| e.path())
        .filter(|p| p.extension().and_then(|e| e.to_str()) == Some("seg"))
        .collect();
    entries.sort();

    let seg_path = &entries[0];
    let data = std::fs::read(seg_path).unwrap();
    // Truncate by 20 bytes — enough to corrupt the last record
    let truncated = &data[..data.len() - 20];
    std::fs::write(seg_path, truncated).unwrap();

    // Replay should recover most messages
    let replayed = storage.replay(&ns, &topic).await.unwrap();
    assert!(
        replayed.len() >= 8,
        "expected at least 8 messages, got {}",
        replayed.len()
    );
    assert!(
        replayed.len() < 10,
        "expected fewer than 10 messages after truncation, got {}",
        replayed.len()
    );
    // Verify order
    for (i, msg) in replayed.iter().enumerate() {
        assert_eq!(msg.offset, TopicOffset(i as u64));
    }
}

#[tokio::test]
async fn segment_rotation_creates_new_file() {
    let tmp = TempDir::new().unwrap();
    // Small segment max to force rotation
    let storage = FsStorage::with_segment_max_bytes(tmp.path().join("data"), 256);
    storage.init().await.unwrap();

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

    // Write 20 messages with 64-byte payloads to force multiple segments
    let payload = vec![0xABu8; 64];
    for i in 0..20u64 {
        let msg = make_msg(i, &payload);
        storage.append(&ns, &topic, &msg).await.unwrap();
    }

    // Shutdown flushes all writers
    storage.shutdown().await.unwrap();

    // Verify multiple segment files were created
    let seg_dir = tmp
        .path()
        .join("data")
        .join("test-ns")
        .join("test-topic")
        .join("segments");
    let seg_count = std::fs::read_dir(&seg_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().and_then(|ext| ext.to_str()) == Some("seg"))
        .count();
    assert!(seg_count > 1, "expected multiple segments, got {seg_count}");

    // Fresh instance with init pre-populates writers and replays correctly
    let storage2 = FsStorage::new(tmp.path().join("data"));
    storage2.init().await.unwrap();
    let replayed = storage2.replay(&ns, &topic).await.unwrap();
    assert_eq!(replayed.len(), 20);
    for (i, msg) in replayed.iter().enumerate() {
        assert_eq!(msg.offset, TopicOffset(i as u64));
        assert_eq!(msg.envelope.payload, payload);
    }
}

#[tokio::test]
async fn shutdown_persists_multiple_topics() {
    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

    let ns = NamespaceId("test-ns".into());
    let topic_a = TopicName("topic-a".into());
    let topic_b = TopicName("topic-b".into());

    for i in 0..5u64 {
        let msg_a = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(i + 1),
                ns.clone(),
                topic_a.clone(),
                format!("a-{i}").into_bytes(),
                now_millis(),
            )),
            offset: TopicOffset(i),
        };
        storage.append(&ns, &topic_a, &msg_a).await.unwrap();

        let msg_b = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(100 + i),
                ns.clone(),
                topic_b.clone(),
                format!("b-{i}").into_bytes(),
                now_millis(),
            )),
            offset: TopicOffset(i),
        };
        storage.append(&ns, &topic_b, &msg_b).await.unwrap();
    }

    storage.shutdown().await.unwrap();

    // Fresh instance — verify both topics survived
    let storage2 = FsStorage::new(tmp.path().join("data"));
    storage2.init().await.unwrap();

    let replayed_a = storage2.replay(&ns, &topic_a).await.unwrap();
    assert_eq!(replayed_a.len(), 5);
    for (i, msg) in replayed_a.iter().enumerate() {
        assert_eq!(msg.envelope.payload, format!("a-{i}").into_bytes());
    }

    let replayed_b = storage2.replay(&ns, &topic_b).await.unwrap();
    assert_eq!(replayed_b.len(), 5);
    for (i, msg) in replayed_b.iter().enumerate() {
        assert_eq!(msg.envelope.payload, format!("b-{i}").into_bytes());
    }
}
