use std::{io::Write as _, path::PathBuf};

use super::*;

fn queue_path(dir: &tempfile::TempDir) -> PathBuf {
    dir.path().join("queue")
}

#[test]
fn roundtrip_fifo() {
    let dir = tempfile::tempdir().unwrap();
    let queue: PersistentQueue<String> = PersistentQueue::open(queue_path(&dir)).unwrap();

    assert!(queue.is_empty());
    assert!(queue.get().unwrap().is_none());

    queue.put(&String::from("one")).unwrap();
    queue.put(&String::from("two")).unwrap();
    queue.put(&String::from("three")).unwrap();

    assert!(!queue.is_empty());
    assert_eq!(queue.pending(), 3);

    let lease = queue.get().unwrap().unwrap();
    assert_eq!(&*lease, "one");
    assert_eq!(queue.pending(), 2);
    assert_eq!(queue.in_flight(), 1);
    assert_eq!(lease.ack(), "one");
    assert_eq!(queue.in_flight(), 0);

    assert_eq!(queue.get().unwrap().unwrap().ack(), "two");
    assert_eq!(queue.get().unwrap().unwrap().ack(), "three");

    assert!(queue.is_empty());
    assert!(queue.get().unwrap().is_none());
}

#[test]
fn dropped_lease_is_redelivered() {
    let dir = tempfile::tempdir().unwrap();
    let queue: PersistentQueue<u32> = PersistentQueue::open(queue_path(&dir)).unwrap();

    queue.put(&1).unwrap();
    queue.put(&2).unwrap();

    let lease = queue.get().unwrap().unwrap();
    assert_eq!(*lease, 1);
    drop(lease);
    assert_eq!(queue.pending(), 2);

    // The returned value is delivered again, ahead of the rest
    let lease = queue.get().unwrap().unwrap();
    assert_eq!(*lease, 1);
    lease.release();

    assert_eq!(queue.get().unwrap().unwrap().ack(), 1);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 2);
    assert!(queue.is_empty());
}

#[test]
fn survives_reopen() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    {
        let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();

        queue.put(&1).unwrap();
        queue.put(&2).unwrap();
        queue.put(&3).unwrap();

        assert_eq!(queue.get().unwrap().unwrap().ack(), 1);
        // Dropping the queue releases the lock and performs a final sync
    }

    let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();

    assert_eq!(queue.pending(), 2);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 2);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 3);
    assert!(queue.is_empty());
}

#[test]
fn double_open_is_refused() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    let _queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();

    match PersistentQueue::<u32>::open(&path) {
        Err(Error::QueueLocked) => {}
        other => panic!("expected Error::QueueLocked, got {other:?}"),
    }
}

#[test]
fn unsynced_acks_redeliver_after_crash() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);
    let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();

    queue.put(&1).unwrap();
    queue.put(&2).unwrap();
    queue.sync().unwrap();

    queue.get().unwrap().unwrap().ack();

    // Copying the file while the queue is open simulates a crash: the
    // in-memory state is discarded and only the on-disk state survives.
    // The ack has not been synced, so it must not have reached the disk.
    let crashed = dir.path().join("crashed");
    std::fs::copy(&path, &crashed).unwrap();

    let recovered: PersistentQueue<u32> = PersistentQueue::open(&crashed).unwrap();
    assert_eq!(recovered.pending(), 2);
    assert_eq!(recovered.get().unwrap().unwrap().ack(), 1);
    assert_eq!(recovered.get().unwrap().unwrap().ack(), 2);

    // After a sync the ack is durable and survives the same "crash"
    queue.sync().unwrap();

    let synced = dir.path().join("synced");
    std::fs::copy(&path, &synced).unwrap();

    let recovered: PersistentQueue<u32> = PersistentQueue::open(&synced).unwrap();
    assert_eq!(recovered.pending(), 1);
    assert_eq!(recovered.get().unwrap().unwrap().ack(), 2);
}

#[test]
fn torn_tail_is_discarded() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    {
        let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();
        queue.put(&1).unwrap();
        queue.put(&2).unwrap();
    }

    // Garbage after the last record, as left by an interrupted append
    let mut file = File::options().append(true).open(&path).unwrap();
    file.write_all(&[0xAB; 7]).unwrap();
    drop(file);

    let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();
    assert_eq!(queue.pending(), 2);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 1);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 2);

    // The truncated file remains fully usable
    queue.put(&3).unwrap();
    assert_eq!(queue.get().unwrap().unwrap().ack(), 3);
}

#[test]
fn oversized_length_in_tail_is_discarded() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    {
        let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();
        queue.put(&1).unwrap();
    }

    // A record header whose length points past the end of the file, as
    // left by a crash between writing the header and the payload
    let mut file = File::options().append(true).open(&path).unwrap();
    RecordHeader {
        state: STATE_LIVE,
        length: 1000,
        crc: 0,
    }
    .store(&mut file)
    .unwrap();
    drop(file);

    let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();
    assert_eq!(queue.pending(), 1);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 1);
}

