// Daemon components test suite (Catch2)
// Consolidates: SocketServer lifecycle, socket utils, resource pool
// Covers: IPC infrastructure, socket path resolution, resource lifecycle management

// pi-lens-ignore: fatal error
#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>

#include <atomic>
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <future>
#include <iostream>
#include <optional>
#include <thread>
#include <vector>
#include "../../common/test_helpers_catch2.h"
#include <yams/compat/unistd.h>

#include <yams/daemon/components/IOCoordinator.h>
#include <yams/daemon/components/LifecycleComponent.h>
#include <yams/daemon/components/ResourceGovernor.h>
#include <yams/daemon/components/SearchEngineManager.h>
#include <yams/daemon/components/SocketServer.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/ipc/socket_utils.h>
#include <yams/daemon/resource/resource_pool.h>

#ifdef _WIN32
inline int geteuid() {
    return 1000;
}
inline int getuid() {
    return 1000;
}
#endif

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

#ifndef _WIN32
#include <sys/un.h>
#endif

// =============================================================================
// Test Helpers
// =============================================================================

namespace {
fs::path makeTempRuntimeDir(const std::string& name) {
    auto base = fs::temp_directory_path();
#ifndef _WIN32
    constexpr std::size_t maxUnixPath = sizeof(sockaddr_un::sun_path) - 1;
    auto candidate = base / "yams-component-tests" / name;
    auto candidateWithSocket = candidate / "ipc.sock";
    std::cerr << "candidate path: " << candidateWithSocket
              << " len=" << candidateWithSocket.native().size() << " limit=" << maxUnixPath << "\n";
    if (candidateWithSocket.native().size() >= maxUnixPath) {
        base = fs::path("/tmp");
        candidate = base / "yams-component-tests" / name;
        candidateWithSocket = candidate / "ipc.sock";
        std::cerr << "fallback path: " << candidateWithSocket
                  << " len=" << candidateWithSocket.native().size() << "\n";
    }
#endif
    auto dir = base / "yams-component-tests" / name;
    std::error_code ec;
    fs::create_directories(dir, ec);
    return dir;
}

TEST_CASE("LifecycleComponent test safe-instance guard overrides production aggressive mode",
          "[daemon][lifecycle][single-instance]") {
    yams::test::ScopedEnvVar safeGuard{"YAMS_TEST_SAFE_SINGLE_INSTANCE", std::string{"1"}};
    yams::test::ScopedEnvVar killOthers{"YAMS_DAEMON_KILL_OTHERS", std::string{"1"}};
    CHECK_FALSE(LifecycleComponent::testingAggressiveModeEnabled());

    safeGuard.unset();
    CHECK((LifecycleComponent::testingAggressiveModeEnabled()));

    killOthers.set("0");
    CHECK_FALSE(LifecycleComponent::testingAggressiveModeEnabled());
}

TEST_CASE("SearchEngineManager snapshots topology compatibility policy once",
          "[daemon][search][config][topology][snapshot]") {
    yams::test::ScopedEnvVar canonicalConfig{"YAMS_CONFIG", std::nullopt};
    yams::test::ScopedEnvVar compatibilityConfig{"YAMS_CONFIG_PATH", std::nullopt};
    yams::test::ScopedEnvVar maxClusters{"YAMS_SEARCH_TOPOLOGY_MAX_CLUSTERS", std::string{"3"}};

    SearchEngineManager manager;
    maxClusters.set("7");

    REQUIRE(manager.topologyMaxClustersSnapshot().has_value());
    CHECK((*manager.topologyMaxClustersSnapshot() == 3));
}

TEST_CASE("SearchEngineManager runtime executor is isolated from unavailable shared workers",
          "[daemon][search][executor][liveness]") {
    constexpr std::size_t kRequestConcurrency = 8;
    boost::asio::io_context unavailableSharedContext;
    std::atomic<std::size_t> sharedWorkRan{0};
    for (std::size_t i = 0; i < kRequestConcurrency; ++i) {
        boost::asio::post(unavailableSharedContext, [&sharedWorkRan]() {
            sharedWorkRan.fetch_add(1, std::memory_order_release);
        });
    }

    SearchEngineManager manager;
    std::vector<std::promise<void>> searchWorkCompleted(kRequestConcurrency);
    std::vector<std::future<void>> searchWorkFutures;
    searchWorkFutures.reserve(kRequestConcurrency);
    for (std::size_t i = 0; i < searchWorkCompleted.size(); ++i) {
        searchWorkFutures.push_back(searchWorkCompleted[i].get_future());
        boost::asio::post(manager.testingRuntimeExecutor(),
                          [&searchWorkCompleted, i]() { searchWorkCompleted[i].set_value(); });
    }

    const auto deadline = std::chrono::steady_clock::now() + 500ms;
    for (auto& future : searchWorkFutures) {
        CHECK((future.wait_until(deadline) == std::future_status::ready));
    }
    CHECK((sharedWorkRan.load(std::memory_order_acquire) == 0));

    unavailableSharedContext.run();
    CHECK((sharedWorkRan.load(std::memory_order_acquire) == kRequestConcurrency));
}

std::string randomSuffix() {
    return std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
}

template <typename ResultLike> bool isPermissionDenied(const ResultLike& result) {
    if (result) {
        return false;
    }
    std::string_view message{result.error().message};
    return message.find("Operation not permitted") != std::string_view::npos ||
           message.find("Permission denied") != std::string_view::npos;
}

// Test resource for pool testing
class TestResource {
public:
    TestResource() { creationCount_++; }

