// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (c) 2025 Trevon Helm
// Consolidated background processing tests (3 → 1): PostIngestQueue, InternalEventBus, MPMC
// concurrency Covers: queue lifecycle, async processing, bus integration, stress testing, MPMC
// correctness

// pi-lens-ignore: fatal error
#include <spdlog/sinks/ostream_sink.h>
#include <spdlog/spdlog.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdlib>
#include <filesystem>
#include <memory>
#include <mutex>
#include <optional>
#include <span>
#include <sstream>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
#include "../../common/test_helpers_catch2.h"
#include <catch2/catch_test_macros.hpp>
#include <yams/api/content_store.h>
#include <yams/daemon/components/DaemonLifecycleFsm.h>
#include <yams/daemon/components/IngestService.h>
#include <yams/daemon/components/InternalEventBus.h>
#include <yams/daemon/components/PostIngestQueue.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/TuneAdvisor.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/extraction/content_extractor.h>
#include <yams/metadata/metadata_repository.h>

using namespace yams;
using namespace yams::daemon;

namespace {

/// PostIngestQueue::processed() counts prepared tasks before their content is committed
/// (commitBatchResults() runs after the counter advances), so it is not a completion barrier for
/// assertions on repository state. Wait for the committed observable with a bounded deadline.
template <typename Predicate>
void waitForCommittedState(Predicate&& ready,
                           std::chrono::milliseconds timeout = std::chrono::seconds(10)) {
    const auto deadline = std::chrono::steady_clock::now() + timeout;
    while (!ready() && std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

void stopAndResetQueue(std::unique_ptr<PostIngestQueue>& queue) {
    if (!queue) {
        return;
    }
    queue->stop();
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (queue->started() && std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    queue.reset();
}

/// Drain and remove a channel, so the next requester gets a queue at the capacity it declares.
///
/// get_or_create_channel() returns an existing channel and ignores the requested capacity, so a
/// channel created here at some incidental capacity pins that capacity for every later case: a
/// queue built for a larger ring silently received the smaller one, the producer blocked until the
/// poller drained, and batch boundaries became a function of runner load (flaky exact-count
/// assertions). Dropping the channel is what makes each case's own capacity take effect.
template <typename T> void drainAndRemoveChannel(const std::string& name) {
    auto& bus = InternalEventBus::instance();
    if (auto existing = bus.get_channel<T>(name)) {
        T drain;
        while (existing->try_pop(drain)) {
        }
    }
    bus.remove_channel(name);
}

/// Removes a channel when the enclosing scope ends.
///
/// A case may deliberately size a channel smaller than production does, to fill it quickly or to
/// force backpressure. What it must not do is leave that ring registered: the bus is a process-wide
/// singleton, so a later case asking for the production capacity silently receives the smaller ring
/// and its producer can block. Pair this with the deliberate capacity.
class ScopedChannelRemoval {
public:
    explicit ScopedChannelRemoval(std::string name) : name_(std::move(name)) {}
    ~ScopedChannelRemoval() { InternalEventBus::instance().remove_channel(name_); }
    ScopedChannelRemoval(const ScopedChannelRemoval&) = delete;
    ScopedChannelRemoval& operator=(const ScopedChannelRemoval&) = delete;
    ScopedChannelRemoval(ScopedChannelRemoval&&) = delete;
    ScopedChannelRemoval& operator=(ScopedChannelRemoval&&) = delete;

private:
    std::string name_;
};

/// Drain leftover items from the singleton post_ingest channel to prevent cross-test interference
/// (tasks enqueued by a previous test that the previous queue didn't consume before stopping).
void drainPostIngestChannel() {
    drainAndRemoveChannel<InternalEventBus::PostIngestTask>("post_ingest");
}

void drainEmbedJobsChannel() {
    drainAndRemoveChannel<InternalEventBus::EmbedJob>("embed_jobs");
}

void drainStoreDocumentChannel() {
    drainAndRemoveChannel<InternalEventBus::StoreDocumentTask>("store_document_tasks");
}

class StoreDocumentChannelGuard {
public:
    StoreDocumentChannelGuard() { drainStoreDocumentChannel(); }
    ~StoreDocumentChannelGuard() { drainStoreDocumentChannel(); }
};

class PostIngestBatchSizeGuard {
public:
    explicit PostIngestBatchSizeGuard(std::uint32_t batchSize) {
        TuneAdvisor::setPostIngestBatchSize(batchSize);
    }
    ~PostIngestBatchSizeGuard() { TuneAdvisor::setPostIngestBatchSize(0); }
};

using yams::test::ScopedEnvVar;

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

TEST_CASE("Ingest storage batching is independent from post-ingest transaction batching",
          "[daemon][ingest][batching]") {
    PostIngestBatchSizeGuard postIngestBatchGuard(16);

    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 20.0, true, true, false) == 64);
    CHECK(TuneAdvisor::postIngestBatchSize() == 16);
}

TEST_CASE("Ingest storage batching never exceeds its configured maximum under pressure",
          "[daemon][ingest][batching]") {
    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 81.0, true, true, false) == 8);
    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 61.0, true, true, false) == 10);
    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 41.0, true, true, false) == 12);
    CHECK(IngestService::testing_resolveStoreBatchLimit(4, 81.0, true, true, false) == 4);
    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 20.0, false, true, false) == 32);
    CHECK(IngestService::testing_resolveStoreBatchLimit(64, 20.0, false, false, false) == 4);
    CHECK(IngestService::testing_resolveStoreBatchLimit(2, 20.0, false, true, false) == 2);
    CHECK(IngestService::testing_resolveStoreBatchLimit(0, 20.0, true, true, false) == 1);
    CHECK(IngestService::testing_resolveStoreBatchLimit(1024, 20.0, true, true, false) == 256);
    CHECK(IngestService::testing_resolveStoreBatchLimit(32, 81.0, false, true, true) == 32);
    CHECK(IngestService::testing_resolveStoreBatchLimit(4, 81.0, false, true, true) == 4);
}

class SpdlogCaptureGuard {
public:
    explicit SpdlogCaptureGuard(spdlog::level::level_enum level)
        : previousLogger_(spdlog::default_logger()), previousLevel_(spdlog::get_level()) {
        auto sink = std::make_shared<spdlog::sinks::ostream_sink_mt>(stream_);
        logger_ = std::make_shared<spdlog::logger>("daemon_background_capture", sink);
        logger_->set_level(level);
        spdlog::set_default_logger(logger_);
        spdlog::set_level(level);
    }

    ~SpdlogCaptureGuard() {
        if (previousLogger_) {
            spdlog::set_default_logger(previousLogger_);
        }
        spdlog::set_level(previousLevel_);
    }

    std::string str() const { return stream_.str(); }

private:
    std::ostringstream stream_;
    std::shared_ptr<spdlog::logger> previousLogger_;
    std::shared_ptr<spdlog::logger> logger_;
    spdlog::level::level_enum previousLevel_;
};

// RAII guard for InternalEventBus configuration
class BusToggleGuard {
public:
    explicit BusToggleGuard(bool enable) {
        prev_ = TuneAdvisor::useInternalBusForPostIngest();
        TuneAdvisor::setUseInternalBusForPostIngest(enable);
    }
    ~BusToggleGuard() { TuneAdvisor::setUseInternalBusForPostIngest(prev_); }

private:
    bool prev_{false};
};

class PostIngestBatchGuard {
public:
    explicit PostIngestBatchGuard(uint32_t size) {
        prev_ = TuneAdvisor::postIngestBatchSize();
        TuneAdvisor::setPostIngestBatchSize(size);
    }
    ~PostIngestBatchGuard() { TuneAdvisor::setPostIngestBatchSize(prev_); }

private:
    uint32_t prev_{0};
};

