v_queue 0.3.0

use crate::consumer::Consumer;
use crate::queue::Queue;
use crate::record::{ErrorQueue, Mode, MsgType, HEADER_SIZE, MAGIC_MARKER_BYTES, MAGIC_MARKER_V3_BYTES};
use std::fs::OpenOptions;
use std::io::{Seek, SeekFrom};
use std::time::Duration;
use std::{fs, thread};

fn check_message_integrity(received_numbers: &[i32]) {
    println!("{:?}", received_numbers);
    for i in 1..received_numbers.len() {
        assert_eq!(received_numbers[i], received_numbers[i - 1] + 1);
    }
}

#[test]
fn test_read_only_mode() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_queue";

    // Создаем очередь в режиме ReadWrite и записываем сообщения
    let mut queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
    let num_messages = 5;
    for i in 0..num_messages {
        let msg = format!("Message {}", i);
        queue.push(msg.as_bytes(), MsgType::String).unwrap();
    }

    // Закрываем очередь и открываем ее в режиме ReadOnly
    drop(queue);
    let mut queue = Queue::new(&base_path, queue_name, Mode::Read).unwrap();

    // Пытаемся записать сообщение в очередь
    let message = "Hello, world!".as_bytes();
    let msg_type = MsgType::String;
    let result = queue.push(message, msg_type);

    // Проверяем, что запись не производится и возвращается ошибка Other
    assert!(result.is_err());
    assert_eq!(result.unwrap_err(), ErrorQueue::NotReady);

    // Создаем потребителя и проверяем, что сообщения доступны для чтения
    let consumer_name = "consumer";
    let mut consumer = Consumer::new(&base_path, consumer_name, queue_name).unwrap();
    let mut received_messages = Vec::new();
    while consumer.pop_header() {
        let msg_size = consumer.record_len();
        let mut msg = vec![0; msg_size];
        if consumer.pop_body(&mut msg).is_ok() {
            let message = String::from_utf8(msg).unwrap();
            received_messages.push(message);
            consumer.commit();
        } else {
            break;
        }
    }

    // Проверяем, что все сообщения были прочитаны
    assert_eq!(received_messages.len(), num_messages as usize);
    for (i, msg) in received_messages.iter().enumerate() {
        assert_eq!(msg, &format!("Message {}", i));
    }
}

#[test]
fn test_queue_consumer_interaction() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_queue";

    // Создаем очередь и записываем сообщения с номерами
    let mut queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
    let num_messages = 10;
    for i in 0..num_messages {
        let msg = format!("{}", i);
        queue.push(msg.as_bytes(), MsgType::String).unwrap();
    }

    // Создаем несколько потребителей
    let num_consumers = 3;
    let mut consumers = Vec::new();
    for i in 0..num_consumers {
        let consumer_name = format!("consumer_{}", i);
        let consumer = Consumer::new(&base_path, &consumer_name, queue_name).unwrap();
        consumers.push(consumer);
    }

    // Каждый потребитель читает свою часть сообщений
    let mut received_numbers = vec![Vec::new(); num_consumers];
    for (i, consumer) in consumers.iter_mut().enumerate() {
        while consumer.pop_header() {
            let msg_size = consumer.record_len();
            let mut msg = vec![0; msg_size];
            if consumer.pop_body(&mut msg).is_ok() {
                let number = String::from_utf8(msg).unwrap().parse::<i32>().unwrap();
                received_numbers[i].push(number);
                consumer.commit();
            } else {
                break;
            }
        }
    }

    // Проверяем, что каждый потребитель получил свою часть сообщений без пропусков
    for numbers in received_numbers.iter() {
        check_message_integrity(numbers);
    }

    // Закрываем очередь и открываем ее снова для создания нового сегмента
    drop(queue);
    let mut queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();

    // Записываем новые сообщения с номерами в очередь
    let num_new_messages = 5;
    for i in num_messages..num_messages + num_new_messages {
        let msg = format!("{}", i);
        queue.push(msg.as_bytes(), MsgType::String).unwrap();
    }

    // Ждем, пока потребители прочитают новые сообщения
    thread::sleep(Duration::from_millis(100));

    // Каждый потребитель читает свою часть новых сообщений
    let mut new_received_numbers = vec![Vec::new(); num_consumers];
    for (i, consumer) in consumers.iter_mut().enumerate() {
        while consumer.pop_header() {
            let msg_size = consumer.record_len();
            let mut msg = vec![0; msg_size];
            if consumer.pop_body(&mut msg).is_ok() {
                let number = String::from_utf8(msg).unwrap().parse::<i32>().unwrap();
                new_received_numbers[i].push(number);
                consumer.commit();
            } else {
                break;
            }
        }
    }

    // Проверяем, что каждый потребитель получил свою часть новых сообщений без пропусков
    for numbers in new_received_numbers.iter() {
        check_message_integrity(numbers);
    }
}

