#include <spdlog/spdlog.h>
#include <catch2/catch_test_macros.hpp>
#include <yams/crypto/hasher.h>
#include <yams/metadata/connection_pool.h>
#include <yams/metadata/database.h>
#include <yams/storage/storage_engine.h>

#include <array>
#include <atomic>
#include <barrier>
#include <chrono>
#include <iostream>
#include <latch>
#include <random>
#include <thread>
#include <vector>

using namespace yams;
using namespace yams::storage;
using namespace std::chrono_literals;

// Helper function to generate random bytes
static std::vector<std::byte> generateRandomBytes(size_t size) {
    std::vector<std::byte> data(size);
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<> dis(0, 255);
    for (auto& b : data) {
        b = std::byte{static_cast<uint8_t>(dis(gen))};
    }
    return data;
}

struct ConcurrentStorageTest {
    std::unique_ptr<StorageEngine> storage;
    std::filesystem::path storagePath;

    ConcurrentStorageTest() {
        storagePath = std::filesystem::temp_directory_path() / "yams_stress_concurrent_test";
        std::filesystem::create_directories(storagePath);

        StorageConfig config{.basePath = storagePath,
                             .shardDepth = 2,
                             .mutexPoolSize = 1024,
                             .maxConcurrentReaders = 1000,
                             .maxConcurrentWriters = 100};

        storage = std::make_unique<StorageEngine>(std::move(config));
    }

    ~ConcurrentStorageTest() {
        storage.reset();
        std::filesystem::remove_all(storagePath);
    }
};

double runPoolWorkload(yams::metadata::ConnectionPool& pool, std::size_t workers,
                       std::size_t opsPerThread) {
    std::atomic<std::size_t> errors{0};
    auto start = std::chrono::steady_clock::now();
    std::vector<std::thread> threads;
    threads.reserve(workers);
    for (std::size_t t = 0; t < workers; ++t) {
        threads.emplace_back([&, threadIndex = t]() {
            for (std::size_t op = 0; op < opsPerThread; ++op) {
                auto res =
                    pool.withConnection([&](yams::metadata::Database& db) -> yams::Result<void> {
                        auto stmtRes = db.prepare("INSERT INTO kv(test_key, value) VALUES (?, ?);");
                        if (!stmtRes)
                            return stmtRes.error();
                        auto stmt = std::move(stmtRes.value());
                        auto bindRes = stmt.bindAll(static_cast<int>(threadIndex),
                                                    std::string("payload-") + std::to_string(op));
                        if (!bindRes)
                            return bindRes.error();
                        return stmt.execute();
                    });
                if (!res)
                    errors.fetch_add(1, std::memory_order_relaxed);
            }
        });
    }
    for (auto& th : threads)
        th.join();
    auto end = std::chrono::steady_clock::now();
    auto elapsed = std::chrono::duration<double>(end - start).count();
    CHECK(errors.load() == 0U);
    double ops = static_cast<double>(workers * opsPerThread);
    return (elapsed > 0.0) ? ops / elapsed : ops;
}

