/*
 * SPDX-FileCopyrightText: 2026 Copyright (c) Contributors to the Eclipse Foundation
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Apache License Version 2.0 which is available at
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * SPDX-License-Identifier: Apache-2.0
 */

//! Integration tests for the local filesystem storage backend.

use cda_interfaces::storage_api::{
    Collection as _, CollectionName, RandomAccessData as _, Storage, StorageError,
};
use cda_storage::LocalStorage;

/// Create a fresh `LocalStorage` in a unique temp directory.
fn create_test_storage() -> (LocalStorage, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("Failed to create temp dir");
    let storage = LocalStorage::new(dir.path()).expect("Failed to create LocalStorage");
    (storage, dir)
}

#[tokio::test]
async fn write_read_roundtrip() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"hello world";
    collection
        .write(&mut tx, "greeting", &mut data)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let handle = collection.read("greeting").await.unwrap();
    let mut buf = vec![0u8; 11];
    let n = handle.read_at(0, &mut buf).unwrap();
    assert_eq!(n, 11);
    assert_eq!(&buf, b"hello world");
}

#[tokio::test]
async fn read_nonexistent_key_returns_not_found() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let result = collection.read("nonexistent").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));
}

#[tokio::test]
async fn delete_removes_key_after_commit() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    // Write a key.
    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"to be deleted";
    collection
        .write(&mut tx, "doomed", &mut data)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Delete the key.
    let mut tx = storage.begin_transaction().unwrap();
    collection.delete(&mut tx, "doomed").await.unwrap();
    tx.commit().await.unwrap();

    let result = collection.read("doomed").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));
}

#[tokio::test]
async fn delete_all_removes_all_keys() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    // Write multiple keys.
    let mut tx = storage.begin_transaction().unwrap();
    let mut d1: &[u8] = b"one";
    let mut d2: &[u8] = b"two";
    collection.write(&mut tx, "key1", &mut d1).await.unwrap();
    collection.write(&mut tx, "key2", &mut d2).await.unwrap();
    tx.commit().await.unwrap();

    // Delete all.
    let mut tx = storage.begin_transaction().unwrap();
    collection.delete_all(&mut tx).await.unwrap();
    tx.commit().await.unwrap();

    let is_empty = collection.is_empty().await.unwrap();
    assert!(is_empty);
}

// Transaction semantics
#[tokio::test]
async fn rollback_discards_writes() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"should not persist";
    collection
        .write(&mut tx, "ephemeral", &mut data)
        .await
        .unwrap();
    tx.rollback();

    let result = collection.read("ephemeral").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));
}

#[tokio::test]
async fn drop_without_commit_is_implicit_rollback() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    {
        let mut tx = storage.begin_transaction().unwrap();
        let mut data: &[u8] = b"dropped";
        collection.write(&mut tx, "ghost", &mut data).await.unwrap();
        // tx is dropped here without commit.
    }

    let result = collection.read("ghost").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));
}

#[tokio::test]
async fn no_read_your_writes() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"uncommitted";
    collection
        .write(&mut tx, "pending", &mut data)
        .await
        .unwrap();

    // Read should not see the uncommitted write.
    let result = collection.read("pending").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));

    tx.commit().await.unwrap();

    // Now it should be visible.
    let handle = collection.read("pending").await.unwrap();
    assert_eq!(handle.data_size().unwrap(), 11);
}

#[tokio::test]
async fn single_transaction_enforcement() {
    let (storage, _dir) = create_test_storage();

    let _tx = storage.begin_transaction().unwrap();
    let result = storage.begin_transaction();
    assert!(matches!(result, Err(StorageError::TransactionBusy)));
}

#[tokio::test]
async fn can_begin_transaction_after_previous_commits() {
    let (storage, _dir) = create_test_storage();

    let tx = storage.begin_transaction().unwrap();
    tx.commit().await.unwrap();

    // Should be able to begin a new transaction now.
    let tx2 = storage.begin_transaction().unwrap();
    tx2.commit().await.unwrap();
}