#[test]
fn test_consumer_reconnect() {
    println!("test_consumer_reconnect");
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_queue";

    // Создаем очередь и записываем сообщения
    let mut queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
    let num_messages = 10;
    for i in 0..num_messages {
        let msg = format!("{}", i);
        println!("push {}", msg);
        queue.push(msg.as_bytes(), MsgType::String).unwrap();
    }

    // Создаем потребителя и читаем часть сообщений
    let consumer_name = "consumer";
    let mut consumer = Consumer::new(&base_path, consumer_name, queue_name).unwrap();
    let num_read_messages = 5;
    for _ in 0..num_read_messages {
        if consumer.pop_header() {
            let msg_size = consumer.record_len();
            let mut msg = vec![0; msg_size];
            consumer.pop_body(&mut msg).unwrap();
            consumer.commit();
        }
    }

    // Закрываем потребителя и создаем нового с тем же именем
    drop(consumer);
    let mut consumer = Consumer::new(&base_path, consumer_name, queue_name).unwrap();

    // Читаем оставшиеся сообщения
    let mut received_numbers = Vec::new();
    while consumer.pop_header() {
        let msg_size = consumer.record_len();
        let mut msg = vec![0; msg_size];
        if consumer.pop_body(&mut msg).is_ok() {
            let number = String::from_utf8(msg).unwrap().parse::<i32>().unwrap();
            received_numbers.push(number);
            consumer.commit();
        } else {
            break;
        }
    }

    // Проверяем, что оставшиеся сообщения были получены без пропусков
    check_message_integrity(&received_numbers);
}

use uuid::Uuid;

fn create_unique_queue_path(base_path: &str, prefix: &str) -> String {
    let uuid = Uuid::new_v4().to_string();
    let path = format!("{}/{}_{}", base_path, prefix, uuid);
    fs::remove_dir_all(&base_path).unwrap_or_default();
    path
}

#[test]
fn test_queue_empty() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_queue";

    // Создаем очередь без сообщений
    let _queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();

    // Создаем потребителя и пытаемся прочитать сообщения
    let consumer_name = "consumer";
    let mut consumer = Consumer::new(&base_path, consumer_name, queue_name).unwrap();

    // Проверяем, что метод pop_header возвращает false, если очередь пуста
    assert!(!consumer.pop_header());
}

#[test]
fn test_multiple_queues() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name_1 = "test_queue_1";
    let queue_name_2 = "test_queue_2";

    // Создаем две очереди и записываем сообщения в каждую
    let mut queue_1 = Queue::new(&base_path, queue_name_1, Mode::ReadWrite).unwrap();
    let mut queue_2 = Queue::new(&base_path, queue_name_2, Mode::ReadWrite).unwrap();
    let num_messages = 5;
    for i in 0..num_messages {
        let msg = format!("{}", i);
        queue_1.push(msg.as_bytes(), MsgType::String).unwrap();
        queue_2.push(msg.as_bytes(), MsgType::String).unwrap();
    }

    // Создаем потребителей для каждой очереди
    let consumer_name_1 = "consumer_1";
    let consumer_name_2 = "consumer_2";
    let mut consumer_1 = Consumer::new(&base_path, consumer_name_1, queue_name_1).unwrap();
    let mut consumer_2 = Consumer::new(&base_path, consumer_name_2, queue_name_2).unwrap();

    // Читаем сообщения из каждой очереди
    let mut received_numbers_1 = Vec::new();
    let mut received_numbers_2 = Vec::new();
    while consumer_1.pop_header() {
        let msg_size = consumer_1.record_len();
        let mut msg = vec![0; msg_size];
        if consumer_1.pop_body(&mut msg).is_ok() {
            let number = String::from_utf8(msg).unwrap().parse::<i32>().unwrap();
            received_numbers_1.push(number);
            consumer_1.commit();
        } else {
            break;
        }
    }
    while consumer_2.pop_header() {
        let msg_size = consumer_2.record_len();
        let mut msg = vec![0; msg_size];
        if consumer_2.pop_body(&mut msg).is_ok() {
            let number = String::from_utf8(msg).unwrap().parse::<i32>().unwrap();
            received_numbers_2.push(number);
            consumer_2.commit();
        } else {
            break;
        }
    }

    // Проверяем, что сообщения были получены из соответствующих очередей без пропусков
    check_message_integrity(&received_numbers_1);
    check_message_integrity(&received_numbers_2);
}