    bool isValid() const { return valid_; }
    void invalidate() { valid_ = false; }

    void doWork() const {
        workCount_++;
        std::this_thread::sleep_for(1ms);
    }

    static void resetCounters() {
        creationCount_ = 0;
        workCount_ = 0;
    }

    static int getCreationCount() { return creationCount_; }
    static int getWorkCount() { return workCount_; }

private:
    bool valid_{true};
    static std::atomic<int> creationCount_;
    static std::atomic<int> workCount_;
};

std::atomic<int> TestResource::creationCount_{0};
std::atomic<int> TestResource::workCount_{0};

using yams::test::ScopedEnvVar;
} // namespace

// =============================================================================
// SocketServer Lifecycle Tests
// =============================================================================

TEST_CASE("SocketServer: Lifecycle management", "[daemon][components][socket]") {
    auto runtimeDir = makeTempRuntimeDir("socket-lifecycle-" + randomSuffix());
    auto socketPath = runtimeDir / "ipc.sock";

    SocketServer::Config config;
    config.socketPath = socketPath;
    config.connectionTimeout = 1500ms;

    StateComponent state;
    state.stats.startTime = std::chrono::steady_clock::now();

    SECTION("Start and stop clears stopping flag") {
        WorkCoordinator coordinator;
        coordinator.start(2);
        IOCoordinator io;
        io.start();
        SocketServer server(config, &io, &coordinator, nullptr, &state);

        auto first = server.start();
        if (!first) {
            UNSCOPED_INFO(first.error().message);
        }
        if (!first && isPermissionDenied(first)) {
            SKIP("UNIX domain sockets not permitted on this system");
        }
        REQUIRE((first));

        auto stopped = server.stop();
        REQUIRE((stopped));

        // Second start should work (stopping flag cleared)
        auto second = server.start();
        if (!second) {
            UNSCOPED_INFO(second.error().message);
        }
        if (!second && isPermissionDenied(second)) {
            SKIP("UNIX domain sockets not permitted on this system");
        }
        REQUIRE((second));

        REQUIRE((server.stop()));

        io.stop();
        io.join();
        coordinator.stop();
        coordinator.join();
    }

    SECTION("Multiple start attempts are safe") {
        WorkCoordinator coordinator;
        coordinator.start(2);
        IOCoordinator io;
        io.start();
        SocketServer server(config, &io, &coordinator, nullptr, &state);

        auto first = server.start();
        if (!first) {
            UNSCOPED_INFO(first.error().message);
        }
        if (!first && isPermissionDenied(first)) {
            SKIP("UNIX domain sockets not permitted on this system");
        }
        REQUIRE((first));

        // Second start while running should fail gracefully
        auto second = server.start();
        REQUIRE((!second)); // Already running

        REQUIRE((server.stop()));

        io.stop();
        io.join();
        coordinator.stop();
        coordinator.join();
    }

    SECTION("Multi-threaded main and proxy accept shutdown is serialized") {
        WorkCoordinator coordinator;
        coordinator.start(2);
        IOCoordinator io(IOCoordinator::Config{.num_threads = 4});
        io.start();

        auto proxyConfig = config;
        proxyConfig.proxySocketPath = runtimeDir / "proxy.sock";
        for (int iteration = 0; iteration < 16; ++iteration) {
            SocketServer server(proxyConfig, &io, &coordinator, nullptr, &state);
            auto started = server.start();
            if (!started) {
                UNSCOPED_INFO(started.error().message);
            }
            if (!started && isPermissionDenied(started)) {
                SKIP("UNIX domain sockets not permitted on this system");
            }
            REQUIRE((started));

            auto firstStop = std::async(std::launch::async, [&server] { return server.stop(); });
            auto secondStop = std::async(std::launch::async, [&server] { return server.stop(); });
            REQUIRE((firstStop.get()));
            REQUIRE((secondStop.get()));
        }

        io.stop();
        io.join();
        coordinator.stop();
        coordinator.join();
    }

    // Cleanup
    std::error_code ec;
    fs::remove_all(runtimeDir, ec);
}