#[tokio::test]
async fn can_begin_transaction_after_previous_rollback() {
    let (storage, _dir) = create_test_storage();

    let tx = storage.begin_transaction().unwrap();
    tx.rollback();

    let tx2 = storage.begin_transaction().unwrap();
    tx2.commit().await.unwrap();
}

#[tokio::test]
async fn opening_missing_root_fails_without_creating_it() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("not_mounted");

    let result = LocalStorage::new(&root);

    assert!(
        matches!(result, Err(StorageError::Io(ref e)) if e.kind() == std::io::ErrorKind::NotFound)
    );
    assert!(!root.exists());
}

#[tokio::test]
async fn writes_never_recreate_a_vanished_root() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("mounted");
    std::fs::create_dir(&root).unwrap();
    let storage = LocalStorage::new(&root).unwrap();

    // E.g. the storage is unmounted while the CDA runs.
    std::fs::remove_dir(&root).unwrap();

    assert!(matches!(
        storage.begin_transaction(),
        Err(StorageError::Io(_))
    ));
    assert!(matches!(
        storage
            .get_or_create_collection(&CollectionName::DiagnosticDatabase)
            .await,
        Err(StorageError::Io(_))
    ));
    assert!(!root.exists());

    // A failed start must not leave the storage busy.
    std::fs::create_dir(&root).unwrap();
    storage.begin_transaction().unwrap();
}

#[tokio::test]
async fn case_insensitive_keys() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"case test";
    collection
        .write(&mut tx, "FooBar", &mut data)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Reading with different cases should find the same entry.
    let handle = collection.read("foobar").await.unwrap();
    assert_eq!(handle.data_size().unwrap(), 9);

    let handle = collection.read("FOOBAR").await.unwrap();
    assert_eq!(handle.data_size().unwrap(), 9);
}

// Collection management
#[tokio::test]
async fn get_nonexistent_collection_returns_not_found() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::Custom("nonexistent".to_string());
    let result = storage.get_collection(&name).await;
    assert!(matches!(result, Err(StorageError::CollectionNotFound(_))));
}

#[tokio::test]
async fn create_and_delete_collection() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::Custom("temp_collection".to_string());

    let mut tx = storage.begin_transaction().unwrap();
    let collection = storage.create_collection(&mut tx, &name).await.unwrap();

    // Write data into the new collection (before commit, the dir doesn't exist yet for reads
    // but the write goes to staging).
    let mut data: &[u8] = b"in new collection";
    collection.write(&mut tx, "item", &mut data).await.unwrap();
    tx.commit().await.unwrap();

    // Should be readable now. Drop the handle before the next transaction to avoid deadlock.
    let collection = storage.get_collection(&name).await.unwrap();
    {
        let handle = collection.read("item").await.unwrap();
        assert_eq!(handle.data_size().unwrap(), 17);
    }

    // Delete the collection.
    let mut tx = storage.begin_transaction().unwrap();
    storage.delete_collection(&mut tx, &name).await.unwrap();
    tx.commit().await.unwrap();

    let result = storage.get_collection(&name).await;
    assert!(matches!(result, Err(StorageError::CollectionNotFound(_))));
}

#[tokio::test]
async fn copy_collection() {
    let (storage, _dir) = create_test_storage();
    let source = CollectionName::DiagnosticDatabase;
    let dest = CollectionName::DiagnosticDatabaseBackup;

    // Write some data to source.
    let source_col = storage.get_or_create_collection(&source).await.unwrap();
    let mut tx = storage.begin_transaction().unwrap();
    let mut d1: &[u8] = b"alpha";
    let mut d2: &[u8] = b"beta";
    source_col.write(&mut tx, "a", &mut d1).await.unwrap();
    source_col.write(&mut tx, "b", &mut d2).await.unwrap();
    tx.commit().await.unwrap();

    // Copy source to dest.
    let mut tx = storage.begin_transaction().unwrap();
    storage
        .copy_collection(&mut tx, &source, &dest)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Verify dest has the same data.
    let dest_col = storage.get_collection(&dest).await.unwrap();
    let handle_a = dest_col.read("a").await.unwrap();
    let mut buf_a = vec![0u8; 5];
    handle_a.read_at(0, &mut buf_a).unwrap();
    assert_eq!(&buf_a, b"alpha");

    let handle_b = dest_col.read("b").await.unwrap();
    let mut buf_b = vec![0u8; 4];
    handle_b.read_at(0, &mut buf_b).unwrap();
    assert_eq!(&buf_b, b"beta");
}