class PostIngestConcurrencyGuard {
public:
    PostIngestConcurrencyGuard(uint32_t totalConcurrent, uint32_t extractionConcurrent) {
        prevHw_ = TuneAdvisor::hardwareConcurrency();
        prevTotal_ = TuneAdvisor::postIngestTotalConcurrent();
        prevExtraction_ = TuneAdvisor::postExtractionConcurrent();
        TuneAdvisor::setHardwareConcurrencyForTests(32);
        TuneAdvisor::setPostIngestTotalConcurrent(totalConcurrent);
        TuneAdvisor::setPostExtractionConcurrent(extractionConcurrent);
        TuneAdvisor::setPostExtractionConcurrentDynamicCap(UINT32_MAX);
    }
    ~PostIngestConcurrencyGuard() {
        TuneAdvisor::setHardwareConcurrencyForTests(prevHw_);
        TuneAdvisor::setPostIngestTotalConcurrent(prevTotal_);
        TuneAdvisor::setPostExtractionConcurrent(prevExtraction_);
        TuneAdvisor::setPostExtractionConcurrentDynamicCap(UINT32_MAX);
    }

private:
    uint32_t prevHw_{0};
    uint32_t prevTotal_{0};
    uint32_t prevExtraction_{0};
};

class PostIngestStageActivityGuard {
public:
    explicit PostIngestStageActivityGuard(TuneAdvisor::PostIngestStage stage)
        : stage_(stage), token_(TuneAdvisor::acquirePostIngestStageActivity(stage)) {}
    ~PostIngestStageActivityGuard() { TuneAdvisor::releasePostIngestStageActivity(stage_, token_); }

    PostIngestStageActivityGuard(const PostIngestStageActivityGuard&) = delete;
    PostIngestStageActivityGuard& operator=(const PostIngestStageActivityGuard&) = delete;

private:
    TuneAdvisor::PostIngestStage stage_;
    TuneAdvisor::PostIngestStageActivityToken token_;
};

class WorkCoordinatorThreadsGuard {
public:
    explicit WorkCoordinatorThreadsGuard(uint32_t threads)
        : previous_(TuneAdvisor::workCoordinatorThreads()) {
        TuneAdvisor::setWorkCoordinatorThreads(threads);
    }
    ~WorkCoordinatorThreadsGuard() { TuneAdvisor::setWorkCoordinatorThreads(previous_); }

private:
    uint32_t previous_{0};
};

// Unified StubContentStore (thread-safe, supports all required operations)
class StubContentStore : public api::IContentStore {
public:
    void setContent(const std::string& hash, const std::string& data) {
        std::lock_guard<std::mutex> lk(mu_);
        ByteVector bytes;
        bytes.reserve(data.size());
        for (unsigned char c : data) {
            bytes.push_back(static_cast<std::byte>(c));
        }
        blobs_[hash] = bytes;
    }

    Result<std::vector<std::byte>> retrieveBytes(const std::string& hash) override {
        std::lock_guard<std::mutex> lk(mu_);
        auto it = blobs_.find(hash);
        if (it == blobs_.end()) {
            return Error{ErrorCode::NotFound, "content not found"};
        }
        return it->second;
    }

    Result<std::vector<std::byte>> retrieveBytesPrefix(const std::string& hash,
                                                       std::size_t maxBytes) override {
        auto r = retrieveBytes(hash);
        if (!r)
            return r;
        auto& v = r.value();
        if (v.size() > maxBytes)
            v.resize(maxBytes);
        return r;
    }

    Result<api::IContentStore::RawContent> retrieveRaw(const std::string& hash) override {
        auto bytes = retrieveBytes(hash);
        if (!bytes) {
            return bytes.error();
        }
        api::IContentStore::RawContent raw;
        raw.data = bytes.value();
        return raw;
    }

    std::future<Result<api::IContentStore::RawContent>>
    retrieveRawAsync(const std::string& hash) override {
        return std::async(std::launch::deferred, [this, hash]() { return retrieveRaw(hash); });
    }

    Result<api::RetrieveResult> retrieveStream(const std::string& hash, std::ostream& os,
                                               api::ProgressCallback) override {
        std::lock_guard<std::mutex> lk(mu_);
        auto it = blobs_.find(hash);
        if (it == blobs_.end()) {
            return Error{ErrorCode::NotFound, "no blob"};
        }
        for (auto b : it->second) {
            os.put(static_cast<char>(b));
        }
        api::RetrieveResult r{};
        r.found = true;
        r.size = static_cast<uint64_t>(it->second.size());
        return r;
    }