TEST_CASE("SocketServer: main-socket admission ignores proxy connection load",
          "[daemon][components][socket][admission]") {
    SocketServer::Config config;
    config.socketPath = "/tmp/yams-socket-admission-test.sock";
    config.maxConnections = 4;
    config.proxyMaxConnections = 64;

    StateComponent state;
    SocketServer server(config, nullptr, nullptr, nullptr, &state);
    REQUIRE((server.resizeConnectionSlots(config.maxConnections)));

    AddDocumentRequest addRequest;
    addRequest.path = "src";
    addRequest.recursive = true;

    SECTION("Main-socket write admission keys off main connections, not total connections") {
        server.testing_setConnectionCounts(/*mainActive=*/0, /*proxyActive=*/10);

        CHECK((server.activeConnections() == 10));
        CHECK((server.proxyActiveConnections() == 10));
        CHECK((server.testing_mainSocketAdmissionVerdict(addRequest) ==
               SocketAdmissionVerdict::admit));
        CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());
        CHECK((server.getSlotUtilization() == 0.0));
        server.testing_setConnectionCounts(0, 0);
    }

    SECTION("Main-socket pressure still rejects once main connections cross the limit bands") {
        server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/0);

        CHECK((server.testing_mainSocketAdmissionVerdict(addRequest) ==
               SocketAdmissionVerdict::reject));
        server.testing_setConnectionCounts(0, 0);
    }

    SECTION("Control requests remain bypassed on the main socket") {
        server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/20);

        CHECK((server.testing_mainSocketAdmissionVerdict(StatusRequest{}) ==
               SocketAdmissionVerdict::bypass));
        server.testing_setConnectionCounts(0, 0);
    }

    SECTION("Emergency guard is driven only by main active connections") {
        const auto hardLimit = AdmissionPolicy::emergencySessionLimit(config.maxConnections);

        server.testing_setConnectionCounts(/*mainActive=*/hardLimit - 1, /*proxyActive=*/hardLimit);
        CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());

        server.testing_setConnectionCounts(/*mainActive=*/hardLimit, /*proxyActive=*/0);
        CHECK((server.testing_mainSocketEmergencyGuardRejects()));
        server.testing_setConnectionCounts(0, 0);
    }
}