TEST_CASE_METHOD(ConcurrentStorageTest, "ThousandConcurrentReaders",
                 "[stress][concurrentstoragetest]") {
    // Pre-store test objects with varying sizes
    constexpr size_t numObjects = 100;
    std::vector<std::pair<std::string, std::vector<std::byte>>> testObjects;

    // Test multiple object size categories
    std::vector<size_t> sizeBuckets = {512, 1024, 4096, 8192, 16384};

    for (size_t i = 0; i < numObjects; ++i) {
        size_t sizeCategory = sizeBuckets[i % sizeBuckets.size()];
        auto data = generateRandomBytes(sizeCategory + (i % 100));
        auto hasher = crypto::createSHA256Hasher();
        auto hash = hasher->hash(data);

        testObjects.emplace_back(hash, data);
        REQUIRE(storage->store(hash, data).has_value());
    }

    // Test with different reader counts to validate scalability
    std::vector<size_t> readerCounts = {100, 500, 1000};

    for (size_t numReaders : readerCounts) {
        std::cout << "\n=== Testing with " << numReaders << " concurrent readers ===" << std::endl;

        constexpr size_t readsPerThread = 100;

        std::atomic<size_t> totalReads{0};
        std::atomic<size_t> successfulReads{0};
        std::atomic<bool> hasErrors{false};

        // Use C++20 barrier for synchronization
        std::barrier startBarrier(numReaders + 1);

        auto startTime = std::chrono::steady_clock::now();

        std::vector<std::thread> readers;
        readers.reserve(numReaders);

        for (size_t i = 0; i < numReaders; ++i) {
            readers.emplace_back([&, threadId = i]() {
                std::random_device rd;
                std::mt19937 gen(rd() + threadId);
                std::uniform_int_distribution<size_t> dis(0, numObjects - 1);

                // Wait for all threads to be ready
                startBarrier.arrive_and_wait();

                for (size_t j = 0; j < readsPerThread; ++j) {
                    auto& [hash, expectedData] = testObjects[dis(gen)];

                    totalReads.fetch_add(1);
                    auto result = storage->retrieve(hash);

                    if (!result.has_value()) {
                        hasErrors.store(true);
                        spdlog::error("Read failed for hash: {}", hash);
                    } else if (result.value() != expectedData) {
                        hasErrors.store(true);
                        spdlog::error("Data mismatch for hash: {}", hash);
                    } else {
                        successfulReads.fetch_add(1);
                    }
                }
            });
        }

        // Start all threads simultaneously
        startBarrier.arrive_and_wait();

        // Wait for all readers to complete
        for (auto& reader : readers) {
            reader.join();
        }

        auto endTime = std::chrono::steady_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);

        // Verify results
        INFO("Errors with " << numReaders << " readers");
        CHECK_FALSE(hasErrors.load());
        CHECK(successfulReads.load() == numReaders * readsPerThread);

        // Performance check
        auto readsPerSecond = (totalReads.load() * 1000) / duration.count();
        spdlog::info("Completed {} reads in {} ms ({} reads/sec)", totalReads.load(),
                     duration.count(), readsPerSecond);

        // Verify < 10ms latency requirement
        auto avgLatency = duration.count() / static_cast<double>(readsPerThread);
        INFO("Average latency too high with " << numReaders << " readers");
        CHECK(avgLatency < 15.0);
    }
}

TEST_CASE_METHOD(ConcurrentStorageTest, "HundredConcurrentWriters",
                 "[stress][concurrentstoragetest]") {
    constexpr size_t numWriters = 100;
    constexpr size_t writesPerThread = 50;

    std::atomic<size_t> totalWrites{0};
    std::atomic<size_t> successfulWrites{0};
    std::atomic<bool> hasErrors{false};

    // C++20 latch for completion tracking
    std::latch completionLatch(numWriters);

    auto startTime = std::chrono::steady_clock::now();

    std::vector<std::thread> writers;
    writers.reserve(numWriters);

    for (size_t i = 0; i < numWriters; ++i) {
        writers.emplace_back([&, threadId = i]() {
            for (size_t j = 0; j < writesPerThread; ++j) {
                // Generate unique data for each write
                auto data = generateRandomBytes(512 + (threadId * 100 + j) % 1024);
                auto hasher = crypto::createSHA256Hasher();
                auto hash = hasher->hash(data);

                totalWrites.fetch_add(1);
                auto result = storage->store(hash, data);

                if (!result.has_value()) {
                    hasErrors.store(true);
                    spdlog::error("Write failed for thread {} iteration {}", threadId, j);
                } else {
                    successfulWrites.fetch_add(1);
                }
            }

            completionLatch.count_down();
        });
    }

    // Wait for all writers
    for (auto& writer : writers) {
        writer.join();
    }

    auto endTime = std::chrono::steady_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);

    // Verify results
    CHECK_FALSE(hasErrors.load());
    CHECK(successfulWrites.load() == numWriters * writesPerThread);

    // Check storage statistics
    auto stats = storage->getStats();
    CHECK(stats.writeOperations.load() == numWriters * writesPerThread);

    spdlog::info("Completed {} writes in {} ms ({} writes/sec)", totalWrites.load(),
                 duration.count(), (totalWrites.load() * 1000) / duration.count());
}