#[tokio::test]
async fn copy_collection_replaces_existing_dest() {
    let (storage, _dir) = create_test_storage();
    let source = CollectionName::DiagnosticDatabase;
    let dest = CollectionName::DiagnosticDatabaseBackup;

    // Write data to source.
    let source_col = storage.get_or_create_collection(&source).await.unwrap();
    let mut tx = storage.begin_transaction().unwrap();
    let mut d1: &[u8] = b"alpha";
    source_col.write(&mut tx, "a", &mut d1).await.unwrap();
    tx.commit().await.unwrap();

    // Write different data to dest (pre-existing content that should be replaced).
    let dest_col = storage.get_or_create_collection(&dest).await.unwrap();
    let mut tx = storage.begin_transaction().unwrap();
    let mut d_old: &[u8] = b"old_stuff";
    let mut d_extra: &[u8] = b"extra_file";
    dest_col.write(&mut tx, "a", &mut d_old).await.unwrap();
    dest_col
        .write(&mut tx, "extra", &mut d_extra)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Copy source -> dest. Dest should be fully replaced.
    let mut tx = storage.begin_transaction().unwrap();
    storage
        .copy_collection(&mut tx, &source, &dest)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Verify dest has exactly source's contents (key "a" with "alpha"), not the old data.
    let dest_col = storage.get_collection(&dest).await.unwrap();
    let handle_a = dest_col.read("a").await.unwrap();
    let mut buf = vec![0u8; 5];
    handle_a.read_at(0, &mut buf).unwrap();
    assert_eq!(&buf, b"alpha");

    // The "extra" key that was only in dest should be gone (replaced, not merged).
    let result = dest_col.read("extra").await;
    assert!(matches!(result, Err(StorageError::KeyNotFound(_))));
}

// Metadata, list, len
#[tokio::test]
async fn metadata_returns_correct_size() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"twelve chars";
    collection.write(&mut tx, "sized", &mut data).await.unwrap();
    tx.commit().await.unwrap();

    let meta = collection.metadata("sized").await.unwrap();
    assert_eq!(meta.name, "sized");
    assert_eq!(meta.data_size, 12);
}

#[tokio::test]
async fn list_and_len_reflect_committed_state() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    assert!(collection.is_empty().await.unwrap());
    assert_eq!(collection.len().await.unwrap(), 0);

    let mut tx = storage.begin_transaction().unwrap();
    let mut d1: &[u8] = b"x";
    let mut d2: &[u8] = b"y";
    let mut d3: &[u8] = b"z";
    collection.write(&mut tx, "one", &mut d1).await.unwrap();
    collection.write(&mut tx, "two", &mut d2).await.unwrap();
    collection.write(&mut tx, "three", &mut d3).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(collection.len().await.unwrap(), 3);
    assert!(!collection.is_empty().await.unwrap());

    let mut keys = collection.list().await.unwrap();
    keys.sort();
    assert_eq!(keys, vec!["one", "three", "two"]);
}