// =============================================================================
// Socket Path Resolution Tests
// =============================================================================

TEST_CASE("Socket path resolution: Priority ordering", "[daemon][components][socket][utils]") {
    auto tempDir = makeTempRuntimeDir("socket-utils-" + randomSuffix());
    auto configBase = tempDir / "config";
    auto configDir = configBase / "yams";
    fs::create_directories(configDir);

    ScopedEnvVar yamsConfigGuard("YAMS_CONFIG", std::nullopt);
    ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET", std::nullopt);
    ScopedEnvVar yamsSockCompatibilityGuard("YAMS_DAEMON_SOCKET_PATH", std::nullopt);
    ScopedEnvVar xdgRuntimeGuard("XDG_RUNTIME_DIR", std::nullopt);
    ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME", configBase.string());
    ScopedEnvVar homeGuard("HOME", tempDir.string());

    SECTION("Environment variable has highest priority") {
        auto testPath = "/tmp/custom-daemon.sock";
        yamsSockGuard.set(testPath);

        auto result = socket_utils::resolve_socket_path();
        REQUIRE((result.string() == testPath));

        auto resultConfigFirst = socket_utils::resolve_socket_path_config_first();
        REQUIRE((resultConfigFirst.string() == testPath));
    }

#ifndef _WIN32
    SECTION("XDG_RUNTIME_DIR used when env var not set") {
        if (::geteuid() == 0) {
            // Root runs default to /var/run per resolve_socket_path().
            // This test specifically targets non-root behavior.
            SKIP("XDG_RUNTIME_DIR precedence test is for non-root users");
        }
        yamsSockGuard.unset();

        auto xdgDir = tempDir / "runtime";
        fs::create_directories(xdgDir);
        xdgRuntimeGuard.set(xdgDir.string());

        auto result = socket_utils::resolve_socket_path();
        REQUIRE((result.parent_path() == xdgDir));
        REQUIRE((result.filename() == "yams-daemon.sock"));
    }

    SECTION("/tmp fallback for non-root when XDG not available") {
        yamsSockGuard.unset();
        xdgRuntimeGuard.unset();

        if (::geteuid() != 0) {
            auto result = socket_utils::resolve_socket_path();
            REQUIRE((result.parent_path() == "/tmp"));
            // Socket name includes UID for non-root users
            auto uid = std::to_string(::getuid());
            REQUIRE((result.filename() == "yams-daemon-" + uid + ".sock"));
        }
    }
#endif // !_WIN32

    // Cleanup
    std::error_code ec;
    fs::remove_all(tempDir, ec);
}

TEST_CASE("Socket path resolution: Config file reading", "[daemon][components][socket][utils]") {
    auto tempDir = makeTempRuntimeDir("socket-config-" + randomSuffix());
    auto configBase = tempDir / "config";
    auto configDir = configBase / "yams";
    fs::create_directories(configDir);

    auto configFile = configDir / "config.toml";

    ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME", configBase.string());
    ScopedEnvVar yamsConfigGuard("YAMS_CONFIG", std::nullopt);
    ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET", std::nullopt);
    ScopedEnvVar yamsSockCompatibilityGuard("YAMS_DAEMON_SOCKET_PATH", std::nullopt);

#ifndef _WIN32
    SECTION("Explicit YAMS_CONFIG socket_path is used") {
        const auto explicitConfig = tempDir / "explicit.toml";
        std::ofstream out(explicitConfig);
        out << "[daemon]\n";
        out << "socket_path = \"/opt/yams/explicit.sock\"\n";
        out.close();

        yamsConfigGuard.set(explicitConfig.string());
        yamsSockGuard.unset();

        auto result = socket_utils::resolve_socket_path_config_first();
        REQUIRE((result.string() == "/opt/yams/explicit.sock"));
    }

    SECTION("Config file socket_path is used") {
        std::ofstream out(configFile);
        out << "[daemon]\n";
        out << "socket_path = \"/opt/yams/daemon.sock\"\n";
        out.close();

        xdgConfigGuard.set(configBase.string());
        yamsSockGuard.unset();

        auto result = socket_utils::resolve_socket_path_config_first();
        REQUIRE((result.string() == "/opt/yams/daemon.sock"));
    }
#endif // !_WIN32

    SECTION("Invalid config falls back to default") {
        std::ofstream out(configFile);
        out << "invalid toml content ][[\n";
        out.close();

        xdgConfigGuard.set(configBase.string());
        yamsSockGuard.unset();

        // Should fall back to default resolution (doesn't crash)
        auto result = socket_utils::resolve_socket_path_config_first();
        REQUIRE((!result.empty()));
    }

    // Cleanup
    std::error_code ec;
    fs::remove_all(tempDir, ec);
}