    Result<api::StoreResult> store(const std::filesystem::path& path, const api::ContentMetadata&,
                                   api::ProgressCallback) override {
        singleCalls_.fetch_add(1, std::memory_order_relaxed);
        storedItems_.fetch_add(1, std::memory_order_release);
        storeCv_.notify_all();

        api::StoreResult result;
        result.contentHash = "single-hash";
        std::error_code error;
        result.bytesStored = std::filesystem::file_size(path, error);
        if (error) {
            return Error{ErrorCode::IOError, error.message()};
        }
        return result;
    }
    Result<api::RetrieveResult> retrieve(const std::string&, const std::filesystem::path&,
                                         api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }
    Result<api::StoreResult> storeStream(std::istream&, const api::ContentMetadata&,
                                         api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }
    Result<api::StoreResult> storeBytes(std::span<const std::byte>,
                                        const api::ContentMetadata&) override {
        return ErrorCode::NotImplemented;
    }
    Result<bool> exists(const std::string&) const override { return ErrorCode::NotImplemented; }
    Result<bool> remove(const std::string&) override { return ErrorCode::NotImplemented; }
    Result<api::ContentMetadata> getMetadata(const std::string&) const override {
        return ErrorCode::NotImplemented;
    }
    Result<void> updateMetadata(const std::string&, const api::ContentMetadata&) override {
        return ErrorCode::NotImplemented;
    }
    std::vector<Result<api::StoreResult>>
    storeBatch(const std::vector<std::filesystem::path>& paths,
               const std::vector<api::ContentMetadata>&) override {
        batchCalls_.fetch_add(1, std::memory_order_relaxed);
        lastBatchSize_.store(paths.size(), std::memory_order_relaxed);
        storedItems_.fetch_add(paths.size(), std::memory_order_release);
        storeCv_.notify_all();

        std::vector<Result<api::StoreResult>> results;
        results.reserve(paths.size());
        for (std::size_t i = 0; i < paths.size(); ++i) {
            api::StoreResult result;
            result.contentHash = "batch-hash-" + std::to_string(i);
            std::error_code error;
            result.bytesStored = std::filesystem::file_size(paths[i], error);
            if (error) {
                results.emplace_back(Error{ErrorCode::IOError, error.message()});
            } else {
                results.emplace_back(std::move(result));
            }
        }
        return results;
    }
    std::vector<Result<bool>> removeBatch(const std::vector<std::string>&) override { return {}; }
    api::ContentStoreStats getStats() const override { return {}; }
    api::HealthStatus checkHealth() const override { return {}; }
    Result<void> verify(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
    Result<void> compact(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
    Result<void> garbageCollect(api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }

    [[nodiscard]] std::size_t batchCalls() const {
        return batchCalls_.load(std::memory_order_relaxed);
    }

    [[nodiscard]] std::size_t singleCalls() const {
        return singleCalls_.load(std::memory_order_relaxed);
    }

    [[nodiscard]] std::size_t lastBatchSize() const {
        return lastBatchSize_.load(std::memory_order_relaxed);
    }

    bool waitForStoredItems(std::size_t expected, std::chrono::steady_clock::time_point deadline) {
        std::unique_lock lock(storeMutex_);
        return storeCv_.wait_until(lock, deadline, [this, expected] {
            return storedItems_.load(std::memory_order_acquire) >= expected;
        });
    }

private:
    mutable std::mutex mu_;
    std::unordered_map<std::string, ByteVector> blobs_;
    std::atomic<std::size_t> singleCalls_{0};
    std::atomic<std::size_t> batchCalls_{0};
    std::atomic<std::size_t> lastBatchSize_{0};
    std::atomic<std::size_t> storedItems_{0};
    std::mutex storeMutex_;
    std::condition_variable storeCv_;
};

// Minimal MetadataRepository stub for testing
std::shared_ptr<metadata::ConnectionPool> getTestPool() {
    static std::shared_ptr<metadata::ConnectionPool> pool = [] {
        metadata::ConnectionPoolConfig cfg{};
        cfg.enableWAL = false;
        cfg.enableForeignKeys = false;
        auto p = std::make_shared<metadata::ConnectionPool>(":memory:", cfg);
        (void)p->initialize();
        return p;
    }();
    return pool;
}

class StubMetadataRepository : public metadata::MetadataRepository {
public:
    StubMetadataRepository() : metadata::MetadataRepository(*getTestPool()) {}

    void setDocument(metadata::DocumentInfo doc) {
        std::lock_guard<std::mutex> lk(mu_);
        docsByHash_[doc.sha256Hash] = doc;
        docsById_[doc.id] = doc;
        lastUpdated_ = doc;
    }

    bool contentInserted() const {
        std::lock_guard<std::mutex> lk(mu_);
        return contentInserted_;
    }

    metadata::DocumentContent lastContent() const {
        std::lock_guard<std::mutex> lk(mu_);
        return lastContent_;
    }

    metadata::DocumentInfo lastUpdated() const {
        std::lock_guard<std::mutex> lk(mu_);
        return lastUpdated_;
    }

    Result<std::optional<metadata::DocumentInfo>>
    getDocumentByHash(const std::string& hash) override {
        std::lock_guard<std::mutex> lk(mu_);
        auto it = docsByHash_.find(hash);
        if (it != docsByHash_.end()) {
            return std::optional<metadata::DocumentInfo>(it->second);
        }
        return std::optional<metadata::DocumentInfo>(std::nullopt);
    }

    Result<std::optional<metadata::DocumentInfo>> getDocument(int64_t id) override {
        std::lock_guard<std::mutex> lk(mu_);
        auto it = docsById_.find(id);
        if (it != docsById_.end()) {
            return std::optional<metadata::DocumentInfo>(it->second);
        }
        return std::optional<metadata::DocumentInfo>(std::nullopt);
    }

    Result<void> updateDocument(const metadata::DocumentInfo& info) override {
        std::lock_guard<std::mutex> lk(mu_);
        docsByHash_[info.sha256Hash] = info;
        docsById_[info.id] = info;
        lastUpdated_ = info;
        return Result<void>();
    }

    Result<void> insertContent(const metadata::DocumentContent& content) override {
        std::lock_guard<std::mutex> lk(mu_);
        lastContent_ = content;
        contentInserted_ = true;
        return Result<void>();
    }

    Result<void> indexDocumentContent(int64_t, const std::string&, const std::string& content,
                                      const std::string&) override {
        std::lock_guard<std::mutex> lk(mu_);
        indexedContent_ = content;
        return Result<void>();
    }

    Result<void> indexDocumentContentTrusted(int64_t, const std::string&,
                                             const std::string& content,
                                             const std::string&) override {
        std::lock_guard<std::mutex> lk(mu_);
        indexedContent_ = content;
        return Result<void>();
    }

    Result<void> updateDocumentExtractionStatus(int64_t docId, bool contentExtracted,
                                                metadata::ExtractionStatus status,
                                                const std::string& = "") override {
        std::lock_guard<std::mutex> lk(mu_);
        auto it = docsById_.find(docId);
        if (it != docsById_.end()) {
            it->second.contentExtracted = contentExtracted;
            it->second.extractionStatus = status;
            lastUpdated_ = it->second;
            docsByHash_[it->second.sha256Hash] = it->second;
        }
        return Result<void>();
    }

    void addSymSpellTerm(std::string_view, int64_t) override { /* no-op for mock */ }

    Result<std::vector<std::string>> getDocumentTags(int64_t) override {
        return std::vector<std::string>{};
    }

    Result<void> setMetadataBatch(
        const std::vector<std::tuple<int64_t, std::string, metadata::MetadataValue>>&) override {
        return Result<void>();
    }

    Result<void> batchUpdateDocumentRepairStatuses(const std::vector<std::string>&,
                                                   metadata::RepairStatus) override {
        return Result<void>();
    }

    void signalCorpusStatsStale() override { /* no-op for mock */ }

    Result<void>
    batchInsertContentAndIndex(const std::vector<metadata::BatchContentEntry>& entries) override {
        std::lock_guard<std::mutex> lk(mu_);
        batchWriteSizes_.push_back(entries.size());
        for (const auto& entry : entries) {
            batchInsertedDocIds_.push_back(entry.documentId);
            contentInserted_ = true;
            // Store last content for verification
            lastContent_.documentId = entry.documentId;
            lastContent_.contentText = entry.contentText;
            lastContent_.contentLength = static_cast<int64_t>(entry.contentText.size());
            lastContent_.extractionMethod = entry.extractionMethod;
            lastContent_.language = entry.language;
            // Update document status
            auto it = docsById_.find(entry.documentId);
            if (it != docsById_.end()) {
                it->second.contentExtracted = true;
                it->second.extractionStatus = metadata::ExtractionStatus::Success;
                docsByHash_[it->second.sha256Hash] = it->second;
                lastUpdated_ = it->second;
            }
        }
        return Result<void>();
    }

    Result<std::unordered_map<std::string, metadata::DocumentInfo>>
    batchGetDocumentsByHash(const std::vector<std::string>& hashes) override {
        std::lock_guard<std::mutex> lk(mu_);
        batchGetCalls_++;
        std::unordered_map<std::string, metadata::DocumentInfo> out;
        out.reserve(hashes.size());
        for (const auto& hash : hashes) {
            auto it = docsByHash_.find(hash);
            if (it != docsByHash_.end()) {
                out.emplace(hash, it->second);
            }
        }
        return out;
    }

    Result<std::unordered_map<int64_t, metadata::DocumentContent>>
    batchGetContent(const std::vector<int64_t>& documentIds) override {
        std::lock_guard<std::mutex> lk(mu_);
        std::unordered_map<int64_t, metadata::DocumentContent> out;
        out.reserve(documentIds.size());
        for (auto id : documentIds) {
            auto it = docsById_.find(id);
            if (it == docsById_.end()) {
                continue;
            }
            metadata::DocumentContent content{};
            content.documentId = id;
            content.contentText = indexedContent_;
            content.contentLength = static_cast<int64_t>(indexedContent_.size());
            content.extractionMethod = "post_ingest";
            content.language = "en";
            out.emplace(id, content);
        }
        return out;
    }

    Result<std::unordered_map<int64_t, std::vector<std::string>>>
    batchGetDocumentTags(std::span<const int64_t> documentIds) override {
        std::lock_guard<std::mutex> lk(mu_);
        std::unordered_map<int64_t, std::vector<std::string>> tags;
        tags.reserve(documentIds.size());
        for (auto id : documentIds) {
            tags.emplace(id, std::vector<std::string>{});
        }
        return tags;
    }

    std::size_t batchGetCalls() const {
        std::lock_guard<std::mutex> lk(mu_);
        return batchGetCalls_;
    }

    std::vector<int64_t> batchInsertedDocIds() const {
        std::lock_guard<std::mutex> lk(mu_);
        return batchInsertedDocIds_;
    }

    std::size_t maxBatchWriteSize() const {
        std::lock_guard<std::mutex> lk(mu_);
        std::size_t maxSize = 0;
        for (std::size_t n : batchWriteSizes_) {
            maxSize = std::max(maxSize, n);
        }
        return maxSize;
    }

private:
    mutable std::mutex mu_;
    metadata::DocumentInfo lastUpdated_{};
    metadata::DocumentContent lastContent_{};
    std::string indexedContent_;
    bool contentInserted_{false};
    std::size_t batchGetCalls_{0};
    std::vector<int64_t> batchInsertedDocIds_{};
    std::vector<std::size_t> batchWriteSizes_{};
    std::unordered_map<std::string, metadata::DocumentInfo> docsByHash_{};
    std::unordered_map<int64_t, metadata::DocumentInfo> docsById_{};
};

/// Holds the first content commit so a test can observe queue counters while content is provably
/// uncommitted. Always released on destruction so a failing case cannot strand the writer thread.
class GatedInsertRepository : public StubMetadataRepository {
public:
    Result<void>
    batchInsertContentAndIndex(const std::vector<metadata::BatchContentEntry>& entries) override {
        {
            std::unique_lock<std::mutex> lock(gateMutex_);
            insertEntered_ = true;
            gateCv_.notify_all();
            gateCv_.wait(lock, [this] { return released_; });
        }
        return StubMetadataRepository::batchInsertContentAndIndex(entries);
    }

    bool waitForInsertEntered(std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(gateMutex_);
        return gateCv_.wait_for(lock, timeout, [this] { return insertEntered_; });
    }

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

    ~GatedInsertRepository() override { release(); }

private:
    std::mutex gateMutex_;
    std::condition_variable gateCv_;
    bool insertEntered_{false};
    bool released_{false};
};

/// Releases the insert gate before the owning queue is destroyed, so a failing assertion cannot
/// deadlock the queue's writer-thread join on a still-blocked repository call.
struct InsertGateRelease {
    std::shared_ptr<GatedInsertRepository> repo;
    ~InsertGateRelease() {
        if (repo) {
            repo->release();
        }
    }
};

class StubExtractor : public extraction::IContentExtractor {
public:
    bool supports(const std::string& mime, const std::string&) const override {
        return mime == "text/plain";
    }

    std::optional<std::string> extractText(const std::vector<std::byte>& bytes, const std::string&,
                                           const std::string&) override {
        std::string text;
        text.reserve(bytes.size());
        for (auto b : bytes) {
            text.push_back(static_cast<char>(b));
        }
        return text;
    }
};

struct StoreCallSnapshot {
    std::size_t singleCalls{0};
    std::size_t batchCalls{0};
    std::size_t lastBatchSize{0};
};

StoreCallSnapshot runQueuedDocuments(std::size_t count, std::uint32_t batchSize = 0) {
    StoreDocumentChannelGuard channelGuard;
    std::optional<PostIngestBatchSizeGuard> batchSizeGuard;
    if (batchSize != 0) {
        batchSizeGuard.emplace(batchSize);
    }
    WorkCoordinatorThreadsGuard threadsGuard(1);
    yams::test::TempDirGuard testDir("yams_ingest_service_batch_");

    auto contentStore = std::make_shared<StubContentStore>();
    {
        DaemonConfig config;
        config.dataDir = testDir.path();
        StateComponent state;
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager serviceManager(config, state, lifecycleFsm);
        serviceManager.__test_setContentStore(contentStore);

        auto channel =
            InternalEventBus::instance().get_or_create_channel<InternalEventBus::StoreDocumentTask>(
                "store_document_tasks", 64);
        for (std::size_t i = 0; i < count; ++i) {
            InternalEventBus::StoreDocumentTask task;
            task.request.name = "queued-" + std::to_string(i) + ".txt";
            task.request.content = "queued content " + std::to_string(i);
            REQUIRE(channel->try_push(std::move(task)));
        }

        IngestService ingestService(&serviceManager, serviceManager.getWorkCoordinator());
        ingestService.start();
        CHECK(contentStore->waitForStoredItems(count, std::chrono::steady_clock::now() +
                                                          std::chrono::seconds(2)));
        ingestService.stop();
    }

    return {.singleCalls = contentStore->singleCalls(),
            .batchCalls = contentStore->batchCalls(),
            .lastBatchSize = contentStore->lastBatchSize()};
}

bool isMpmcEnabled() {
    const char* m = std::getenv("YAMS_INTERNAL_BUS_MPMC");
    return m && std::string(m) == "1";
}

} // namespace

// =============================================================================
// IngestService Tests
// =============================================================================

TEST_CASE("IngestService preserves single and batched queued storage paths",
          "[daemon][background][ingest][batch]") {
    SpdlogCaptureGuard logGuard(spdlog::level::err);

    SECTION("one task keeps the exact single-store path") {
        const auto calls = runQueuedDocuments(1);
        CHECK(calls.singleCalls == 1);
        CHECK(calls.batchCalls == 0);
    }

    SECTION("two tasks cross one batch boundary") {
        const auto calls = runQueuedDocuments(2);
        CHECK(calls.singleCalls == 0);
        CHECK(calls.batchCalls == 1);
        CHECK(calls.lastBatchSize == 2);
    }

    SECTION("one drained admission batch crosses one storage boundary") {
        constexpr std::size_t kDocuments = 12;
        const auto calls = runQueuedDocuments(kDocuments);
        CHECK(calls.singleCalls == 0);
        CHECK(calls.batchCalls == 1);
        CHECK(calls.lastBatchSize == kDocuments);
    }

    SECTION("configured batch size also bounds the storage drain") {
        constexpr std::size_t kDocuments = 24;
        const auto calls = runQueuedDocuments(kDocuments, 32);
        CHECK(calls.singleCalls == 0);
        CHECK(calls.batchCalls == 1);
        CHECK(calls.lastBatchSize == kDocuments);
    }
}

TEST_CASE("IngestService suspends instead of blocking its WorkCoordinator worker",
          "[daemon][background][ingest][executor-progress]") {
    SpdlogCaptureGuard logGuard(spdlog::level::err);
    StoreDocumentChannelGuard channelGuard;
    WorkCoordinatorThreadsGuard threadsGuard(2);
    yams::test::TempDirGuard testDir("yams_ingest_service_executor_progress_");

    DaemonConfig config;
    config.dataDir = testDir.path();
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager serviceManager(config, state, lifecycleFsm);
    auto contentStore = std::make_shared<StubContentStore>();
    serviceManager.__test_setContentStore(contentStore);

    auto blockerStarted = std::make_shared<std::promise<void>>();
    auto blockerStartedFuture = blockerStarted->get_future();
    auto releaseBlocker = std::make_shared<std::promise<void>>();
    auto releaseFuture = releaseBlocker->get_future().share();
    boost::asio::post(serviceManager.getWorkCoordinator()->getExecutor(),
                      [blockerStarted, releaseFuture] {
                          blockerStarted->set_value();
                          releaseFuture.wait();
                      });
    REQUIRE(blockerStartedFuture.wait_for(std::chrono::seconds(1)) == std::future_status::ready);

    auto channel =
        InternalEventBus::instance().get_or_create_channel<InternalEventBus::StoreDocumentTask>(
            "store_document_tasks", 64);
    InternalEventBus::StoreDocumentTask task;
    task.request.name = "executor-progress.txt";
    task.request.content = "executor progress";
    REQUIRE(channel->try_push(std::move(task)));

    IngestService ingestService(&serviceManager, serviceManager.getWorkCoordinator());
    ingestService.start();
    const bool stored = contentStore->waitForStoredItems(1, std::chrono::steady_clock::now() +
                                                                std::chrono::seconds(1));

    releaseBlocker->set_value();
    ingestService.stop();
    CHECK(stored);
}

// =============================================================================
// PostIngestQueue Tests
// =============================================================================

TEST_CASE("PostIngestQueue full KG channel does not block WorkCoordinator progress",
          "[daemon][background][queue][kg][backpressure]") {
    WorkCoordinator coordinator;
    coordinator.start(1);

    auto kgChannel =
        InternalEventBus::instance().get_or_create_channel<InternalEventBus::KgJob>("kg_jobs", 32);
    std::size_t fillerId = 0;
    while (!kgChannel->full()) {
        InternalEventBus::KgJob filler;
        filler.hash = "kg-filler-" + std::to_string(fillerId++);
        REQUIRE(kgChannel->try_push(std::move(filler)));
    }

    {
        PostIngestQueue queue(nullptr, nullptr, {}, nullptr, nullptr, &coordinator, nullptr, 32);
        InternalEventBus::KgJob job;
        job.hash = "kg-pending-target";
        queue.testing_deferKgJob(std::move(job));
        queue.testing_schedulePendingKgDrain();
        REQUIRE(queue.testing_pendingKgJobs() == 1);
        CHECK_FALSE(queue.tryEnqueue(PostIngestQueue::Task{.hash = "upstream-backpressure"}));

        auto sentinelRan = std::make_shared<std::promise<void>>();
        auto sentinelFuture = sentinelRan->get_future();
        boost::asio::post(coordinator.getExecutor(), [sentinelRan] { sentinelRan->set_value(); });
        REQUIRE(sentinelFuture.wait_for(std::chrono::milliseconds(250)) ==
                std::future_status::ready);

        // Explicit disablement cancels deferred work outside the pipeline without counting a drop.
        const auto droppedBefore = InternalEventBus::instance().kgDropped();
        queue.setKnowledgeGraphEnabled(false);
        CHECK(queue.testing_pendingKgJobs() == 0);
        CHECK(InternalEventBus::instance().kgDropped() == droppedBefore);
        queue.stop();
    }

    coordinator.stop();
    coordinator.join();

    InternalEventBus::KgJob drained;
    while (kgChannel->try_pop(drained)) {
    }
    // Drop the ring as well as its contents. The 32 here is deliberate - this case wants a channel
    // it can fill quickly - but leaving it registered pins kg_jobs to 32 for every later case, and
    // a queue configured with capacity_ == 0 asks for boundedStageChannelCapacity(16384) instead.
    InternalEventBus::instance().remove_channel("kg_jobs");
}

TEST_CASE("PostIngestQueue accounts deferred KG jobs when its scheduler is unavailable",
          "[daemon][background][queue][kg][backpressure]") {
    WorkCoordinator coordinator;
    PostIngestQueue queue(nullptr, nullptr, {}, nullptr, nullptr, &coordinator, nullptr, 32);
    const auto droppedBefore = InternalEventBus::instance().kgDropped();

    InternalEventBus::KgJob job;
    job.hash = "kg-unscheduled-target";
    queue.testing_deferKgJob(std::move(job));
    queue.testing_schedulePendingKgDrain();

    CHECK(queue.testing_pendingKgJobs() == 0);
    CHECK(InternalEventBus::instance().kgDropped() == droppedBefore + 1);
}

TEST_CASE("PostIngestQueue: Basic lifecycle and task processing", "[daemon][background][queue]") {
    BusToggleGuard busGuard(false); // Disable bus for direct queue testing
    drainPostIngestChannel();

    // WorkCoordinator is required for PostIngestQueue strand creation
    WorkCoordinator coordinator;
    coordinator.start(2);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();

    metadata::DocumentInfo doc{};
    doc.id = 101;
    doc.fileName = "doc.txt";
    doc.fileExtension = ".txt";
    doc.sha256Hash = "hash-123";
    doc.mimeType = "text/plain";
    doc.indexedTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    metadataRepo->setDocument(doc);

    const std::string payload = "Hello from post-ingest";
    store->setContent(doc.sha256Hash, payload);

    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    SECTION("Process single task successfully") {
        drainEmbedJobsChannel();
        // 2048 is this case's own choice; the guard keeps it from outliving the heading.
        ScopedChannelRemoval embedChannelRemoval{"embed_jobs"};
        auto embedChannel =
            InternalEventBus::instance().get_or_create_channel<InternalEventBus::EmbedJob>(
                "embed_jobs", 2048);

        auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                       nullptr, &coordinator, nullptr, 32);
        queue->start(); // Start the channel poller

        // Wait for channel poller coroutine to start before enqueueing
        auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
        while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        REQUIRE(queue->started());

        PostIngestQueue::Task task{.hash = doc.sha256Hash, .mime = doc.mimeType};
        task.noEmbeddings = true;
        REQUIRE(queue->tryEnqueue(std::move(task)));

        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
        while (queue->processed() < 1 && std::chrono::steady_clock::now() < deadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }

        REQUIRE((queue->processed() == 1));
        REQUIRE((queue->failed() == 0));
        REQUIRE((queue->entityInFlight() == 0));
        // The content insert is a later pipeline stage than processed(); on slow platforms it can
        // lag, so poll instead of asserting immediately.
        auto contentDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
        while (!metadataRepo->contentInserted() &&
               std::chrono::steady_clock::now() < contentDeadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        REQUIRE(metadataRepo->contentInserted());

        auto content = metadataRepo->lastContent();
        REQUIRE((content.documentId == doc.id));
        REQUIRE((content.contentText == payload));
        REQUIRE((content.contentLength == static_cast<int64_t>(payload.size())));
        REQUIRE((content.extractionMethod == "post_ingest"));
        REQUIRE_FALSE(content.language.empty());

        auto updated = metadataRepo->lastUpdated();
        REQUIRE(updated.contentExtracted);
        REQUIRE((updated.extractionStatus == metadata::ExtractionStatus::Success));

        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        InternalEventBus::EmbedJob embedJob;
        CHECK_FALSE(embedChannel->try_pop(embedJob));

        stopAndResetQueue(queue);
    }

    SECTION("Stop wakes title-inactive poller promptly") {
        auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                       nullptr, &coordinator, nullptr, 32);
        queue->start();

        auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
        while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        REQUIRE(queue->started());

        std::this_thread::sleep_for(std::chrono::milliseconds(80));

        const auto stopStart = std::chrono::steady_clock::now();
        queue->stop();
        const auto stopElapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now() - stopStart);