// ---------------------------------------------------------------------------
// Tests for queue-reliability-fixes plan.
//
// These tests document the expected behaviour after the corresponding phase
// of the queue-reliability-fixes plan is applied. Tests that depend on a
// not-yet-applied fix are gated with #[ignore] and an explanation; remove the
// attribute when the corresponding fix lands. They can still be executed
// explicitly with `cargo test -- --ignored`.
// ---------------------------------------------------------------------------

// Test for phase 1.2: push must report write failure and not advance in-memory
// state past the last successfully persisted message. We force the failure by
// swapping the queue file handle for one pointing at /dev/full, which always
// returns ENOSPC on write (Linux only). This does not exercise the file
// truncation rollback path directly — that requires a partial write that we
// cannot trigger without root or API changes — but it pins down the user
// visible contract: failed push returns Err and the queue keeps its previous
// invariants.
#[test]
fn test_push_rollback_on_write_error() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_rollback";

    let mut queue = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
    queue.push(b"valid_message", MsgType::String).expect("first push must succeed");

    let saved_right_edge = queue.right_edge;
    let saved_count = queue.count_pushed;
    let queue_file_path = format!("{}/{}-{}/{}_queue", base_path, queue_name, queue.id, queue_name);
    let saved_file_size = fs::metadata(&queue_file_path).unwrap().len();

    let dev_full = match OpenOptions::new().write(true).open("/dev/full") {
        Ok(f) => f,
        Err(_) => {
            eprintln!("test_push_rollback_on_write_error: skipped, /dev/full not available");
            return;
        },
    };
    let original_handle = std::mem::replace(&mut queue.ff_queue, dev_full);

    let result = queue.push(b"this push will fail", MsgType::String);
    assert!(result.is_err(), "push to /dev/full must return Err, got {:?}", result);
    assert_eq!(queue.right_edge, saved_right_edge, "right_edge must not advance past the last successful push");
    assert_eq!(queue.count_pushed, saved_count, "count_pushed must not advance past the last successful push");

    // Restore the real file handle and verify the queue is still usable.
    queue.ff_queue = original_handle;
    queue.ff_queue.seek(SeekFrom::Start(saved_right_edge)).unwrap();

    // The on-disk queue file size must not have changed while /dev/full was active.
    let file_size_after = fs::metadata(&queue_file_path).unwrap().len();
    assert_eq!(file_size_after, saved_file_size, "queue file must not change while writes were redirected to /dev/full");

    queue.push(b"recovery_message", MsgType::String).expect("post-recovery push must succeed");
    drop(queue);

    let mut consumer = Consumer::new(&base_path, "rollback_consumer", queue_name).unwrap();
    let mut received = Vec::new();
    while consumer.pop_header() {
        let mut msg = vec![0; consumer.record_len()];
        consumer.pop_body(&mut msg).unwrap();
        received.push(String::from_utf8(msg).unwrap());
        consumer.commit();
    }

    assert_eq!(received, vec!["valid_message".to_string(), "recovery_message".to_string()]);
}

// Test for phase 1.3: corrupted CRC inside info_push of a finished part must be
// detected on subsequent open and surfaced as Err(InvalidChecksum) instead of
// being silently swallowed. We open the queue in Read mode so that the queue.id
// taken from info_queue is the same one whose info_push we corrupted.
#[test]
fn test_queue_new_fails_on_corrupted_info_push_crc() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_info_push_corrupt";

    {
        let mut q = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
        for i in 0..3 {
            q.push(format!("msg{}", i).as_bytes(), MsgType::String).unwrap();
        }
    }

    let info_push_path = format!("{}/{}-0/{}_info_push", base_path, queue_name, queue_name);
    let mut content = fs::read(&info_push_path).expect("info_push must exist");
    let len = content.len();
    assert!(len > 1 && content[len - 1] == b'\n', "info_push must end with newline");
    let target = len - 2; // last digit of CRC, just before the trailing newline
    content[target] = if content[target] == b'9' { b'0' } else { content[target] + 1 };
    fs::write(&info_push_path, &content).unwrap();

    match Queue::new(&base_path, queue_name, Mode::Read) {
        Err(e) => assert_eq!(e, ErrorQueue::InvalidChecksum),
        Ok(_) => panic!("Queue::new(Read) must return Err when info_push CRC is corrupted, got Ok"),
    }
}