TEST_CASE_METHOD(ConcurrentStorageTest, "MixedReadWriteWorkload",
                 "[stress][concurrentstoragetest]") {
    // Test multiple read/write ratios
    struct WorkloadConfig {
        size_t numReaders;
        size_t numWriters;
        const char* description;
    };

    std::vector<WorkloadConfig> workloads = {{160, 40, "80/20 read/write (typical)"},
                                             {180, 20, "90/10 read/write (read-heavy)"},
                                             {100, 100, "50/50 read/write (balanced)"},
                                             {50, 150, "25/75 read/write (write-heavy)"}};

    for (const auto& workload : workloads) {
        spdlog::info("Testing workload: {}", workload.description);

        const size_t numThreads = workload.numReaders + workload.numWriters;
        constexpr size_t operationsPerThread = 100;

        // Pre-populate with some data
        std::vector<std::string> existingHashes;
        for (size_t i = 0; i < 500; ++i) {
            auto data = generateRandomBytes(1024 + (i % 1024));
            auto hasher = crypto::createSHA256Hasher();
            auto hash = hasher->hash(data);
            existingHashes.push_back(hash);
            storage->store(hash, data);
        }

        std::atomic<size_t> readOps{0};
        std::atomic<size_t> writeOps{0};
        std::atomic<size_t> conflicts{0};

        std::barrier startBarrier(numThreads + 1);
        std::vector<std::thread> threads;

        // Reader threads
        for (size_t i = 0; i < workload.numReaders; ++i) {
            threads.emplace_back([&]() {
                std::random_device rd;
                std::mt19937 gen(rd());
                std::uniform_int_distribution<size_t> dis(0, existingHashes.size() - 1);

                startBarrier.arrive_and_wait();

                for (size_t j = 0; j < operationsPerThread; ++j) {
                    auto& hash = existingHashes[dis(gen)];
                    storage->retrieve(hash);
                    readOps.fetch_add(1);
                }
            });
        }

        // Writer threads
        for (size_t i = 0; i < workload.numWriters; ++i) {
            threads.emplace_back([&, threadId = i]() {
                startBarrier.arrive_and_wait();

                for (size_t j = 0; j < operationsPerThread; ++j) {
                    auto data = generateRandomBytes(2048);
                    auto hasher = crypto::createSHA256Hasher();
                    auto hash = hasher->hash(data);

                    auto result = storage->store(hash, data);
                    if (result.has_value()) {
                        writeOps.fetch_add(1);
                    } else {
                        conflicts.fetch_add(1);
                    }
                }
            });
        }

        auto startTime = std::chrono::steady_clock::now();
        startBarrier.arrive_and_wait();

        for (auto& thread : threads) {
            thread.join();
        }

        auto endTime = std::chrono::steady_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);

        spdlog::info("  Completed: {} reads, {} writes in {} ms", readOps.load(), writeOps.load(),
                     duration.count());

        INFO("Failed for workload: " << workload.description);
        CHECK(readOps.load() == workload.numReaders * operationsPerThread);
        INFO("Failed for workload: " << workload.description);
        CHECK(writeOps.load() == workload.numWriters * operationsPerThread);
        INFO("Conflicts in workload: " << workload.description);
        CHECK(conflicts.load() == 0u);

        // Cleanup for next workload
        existingHashes.clear();
    }
}