        INFO("stopElapsedMs=" << stopElapsed.count());
        CHECK((stopElapsed < std::chrono::milliseconds(150)));
        CHECK_FALSE(queue->started());

        queue.reset();
    }

    SECTION("Queue shutdown drains pending tasks") {
        auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                       nullptr, &coordinator, nullptr, 32);
        queue->start(); // Start the channel poller

        // Wait for channel poller coroutine to start before enqueueing
        auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
        while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        REQUIRE(queue->started());

        PostIngestQueue::Task task{.hash = doc.sha256Hash, .mime = doc.mimeType};
        REQUIRE(queue->tryEnqueue(task));

        stopAndResetQueue(queue);
        SUCCEED("Queue shutdown completed without hang");
    }

    // Cleanup coordinator at test end
    coordinator.stop();
    coordinator.join();
}

TEST_CASE("PostIngestQueue: Batch uses batched metadata lookup and enqueues embeds",
          "[daemon][background][queue][batch]") {
    BusToggleGuard busGuard(false);
    drainPostIngestChannel();
    PostIngestBatchGuard batchGuard(4);
    PostIngestStageActivityGuard embedActivity(TuneAdvisor::PostIngestStage::Embed);

    // Ensure embed chunking policy is exercised through the PostIngestQueue -> embed_jobs path.
    ScopedEnvVar chunkStrategy("YAMS_EMBED_CHUNK_STRATEGY", "fixed");
    ScopedEnvVar chunkTarget("YAMS_EMBED_CHUNK_TARGET", "16");
    ScopedEnvVar chunkMin("YAMS_EMBED_CHUNK_MIN", "8");
    ScopedEnvVar chunkMax("YAMS_EMBED_CHUNK_MAX", "32");
    ScopedEnvVar chunkOverlap("YAMS_EMBED_CHUNK_OVERLAP", "0");
    ScopedEnvVar preserveSent("YAMS_EMBED_CHUNK_PRESERVE_SENTENCES", "0");

    WorkCoordinator coordinator;
    coordinator.start(2);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    const std::string payload = []() {
        std::string out;
        out.reserve(2048);
        for (int i = 0; i < 256; ++i) {
            out += "0123456789abcdef";
        }
        return out;
    }();
    std::vector<metadata::DocumentInfo> docs;
    docs.reserve(8);
    for (int i = 0; i < 8; ++i) {
        metadata::DocumentInfo doc{};
        doc.id = 200 + i;
        doc.fileName = "doc.txt";
        doc.fileExtension = ".txt";
        doc.sha256Hash = "batch-" + std::to_string(i);
        doc.mimeType = "text/plain";
        doc.indexedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        metadataRepo->setDocument(doc);
        store->setContent(doc.sha256Hash, payload);
        docs.push_back(doc);
    }

    // Establish 2048 for this case before guarding it. Without this drain the name may already be
    // registered at production size, and get_or_create_channel would silently keep that larger ring
    // instead - so the guard would protect a capacity this case never actually got.
    drainEmbedJobsChannel();
    ScopedChannelRemoval embedChannelRemoval{"embed_jobs"};
    auto embedChannel =
        InternalEventBus::instance().get_or_create_channel<InternalEventBus::EmbedJob>("embed_jobs",
                                                                                       2048);
    InternalEventBus::EmbedJob pre;
    while (embedChannel->try_pop(pre)) {
    }

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, &coordinator, nullptr, 32);
    queue->start();

    // Wait for channel poller coroutine to start before enqueueing
    auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(queue->started());

    for (const auto& doc : docs) {
        PostIngestQueue::Task task{.hash = doc.sha256Hash, .mime = doc.mimeType};
        REQUIRE(queue->tryEnqueue(task));
    }

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
    while ((queue->processed() < docs.size() ||
            queue->metricsSnapshot().batches.embedDocsEmitted < docs.size()) &&
           std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    REQUIRE((queue->processed() == docs.size()));
    REQUIRE((queue->metricsSnapshot().batches.embedDocsEmitted == docs.size()));
    // Note: With parallel processing and caching, individual lookups may be used instead of batch.
    // The dispatch metric is the completion barrier for the subsequent embed channel inspection.

    std::size_t jobCount = 0;
    std::size_t docsWithPreparedChunks = 0;
    std::size_t docsWithMultipleChunks = 0;
    InternalEventBus::EmbedJob job;
    while (embedChannel->try_pop(job)) {
        jobCount++;

        for (const auto& doc : job.preparedDocs) {
            if (doc.chunks.empty()) {
                continue;
            }
            docsWithPreparedChunks++;
            if (doc.chunks.size() >= 2) {
                docsWithMultipleChunks++;
            }
            for (const auto& chunk : doc.chunks) {
                CHECK_FALSE(chunk.chunkId.empty());
                CHECK_FALSE(chunk.content.empty());
            }
        }
    }

    // PostIngestQueue should enqueue embedding jobs for successfully extracted documents.
    REQUIRE((jobCount > 0));

    // Phase 2: embed jobs should include prepared doc chunks for at least some docs.
    REQUIRE((docsWithPreparedChunks > 0));
    // With fixed chunking target=16 and a large payload, at least one doc should have >1 chunk.
    REQUIRE((docsWithMultipleChunks > 0));

    stopAndResetQueue(queue);
    coordinator.stop();
    coordinator.join();
}