// Test for phase 1.3 applied to info_queue. Same idea as the test above but for
// the per-queue file written by put_info_queue.
#[test]
fn test_queue_new_fails_on_corrupted_info_queue_crc() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_info_queue_corrupt";

    {
        let mut q = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
        q.push(b"msg", MsgType::String).unwrap();
    }

    let info_queue_path = format!("{}/{}_info_queue", base_path, queue_name);
    let mut content = fs::read(&info_queue_path).expect("info_queue must exist");
    let len = content.len();
    assert!(len > 1 && content[len - 1] == b'\n', "info_queue must end with newline");
    let target = len - 2;
    content[target] = if content[target] == b'9' { b'0' } else { content[target] + 1 };
    fs::write(&info_queue_path, &content).unwrap();

    match Queue::new(&base_path, queue_name, Mode::Read) {
        Err(e) => assert_eq!(e, ErrorQueue::InvalidChecksum),
        Ok(_) => panic!("Queue::new(Read) must return Err when info_queue CRC is corrupted, got Ok"),
    }
}

// Test for phase 1.1: the advisory lock acquired by Queue::new(ReadWrite) must
// stay alive for the whole lifetime of the Queue. A second concurrent
// ReadWrite open on the same queue must therefore fail with AlreadyOpen.
// Currently the lock file handle is dropped at the end of Queue::new so the
// second open succeeds.
#[test]
fn test_double_open_returns_already_open() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_double_open";

    let q1 = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();

    match Queue::new(&base_path, queue_name, Mode::ReadWrite) {
        Err(e) => assert_eq!(e, ErrorQueue::AlreadyOpen, "expected AlreadyOpen, got {:?}", e),
        Ok(_) => panic!("second Queue::new(ReadWrite) must fail while first instance is alive"),
    }

    // Read-only access must remain possible alongside an active writer.
    if Queue::new(&base_path, queue_name, Mode::Read).is_err() {
        panic!("Queue::new(Read) must succeed alongside an active ReadWrite handle");
    }

    drop(q1);

    if Queue::new(&base_path, queue_name, Mode::ReadWrite).is_err() {
        panic!("re-open in ReadWrite mode must succeed after the first instance was dropped");
    }
}

// Test for phase 2.1: Consumer must keep its persisted state in a file whose
// path is derived from the actual queue name, not from the hardcoded
// "individuals-flow" prefix. After a commit + drop + reconnect cycle the
// consumer must resume from the last committed position. Currently the
// existence check uses the hardcoded path while subsequent commit() writes go
// to the per-queue path, so on reconnect the consumer reads back the empty
// hardcoded file and restarts from position 0.
#[test]
fn test_consumer_reconnect_arbitrary_queue_name() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "arbitrary_queue_name"; // intentionally not "individuals-flow"

    {
        let mut q = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
        for i in 0..5 {
            q.push(format!("m{}", i).as_bytes(), MsgType::String).unwrap();
        }
    }

    {
        let mut c = Consumer::new(&base_path, "consumer_x", queue_name).unwrap();
        for _ in 0..2 {
            assert!(c.pop_header(), "pop_header must succeed on first session");
            let mut msg = vec![0; c.record_len()];
            c.pop_body(&mut msg).unwrap();
            assert!(c.commit(), "commit must succeed");
        }
    }

    let mut c = Consumer::new(&base_path, "consumer_x", queue_name).unwrap();
    let mut received = Vec::new();
    while c.pop_header() {
        let mut msg = vec![0; c.record_len()];
        c.pop_body(&mut msg).unwrap();
        received.push(String::from_utf8(msg).unwrap());
        c.commit();
    }

    assert_eq!(received, vec!["m2".to_string(), "m3".to_string(), "m4".to_string()]);
}

// ---------------------------------------------------------------------------
// Tests for the v3 per-part compressed format (0.3.0).
// ---------------------------------------------------------------------------

