// Catch2 migration of connection pool tests
// Migration: yams-3s4 / yams-aqc (metadata tests)
// Covers:
//   - connection_pool_maintenance_smoke_test.cpp
//   - connection_pool_waiting_guard_test.cpp

#include <catch2/catch_test_macros.hpp>

#include <sqlite3.h>
#include <spdlog/sinks/ostream_sink.h>
#include <spdlog/spdlog.h>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <filesystem>
#include <future>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>
#include <yams/metadata/connection_pool.h>
#include <yams/metadata/database.h>

using namespace std::chrono_literals;
using namespace yams::metadata;
namespace fs = std::filesystem;

namespace {
std::filesystem::path make_db_path(const std::string& prefix) {
    const char* t = std::getenv("YAMS_TEST_TMPDIR");
    auto base = (t && *t) ? std::filesystem::path(t) : std::filesystem::temp_directory_path();
    std::error_code ec;
    std::filesystem::create_directories(base, ec);
    auto ts = std::chrono::steady_clock::now().time_since_epoch().count();
    auto p = base / (prefix + std::to_string(ts) + ".db");
    std::filesystem::remove(p, ec);
    return p;
}

class CreationGate {
public:
    explicit CreationGate(std::size_t passThroughCalls = 0) : passThroughCalls_(passThroughCalls) {}

    ~CreationGate() { release(); }

    void arriveAndWait() {
        std::unique_lock<std::mutex> lock(mutex_);
        ++calls_;
        cv_.notify_all();
        if (calls_ <= passThroughCalls_) {
            return;
        }
        cv_.wait(lock, [this] { return released_; });
    }

    bool waitForCalls(std::size_t count, std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout, [this, count] { return calls_ >= count; });
    }

    void markContenderCompleted() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            contenderCompleted_ = true;
        }
        cv_.notify_all();
    }

    bool waitForCallsOrContender(std::size_t count, std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout,
                            [this, count] { return calls_ >= count || contenderCompleted_; });
    }

    [[nodiscard]] std::size_t calls() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return calls_;
    }

    [[nodiscard]] bool contenderCompleted() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return contenderCompleted_;
    }

    void release() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            released_ = true;
        }
        cv_.notify_all();
    }

private:
    const std::size_t passThroughCalls_;
    mutable std::mutex mutex_;
    std::condition_variable cv_;
    std::size_t calls_{0};
    bool contenderCompleted_{false};
    bool released_{false};
};

struct CreationGateRelease {
    std::shared_ptr<CreationGate> gate;
    ~CreationGateRelease() { gate->release(); }
};
} // namespace

// ============================================================================
// ConnectionPoolMaintenance - pruneIdleConnections
// ============================================================================
TEST_CASE("Connection pool prunes idle connections to min", "[metadata][connection_pool][slow]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 4;
    cfg.idleTimeout = std::chrono::seconds(1);
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_mt_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Acquire and release to create extra connections
    {
        auto c1 = pool.acquire();
        REQUIRE(c1.has_value());
    }
    {
        auto c2 = pool.acquire();
        REQUIRE(c2.has_value());
    }

    // Let them become idle
    std::this_thread::sleep_for(1500ms);
    pool.pruneIdleConnections();

    auto stats = pool.getStats();
    // After prune, available connections should not exceed minConnections by much
    CHECK((stats.availableConnections <= cfg.minConnections));

    pool.shutdown();
}