TEST_CASE("PostIngestQueue: Parallel extraction preserves per-task identity",
          "[daemon][background][queue][batch][regression]") {
    BusToggleGuard busGuard(false);
    drainPostIngestChannel();
    PostIngestBatchGuard batchGuard(64);
    PostIngestConcurrencyGuard concurrencyGuard(64, 16);

    WorkCoordinator coordinator;
    coordinator.start(4);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};
    if (TuneAdvisor::postExtractionConcurrent() <= 1) {
        SKIP("Parallel extraction concurrency not available in this test runtime");
    }

    constexpr int64_t kDocBaseId = 5000;
    // InternalEventBus channels are process-wide and may have been created by
    // earlier sections with a smaller capacity than this queue's constructor
    // request. Keep the regression batch below the smallest test channel cap so
    // identity coverage is independent of singleton creation order.
    constexpr int kDocCount = 24;
    std::vector<metadata::DocumentInfo> docs;
    docs.reserve(kDocCount);
    for (int i = 0; i < kDocCount; ++i) {
        metadata::DocumentInfo doc{};
        doc.id = kDocBaseId + i;
        doc.fileName = "doc-" + std::to_string(i) + ".txt";
        doc.fileExtension = ".txt";
        doc.sha256Hash = "parallel-hash-" + std::to_string(i);
        doc.mimeType = "text/plain";
        doc.indexedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        metadataRepo->setDocument(doc);
        store->setContent(doc.sha256Hash, "payload-" + std::to_string(i));
        docs.push_back(doc);
    }

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, &coordinator, nullptr, 512);
    queue->start();

    auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(queue->started());

    for (const auto& doc : docs) {
        PostIngestQueue::Task task{.hash = doc.sha256Hash, .mime = doc.mimeType};
        REQUIRE(queue->tryEnqueue(task));
    }

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(8);
    while (queue->processed() < docs.size() && std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    REQUIRE((queue->processed() == docs.size()));
    REQUIRE((queue->failed() == 0));

    waitForCommittedState([&] { return metadataRepo->batchInsertedDocIds().size() >= docs.size(); },
                          std::chrono::seconds(8));
    auto insertedDocIds = metadataRepo->batchInsertedDocIds();
    REQUIRE((insertedDocIds.size() == docs.size()));

    std::sort(insertedDocIds.begin(), insertedDocIds.end());
    REQUIRE(
        (std::adjacent_find(insertedDocIds.begin(), insertedDocIds.end()) == insertedDocIds.end()));
    for (int i = 0; i < kDocCount; ++i) {
        REQUIRE((insertedDocIds[static_cast<std::size_t>(i)] == kDocBaseId + i));
    }

    stopAndResetQueue(queue);
    coordinator.stop();
    coordinator.join();
}