TEST_CASE_METHOD(ConcurrentStorageTest, "HighContentionSameObjects",
                 "[stress][concurrentstoragetest]") {
    // Test with different contention levels
    struct ContentionConfig {
        size_t numThreads;
        size_t numObjects;
        const char* description;
    };

    std::vector<ContentionConfig> configs = {
        {50, 10, "High contention (50 threads, 10 objects)"},
        {100, 20, "Very high contention (100 threads, 20 objects)"},
        {25, 50, "Moderate contention (25 threads, 50 objects)"},
        {50, 100, "Low contention (50 threads, 100 objects)"}};

    for (const auto& config : configs) {
        spdlog::info("Testing: {}", config.description);

        constexpr size_t opsPerThread = 200;

        // Create a set of objects
        std::vector<std::pair<std::string, std::vector<std::byte>>> objects;
        for (size_t i = 0; i < config.numObjects; ++i) {
            auto data = generateRandomBytes(4096 + (i % 1024));
            auto hasher = crypto::createSHA256Hasher();
            auto hash = hasher->hash(data);
            objects.emplace_back(hash, data);
            storage->store(hash, data);
        }

        std::atomic<size_t> operations{0};
        std::atomic<bool> dataCorruption{false};
        std::atomic<size_t> readOps{0};
        std::atomic<size_t> writeOps{0};

        std::vector<std::thread> threads;
        for (size_t i = 0; i < config.numThreads; ++i) {
            threads.emplace_back([&]() {
                std::random_device rd;
                std::mt19937 gen(rd());
                std::uniform_int_distribution<size_t> objDis(0, config.numObjects - 1);
                std::uniform_int_distribution<int> opDis(0, 9);

                for (size_t j = 0; j < opsPerThread; ++j) {
                    auto& [hash, expectedData] = objects[objDis(gen)];

                    if (opDis(gen) < 8) { // 80% reads
                        auto result = storage->retrieve(hash);
                        if (!result.has_value() || result.value() != expectedData) {
                            dataCorruption.store(true);
                        }
                        readOps.fetch_add(1);
                    } else { // 20% re-writes (same data)
                        storage->store(hash, expectedData);
                        writeOps.fetch_add(1);
                    }

                    operations.fetch_add(1);
                }
            });
        }

        for (auto& thread : threads) {
            thread.join();
        }

        spdlog::info("  Ops: {} ({} reads, {} writes)", operations.load(), readOps.load(),
                     writeOps.load());

        INFO("Data corruption in: " << config.description);
        CHECK_FALSE(dataCorruption.load());
        CHECK(operations.load() == config.numThreads * opsPerThread);
    }
}

TEST_CASE_METHOD(ConcurrentStorageTest, "RapidCreateDeleteCycles",
                 "[stress][concurrentstoragetest]") {
    constexpr size_t numThreads = 20;
    constexpr size_t cyclesPerThread = 50;

    std::atomic<size_t> creates{0};
    std::atomic<size_t> deletes{0};
    std::atomic<bool> errors{false};

    std::vector<std::thread> threads;
    for (size_t i = 0; i < numThreads; ++i) {
        threads.emplace_back([&, threadId = i]() {
            for (size_t j = 0; j < cyclesPerThread; ++j) {
                // Generate unique data
                auto data = generateRandomBytes(1024 + threadId * 100 + j);
                auto hasher = crypto::createSHA256Hasher();
                auto hash = hasher->hash(data);

                // Create
                if (!storage->store(hash, data).has_value()) {
                    errors.store(true);
                    continue;
                }
                creates.fetch_add(1);

                // Verify
                auto result = storage->retrieve(hash);
                if (!result.has_value() || result.value() != data) {
                    errors.store(true);
                    continue;
                }

                // Delete
                if (!storage->remove(hash).has_value()) {
                    errors.store(true);
                    continue;
                }
                deletes.fetch_add(1);

                // Verify deletion
                auto existsRes = storage->exists(hash);
                if (!existsRes || existsRes.value()) {
                    errors.store(true);
                }
            }
        });
    }

    for (auto& thread : threads) {
        thread.join();
    }

    CHECK_FALSE(errors.load());
    CHECK(creates.load() == numThreads * cyclesPerThread);
    CHECK(deletes.load() == numThreads * cyclesPerThread);

    // Final storage should be empty
    auto stats = storage->getStats();
    CHECK(stats.totalObjects.load() == 0u);
}