// A v3 part trains a per-part dictionary during warmup, then compresses
// subsequent message bodies. This test pushes enough small, similar messages
// to cross the warmup saturation threshold, then verifies that:
//   * a dictionary file was frozen for the part,
//   * every message round-trips byte-for-byte through a consumer (covering both
//     the raw warmup records and the later compressed records),
//   * the stored part is smaller than the same records would take uncompressed.
#[test]
fn test_v3_compression_roundtrip() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_compress";

    let total = 5000usize;
    let mut sent = Vec::with_capacity(total);
    {
        let mut q = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
        for i in 0..total {
            let msg = format!(
                "user-event id={} type=click session=sess-{:04} payload={{\"k\":\"value-{}\",\"n\":{},\"ok\":true}}",
                i,
                i % 50,
                i % 13,
                i
            );
            q.push(msg.as_bytes(), MsgType::String).unwrap();
            sent.push(msg);
        }
    }

    // The dictionary must have been trained and frozen for this part.
    let dict_path = format!("{}/{}-0/{}_dict", base_path, queue_name, queue_name);
    assert!(std::path::Path::new(&dict_path).exists(), "expected a trained dictionary at {}", dict_path);

    // Round-trip: every message must read back unchanged.
    let mut consumer = Consumer::new(&base_path, "c", queue_name).unwrap();
    let mut received = Vec::with_capacity(total);
    while consumer.pop_header() {
        let mut msg = vec![0; consumer.record_len()];
        if consumer.pop_body(&mut msg).is_ok() {
            received.push(String::from_utf8(msg).unwrap());
            consumer.commit();
        } else {
            break;
        }
    }
    assert_eq!(received.len(), total, "all messages must be read back");
    assert_eq!(received, sent, "messages must round-trip unchanged");

    // Compactness: the stored part must be smaller than the same records would
    // occupy uncompressed (HEADER_SIZE + body per record).
    let queue_file = format!("{}/{}-0/{}_queue", base_path, queue_name, queue_name);
    let stored = fs::metadata(&queue_file).unwrap().len();
    let raw: u64 = sent.iter().map(|m| (HEADER_SIZE + m.len()) as u64).sum();
    assert!(stored < raw, "compressed part ({} bytes) must be smaller than raw ({} bytes)", stored, raw);
}

// A v3 part must be invisible to a 0.2.9 reader: the per-part info_push uses a
// version token that the old parser rejects, and records use the v3 magic
// marker so the old resync scanner cannot latch onto them.
#[test]
fn test_v3_blocks_old_format_reader() {
    let base_path = create_unique_queue_path("./test-tmp", "queue");
    let queue_name = "test_v3_lockout";

    {
        let mut q = Queue::new(&base_path, queue_name, Mode::ReadWrite).unwrap();
        for i in 0..5 {
            q.push(format!("msg-{}", i).as_bytes(), MsgType::String).unwrap();
        }
    }

    // The info_push must carry the v3 token. The old 0.2.9 parser reads the
    // second ';'-separated field as a u64 (right_edge); with the token in front
    // that field is the queue name, so the parse fails and the old reader
    // refuses the part instead of reading or seeing its records.
    let info_push_path = format!("{}/{}-0/{}_info_push", base_path, queue_name, queue_name);
    let content = fs::read_to_string(&info_push_path).unwrap();
    let line = content.lines().next().unwrap();
    assert!(line.starts_with("v3;"), "v3 info_push must carry the version token, got [{}]", line);
    let fields: Vec<&str> = line.split(';').collect();
    assert!(fields.len() >= 2, "info_push line must have several fields");
    assert!(
        fields[1].parse::<u64>().is_err(),
        "the old parser expects a u64 as the second field; the v3 token must break that, got [{}]",
        fields[1]
    );

    // Records must use the v3 magic marker (offset 12..16 of the header), so a
    // reader scanning for the v2 marker cannot resync into v3 data.
    let queue_file = format!("{}/{}-0/{}_queue", base_path, queue_name, queue_name);
    let bytes = fs::read(&queue_file).unwrap();
    assert!(bytes.len() >= HEADER_SIZE, "queue file must contain at least one record");
    let magic = &bytes[12..16];
    assert_eq!(magic, &MAGIC_MARKER_V3_BYTES, "records must use the v3 magic marker");
    assert_ne!(magic, &MAGIC_MARKER_BYTES, "records must not use the v2 magic marker");
}