#[test]
fn corrupt_payload_is_discarded() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    {
        let queue: PersistentQueue<String> = PersistentQueue::open(&path).unwrap();
        queue.put(&String::from("aaaa")).unwrap();
        queue.put(&String::from("bbbb")).unwrap();
        queue.put(&String::from("cccc")).unwrap();
    }

    // Flip a bit near the end of the file, inside the last payload
    let mut contents = std::fs::read(&path).unwrap();
    let target = contents.len() - 2;
    contents[target] ^= 0x01;
    std::fs::write(&path, &contents).unwrap();

    let queue: PersistentQueue<String> = PersistentQueue::open(&path).unwrap();
    assert_eq!(queue.pending(), 2);
    assert_eq!(queue.get().unwrap().unwrap().ack(), "aaaa");
    assert_eq!(queue.get().unwrap().unwrap().ack(), "bbbb");
    assert!(queue.is_empty());
}

#[test]
fn non_queue_file_is_rejected() {
    let dir = tempfile::tempdir().unwrap();

    let short = dir.path().join("short");
    std::fs::write(&short, b"hello").unwrap();
    assert!(matches!(
        PersistentQueue::<u32>::open(&short),
        Err(Error::InvalidFile)
    ));

    let long = dir.path().join("long");
    std::fs::write(&long, [0x55; 64]).unwrap();
    assert!(matches!(
        PersistentQueue::<u32>::open(&long),
        Err(Error::InvalidFile)
    ));
}

#[test]
fn drained_queue_file_is_reset() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);
    let queue: PersistentQueue<String> = PersistentQueue::open(&path).unwrap();

    for _ in 0..10 {
        queue.put(&"x".repeat(1000)).unwrap();
    }

    while let Some(lease) = queue.get().unwrap() {
        lease.ack();
    }

    queue.sync().unwrap();

    assert_eq!(
        std::fs::metadata(&path).unwrap().len(),
        FILE_HEADER_BYTES,
        "a fully drained queue file should shrink back to just its header"
    );

    queue.put(&String::from("again")).unwrap();
    assert_eq!(queue.get().unwrap().unwrap().ack(), "again");
}

#[test]
fn compaction_shrinks_file_and_preserves_leases() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);
    let queue: PersistentQueue<String> = PersistentQueue::open(&path).unwrap();

    let big = "x".repeat(8192);

    for _ in 0..200 {
        queue.put(&big).unwrap();
    }
    queue.put(&String::from("marker")).unwrap();

    for _ in 0..200 {
        let lease = queue.get().unwrap().unwrap();
        assert_eq!(lease.len(), 8192);
        lease.ack();
    }

    // Hold a lease across the compaction to exercise offset remapping
    let marker = queue.get().unwrap().unwrap();
    assert_eq!(&*marker, "marker");

    let before = std::fs::metadata(&path).unwrap().len();
    assert!(before > 200 * 8192);

    queue.sync().unwrap();

    let after = std::fs::metadata(&path).unwrap().len();
    assert!(
        after < 10_000,
        "compaction should have shrunk the file, but it is {after} bytes"
    );

    assert_eq!(marker.ack(), "marker");
    queue.sync().unwrap();
    assert_eq!(
        std::fs::metadata(&path).unwrap().len(),
        FILE_HEADER_BYTES,
        "acking the last value and syncing should reset the file"
    );

    queue.put(&String::from("after")).unwrap();
    assert_eq!(queue.get().unwrap().unwrap().ack(), "after");
}

#[test]
fn compaction_remaps_requeued_records() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);
    let queue: PersistentQueue<String> = PersistentQueue::open(&path).unwrap();

    let big = "x".repeat(8192);

    for _ in 0..200 {
        queue.put(&big).unwrap();
    }
    queue.put(&String::from("x-requeued")).unwrap();
    queue.put(&String::from("y-leased")).unwrap();

    for _ in 0..200 {
        queue.get().unwrap().unwrap().ack();
    }

    let lease_x = queue.get().unwrap().unwrap();
    assert_eq!(&*lease_x, "x-requeued");
    let lease_y = queue.get().unwrap().unwrap();
    assert_eq!(&*lease_y, "y-leased");

    lease_x.release();

    queue.sync().unwrap();
    assert!(std::fs::metadata(&path).unwrap().len() < 10_000);

    // The requeued record survived the compaction and is delivered first
    let lease_x = queue.get().unwrap().unwrap();
    assert_eq!(&*lease_x, "x-requeued");
    assert_eq!(lease_x.ack(), "x-requeued");
    assert_eq!(lease_y.ack(), "y-leased");

    queue.sync().unwrap();
    assert!(queue.is_empty());
    assert_eq!(std::fs::metadata(&path).unwrap().len(), FILE_HEADER_BYTES);
}