TEST_CASE_METHOD(ConcurrentStorageTest, "LatencyUnderLoad", "[stress][concurrentstoragetest]") {
    // Pre-populate storage
    for (size_t i = 0; i < 10000; ++i) {
        auto data = generateRandomBytes(512);
        auto hasher = crypto::createSHA256Hasher();
        auto hash = hasher->hash(data);
        storage->store(hash, data);
    }

    // Measure individual operation latencies under concurrent load
    constexpr size_t numBackgroundThreads = 50;
    constexpr size_t numMeasurements = 1000;

    std::atomic<bool> stopFlag{false};
    std::vector<std::thread> backgroundThreads;

    // Start background load
    for (size_t i = 0; i < numBackgroundThreads; ++i) {
        backgroundThreads.emplace_back([&]() {
            while (!stopFlag.load()) {
                auto data = generateRandomBytes(1024);
                auto hasher = crypto::createSHA256Hasher();
                auto hash = hasher->hash(data);
                storage->store(hash, data);
                storage->retrieve(hash);
            }
        });
    }

    // Measure latencies
    std::vector<double> latencies;
    latencies.reserve(numMeasurements);

    auto data = generateRandomBytes(1024);
    auto hasher = crypto::createSHA256Hasher();
    auto hash = hasher->hash(data);
    storage->store(hash, data);

    for (size_t i = 0; i < numMeasurements; ++i) {
        auto start = std::chrono::high_resolution_clock::now();
        auto exists = storage->exists(hash);
        auto end = std::chrono::high_resolution_clock::now();

        REQUIRE(exists.has_value());

        auto latency = std::chrono::duration<double, std::milli>(end - start).count();
        latencies.push_back(latency);
    }

    // Stop background threads
    stopFlag.store(true);
    for (auto& thread : backgroundThreads) {
        thread.join();
    }

    // Calculate statistics
    std::sort(latencies.begin(), latencies.end());
    double p50 = latencies[latencies.size() / 2];
    double p95 = latencies[latencies.size() * 95 / 100];
    double p99 = latencies[latencies.size() * 99 / 100];

    spdlog::info(
        "Existence check latencies under load - P50: {:.2f}ms, P95: {:.2f}ms, P99: {:.2f}ms", p50,
        p95, p99);

    // Verify < 10ms requirement
    CHECK(p99 < 10.0);
}

TEST_CASE_METHOD(ConcurrentStorageTest, "ConnectionPoolWorkerSweep",
                 "[stress][concurrentstoragetest]") {
    const auto dbPath = storagePath / "pool_bench.db";
    yams::metadata::Database bootstrap;
    REQUIRE(bootstrap.open(dbPath.string(), yams::metadata::ConnectionMode::Create));
    REQUIRE(bootstrap.execute("PRAGMA journal_mode=WAL;").has_value());
    REQUIRE(bootstrap.execute("CREATE TABLE IF NOT EXISTS kv (test_key INTEGER, value TEXT);")
                .has_value());
    bootstrap.close();

    yams::metadata::ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 8;
    cfg.busyTimeout = std::chrono::milliseconds(1000);
    yams::metadata::ConnectionPool pool(dbPath.string(), cfg);
    REQUIRE(pool.initialize());

    const std::array<std::size_t, 4> workerSweep{1, 2, 4, 8};
    const std::size_t opsPerThread = 150;
    double baseline = 0.0;

    for (std::size_t workers : workerSweep) {
        auto resetRes = pool.withConnection([](yams::metadata::Database& db) -> yams::Result<void> {
            return db.execute("DELETE FROM kv;");
        });
        REQUIRE(resetRes);

        double throughput = runPoolWorkload(pool, workers, opsPerThread);
        spdlog::info("pool sweep workers={} throughput={}", workers, throughput);
        if (baseline == 0.0) {
            baseline = throughput;
        } else if (baseline > 0.0) {
            double warnThreshold = baseline * 0.9;
            double cautionThreshold = baseline * 0.5;
            double failThreshold = baseline * 0.25;
            if (throughput < failThreshold) {
                FAIL_CHECK("connection pool regression for " + std::to_string(workers) +
                           " workers (" + std::to_string(throughput) + " < " +
                           std::to_string(failThreshold) + ")");
            } else if (throughput < cautionThreshold) {
                spdlog::warn(
                    "connection pool throughput dipped below 50% for workers={} ({} vs {} base)",
                    workers, throughput, baseline);
            } else if (throughput < warnThreshold) {
                spdlog::info(
                    "connection pool throughput dipped below 90% for workers={} ({} vs {} base)",
                    workers, throughput, baseline);
            }
        }
    }

    pool.shutdown();
    std::filesystem::remove(dbPath);
}