TEST_CASE("PostIngestQueue: enqueueBatch submits all tasks without loss",
          "[daemon][background][queue][batch][enqueue]") {
    BusToggleGuard busGuard(false);
    drainPostIngestChannel();
    PostIngestBatchGuard batchGuard(8);
    PostIngestConcurrencyGuard concurrencyGuard(32, 8);

    WorkCoordinator coordinator;
    coordinator.start(4);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    constexpr int64_t kDocBaseId = 7000;
    constexpr int kDocCount = 64;
    std::vector<PostIngestQueue::Task> tasks;
    tasks.reserve(kDocCount);
    for (int i = 0; i < kDocCount; ++i) {
        metadata::DocumentInfo doc{};
        doc.id = kDocBaseId + i;
        doc.fileName = "batch-doc-" + std::to_string(i) + ".txt";
        doc.fileExtension = ".txt";
        doc.sha256Hash = "batch-enqueue-hash-" + std::to_string(i);
        doc.mimeType = "text/plain";
        doc.indexedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        metadataRepo->setDocument(doc);
        store->setContent(doc.sha256Hash, "batch-payload-" + std::to_string(i));

        tasks.push_back(PostIngestQueue::Task{.hash = doc.sha256Hash, .mime = doc.mimeType});
    }

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, &coordinator, nullptr, 64);
    queue->setBatchCoalesceWindow(std::chrono::milliseconds(20));
    queue->start();
    auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(queue->started());

    bool highPriority = false;
    SECTION("normal bulk work uses the configured batch cap") {
        queue->enqueueBatch(std::move(tasks));
    }
    SECTION("priority work uses the interactive quota") {
        highPriority = true;
        queue->enqueueRpcBatch(std::move(tasks));
    }

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    while (queue->processed() < kDocCount && std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    REQUIRE((queue->processed() == kDocCount));
    REQUIRE((queue->failed() == 0));

    waitForCommittedState([&] {
        return metadataRepo->batchInsertedDocIds().size() >= static_cast<std::size_t>(kDocCount);
    });
    auto insertedDocIds = metadataRepo->batchInsertedDocIds();
    REQUIRE((insertedDocIds.size() == static_cast<std::size_t>(kDocCount)));
    std::sort(insertedDocIds.begin(), insertedDocIds.end());
    REQUIRE(
        (std::adjacent_find(insertedDocIds.begin(), insertedDocIds.end()) == insertedDocIds.end()));

    const auto metrics = queue->metricsSnapshot();
    const auto expectedBatches = highPriority ? 16u : 8u;
    const auto expectedMaxBatchSize = highPriority ? 4u : 8u;
    CHECK(metrics.batches.extractionBatches == expectedBatches);
    CHECK(metrics.batches.contentIndexCalls == expectedBatches);
    CHECK(metrics.batches.contentIndexMaxEntries == expectedMaxBatchSize);

    stopAndResetQueue(queue);
    coordinator.stop();
    coordinator.join();
}