#[test]
fn structured_values_roundtrip() {
    #[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
    struct Job {
        id: u64,
        path: PathBuf,
        chunks: Vec<[u8; 4]>,
    }

    let dir = tempfile::tempdir().unwrap();
    let queue: PersistentQueue<Job> = PersistentQueue::open(queue_path(&dir)).unwrap();

    let job = Job {
        id: 42,
        path: PathBuf::from("/some/where"),
        chunks: vec![[1, 2, 3, 4], [5, 6, 7, 8]],
    };

    queue.put(&job).unwrap();

    assert_eq!(queue.get().unwrap().unwrap().ack(), job);
}

/// Retaining keeps the order of what survives and loses nothing else,
/// and what it leaves behind is what a later session finds.
#[test]
fn retaining_drops_the_rest_and_keeps_the_order() {
    let dir = tempfile::tempdir().unwrap();
    let path = queue_path(&dir);

    {
        let queue: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();

        for value in 1..=9 {
            queue.put(&value).unwrap();
        }

        assert_eq!(queue.retain(|value| value % 3 != 0).unwrap(), 3);
        assert_eq!(queue.pending(), 6);
        assert_eq!(queue.peek(9).unwrap(), vec![1, 2, 4, 5, 7, 8]);

        // Keeping everything is not the same as doing nothing, but it
        // must look like it from the outside
        assert_eq!(queue.retain(|_| true).unwrap(), 0);
        assert_eq!(queue.peek(9).unwrap(), vec![1, 2, 4, 5, 7, 8]);

        queue.sync().unwrap();
    }

    let reopened: PersistentQueue<u32> = PersistentQueue::open(&path).unwrap();
    assert_eq!(reopened.peek(9).unwrap(), vec![1, 2, 4, 5, 7, 8]);

    // And an empty queue is an ordinary case, not an edge one
    assert_eq!(reopened.retain(|_| false).unwrap(), 6);
    assert_eq!(reopened.retain(|_| false).unwrap(), 0);
    assert!(reopened.is_empty());
}

/// Inspecting walks everything the queue holds without taking any of it.
#[test]
fn inspecting_walks_the_queue_without_taking_it() {
    let dir = tempfile::tempdir().unwrap();
    let queue: PersistentQueue<u32> = PersistentQueue::open(queue_path(&dir)).unwrap();

    for value in 1..=5 {
        queue.put(&value).unwrap();
    }

    let mut seen = Vec::new();
    queue.inspect(|value| seen.push(*value)).unwrap();

    assert_eq!(seen, vec![1, 2, 3, 4, 5]);
    assert_eq!(queue.pending(), 5, "inspecting delivers nothing");
}

/// Peeking says what the next deliveries will be without becoming one:
/// the values stay pending, and in the order `get` will hand them over.
#[test]
fn peeking_names_the_front_without_taking_it() {
    let dir = tempfile::tempdir().unwrap();
    let queue: PersistentQueue<u32> = PersistentQueue::open(queue_path(&dir)).unwrap();

    assert!(queue.peek(4).unwrap().is_empty(), "nothing queued yet");

    for value in 1..=5 {
        queue.put(&value).unwrap();
    }

    assert_eq!(queue.peek(3).unwrap(), vec![1, 2, 3]);
    assert_eq!(
        queue.peek(50).unwrap(),
        vec![1, 2, 3, 4, 5],
        "asking for more than there is"
    );
    assert_eq!(queue.pending(), 5, "peeking delivers nothing");

    // What is out on a lease is no longer at the front, and what is
    // acknowledged never comes back to it
    let lease = queue.get().unwrap().unwrap();
    assert_eq!(queue.peek(2).unwrap(), vec![2, 3]);
    lease.ack();
    assert_eq!(queue.peek(2).unwrap(), vec![2, 3]);

    // A dropped lease is redelivered ahead of the rest, and peeking says
    // so before it happens
    let lease = queue.get().unwrap().unwrap();
    assert_eq!(*lease, 2);
    let held = queue.get().unwrap().unwrap();
    assert_eq!(*held, 3);
    drop(lease);

    assert_eq!(queue.peek(3).unwrap(), vec![2, 4, 5]);
    assert_eq!(queue.get().unwrap().unwrap().ack(), 2);
    assert_eq!(queue.peek(3).unwrap(), vec![4, 5]);

    // Peeking is still honest across a reopen, where everything
    // unacknowledged becomes deliverable again
    held.release();
    drop(queue);

    let queue: PersistentQueue<u32> = PersistentQueue::open(queue_path(&dir)).unwrap();
    assert_eq!(queue.peek(4).unwrap(), vec![3, 4, 5]);
}