// New comprehensive stress test: Variable object sizes under concurrent access
TEST_CASE_METHOD(ConcurrentStorageTest, "VariableObjectSizesConcurrent",
                 "[stress][concurrentstoragetest]") {
    spdlog::info("Testing variable object sizes under concurrent access");

    struct SizeClass {
        size_t minSize;
        size_t maxSize;
        size_t count;
        const char* name;
    };

    std::vector<SizeClass> sizeClasses = {{100, 1024, 100, "Tiny (100B-1KB)"},
                                          {1024, 10 * 1024, 100, "Small (1KB-10KB)"},
                                          {10 * 1024, 100 * 1024, 50, "Medium (10KB-100KB)"},
                                          {100 * 1024, 1024 * 1024, 20, "Large (100KB-1MB)"}};

    // Pre-populate with objects of varying sizes
    std::vector<std::pair<std::string, std::vector<std::byte>>> allObjects;

    for (const auto& sizeClass : sizeClasses) {
        std::random_device rd;
        std::mt19937 gen(rd());
        std::uniform_int_distribution<size_t> sizeDist(sizeClass.minSize, sizeClass.maxSize);

        for (size_t i = 0; i < sizeClass.count; ++i) {
            auto data = generateRandomBytes(sizeDist(gen));
            auto hasher = crypto::createSHA256Hasher();
            auto hash = hasher->hash(data);
            allObjects.emplace_back(hash, data);
            REQUIRE(storage->store(hash, data).has_value());
        }
    }

    spdlog::info("Pre-populated {} objects across {} size classes", allObjects.size(),
                 sizeClasses.size());

    // Concurrent access test
    constexpr size_t numThreads = 50;
    constexpr size_t opsPerThread = 100;

    std::atomic<size_t> successfulOps{0};
    std::atomic<size_t> failedOps{0};
    std::atomic<size_t> corruptedReads{0};

    std::barrier startBarrier(numThreads + 1);
    std::vector<std::thread> threads;

    for (size_t i = 0; i < numThreads; ++i) {
        threads.emplace_back([&]() {
            std::random_device rd;
            std::mt19937 gen(rd());
            std::uniform_int_distribution<size_t> objDist(0, allObjects.size() - 1);

            startBarrier.arrive_and_wait();

            for (size_t j = 0; j < opsPerThread; ++j) {
                auto& [hash, expectedData] = allObjects[objDist(gen)];

                auto result = storage->retrieve(hash);
                if (!result.has_value()) {
                    failedOps.fetch_add(1);
                } else if (result.value() != expectedData) {
                    corruptedReads.fetch_add(1);
                } else {
                    successfulOps.fetch_add(1);
                }
            }
        });
    }

    startBarrier.arrive_and_wait();

    for (auto& thread : threads) {
        thread.join();
    }

    spdlog::info("Variable size test: {} successful, {} failed, {} corrupted", successfulOps.load(),
                 failedOps.load(), corruptedReads.load());

    CHECK(successfulOps.load() == numThreads * opsPerThread);
    CHECK(failedOps.load() == 0u);
    CHECK(corruptedReads.load() == 0u);
}

// New stress test: Burst traffic patterns
TEST_CASE_METHOD(ConcurrentStorageTest, "BurstTrafficPattern", "[stress][concurrentstoragetest]") {
    spdlog::info("Testing burst traffic patterns");

    // Pre-populate
    std::vector<std::pair<std::string, std::vector<std::byte>>> objects;
    for (size_t i = 0; i < 100; ++i) {
        auto data = generateRandomBytes(2048);
        auto hasher = crypto::createSHA256Hasher();
        auto hash = hasher->hash(data);
        objects.emplace_back(hash, data);
        storage->store(hash, data);
    }

    struct BurstConfig {
        size_t numThreads;
        size_t opsPerThread;
        std::chrono::milliseconds burstDuration;
        std::chrono::milliseconds idleDuration;
        const char* name;
    };

    std::vector<BurstConfig> bursts = {{100, 50, 100ms, 200ms, "Quick burst"},
                                       {200, 25, 50ms, 150ms, "Intense burst"},
                                       {50, 100, 200ms, 100ms, "Sustained burst"}};

    for (const auto& burst : bursts) {
        spdlog::info("  Testing: {}", burst.name);

        std::atomic<size_t> totalOps{0};
        std::atomic<bool> hasError{false};

        // Burst phase
        std::vector<std::thread> threads;
        for (size_t i = 0; i < burst.numThreads; ++i) {
            threads.emplace_back([&]() {
                std::random_device rd;
                std::mt19937 gen(rd());
                std::uniform_int_distribution<size_t> objDist(0, objects.size() - 1);

                for (size_t j = 0; j < burst.opsPerThread; ++j) {
                    auto& [hash, expectedData] = objects[objDist(gen)];
                    auto result = storage->retrieve(hash);

                    if (!result.has_value() || result.value() != expectedData) {
                        hasError.store(true);
                    }
                    totalOps.fetch_add(1);
                }
            });
        }

        auto start = std::chrono::steady_clock::now();
        for (auto& thread : threads) {
            thread.join();
        }
        auto elapsed = std::chrono::steady_clock::now() - start;

        spdlog::info("    Completed {} ops in {}ms", totalOps.load(),
                     std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count());

        INFO("Errors in burst: " << burst.name);
        CHECK_FALSE(hasError.load());
        CHECK(totalOps.load() == burst.numThreads * burst.opsPerThread);

        // Idle phase
        std::this_thread::sleep_for(burst.idleDuration);
    }
}