TEST_CASE("Socket path resolution: proxy path derived from main socket",
          "[daemon][components][socket][utils]") {
    const fs::path mainSocket{"/tmp/yams-daemon.sock"};
    const auto proxySocket = socket_utils::derive_proxy_socket_path(mainSocket);
    REQUIRE((proxySocket == fs::path{"/tmp/yams-daemon.proxy.sock"}));

    const fs::path nestedSocket{"/run/user/1000/custom.sock"};
    const auto nestedProxy = socket_utils::derive_proxy_socket_path(nestedSocket);
    REQUIRE((nestedProxy == fs::path{"/run/user/1000/custom.proxy.sock"}));
}

// =============================================================================
// Resource Pool Tests
// =============================================================================

TEST_CASE("ResourcePool: Basic lifecycle", "[daemon][components][pool]") {
    TestResource::resetCounters();

    PoolConfig<TestResource> config;
    config.minSize = 3;
    config.maxSize = 5;
    config.maxIdle = 3;
    config.idleTimeout = 1s;
    config.acquisitionTimeout = 1s;
    config.validateOnAcquire = true;
    config.preCreateResources = true;

    SECTION("Pool pre-creates minimum resources") {
        using namespace yams;
        auto pool = std::make_unique<ResourcePool<TestResource>>(
            config,
            [](const std::string&) -> Result<std::shared_ptr<TestResource>> {
                return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
            },
            [](const TestResource& r) { return r.isValid(); });

        std::this_thread::sleep_for(50ms); // Allow async creation
        REQUIRE((TestResource::getCreationCount() >= static_cast<int>(config.minSize)));
    }

    TestResource::resetCounters();

    SECTION("Resource acquisition and release") {
        using namespace yams;
        auto pool = std::make_unique<ResourcePool<TestResource>>(
            config,
            [](const std::string&) -> Result<std::shared_ptr<TestResource>> {
                return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
            },
            [](const TestResource& r) { return r.isValid(); });

        std::this_thread::sleep_for(50ms);

        // Acquire resource (no ID parameter, timeout is optional)
        auto handleResult = pool->acquire();
        REQUIRE((handleResult.has_value()));
        auto& handle = handleResult.value();
        REQUIRE((handle.isValid()));
        REQUIRE((handle->isValid()));

        // Release by destroying handle (handle goes out of scope)
    }

    TestResource::resetCounters();

    SECTION("Invalid resources are replaced") {
        using namespace yams;
        auto pool = std::make_unique<ResourcePool<TestResource>>(
            config,
            [](const std::string&) -> Result<std::shared_ptr<TestResource>> {
                return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
            },
            [](const TestResource& r) { return r.isValid(); });

        std::this_thread::sleep_for(50ms);

        // Acquire and invalidate
        {
            auto handleResult = pool->acquire();
            REQUIRE((handleResult.has_value()));
            auto& handle = handleResult.value();
            handle->invalidate();
            // handle goes out of scope and releases
        }

        // Next acquisition should get valid resource (invalid one discarded)
        auto handle2Result = pool->acquire();
        REQUIRE((handle2Result.has_value()));
        auto& handle2 = handle2Result.value();
        REQUIRE((handle2->isValid()));
    }
}