// Characterizes the ordering that makes a bare wait on processed() racy: the counter advances
// once tasks are prepared, before commitBatchResults() writes their content.
TEST_CASE("PostIngestQueue: processed() advances before content is committed",
          "[daemon][background][queue][batch][ordering]") {
    BusToggleGuard busGuard(false);
    drainPostIngestChannel();
    PostIngestBatchGuard batchGuard(8);
    PostIngestConcurrencyGuard concurrencyGuard(32, 8);

    WorkCoordinator coordinator;
    coordinator.start(4);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<GatedInsertRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    constexpr int64_t kDocBaseId = 9000;
    constexpr int kDocCount = 16;
    std::vector<PostIngestQueue::Task> tasks;
    tasks.reserve(kDocCount);
    for (int i = 0; i < kDocCount; ++i) {
        metadata::DocumentInfo doc{};
        doc.id = kDocBaseId + i;
        doc.fileName = "gated-doc-" + std::to_string(i) + ".txt";
        doc.fileExtension = ".txt";
        doc.sha256Hash = "gated-commit-hash-" + std::to_string(i);
        doc.mimeType = "text/plain";
        doc.indexedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        metadataRepo->setDocument(doc);
        store->setContent(doc.sha256Hash, "gated-payload-" + std::to_string(i));
        tasks.push_back(PostIngestQueue::Task{.hash = doc.sha256Hash, .mime = doc.mimeType});
    }

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, &coordinator, nullptr, 64);
    InsertGateRelease releaseOnExit{metadataRepo};
    queue->setBatchCoalesceWindow(std::chrono::milliseconds(20));
    queue->start();
    auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(queue->started());

    queue->enqueueBatch(std::move(tasks));
    REQUIRE(metadataRepo->waitForInsertEntered(std::chrono::seconds(5)));

    // A commit is blocked inside the repository, yet the queue already reports prepared work.
    CHECK((queue->processed() > 0));
    CHECK((metadataRepo->batchInsertedDocIds().empty()));

    metadataRepo->release();
    waitForCommittedState([&] {
        return metadataRepo->batchInsertedDocIds().size() >= static_cast<std::size_t>(kDocCount);
    });
    REQUIRE((queue->processed() == kDocCount));
    REQUIRE((queue->failed() == 0));
    REQUIRE((metadataRepo->batchInsertedDocIds().size() == static_cast<std::size_t>(kDocCount)));

    stopAndResetQueue(queue);
    coordinator.stop();
    coordinator.join();
}

TEST_CASE("PostIngestQueue: keeps multi-doc batches when extraction concurrency is low",
          "[daemon][background][queue][batch][throughput]") {
    BusToggleGuard busGuard(false);
    drainPostIngestChannel();
    PostIngestBatchGuard batchGuard(16);
    PostIngestConcurrencyGuard concurrencyGuard(32, 1);

    WorkCoordinator coordinator;
    coordinator.start(2);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    constexpr int64_t kDocBaseId = 9000;
    constexpr int kDocCount = 48;
    std::vector<PostIngestQueue::Task> tasks;
    tasks.reserve(kDocCount);
    for (int i = 0; i < kDocCount; ++i) {
        metadata::DocumentInfo doc{};
        doc.id = kDocBaseId + i;
        doc.fileName = "low-concurrency-batch-doc-" + std::to_string(i) + ".txt";
        doc.fileExtension = ".txt";
        doc.sha256Hash = "low-concurrency-batch-hash-" + std::to_string(i);
        doc.mimeType = "text/plain";
        doc.indexedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        metadataRepo->setDocument(doc);
        store->setContent(doc.sha256Hash, "payload-" + std::to_string(i));
        tasks.push_back(PostIngestQueue::Task{.hash = doc.sha256Hash, .mime = doc.mimeType});
    }

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, &coordinator, nullptr, 64);
    queue->setBatchCoalesceWindow(std::chrono::milliseconds(20));
    queue->start();
    auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
    while (!queue->started() && std::chrono::steady_clock::now() < startDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(queue->started());

    queue->enqueueBatch(std::move(tasks));

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    while (queue->processed() < kDocCount && std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }

    REQUIRE((queue->processed() == kDocCount));
    REQUIRE((queue->failed() == 0));
    waitForCommittedState([&] {
        return metadataRepo->batchInsertedDocIds().size() >= static_cast<std::size_t>(kDocCount);
    });
    REQUIRE((metadataRepo->maxBatchWriteSize() > 1));

    const auto metrics = queue->metricsSnapshot();
    CHECK(metrics.batches.extractionTasks == kDocCount);
    CHECK(metrics.batches.extractionBatches == 3);
    CHECK(metrics.batches.contentIndexCalls == 3);
    CHECK(metrics.batches.contentIndexEntries == kDocCount);
    CHECK(metrics.batches.contentIndexMaxEntries == 16);

    // This case's result depends on the ring it declared for post_ingest, not on whatever capacity
    // an earlier case left registered for that process-wide singleton channel. A recorded
    // disagreement means the declared capacity was silently ignored - which is exactly what let a
    // 31-slot ring serve a case arranged for 64 and produce 4 batches instead of 3 on a loaded
    // runner. Guards the drain-then-remove behaviour in drainPostIngestChannel().
    const auto capacityMismatches = InternalEventBus::instance().capacity_mismatches();
    CHECK(std::none_of(
        capacityMismatches.begin(), capacityMismatches.end(),
        [](const InternalEventBus::CapacityMismatch& m) { return m.name == "post_ingest"; }));

    stopAndResetQueue(queue);
    coordinator.stop();
    coordinator.join();
}