// New stress test: Long-running sustained load
TEST_CASE_METHOD(ConcurrentStorageTest, "LongRunningSustainedLoad",
                 "[stress][concurrentstoragetest]") {
    spdlog::info("Testing long-running sustained load");

    constexpr size_t numThreads = 20;
    constexpr auto testDuration = 30s; // 30 seconds of sustained load

    // Pre-populate
    std::vector<std::pair<std::string, std::vector<std::byte>>> objects;
    for (size_t i = 0; i < 200; ++i) {
        auto data = generateRandomBytes(1024 + (i % 2048));
        auto hasher = crypto::createSHA256Hasher();
        auto hash = hasher->hash(data);
        objects.emplace_back(hash, data);
        storage->store(hash, data);
    }

    std::atomic<bool> stopFlag{false};
    std::atomic<size_t> totalOperations{0};
    std::atomic<size_t> readOps{0};
    std::atomic<size_t> writeOps{0};
    std::atomic<size_t> errors{0};

    std::vector<std::thread> threads;

    auto startTime = std::chrono::steady_clock::now();

    for (size_t i = 0; i < numThreads; ++i) {
        threads.emplace_back([&, threadId = i]() {
            std::random_device rd;
            std::mt19937 gen(rd() + threadId);
            std::uniform_int_distribution<size_t> objDist(0, objects.size() - 1);
            std::uniform_int_distribution<int> opTypeDist(0, 99);

            while (!stopFlag.load()) {
                int opType = opTypeDist(gen);

                if (opType < 70) { // 70% reads
                    auto& [hash, expectedData] = objects[objDist(gen)];
                    auto result = storage->retrieve(hash);

                    if (!result.has_value() || result.value() != expectedData) {
                        errors.fetch_add(1);
                    }
                    readOps.fetch_add(1);
                } else { // 30% writes (new objects)
                    auto data = generateRandomBytes(1024);
                    auto hasher = crypto::createSHA256Hasher();
                    auto hash = hasher->hash(data);

                    if (storage->store(hash, data).has_value()) {
                        writeOps.fetch_add(1);
                    } else {
                        errors.fetch_add(1);
                    }
                }

                totalOperations.fetch_add(1);
            }
        });
    }

    // Let it run for the test duration
    std::this_thread::sleep_for(testDuration);
    stopFlag.store(true);

    for (auto& thread : threads) {
        thread.join();
    }

    auto endTime = std::chrono::steady_clock::now();
    auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(endTime - startTime).count();

    spdlog::info("Long-running test completed:");
    spdlog::info("  Duration: {}s", elapsed);
    spdlog::info("  Total ops: {}", totalOperations.load());
    spdlog::info("  Reads: {}, Writes: {}", readOps.load(), writeOps.load());
    spdlog::info("  Ops/sec: {}", totalOperations.load() / elapsed);
    spdlog::info("  Errors: {}", errors.load());

    CHECK(errors.load() == 0u);
    CHECK(totalOperations.load() > numThreads * 100); // Should do many operations
}