TEST_CASE("ResourcePool: Concurrency", "[daemon][components][pool][concurrent]") {
    TestResource::resetCounters();

    PoolConfig<TestResource> config;
    config.minSize = 2;
    config.maxSize = 10;
    config.maxIdle = 5;
    config.idleTimeout = std::chrono::seconds(1);
    config.acquisitionTimeout = std::chrono::seconds(1);
    config.validateOnAcquire = true;
    config.preCreateResources = true;

    using namespace yams;
    auto pool = std::make_unique<ResourcePool<TestResource>>(
        config,
        [](const std::string&) -> Result<std::shared_ptr<TestResource>> {
            return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
        },
        [](const TestResource& r) { return r.isValid(); });

    std::this_thread::sleep_for(50ms);

    SECTION("Concurrent acquisition and release") {
        std::vector<std::thread> threads;
        std::atomic<int> successCount{0};

        for (int i = 0; i < 20; ++i) {
            threads.emplace_back([&pool, &successCount]() {
                auto handleResult = pool->acquire();
                if (handleResult.has_value() && handleResult.value().isValid()) {
                    auto& handle = handleResult.value();
                    handle->doWork();
                    successCount++;
                }
            });
        }

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

        // All threads should successfully acquire resources
        REQUIRE((successCount == 20));
        REQUIRE((TestResource::getWorkCount() == 20));
    }
}

#include <yams/daemon/client/asio_connection_pool.h>
#include <yams/daemon/client/global_io_context.h>

TEST_CASE("Daemon components: ResourceGovernor backpressure policy helpers",
          "[daemon][components][governor]") {
    auto& governor = ResourceGovernor::instance();

    constexpr std::uint32_t currentSlots = 16;
    constexpr std::uint32_t minSlots = 4;
    constexpr std::uint32_t maxSlots = 64;
    constexpr std::uint32_t scaleStep = 8;

    // Normalize singleton policy state to avoid cross-test carryover.
    governor.reportDbLockContention(0, 1);
    (void)governor.recommendConnectionSlotTarget(currentSlots, currentSlots / 2, minSlots, maxSlots,
                                                 scaleStep, 1);

    auto unchanged = governor.recommendConnectionSlotTarget(currentSlots, currentSlots, minSlots,
                                                            maxSlots, scaleStep, 3);
    CHECK((unchanged == currentSlots));

    auto shrunk =
        governor.recommendConnectionSlotTarget(currentSlots, 1, minSlots, maxSlots, scaleStep, 1);
    CHECK((shrunk <= currentSlots));
    CHECK((shrunk >= minSlots));

    governor.reportDbLockContention(100, 10);
    CHECK((governor.capKgConcurrencyForDbContention(8) <= 2));
    CHECK((governor.capEmbedConcurrencyForDbContention(8) <= 1));

    governor.reportDbLockContention(0, 10);
    CHECK((governor.capKgConcurrencyForDbContention(8) == 8));
    CHECK((governor.capEmbedConcurrencyForDbContention(4) == 4));

    const auto retryBaseline = governor.recommendRetryAfterMs(0, 100, 0, 1024, 1, 100, 0, 100, 500);
    const auto retryOverloaded = governor.recommendRetryAfterMs(
        200, 100, 4ULL * 1024 * 1024, 1024ULL * 1024, 200, 50, 100, 100, 500);
    CHECK((retryOverloaded >= retryBaseline));
    CHECK((retryOverloaded <= 5000));

    const auto pauseIdle = governor.recommendBackpressureReadPauseMs(100);
    CHECK((pauseIdle >= 1));
    CHECK((pauseIdle <= 5000));

    // Leave singleton policy state neutral for subsequent tests.
    governor.reportDbLockContention(0, 1);
}