TEST_CASE("PostIngestQueue: full-channel enqueueBatch waits only log at debug",
          "[daemon][background][queue][logging]") {
    drainPostIngestChannel();
    SpdlogCaptureGuard logs(spdlog::level::info);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    auto queue = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                   nullptr, nullptr, nullptr, 1);

    std::vector<PostIngestQueue::Task> tasks;
    tasks.push_back(PostIngestQueue::Task{.hash = "log-test-hash-1", .mime = "text/plain"});
    tasks.push_back(PostIngestQueue::Task{.hash = "log-test-hash-2", .mime = "text/plain"});

    std::thread producer([&queue, tasks]() mutable { queue->enqueueBatch(tasks); });

    std::this_thread::sleep_for(std::chrono::milliseconds(350));
    queue->stop();
    producer.join();

    CHECK((logs.str().find("enqueueBatch waiting on full channel") == std::string::npos));
}

// =============================================================================
// InternalEventBus Integration Tests
// =============================================================================

TEST_CASE("PostIngestQueue: InternalEventBus integration and stress",
          "[daemon][background][bus][stress][slow]") {
    if (!isMpmcEnabled()) {
        SKIP("MPMC bus not enabled (set YAMS_INTERNAL_BUS_MPMC=1)");
    }

    BusToggleGuard busGuard(true); // Enable bus routing
    drainPostIngestChannel();

    // WorkCoordinator is required for PostIngestQueue strand creation
    WorkCoordinator coordinator;
    coordinator.start(4);

    auto store = std::make_shared<StubContentStore>();
    auto metadataRepo = std::make_shared<StubMetadataRepository>();
    auto extractor = std::make_shared<StubExtractor>();
    std::vector<std::shared_ptr<extraction::IContentExtractor>> extractors{extractor};

    SECTION("Bus stress: 4 producers × 1000 tasks") {
        constexpr int64_t kDocBaseId = 1000;
        constexpr int kProducers = 4;
        constexpr int kPerProducer = 1000;
        const std::string payload = "hello world";

        auto pq = std::make_unique<PostIngestQueue>(store, metadataRepo, extractors, nullptr,
                                                    nullptr, &coordinator, nullptr, 4096);
        pq->start(); // Start the channel poller

        // Spawn producers that publish into the InternalEventBus channel
        std::vector<std::thread> producers;
        producers.reserve(kProducers);
        for (int p = 0; p < kProducers; ++p) {
            producers.emplace_back([&, p]() {
                auto chan = InternalEventBus::instance()
                                .get_or_create_channel<InternalEventBus::PostIngestTask>(
                                    "post_ingest", 4096);
                for (int i = 0; i < kPerProducer; ++i) {
                    // Unique hash per item
                    std::string hash = "h-" + std::to_string(p) + "-" + std::to_string(i);
                    metadata::DocumentInfo doc{};
                    doc.id = kDocBaseId + (p * kPerProducer + i);
                    doc.fileName = "doc.txt";
                    doc.fileExtension = ".txt";
                    doc.sha256Hash = hash;
                    doc.mimeType = "text/plain";
                    metadataRepo->setDocument(doc);
                    store->setContent(hash, payload);

                    InternalEventBus::PostIngestTask t{.hash = hash, .mime = doc.mimeType};
                    while (!chan->try_push(t)) {
                        std::this_thread::yield();
                    }
                    InternalEventBus::instance().incPostQueued();
                }
            });
        }

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

        const int expected = kProducers * kPerProducer;
        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
        while (static_cast<int>(pq->processed()) < expected &&
               std::chrono::steady_clock::now() < deadline) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }

        REQUIRE((static_cast<int>(pq->processed()) == expected));
        REQUIRE((pq->failed() == 0));
        REQUIRE((pq->entityInFlight() == 0));

        stopAndResetQueue(pq);
    }

    // Cleanup coordinator at test end
    coordinator.stop();
    coordinator.join();
}

// =============================================================================
// MPMC Queue Concurrency Tests
// =============================================================================

TEST_CASE("InternalEventBus: MPMC queue correctness under concurrent load",
          "[daemon][background][concurrency][mpmc]") {
    if (!isMpmcEnabled()) {
        SKIP("MPMC not enabled (set YAMS_INTERNAL_BUS_MPMC=1)");
    }

    SECTION("4 producers × 4 consumers × 5000 items") {
        constexpr int kProducers = 4;
        constexpr int kConsumers = 4;
        constexpr int kPerProducer = 5000;

        SpscQueue<int> q(1024);
        std::atomic<int> produced{0};
        std::atomic<int> consumed{0};
        std::atomic<bool> done{false};

        std::vector<std::thread> producers;
        producers.reserve(kProducers);
        for (int p = 0; p < kProducers; ++p) {
            producers.emplace_back([&]() {
                for (int i = 0; i < kPerProducer; ++i) {
                    while (!q.try_push(1)) {
                        std::this_thread::yield();
                    }
                    produced.fetch_add(1, std::memory_order_relaxed);
                }
            });
        }

        std::vector<std::thread> consumers;
        consumers.reserve(kConsumers);
        for (int c = 0; c < kConsumers; ++c) {
            consumers.emplace_back([&]() {
                int v;
                while (!done.load(std::memory_order_acquire)) {
                    if (q.try_pop(v)) {
                        consumed.fetch_add(v, std::memory_order_relaxed);
                    } else {
                        std::this_thread::yield();
                    }
                }
                // Drain residual items
                while (q.try_pop(v)) {
                    consumed.fetch_add(v, std::memory_order_relaxed);
                }
            });
        }

        for (auto& t : producers) {
            t.join();
        }
        done.store(true, std::memory_order_release);
        for (auto& t : consumers) {
            t.join();
        }

        REQUIRE((produced.load() == kProducers * kPerProducer));
        REQUIRE((consumed.load() == kProducers * kPerProducer));
    }
}

// =============================================================================
// InternalEventBus capacity-mismatch diagnostic
// =============================================================================

TEST_CASE("InternalEventBus records capacity disagreements on channel reuse",
          "[daemon][background][bus][mismatch]") {
    auto& bus = InternalEventBus::instance();
    // A name no production code and no other case in this binary uses.
    constexpr const char* kProbe = "capacity_mismatch_probe";

    bus.remove_channel(kProbe);
    const std::size_t before = bus.capacity_mismatches().size();

    // Creating a channel is not a disagreement.
    auto first = bus.get_or_create_channel<InternalEventBus::Fts5Job>(kProbe, 8);
    REQUIRE((first != nullptr));
    CHECK(first->capacity() == 8);
    CHECK(bus.capacity_mismatches().size() == before);

    // Neither is reusing it at the capacity it already has.
    auto sameAgain = bus.get_or_create_channel<InternalEventBus::Fts5Job>(kProbe, 8);
    CHECK(sameAgain == first);
    CHECK(bus.capacity_mismatches().size() == before);

    // Reuse at a different capacity is recorded, and the registered ring still wins.
    auto reused = bus.get_or_create_channel<InternalEventBus::Fts5Job>(kProbe, 64);
    CHECK(reused == first);
    CHECK(reused->capacity() == 8);

    const auto recorded = bus.capacity_mismatches();
    REQUIRE(recorded.size() == before + 1);
    CHECK(recorded.back().name == kProbe);
    CHECK(recorded.back().actual == 8);
    CHECK(recorded.back().requested == 64);

    // Repeating the same disagreement is not recorded twice.
    (void)bus.get_or_create_channel<InternalEventBus::Fts5Job>(kProbe, 64);
    CHECK(bus.capacity_mismatches().size() == before + 1);

    // A second, different disagreement for the same name is still recorded, so a name is not
    // permanently suppressed by its first entry.
    (void)bus.get_or_create_channel<InternalEventBus::Fts5Job>(kProbe, 128);
    CHECK(bus.capacity_mismatches().size() == before + 2);

    // Dropping the channel drops its records: the list describes live disagreements, so a later
    // assertion cannot depend on which cases ran first.
    bus.remove_channel(kProbe);
    CHECK(bus.capacity_mismatches().size() == before);
}