// Random access data
#[tokio::test]
async fn random_access_read_at_offset() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"0123456789ABCDEF";
    collection.write(&mut tx, "hex", &mut data).await.unwrap();
    tx.commit().await.unwrap();

    let handle = collection.read("hex").await.unwrap();
    assert_eq!(handle.data_size().unwrap(), 16);

    // Read from offset 10.
    let mut buf = vec![0u8; 6];
    let n = handle.read_at(10, &mut buf).unwrap();
    assert_eq!(n, 6);
    assert_eq!(&buf, b"ABCDEF");

    // Read from offset 14..only 2 bytes remaining.
    let mut buf = vec![0u8; 6];
    let n = handle.read_at(14, &mut buf).unwrap();
    assert_eq!(n, 2);
    assert_eq!(buf.get(..2).unwrap(), b"EF");
}

#[tokio::test]
async fn overwrite_existing_key() {
    let (storage, _dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;
    let collection = storage.get_or_create_collection(&name).await.unwrap();

    // Write initial value.
    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"version 1";
    collection
        .write(&mut tx, "config", &mut data)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    // Overwrite with new value.
    let mut tx = storage.begin_transaction().unwrap();
    let mut data: &[u8] = b"version 2";
    collection
        .write(&mut tx, "config", &mut data)
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let handle = collection.read("config").await.unwrap();
    let mut buf = vec![0u8; 9];
    handle.read_at(0, &mut buf).unwrap();
    assert_eq!(&buf, b"version 2");
}

/// Verifies the WAL on-disk format at the byte level: appends a mix of `CreateCollection`,
/// `Write`, and `Delete` operations, then reads them back and checks that each operation is
/// decoded correctly, in order, with its checksum intact and `truncated` reported as `false`.
/// This underpins the crash-recovery guarantees, which depend on being able to faithfully replay
/// exactly the operations that were durably recorded before a crash.
/// [[ test~storage-wal-checksum-round-trip, WAL entries round-trip through checksum-verified encode/decode, test ]]
// WAL checksum round-trip
#[tokio::test]
async fn wal_round_trip_with_checksum_verification() {
    let dir = tempfile::tempdir().unwrap();
    let wal_path = dir.path().join("test.wal");
    cda_storage::wal::create_wal(&wal_path).unwrap();

    let ops = vec![
        cda_interfaces::storage_api::Operation::CreateCollection {
            name: CollectionName::Custom("col_a".to_string()),
        },
        cda_interfaces::storage_api::Operation::Write {
            collection: CollectionName::DiagnosticDatabase,
            key: "my_key".to_string(),
            staged_path: "/tmp/fake.tmp".to_string(),
        },
        cda_interfaces::storage_api::Operation::Delete {
            collection: CollectionName::DiagnosticDatabase,
            key: "old_key".to_string(),
        },
    ];

    for op in &ops {
        cda_storage::wal::append_operation(&wal_path, op).unwrap();
    }

    let wal_data = std::fs::read(&wal_path).unwrap();
    let wal_result = cda_storage::wal::read_wal(&wal_data).unwrap();
    assert!(!wal_result.truncated);
    assert_eq!(wal_result.operations.len(), 3);

    // Verify the operations match using .get() to satisfy clippy::indexing_slicing.
    let read_op_0 = wal_result
        .operations
        .first()
        .expect("Expected 3 operations");
    assert!(matches!(
        read_op_0,
        cda_interfaces::storage_api::Operation::CreateCollection { name }
        if name.as_str() == "col_a"
    ));
    let read_op_1 = wal_result.operations.get(1).expect("Expected 3 operations");
    assert!(matches!(
        read_op_1,
        cda_interfaces::storage_api::Operation::Write { key, .. }
        if key == "my_key"
    ));
    let read_op_2 = wal_result.operations.get(2).expect("Expected 3 operations");
    assert!(matches!(
        read_op_2,
        cda_interfaces::storage_api::Operation::Delete { key, .. }
        if key == "old_key"
    ));
}

/// Writes two valid WAL entries, then truncates the file mid-way through the second entry to
/// simulate a torn write. The reader must return exactly the first (fully valid) operation and
/// report `truncated: true`, rather than erroring out or fabricating data for the incomplete
/// entry. This is the detection mechanism that lets recovery safely distinguish "crash mid-write
/// to the WAL itself" from genuine filesystem corruption.
/// [[ test~storage-wal-truncation-detection, WAL reader detects and stops at a truncated entry, test ]]
#[tokio::test]
async fn wal_stops_at_truncated_entry() {
    let dir = tempfile::tempdir().unwrap();
    let wal_path = dir.path().join("test.wal");
    cda_storage::wal::create_wal(&wal_path).unwrap();

    // Write two valid entries.
    cda_storage::wal::append_operation(
        &wal_path,
        &cda_interfaces::storage_api::Operation::CreateCollection {
            name: CollectionName::Custom("first".to_string()),
        },
    )
    .unwrap();
    cda_storage::wal::append_operation(
        &wal_path,
        &cda_interfaces::storage_api::Operation::CreateCollection {
            name: CollectionName::Custom("second".to_string()),
        },
    )
    .unwrap();

    // Truncate the file to corrupt the second entry (keep first + partial second).
    let data = std::fs::read(&wal_path).unwrap();
    // Write only 80% of the data to truncate the second entry.
    let truncated_len = data.len() * 4 / 5;
    std::fs::write(&wal_path, data.get(..truncated_len).unwrap()).unwrap();

    let wal_data = std::fs::read(&wal_path).unwrap();
    let wal_result = cda_storage::wal::read_wal(&wal_data).unwrap();
    assert!(wal_result.truncated);
    // Should only have the first valid entry.
    assert_eq!(wal_result.operations.len(), 1);
    let read_op_0 = wal_result.operations.first().expect("Expected 1 operation");
    assert!(matches!(
        read_op_0,
        cda_interfaces::storage_api::Operation::CreateCollection { name }
        if name.as_str() == "first"
    ));
}

/// Regression test for a bug where `list()` returned a file's on-disk (possibly mixed-case)
/// name verbatim, while `metadata()`/`read()`/`file_path()` normalize keys to lowercase before
/// resolving them to a path. On a case-sensitive filesystem this mismatch caused every
/// mixed-case file to appear in `list()` but then fail with `KeyNotFound` as soon as any
/// per-key lookup (`metadata`, `read`, `file_path`) was attempted - exactly the "key vanished
/// during iteration" symptom, even with no concurrent writer at all.
///
/// Simulates a file that ended up on disk with a mixed-case name outside of `Collection::write`
/// (e.g. placed there during initial provisioning), then verifies that `list()`, `metadata()`,
/// and `read()` are all consistent with each other.
#[tokio::test]
async fn list_normalizes_mixed_case_filenames_to_match_metadata_and_read() {
    let (storage, dir) = create_test_storage();
    let name = CollectionName::DiagnosticDatabase;

    // Create the collection directory and place a mixed-case file directly on disk, bypassing
    // `Collection::write` (which would have normalized the name to lowercase).
    let collection_dir = dir.path().join("collections").join(name.as_str());
    std::fs::create_dir_all(&collection_dir).unwrap();
    std::fs::write(
        collection_dir.join("FLXC1000_06.18.13.mdd"),
        b"mixed case data",
    )
    .unwrap();

    let collection = storage.get_or_create_collection(&name).await.unwrap();

    // `list()` must return the normalized (lowercase) key, matching what `metadata`/`read`/
    // `file_path` expect.
    let keys = collection.list().await.unwrap();
    assert_eq!(keys, vec!["flxc1000_06.18.13.mdd".to_string()]);

    // Every key returned by `list()` must be resolvable via `metadata()` without a
    // `KeyNotFound` error.
    for key in &keys {
        let meta = collection.metadata(key).await.unwrap();
        assert_eq!(meta.data_size, "mixed case data".len() as u64);
    }

    // Every key returned by `list()` must also be resolvable via `read()`.
    let Some(key) = keys.first() else {
        panic!("expected one key");
    };
    let handle = collection.read(key).await.unwrap();
    let mut buf = vec![0u8; "mixed case data".len()];
    handle.read_at(0, &mut buf).unwrap();
    assert_eq!(&buf, b"mixed case data");
}