// ============================================================================
// ConnectionPoolWaitingGuard - timeout behavior
// ============================================================================
TEST_CASE("Connection pool waiting counter resets on timeout", "[metadata][connection_pool]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;
    cfg.busyTimeout = 50ms;

    ConnectionPool pool(make_db_path("pool_wait_guard_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto first = pool.acquire();
    REQUIRE(first.has_value());

    auto second = pool.acquire(100ms);
    REQUIRE_FALSE(second.has_value());

    auto stats = pool.getStats();
    CHECK((stats.waitingRequests == 0u));
    CHECK((stats.maxObservedWaiting >= 1u));
    CHECK((stats.totalWaitMicros > 0u));
    CHECK((stats.timeoutCount == 1u));

    pool.shutdown();
}

TEST_CASE("Connection pool shutdown interrupt wakes blocked acquires",
          "[metadata][connection_pool][shutdown]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_interrupt_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto first = pool.acquire();
    REQUIRE(first.has_value());

    std::promise<yams::Result<std::unique_ptr<PooledConnection>>> blockedPromise;
    auto blockedFuture = blockedPromise.get_future();

    std::thread blocked([&]() { blockedPromise.set_value(pool.acquire(std::chrono::seconds(5))); });

    const auto waitDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    while (std::chrono::steady_clock::now() < waitDeadline &&
           pool.getStats().waitingRequests == 0u) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE((pool.getStats().waitingRequests > 0u));

    pool.interruptPendingAcquires();

    auto interrupted = blockedFuture.get();
    blocked.join();

    REQUIRE_FALSE(interrupted.has_value());
    CHECK(((interrupted.error().code == yams::ErrorCode::OperationCancelled)));
    CHECK((interrupted.error().message.find("shutdown") != std::string::npos));
    CHECK((pool.getStats().waitingRequests == 0u));

    pool.shutdown();
}

TEST_CASE("Connection pool immediate acquire does not count waiting",
          "[metadata][connection_pool]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_wait_guard_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    {
        auto conn = pool.acquire();
        REQUIRE(conn.has_value());
    }

    auto stats = pool.getStats();
    CHECK((stats.waitingRequests == 0u));
    CHECK((stats.maxObservedWaiting == 0u));
    CHECK((stats.totalWaitMicros == 0u));
    CHECK((stats.timeoutCount == 0u));

    pool.shutdown();
}

TEST_CASE("Connection pool histograms holder durations and warns on slow holds",
          "[metadata][connection_pool]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_holder_hist_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    pool.setSlowHolderThreshold(std::chrono::milliseconds(20));

    {
        auto conn =
            pool.acquire(std::chrono::milliseconds(1000), ConnectionPriority::Normal, "fast_path");
        REQUIRE(conn.has_value());
        CHECK((conn.value()->holderTag() == "fast_path"));
    }

    {
        auto conn =
            pool.acquire(std::chrono::milliseconds(1000), ConnectionPriority::Normal, "slow_path");
        REQUIRE(conn.has_value());
        CHECK((conn.value()->holderTag() == "slow_path"));
        std::this_thread::sleep_for(std::chrono::milliseconds(60));
    }

    auto stats = pool.getStats();
    std::uint64_t total = 0;
    for (auto v : stats.holderDurationBuckets) {
        total += v;
    }
    CHECK((total >= 2u));
    CHECK((stats.slowHolderCount >= 1u));
    CHECK((stats.maxHolderMicros >= 50'000u));

    pool.shutdown();
}

TEST_CASE("Connection pool names the pool in slow-hold warnings", "[metadata][connection_pool]") {
    const auto dbPath = make_db_path("pool_holder_label_").string();
    const auto slowHoldLog = [&](bool readOnly) {
        ConnectionPoolConfig cfg;
        cfg.minConnections = 1;
        cfg.maxConnections = 1;
        cfg.enableWAL = false;
        cfg.readOnly = readOnly;
        ConnectionPool pool(dbPath, cfg);
        REQUIRE(pool.initialize().has_value());
        pool.setSlowHolderThreshold(std::chrono::milliseconds(20));

        std::ostringstream captured;
        auto previous = spdlog::default_logger();
        auto logger = std::make_shared<spdlog::logger>(
            "pool_label_capture", std::make_shared<spdlog::sinks::ostream_sink_mt>(captured));
        spdlog::set_default_logger(logger);
        {
            auto conn = pool.acquire(std::chrono::milliseconds(1000), ConnectionPriority::Normal,
                                     "client_list");
            REQUIRE(conn.has_value());
            std::this_thread::sleep_for(std::chrono::milliseconds(60));
        }
        spdlog::set_default_logger(previous);
        pool.shutdown();
        return captured.str();
    };

    const auto writeLog = slowHoldLog(false);
    CHECK(writeLog.find("slow write-connection hold: tag='client_list'") != std::string::npos);

    // Read pools share the threshold; calling their holds "write" sent a slow read-only
    // listing down the wrong path.
    const auto readLog = slowHoldLog(true);
    CHECK(readLog.find("slow read-connection hold: tag='client_list'") != std::string::npos);
    CHECK(readLog.find("write-connection") == std::string::npos);
}

TEST_CASE("Connection pool records source location for untagged acquire",
          "[metadata][connection_pool][holder-tag]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_holder_tag_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    {
        auto conn = pool.acquire();
        REQUIRE(conn.has_value());
        const auto& tag = conn.value()->holderTag();
        CHECK_FALSE(tag.empty());
        CHECK((tag.find("connection_pool_catch2_test.cpp") != std::string::npos));
        CHECK((tag.find("<untagged>") == std::string::npos));
    }

    pool.shutdown();
}

TEST_CASE("Connection pool timeout holder summary includes withConnection call site",
          "[metadata][connection_pool][holder-tag]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_with_connection_tag_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    std::promise<void> entered;
    std::promise<void> release;
    auto enteredFuture = entered.get_future();
    auto releaseFuture = release.get_future();
    std::atomic<bool> holderSucceeded{false};

    std::thread holder([&] {
        auto result = pool.withConnection([&](Database&) -> yams::Result<void> {
            entered.set_value();
            releaseFuture.wait();
            return yams::Result<void>();
        });
        holderSucceeded.store(result.has_value(), std::memory_order_relaxed);
    });

    if (enteredFuture.wait_for(std::chrono::seconds(5)) != std::future_status::ready) {
        release.set_value();
        holder.join();
        FAIL("withConnection holder did not acquire the connection");
    }

    auto blocked = pool.acquire(std::chrono::milliseconds(50));
    if (blocked.has_value()) {
        release.set_value();
        holder.join();
        FAIL("second acquire unexpectedly succeeded while withConnection held the only connection");
    }
    CHECK((blocked.error().message.find("connection_pool_catch2_test.cpp") != std::string::npos));
    CHECK((blocked.error().message.find("holders=[<untagged>") == std::string::npos));

    release.set_value();
    holder.join();
    CHECK((holderSucceeded.load(std::memory_order_relaxed)));
    pool.shutdown();
}

TEST_CASE("Read-only connection pool opens SQLite read-only",
          "[metadata][connection_pool][readonly]") {
    const auto dbPath = make_db_path("pool_readonly_");

    {
        Database seed;
        REQUIRE(seed.open(dbPath.string(), ConnectionMode::Create).has_value());
        REQUIRE(seed.execute("CREATE TABLE readonly_probe(id INTEGER PRIMARY KEY)").has_value());
    }

    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.readOnly = true;

    ConnectionPool pool(dbPath.string(), cfg);
    REQUIRE(pool.initialize().has_value());

    {
        auto conn = pool.acquire();
        REQUIRE(conn.has_value());
        auto pooled = std::move(conn).value();
        auto& db = **pooled;
        REQUIRE((db.rawHandle() != nullptr));
        CHECK((sqlite3_db_readonly(db.rawHandle(), "main") == 1));

        auto write = db.execute("CREATE TABLE should_not_write(id INTEGER)");
        CHECK_FALSE(write.has_value());
    }

    pool.shutdown();
}

TEST_CASE("Connection pool shutdown while connection is leased does not crash",
          "[metadata][connection_pool][shutdown]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_shutdown_leased_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto conn = pool.acquire();
    REQUIRE(conn.has_value());

    auto statsBefore = pool.getStats();
    CHECK((statsBefore.activeConnections == 1u));

    pool.shutdown();

    auto statsAfter = pool.getStats();
    CHECK((statsAfter.activeConnections == 0u));
    CHECK((statsAfter.totalConnections == 0u));

    auto afterShutdown = pool.acquire(std::chrono::milliseconds(0));
    REQUIRE_FALSE(afterShutdown.has_value());
    CHECK((afterShutdown.error().code == yams::ErrorCode::InvalidState));
}

// ============================================================================
// WAL auto-checkpoint behavior — verifies the fix for unbounded WAL growth.
// When wal_autocheckpoint is enabled (non-zero), SQLite should checkpoint
// WAL pages back to the main DB, keeping the WAL bounded.
// ============================================================================
TEST_CASE("Connection pool enables WAL auto-checkpoint by default",
          "[metadata][connection_pool][wal]") {
    // Reproduces the bug: wal_autocheckpoint was set to 0, which disabled
    // automatic WAL checkpointing entirely.  On a 45GB DB this meant the
    // WAL grew to the same size as the DB, and on restart wal_checkpoint
    // took 1000+ seconds.  The fix: wal_autocheckpoint = 10000 (≈40 MB).
    // Verification: the pool initializes cleanly with WAL enabled and
    // supports read/write operations.
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = true;

    ConnectionPool pool(make_db_path("pool_wal_ac_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto conn = pool.acquire();
    REQUIRE(conn.has_value());
    REQUIRE((*conn.value())->execute("CREATE TABLE IF NOT EXISTS ac_check(x)").has_value());
    REQUIRE((*conn.value())->execute("INSERT INTO ac_check VALUES(42)").has_value());

    pool.shutdown();
}

TEST_CASE("Connection pool WAL mode creates WAL and SHM files",
          "[metadata][connection_pool][wal]") {
    auto dbPath = make_db_path("pool_wal_files_");
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = true;

    ConnectionPool pool(dbPath.string(), cfg);
    REQUIRE(pool.initialize().has_value());

    {
        auto conn = pool.acquire();
        REQUIRE(conn.has_value());

        // Write data to trigger WAL file creation while the pool still owns live connections.
        REQUIRE((*conn.value())->execute("CREATE TABLE IF NOT EXISTS t(x)").has_value());
        REQUIRE((*conn.value())->execute("INSERT INTO t VALUES(1)").has_value());
    }

    // WAL companion files are only guaranteed while at least one connection remains open.
    // After the final close, SQLite may checkpoint and remove them immediately.
    fs::path walPath(dbPath.string() + "-wal");
    fs::path shmPath(dbPath.string() + "-shm");
    CHECK((fs::exists(walPath)));
    CHECK((fs::exists(shmPath)));

    pool.shutdown();

    // Cleanup companion files if SQLite left them behind.
    std::error_code ec;
    fs::remove(walPath, ec);
    fs::remove(shmPath, ec);
}

TEST_CASE("Connection pool reserves capacity before concurrent acquire creation",
          "[metadata][connection_pool][concurrent][capacity]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 0;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    auto gate = std::make_shared<CreationGate>();
    ConnectionPool pool(make_db_path("pool_acquire_capacity_").string(), cfg);
    pool.testing_setBeforeCreateHook([gate] { gate->arriveAndWait(); });
    REQUIRE(pool.initialize().has_value());

    using AcquireResult = decltype(pool.acquire());
    std::future<AcquireResult> firstFuture;
    std::future<AcquireResult> contenderFuture;
    CreationGateRelease releaseOnExit{gate};

    firstFuture = std::async(std::launch::async, [&pool] { return pool.acquire(5s); });
    REQUIRE(gate->waitForCalls(1, 2s));

    contenderFuture = std::async(std::launch::async, [&pool, gate] {
        auto result = pool.acquire(0ms);
        gate->markContenderCompleted();
        return result;
    });
    REQUIRE(gate->waitForCallsOrContender(2, 2s));
    const bool contenderCompletedBeforeRelease = gate->contenderCompleted();

    gate->release();
    auto first = firstFuture.get();
    auto contender = contenderFuture.get();

    REQUIRE(first.has_value());
    CHECK(contenderCompletedBeforeRelease);
    CHECK_FALSE(contender.has_value());
    if (!contender.has_value()) {
        CHECK((contender.error().code == yams::ErrorCode::Timeout));
    }
    CHECK((gate->calls() == 1));
    CHECK((pool.getStats().totalConnections <= cfg.maxConnections));

    pool.shutdown();
}

TEST_CASE("Connection pool reserves capacity across health check and acquire creation",
          "[metadata][connection_pool][concurrent][capacity][health]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.enableWAL = false;

    // initialize() owns the first creation. Every later creation is held at the seam. The
    // maintenance thread may win the post-refresh race; the test deliberately does not care.
    auto gate = std::make_shared<CreationGate>(1);
    ConnectionPool pool(make_db_path("pool_health_capacity_").string(), cfg);
    pool.testing_setBeforeCreateHook([gate] { gate->arriveAndWait(); });
    REQUIRE(pool.initialize().has_value());
    REQUIRE((gate->calls() == 1));

    using HealthResult = decltype(pool.healthCheck());
    using AcquireResult = decltype(pool.acquire());
    std::future<HealthResult> healthFuture;
    std::future<AcquireResult> contenderFuture;
    CreationGateRelease releaseOnExit{gate};

    pool.refreshAll();
    CHECK((pool.getStats().totalConnections == 0));

    healthFuture = std::async(std::launch::async, [&pool] { return pool.healthCheck(); });
    REQUIRE(gate->waitForCalls(2, 2s));

    contenderFuture = std::async(std::launch::async, [&pool, gate] {
        auto result = pool.acquire(0ms);
        gate->markContenderCompleted();
        return result;
    });
    REQUIRE(gate->waitForCallsOrContender(3, 2s));
    const bool contenderCompletedBeforeRelease = gate->contenderCompleted();

    gate->release();
    auto health = healthFuture.get();
    auto contender = contenderFuture.get();

    REQUIRE(health.has_value());
    CHECK(contenderCompletedBeforeRelease);
    CHECK_FALSE(contender.has_value());
    if (!contender.has_value()) {
        CHECK((contender.error().code == yams::ErrorCode::Timeout));
    }
    CHECK((gate->calls() == 2));
    CHECK((pool.getStats().totalConnections <= cfg.maxConnections));

    pool.shutdown();
}

TEST_CASE("Connection pool handles concurrent acquire/release",
          "[metadata][connection_pool][concurrent]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 2;
    cfg.maxConnections = 4;
    cfg.enableWAL = true;
    cfg.busyTimeout = std::chrono::milliseconds(2000);

    ConnectionPool pool(make_db_path("pool_concurrent_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    std::atomic<int> errors{0};
    std::atomic<int> successes{0};

    auto worker = [&](int /*id*/) {
        for (int i = 0; i < 5; ++i) {
            auto conn = pool.acquire(std::chrono::milliseconds(5000));
            if (conn.has_value()) {
                ++successes;
                // Simulate brief work
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                // conn goes out of scope -> returned to pool
            } else {
                ++errors;
            }
        }
    };

    std::thread t1(worker, 1);
    std::thread t2(worker, 2);
    std::thread t3(worker, 3);
    t1.join();
    t2.join();
    t3.join();

    CHECK((successes.load() == 15));
    CHECK((errors.load() == 0));

    auto stats = pool.getStats();
    CHECK((stats.totalConnections <= cfg.maxConnections));

    pool.shutdown();
}

TEST_CASE("Connection pool fails acquire on exhaustion with zero timeout",
          "[metadata][connection_pool][exhaustion]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1; // single-connection pool
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_exhaust_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Hold the only connection
    auto held = pool.acquire();
    REQUIRE(held.has_value());

    // Second acquire with zero timeout should fail
    auto blocked = pool.acquire(std::chrono::milliseconds(0));
    REQUIRE_FALSE(blocked.has_value());
    CHECK((blocked.error().code == yams::ErrorCode::Timeout));

    // Release and re-acquire should succeed
    held = {}; // return to pool
    auto retry = pool.acquire(std::chrono::milliseconds(200));
    REQUIRE(retry.has_value());

    pool.shutdown();
}

TEST_CASE("Connection pool stats reflect active and available counts",
          "[metadata][connection_pool][stats]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 2;
    cfg.maxConnections = 4;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_stats_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // After init, minConnections should be available
    auto s0 = pool.getStats();
    CHECK((s0.availableConnections >= cfg.minConnections));
    CHECK((s0.activeConnections == 0u));

    // Acquire one: available down by 1, active up by 1
    auto conn = pool.acquire();
    REQUIRE(conn.has_value());
    auto s1 = pool.getStats();
    CHECK((s1.availableConnections == s0.availableConnections - 1));
    CHECK((s1.activeConnections == 1u));

    // Return it: stats recover
    conn = {}; // triggers PooledConnection destructor -> returnConnection
    std::this_thread::sleep_for(std::chrono::milliseconds(10)); // let return settle
    auto s2 = pool.getStats();
    CHECK((s2.availableConnections >= s1.availableConnections));
    CHECK((s2.activeConnections == 0u));

    pool.shutdown();
}

TEST_CASE("Connection pool shutdown invalidates all connections",
          "[metadata][connection_pool][lifecycle]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = false;

    auto dbPath = make_db_path("pool_lifecycle_");
    ConnectionPool pool(dbPath.string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Hold a connection before shutdown
    auto conn = pool.acquire();
    REQUIRE(conn.has_value());

    // Shutdown while a connection is held should still succeed
    pool.shutdown();

    // Stats should show zero after shutdown
    auto stats = pool.getStats();
    CHECK((stats.totalConnections == 0u));

    // Re-acquire should fail with InvalidState
    auto retry = pool.acquire(std::chrono::milliseconds(100));
    REQUIRE_FALSE(retry.has_value());
    CHECK((retry.error().code == yams::ErrorCode::InvalidState));

    // Cleanup
    std::error_code ec;
    fs::remove(dbPath, ec);
}

TEST_CASE("Connection pool read-only connections skip WAL pragmas",
          "[metadata][connection_pool][readonly]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.readOnly = true;
    cfg.enableWAL = true; // should be ignored for read-only

    auto dbPath = make_db_path("pool_ro_");
    // Create the DB first so read-only can open it
    {
        Database db;
        REQUIRE(db.open(dbPath.string(), ConnectionMode::Create).has_value());
        REQUIRE(db.execute("CREATE TABLE ro_test(x)").has_value());
        db.close();
    }

    ConnectionPool pool(dbPath.string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Read-only pool should still allow acquire and read
    auto conn = pool.acquire();
    REQUIRE(conn.has_value());
    auto stmt = (*conn.value())->prepare("SELECT COUNT(*) FROM ro_test");
    REQUIRE(stmt.has_value());

    pool.shutdown();

    std::error_code ec;
    fs::remove(dbPath, ec);
}

TEST_CASE("Connection pool verifies WAL journal_mode is active",
          "[metadata][connection_pool][wal]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = true;

    ConnectionPool pool(make_db_path("pool_wal_jm_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto conn = pool.acquire();
    REQUIRE(conn.has_value());

    // Verify WAL mode is active by writing and verifying persistence.
    REQUIRE((*conn.value())->execute("CREATE TABLE IF NOT EXISTS wal_verify(x)").has_value());

    pool.shutdown();
}

TEST_CASE("Connection pool tags connections for diagnostics",
          "[metadata][connection_pool][tagging]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_tags_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Acquire with default tag
    auto conn = pool.acquire();
    REQUIRE(conn.has_value());
    std::string defaultTag = conn.value()->holderTag();
    CHECK_FALSE(defaultTag.empty());

    // Release and re-acquire with explicit tag
    conn = {};
    auto tagged =
        pool.acquire(std::chrono::milliseconds(100), ConnectionPriority::Normal, "read_query");
    REQUIRE(tagged.has_value());
    std::string tag = tagged.value()->holderTag();
    // The explicit tag should be reflected (implementation-dependent format)
    CHECK_FALSE(tag.empty());

    pool.shutdown();
}

TEST_CASE("Connection pool reuses connections across acquire/release cycles",
          "[metadata][connection_pool][reuse]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_reuse_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Round 1: acquire, use, release
    auto conn1 = pool.acquire();
    REQUIRE(conn1.has_value());
    REQUIRE((*conn1.value())->execute("CREATE TABLE IF NOT EXISTS reuse_test(x)").has_value());
    REQUIRE((*conn1.value())->execute("INSERT INTO reuse_test VALUES(1)").has_value());
    conn1 = {};

    // Round 2: re-acquire and verify data persists via execute
    auto conn2 = pool.acquire();
    REQUIRE(conn2.has_value());
    auto result = (*conn2.value())->execute("SELECT COUNT(*) FROM reuse_test WHERE x=1");
    // execute() on SELECT may fail on some implementations; just verify
    // the pool works across acquire/release cycles without crashing.
    (void)result;

    pool.shutdown();
}

TEST_CASE("Connection pool respects busy timeout on contention",
          "[metadata][connection_pool][busy]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 1;
    cfg.busyTimeout = std::chrono::milliseconds(100);
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_busy_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    // Hold the only connection
    auto held = pool.acquire();
    REQUIRE(held.has_value());

    // Second acquire should timeout (no connections available)
    auto t0 = std::chrono::steady_clock::now();
    auto blocked = pool.acquire(std::chrono::milliseconds(200));
    auto elapsed = std::chrono::steady_clock::now() - t0;

    REQUIRE_FALSE(blocked.has_value());
    // Should have waited at least the busyTimeout before failing
    CHECK((elapsed >= cfg.busyTimeout));

    pool.shutdown();
}

namespace {
// Deadline for the bounded long-running query used by the interrupt tests. The SQL function lets
// the statement run for a fixed wall-clock budget independent of machine or build speed.
std::atomic<std::int64_t> gLongQueryDeadlineNs{0};

void keepRunningUntilDeadline(sqlite3_context* context, int, sqlite3_value**) {
    const auto now = std::chrono::steady_clock::now().time_since_epoch();
    const auto nowNs = std::chrono::duration_cast<std::chrono::nanoseconds>(now).count();
    sqlite3_result_int(context, nowNs < gLongQueryDeadlineNs.load() ? 1 : 0);
}

constexpr const char* kLongQuery =
    "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c WHERE "
    "yams_test_keep_running()) SELECT count(*) FROM c";
} // namespace

TEST_CASE("Connection pool interrupts statements running on leased connections",
          "[metadata][connection_pool][interrupt][shutdown]") {
    ConnectionPoolConfig cfg;
    cfg.minConnections = 1;
    cfg.maxConnections = 2;
    cfg.enableWAL = false;

    ConnectionPool pool(make_db_path("pool_interrupt_").string(), cfg);
    REQUIRE(pool.initialize().has_value());

    auto conn = pool.acquire();
    REQUIRE(conn.has_value());
    auto& db = **conn.value();
    REQUIRE(
        (sqlite3_create_function(db.rawHandle(), "yams_test_keep_running", 0, SQLITE_UTF8, nullptr,
                                 keepRunningUntilDeadline, nullptr, nullptr) == SQLITE_OK));

    // An interrupt that lands while the leased connection is idle must not fail its next
    // statement: shutdown interrupts every lease, including ones between statements.
    CHECK((pool.interruptActiveConnections() == 1U));
    CHECK(db.execute("SELECT 1").has_value());

    constexpr auto kQueryBudget = 30s;
    gLongQueryDeadlineNs.store(
        std::chrono::duration_cast<std::chrono::nanoseconds>(
            (std::chrono::steady_clock::now() + kQueryBudget).time_since_epoch())
            .count());
    std::promise<void> started;
    auto startedFuture = started.get_future();
    auto query = std::async(std::launch::async, [&db, &started]() {
        started.set_value();
        return db.execute(kLongQuery);
    });
    REQUIRE((startedFuture.wait_for(5s) == std::future_status::ready));

    // The statement may not have entered sqlite3_step() yet, and SQLite clears an interrupt that
    // arrives before any statement is active, so re-issue until the query returns.
    const auto interruptStart = std::chrono::steady_clock::now();
    while (query.wait_for(10ms) != std::future_status::ready &&
           std::chrono::steady_clock::now() - interruptStart < kQueryBudget) {
        pool.interruptActiveConnections();
    }
    REQUIRE((query.wait_for(0s) == std::future_status::ready));
    const auto elapsed = std::chrono::steady_clock::now() - interruptStart;
    CHECK_FALSE(query.get().has_value());
    CHECK((elapsed < 5s));

    conn = {};
    CHECK((pool.interruptActiveConnections() == 0U));
    pool.shutdown();
}