// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright 2025 YAMS Contributors

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <filesystem>
#include <future>
#include <iterator>
#include <memory>
#include <optional>
#include <sstream>
#include <thread>
#include <variant>
#include <vector>

#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>
#include <catch2/matchers/catch_matchers_floating_point.hpp>

#include <spdlog/sinks/ostream_sink.h>
#include <boost/asio/io_context.hpp>
#include <yams/common/utf8_utils.h>
#include <yams/daemon/components/MetadataWriteFacade.h>
#include <yams/daemon/components/TuneAdvisor.h>
#include <yams/daemon/components/WriteCoordinator.h>
#include <yams/metadata/connection_pool.h>

#include "../../common/metadata_test_db.h"
#include <yams/metadata/content_index_writer.h>
#include <yams/metadata/database.h>
#include <yams/metadata/metadata_insert_writer.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/metadata/migration.h>
#include <yams/metadata/path_utils.h>
#include <yams/storage/corpus_stats.h>

using namespace yams;
using namespace yams::metadata;
using namespace yams::search;

// Catch2's expression-decomposition macros intentionally overload operator<=,
// which triggers false positives from chained-comparison linters on ordinary
// CHECK/REQUIRE assertions in this file.
// NOLINTBEGIN(bugprone-chained-comparison)
namespace {

std::filesystem::path tempDbPath(const char* prefix) {
    static const bool kSuppressInfoLogs = [] {
        spdlog::set_level(spdlog::level::warn);
        return true;
    }();
    (void)kSuppressInfoLogs;
    return yams::test::migrated_metadata_db_template().clone(prefix);
}

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>("metadata_repository_capture", sink);
        logger_->set_level(level);
        logger_->set_pattern("[%l] %v");
        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_;
};

DocumentInfo makeDocumentWithPath(const std::string& path, const std::string& hash,
                                  const std::string& mime = "text/plain") {
    DocumentInfo info;
    info.filePath = path;
    info.fileName = std::filesystem::path(path).filename().string();
    info.fileExtension = std::filesystem::path(path).extension().string();
    info.fileSize = 1234;
    info.sha256Hash = hash;
    info.mimeType = mime;
    info.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    info.modifiedTime = info.createdTime;
    info.indexedTime = info.createdTime;
    populatePathDerivedFields(info);
    info.contentExtracted = true;
    info.extractionStatus = ExtractionStatus::Success;
    return info;
}

BatchContentEntry makeBatchContentEntry(int64_t documentId, const std::string& title,
                                        const std::string& contentText,
                                        const std::string& mimeType = "text/plain",
                                        const std::string& extractionMethod = "test",
                                        const std::string& language = "en") {
    BatchContentEntry entry;
    entry.documentId = documentId;
    entry.title = title;
    entry.contentText = contentText;
    entry.mimeType = mimeType;
    entry.extractionMethod = extractionMethod;
    entry.language = language;
    return entry;
}

std::unique_ptr<yams::daemon::WriteBatch> makeSymSpellBatch(const std::string& source,
                                                            const std::string& indexedTerm) {
    auto batch = std::make_unique<yams::daemon::WriteBatch>();
    batch->source = source;
    batch->ops.emplace_back(yams::daemon::AddSymSpellTermsOp{{indexedTerm}});
    return batch;
}

BatchDocumentInsert makeIndexedBatchInsert(int index) {
    BatchDocumentInsert item;
    item.info = makeDocumentWithPath("repo/writer/doc-" + std::to_string(index) + ".txt",
                                     "writer-hash-" + std::to_string(index));
    item.tags.emplace_back("idx", MetadataValue(std::to_string(index)));
    return item;
}

TEST_CASE("populatePathDerivedFields fills DocumentInfo path index fields", "[metadata][path]") {
    DocumentInfo info;
    info.filePath = "/tmp/../tmp/yams/path-note.md";

    const auto expected = computePathDerivedValues(info.filePath);
    populatePathDerivedFields(info);

    CHECK((info.filePath == expected.normalizedPath));
    CHECK((info.pathPrefix == expected.pathPrefix));
    CHECK((info.reversePath == expected.reversePath));
    CHECK((info.pathHash == expected.pathHash));
    CHECK((info.parentHash == expected.parentHash));
    CHECK((info.pathDepth == expected.pathDepth));
}

struct MetadataRepositoryFixture {
    explicit MetadataRepositoryFixture(std::size_t maxConnections = 2) {
        dbPath_ = tempDbPath("metadata_repo_catch2_test_");

        ConnectionPoolConfig config;
        config.minConnections = 1;
        config.maxConnections = maxConnections;

        pool_ = std::make_unique<ConnectionPool>(dbPath_.string(), config);
        auto initResult = pool_->initialize();
        REQUIRE((initResult.has_value()));

        repository_ = std::make_unique<MetadataRepository>(
            *pool_, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);
    }

    ~MetadataRepositoryFixture() {
        repository_.reset();
        pool_->shutdown();
        pool_.reset();
        yams::test::remove_sqlite_artifacts(dbPath_);
    }

    std::filesystem::path dbPath_;
    std::unique_ptr<ConnectionPool> pool_;
    std::unique_ptr<MetadataRepository> repository_;
};

struct EmbeddingStatusRow {
    bool hasEmbedding = false;
    std::optional<std::string> modelId;
    int64_t chunkCount = 0;
};

Result<std::optional<EmbeddingStatusRow>> getEmbeddingStatusRow(ConnectionPool& pool,
                                                                const std::string& hash) {
    auto connResult = pool.acquire();
    if (!connResult) {
        return connResult.error();
    }

    auto& db = **connResult.value();
    auto stmtResult = db.prepare(R"(
        SELECT COALESCE(des.has_embedding, 0), des.model_id, COALESCE(des.chunk_count, 0)
        FROM documents d
        LEFT JOIN document_embeddings_status des ON d.id = des.document_id
        WHERE d.sha256_hash = ?
    )");
    if (!stmtResult) {
        return stmtResult.error();
    }

    auto& stmt = stmtResult.value();
    if (auto bindResult = stmt.bind(1, hash); !bindResult) {
        return bindResult.error();
    }

    auto stepResult = stmt.step();
    if (!stepResult) {
        return stepResult.error();
    }
    if (!stepResult.value()) {
        return std::optional<EmbeddingStatusRow>{};
    }

    EmbeddingStatusRow row;
    row.hasEmbedding = stmt.getInt(0) != 0;
    if (!stmt.isNull(1)) {
        row.modelId = stmt.getString(1);
    }
    row.chunkCount = stmt.getInt64(2);
    return std::optional<EmbeddingStatusRow>{row};
}

Result<void> setEmbeddingChunkCount(ConnectionPool& pool, const std::string& hash,
                                    int64_t chunkCount) {
    auto connResult = pool.acquire();
    if (!connResult) {
        return connResult.error();
    }

    auto& db = **connResult.value();
    auto stmtResult = db.prepare(R"(
        UPDATE document_embeddings_status
        SET chunk_count = ?
        WHERE document_id = (SELECT id FROM documents WHERE sha256_hash = ?)
    )");
    if (!stmtResult) {
        return stmtResult.error();
    }

    auto& stmt = stmtResult.value();
    if (auto bindCount = stmt.bind(1, chunkCount); !bindCount) {
        return bindCount.error();
    }
    if (auto bindHash = stmt.bind(2, hash); !bindHash) {
        return bindHash.error();
    }
    return stmt.execute();
}

Result<void> overwritePathTreeCentroid(ConnectionPool& pool, std::string_view fullPath,
                                       std::span<const std::byte> blob, int64_t centroidWeight) {
    auto connResult = pool.acquire();
    if (!connResult) {
        return connResult.error();
    }

    auto& db = **connResult.value();
    auto stmtResult = db.prepare(R"(
        UPDATE path_tree_nodes
        SET centroid = ?, centroid_weight = ?
        WHERE full_path = ?
    )");
    if (!stmtResult) {
        return stmtResult.error();
    }

    auto& stmt = stmtResult.value();
    if (auto bindBlob = stmt.bind(1, blob); !bindBlob) {
        return bindBlob.error();
    }
    if (auto bindWeight = stmt.bind(2, centroidWeight); !bindWeight) {
        return bindWeight.error();
    }
    if (auto bindPath = stmt.bind(3, fullPath); !bindPath) {
        return bindPath.error();
    }
    return stmt.execute();
}

} // namespace

TEST_CASE("MetadataRepository: path-derived fields support exact path lookup",
          "[unit][metadata][repository][path]") {
    MetadataRepositoryFixture fix;

    auto docInfo = makeDocumentWithPath("/tmp/../tmp/yams/path-index.md", "path-index-hash");
    REQUIRE((!docInfo.pathHash.empty()));
    REQUIRE((!docInfo.reversePath.empty()));
    REQUIRE((docInfo.pathDepth > 0));

    auto insert = fix.repository_->insertDocument(docInfo);
    REQUIRE((insert.has_value()));

    auto found = fix.repository_->findDocumentByExactPath(docInfo.filePath);
    REQUIRE((found.has_value()));
    REQUIRE((found.value().has_value()));
    CHECK((found.value()->pathHash == docInfo.pathHash));
    CHECK((found.value()->pathPrefix == docInfo.pathPrefix));
    CHECK((found.value()->parentHash == docInfo.parentHash));
    CHECK((found.value()->pathDepth == docInfo.pathDepth));
}

TEST_CASE("MetadataRepository: insert and get document", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "hash123";
    docInfo.fileName = "test.txt";
    docInfo.fileSize = 1024;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto result = fix.repository_->insertDocument(docInfo);
    REQUIRE((result.has_value()));

    auto docId = result.value();
    CHECK((docId > 0));

    auto getResult = fix.repository_->getDocument(docId);
    REQUIRE((getResult.has_value()));
    REQUIRE((getResult.value().has_value()));

    auto retrievedDoc = getResult.value().value();
    CHECK((retrievedDoc.id == docId));
    CHECK((retrievedDoc.sha256Hash == docInfo.sha256Hash));
    CHECK((retrievedDoc.fileName == docInfo.fileName));
    CHECK((retrievedDoc.fileSize == docInfo.fileSize));
    CHECK((retrievedDoc.mimeType == docInfo.mimeType));
}

TEST_CASE("MetadataRepository: dual-pool read/write routing behavior",
          "[unit][metadata][repository][dual-pool]") {
    auto dbPath = tempDbPath("metadata_repo_dual_pool_");

    ConnectionPoolConfig writeCfg;
    writeCfg.minConnections = 1;
    writeCfg.maxConnections = 2;

    ConnectionPoolConfig readCfg;
    readCfg.minConnections = 1;
    readCfg.maxConnections = 2;

    auto writePool = std::make_unique<ConnectionPool>(dbPath.string(), writeCfg);
    auto readPool = std::make_unique<ConnectionPool>(dbPath.string(), readCfg);

    REQUIRE((writePool->initialize().has_value()));
    REQUIRE((readPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *writePool, readPool.get(), MetadataRepository::SchemaBootstrapMode::AssumeReady);

    auto doc = makeDocumentWithPath("/tmp/dual-pool.txt", "dual-pool-hash");
    auto insertResult = repository->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    auto readOk = repository->getDocumentByHash("dual-pool-hash");
    REQUIRE((readOk.has_value()));
    REQUIRE((readOk.value().has_value()));

    readPool->shutdown();

    SpdlogCaptureGuard logCapture(spdlog::level::err);
    auto readAfterShutdown = repository->getDocumentByHash("dual-pool-hash");
    REQUIRE_FALSE(readAfterShutdown.has_value());
    CHECK(((readAfterShutdown.error().code == ErrorCode::OperationCancelled ||
            readAfterShutdown.error().code == ErrorCode::InvalidState)));
    CHECK((readAfterShutdown.error().message.find("shut down") != std::string::npos ||
           readAfterShutdown.error().message.find("shutdown") != std::string::npos));
    CHECK((logCapture.str().find("MetadataRepository::executeQueryOnPool") == std::string::npos));

    MetadataValue value;
    value.value = "still-writeable";
    value.type = MetadataValueType::String;
    auto writeAfterReadPoolDown = repository->setMetadata(docId, "dual_pool_test", value);
    REQUIRE((writeAfterReadPoolDown.has_value()));

    repository.reset();
    writePool->shutdown();
    writePool.reset();
    readPool.reset();

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

TEST_CASE("MetadataRepository: dual-pool session count read route isolation",
          "[unit][metadata][repository][dual-pool]") {
    auto dbPath = tempDbPath("metadata_repo_dual_pool_session_count_");

    ConnectionPoolConfig writeCfg;
    writeCfg.minConnections = 1;
    writeCfg.maxConnections = 2;

    ConnectionPoolConfig readCfg;
    readCfg.minConnections = 1;
    readCfg.maxConnections = 2;

    auto writePool = std::make_unique<ConnectionPool>(dbPath.string(), writeCfg);
    auto readPool = std::make_unique<ConnectionPool>(dbPath.string(), readCfg);

    REQUIRE((writePool->initialize().has_value()));
    REQUIRE((readPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *writePool, readPool.get(), MetadataRepository::SchemaBootstrapMode::AssumeReady);

    auto doc = makeDocumentWithPath("/tmp/dual-pool-session.txt", "dual-pool-session-hash");
    auto insertResult = repository->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    MetadataValue sessionValue;
    sessionValue.value = "sess-123";
    sessionValue.type = MetadataValueType::String;
    REQUIRE((repository->setMetadata(docId, "session_id", sessionValue).has_value()));

    auto countOk = repository->countDocumentsBySessionId("sess-123");
    REQUIRE((countOk.has_value()));
    REQUIRE((countOk.value() == 1));

    readPool->shutdown();

    auto countAfterShutdown = repository->countDocumentsBySessionId("sess-123");
    REQUIRE_FALSE(countAfterShutdown.has_value());

    MetadataValue writeValue;
    writeValue.value = "still-writeable-after-read-pool-down";
    writeValue.type = MetadataValueType::String;
    REQUIRE((repository->setMetadata(docId, "dual_pool_write_check", writeValue).has_value()));

    repository.reset();
    writePool->shutdown();
    writePool.reset();
    readPool.reset();

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

TEST_CASE("MetadataRepository: snapshot metadata helpers round-trip",
          "[unit][metadata][repository][snapshot-metadata]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("repo/releases/a.txt", "snapshot-meta-a");
    auto docB = makeDocumentWithPath("repo/releases/b.txt", "snapshot-meta-b");
    auto docC = makeDocumentWithPath("repo/archive/c.txt", "snapshot-meta-c");

    auto aId = fix.repository_->insertDocument(docA);
    auto bId = fix.repository_->insertDocument(docB);
    auto cId = fix.repository_->insertDocument(docC);
    REQUIRE((aId.has_value()));
    REQUIRE((bId.has_value()));
    REQUIRE((cId.has_value()));

    auto setStringMeta = [&](int64_t docId, const std::string& key, const std::string& value) {
        MetadataValue mv;
        mv.type = MetadataValueType::String;
        mv.value = value;
        REQUIRE((fix.repository_->setMetadata(docId, key, mv).has_value()));
    };

    setStringMeta(aId.value(), "snapshot_id", "snap-1");
    setStringMeta(aId.value(), "snapshot_label", "release-a");
    setStringMeta(aId.value(), "git_commit", "abc123");
    setStringMeta(bId.value(), "snapshot_id", "snap-1");
    setStringMeta(bId.value(), "snapshot_label", "release-a");
    setStringMeta(bId.value(), "git_commit", "abc123");
    setStringMeta(cId.value(), "snapshot_id", "snap-2");
    setStringMeta(cId.value(), "snapshot_label", "release-b");
    setStringMeta(cId.value(), "git_commit", "def456");

    auto bySnapshot = fix.repository_->findDocumentsBySnapshot("snap-1");
    REQUIRE((bySnapshot.has_value()));
    REQUIRE((bySnapshot.value().size() == 2));
    std::vector<std::string> snapshotPaths;
    for (const auto& doc : bySnapshot.value()) {
        snapshotPaths.push_back(doc.filePath);
    }
    std::sort(snapshotPaths.begin(), snapshotPaths.end());
    auto expectedSnapshotPaths = std::vector<std::string>{docA.filePath, docB.filePath};
    std::sort(expectedSnapshotPaths.begin(), expectedSnapshotPaths.end());
    CHECK((snapshotPaths == expectedSnapshotPaths));

    auto byLabel = fix.repository_->findDocumentsBySnapshotLabel("release-a");
    REQUIRE((byLabel.has_value()));
    REQUIRE((byLabel.value().size() == 2));

    auto snapshots = fix.repository_->getSnapshots();
    REQUIRE((snapshots.has_value()));
    CHECK((snapshots.value() == std::vector<std::string>{"snap-1", "snap-2"}));
    auto snapshotsCached = fix.repository_->getSnapshots();
    REQUIRE((snapshotsCached.has_value()));
    CHECK((snapshotsCached.value() == snapshots.value()));

    auto labels = fix.repository_->getSnapshotLabels();
    REQUIRE((labels.has_value()));
    CHECK((labels.value() == std::vector<std::string>{"release-a", "release-b"}));
    auto labelsCached = fix.repository_->getSnapshotLabels();
    REQUIRE((labelsCached.has_value()));
    CHECK((labelsCached.value() == labels.value()));

    auto snap1Info = fix.repository_->getSnapshotInfo("snap-1");
    REQUIRE((snap1Info.has_value()));
    CHECK((snap1Info.value().fileCount == 2));
    CHECK((snap1Info.value().label == "release-a"));
    CHECK((snap1Info.value().gitCommit == "abc123"));
    CHECK((snap1Info.value().createdTime > 0));

    auto batchInfo = fix.repository_->batchGetSnapshotInfo({"snap-1", "snap-2"});
    REQUIRE((batchInfo.has_value()));
    REQUIRE((batchInfo.value().size() == 2));
    CHECK((batchInfo.value().at("snap-1").fileCount == 2));
    CHECK((batchInfo.value().at("snap-1").label == "release-a"));
    CHECK((batchInfo.value().at("snap-2").fileCount == 1));
    CHECK((batchInfo.value().at("snap-2").gitCommit == "def456"));
}

TEST_CASE("MetadataRepository: content and KG intent commit atomically",
          "[unit][metadata][repository][kg-intent]") {
    MetadataRepositoryFixture fix;
    auto first = fix.repository_->insertDocument(makeDocumentWithPath("kg/a.txt", "kg-intent-a"));
    auto second = fix.repository_->insertDocument(makeDocumentWithPath("kg/b.txt", "kg-intent-b"));
    REQUIRE(first.has_value());
    REQUIRE(second.has_value());
    BatchContentEntry a;
    a.documentId = first.value();
    a.contentText = "first KG content";
    a.knowledgeGraphToken = "generation-a";
    BatchContentEntry b;
    b.documentId = second.value();
    b.contentText = "second KG content";
    b.knowledgeGraphToken = "generation-b";
    SECTION("successful extraction retains independently pending KG intent") {
        REQUIRE(fix.repository_->batchInsertContentAndIndex({a, b}).has_value());
        auto marker = fix.repository_->getMetadata(first.value(), "yams:kg_enrichment");
        REQUIRE(marker.has_value());
        REQUIRE(marker.value().has_value());
        CHECK(marker.value()->value == "pending:generation-a");
    }
    SECTION("completion acknowledges only the current pending token") {
        REQUIRE(fix.repository_->batchInsertContentAndIndex({a}).has_value());
        a.knowledgeGraphToken = "generation-new";
        REQUIRE(fix.repository_->batchInsertContentAndIndex({a}).has_value());
        auto stale =
            fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "generation-a");
        REQUIRE(stale.has_value());
        CHECK_FALSE(stale.value());
        auto empty = fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "");
        REQUIRE(empty.has_value());
        CHECK_FALSE(empty.value());
        REQUIRE(fix.pool_
                    ->withConnection([](Database& db) -> Result<void> {
                        return db.execute(
                            "CREATE TRIGGER reject_kg_ack BEFORE UPDATE ON metadata "
                            "WHEN NEW.key='yams:kg_enrichment' AND NEW.value LIKE 'complete:%' "
                            "BEGIN SELECT RAISE(ABORT, 'injected KG ack failure'); END");
                    })
                    .has_value());
        CHECK_FALSE(
            fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "generation-new")
                .has_value());
        auto pending = fix.repository_->getMetadata(first.value(), "yams:kg_enrichment");
        REQUIRE(pending.has_value());
        REQUIRE(pending.value().has_value());
        CHECK(pending.value()->value == "pending:generation-new");
        REQUIRE(fix.pool_
                    ->withConnection(
                        [](Database& db) { return db.execute("DROP TRIGGER reject_kg_ack"); })
                    .has_value());
        auto completed =
            fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "generation-new");
        REQUIRE(completed.has_value());
        CHECK(completed.value());
        auto duplicate =
            fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "generation-new");
        REQUIRE(duplicate.has_value());
        CHECK_FALSE(duplicate.value());
        auto marker = fix.repository_->getMetadata(first.value(), "yams:kg_enrichment");
        REQUIRE(marker.has_value());
        REQUIRE(marker.value().has_value());
        CHECK(marker.value()->value == "complete:generation-new");
    }
    SECTION("KG completion invalidates cached metadata value counts") {
        REQUIRE(fix.repository_->batchInsertContentAndIndex({a}).has_value());
        auto before = fix.repository_->getMetadataValueCounts({"yams:kg_enrichment"}, {});
        REQUIRE(before.has_value());
        REQUIRE(before.value().at("yams:kg_enrichment").size() == 1);
        CHECK(before.value().at("yams:kg_enrichment").front().value == "pending:generation-a");

        auto completed =
            fix.repository_->completeKnowledgeGraphEnrichment(first.value(), "generation-a");
        REQUIRE(completed.has_value());
        REQUIRE(completed.value());

        auto after = fix.repository_->getMetadataValueCounts({"yams:kg_enrichment"}, {});
        REQUIRE(after.has_value());
        REQUIRE(after.value().at("yams:kg_enrichment").size() == 1);
        CHECK(after.value().at("yams:kg_enrichment").front().value == "complete:generation-a");
    }
    SECTION("failed intent write rolls back the entire content batch") {
        REQUIRE(fix.pool_
                    ->withConnection([&](Database& db) -> Result<void> {
                        return db.execute(
                            "CREATE TRIGGER reject_kg_intent BEFORE INSERT ON metadata "
                            "WHEN NEW.key = 'yams:kg_enrichment' AND NEW.document_id = " +
                            std::to_string(second.value()) +
                            " BEGIN SELECT RAISE(ABORT, 'injected KG intent failure'); END");
                    })
                    .has_value());
        auto written = fix.repository_->batchInsertContentAndIndex({a, b});
        CHECK_FALSE(written.has_value());
        for (auto id : {first.value(), second.value()}) {
            auto content = fix.repository_->getContent(id);
            REQUIRE(content.has_value());
            CHECK_FALSE(content.value().has_value());
            auto marker = fix.repository_->getMetadata(id, "yams:kg_enrichment");
            REQUIRE(marker.has_value());
            CHECK_FALSE(marker.value().has_value());
        }
    }
}

TEST_CASE("MetadataRepository: session and tag helpers round-trip",
          "[unit][metadata][repository][session-tags]") {
    // SQLITE_LIMIT_SQL_LENGTH is connection-local; pin fault injection and queries together.
    MetadataRepositoryFixture fix{1};

    auto docA = makeDocumentWithPath("repo/sessions/a.txt", "session-tag-a");
    auto docB = makeDocumentWithPath("repo/sessions/b.txt", "session-tag-b");
    auto docC = makeDocumentWithPath("repo/archive/c.txt", "session-tag-c");

    auto aId = fix.repository_->insertDocument(docA);
    auto bId = fix.repository_->insertDocument(docB);
    auto cId = fix.repository_->insertDocument(docC);
    REQUIRE((aId.has_value()));
    REQUIRE((bId.has_value()));
    REQUIRE((cId.has_value()));

    auto setStringMeta = [&](int64_t docId, const std::string& key, const std::string& value) {
        MetadataValue mv;
        mv.type = MetadataValueType::String;
        mv.value = value;
        REQUIRE((fix.repository_->setMetadata(docId, key, mv).has_value()));
    };

    setStringMeta(aId.value(), "session_id", "sess-a");
    setStringMeta(bId.value(), "session_id", "sess-a");
    setStringMeta(cId.value(), "session_id", "sess-b");

    auto sessionDocs = fix.repository_->findDocumentsBySessionId("sess-a");
    REQUIRE((sessionDocs.has_value()));
    REQUIRE((sessionDocs.value().size() == 2));
    std::vector<std::string> sessionPaths;
    for (const auto& doc : sessionDocs.value()) {
        sessionPaths.push_back(doc.filePath);
    }
    std::sort(sessionPaths.begin(), sessionPaths.end());
    auto expectedSessionPaths = std::vector<std::string>{docA.filePath, docB.filePath};
    std::sort(expectedSessionPaths.begin(), expectedSessionPaths.end());
    CHECK((sessionPaths == expectedSessionPaths));

    auto sessionCount = fix.repository_->countDocumentsBySessionId("sess-a");
    REQUIRE((sessionCount.has_value()));
    CHECK((sessionCount.value() == 2));

    REQUIRE((fix.repository_->removeSessionIdFromDocuments("sess-a").has_value()));
    auto sessionCountAfterRemove = fix.repository_->countDocumentsBySessionId("sess-a");
    REQUIRE((sessionCountAfterRemove.has_value()));
    CHECK((sessionCountAfterRemove.value() == 0));
    auto preservedDoc = fix.repository_->getDocument(aId.value());
    REQUIRE((preservedDoc.has_value()));
    REQUIRE((preservedDoc.value().has_value()));

    setStringMeta(aId.value(), "session_id", "sess-a");
    setStringMeta(bId.value(), "session_id", "sess-a");
    auto deletedCount = fix.repository_->deleteDocumentsBySessionId("sess-a");
    REQUIRE((deletedCount.has_value()));
    CHECK((deletedCount.value() == 2));
    auto deletedA = fix.repository_->getDocument(aId.value());
    auto deletedB = fix.repository_->getDocument(bId.value());
    auto survivingC = fix.repository_->getDocument(cId.value());
    REQUIRE((deletedA.has_value()));
    REQUIRE((deletedB.has_value()));
    REQUIRE((survivingC.has_value()));
    CHECK_FALSE(deletedA.value().has_value());
    CHECK_FALSE(deletedB.value().has_value());
    REQUIRE((survivingC.value().has_value()));

    setStringMeta(cId.value(), "tag:alpha", "alpha");
    setStringMeta(cId.value(), "tag:beta", "beta");

    auto docD = makeDocumentWithPath("repo/tags/d.txt", "session-tag-d");
    auto dId = fix.repository_->insertDocument(docD);
    REQUIRE((dId.has_value()));
    setStringMeta(dId.value(), "tag:alpha", "alpha");
    setStringMeta(dId.value(), "tag", "legacy");

    auto docCTags = fix.repository_->getDocumentTags(cId.value());
    REQUIRE((docCTags.has_value()));
    CHECK((docCTags.value() == std::vector<std::string>{"alpha", "beta"}));

    std::vector<int64_t> tagIds{cId.value(), dId.value(), 999999};
    SECTION("legacy tag values are retained") {}
    SECTION("bounded queries deduplicate IDs across chunks") {
        REQUIRE(fix.pool_
                    ->withConnection([](Database& db) -> Result<void> {
                        sqlite3_limit(db.rawHandle(), SQLITE_LIMIT_VARIABLE_NUMBER, 16);
                        return {};
                    })
                    .has_value());
        for (int64_t i = 1; i < 40; ++i)
            tagIds.push_back(999999 + i);
        tagIds.push_back(cId.value());
        tagIds.push_back(dId.value());
    }
    auto batchTags = fix.repository_->batchGetDocumentTags(tagIds);
    REQUIRE((batchTags.has_value()));
    REQUIRE((batchTags.value().contains(cId.value())));
    REQUIRE((batchTags.value().contains(dId.value())));
    CHECK((batchTags.value().at(cId.value()) == std::vector<std::string>{"alpha", "beta"}));
    CHECK((batchTags.value().at(dId.value()) == std::vector<std::string>{"alpha", "legacy"}));
    CHECK_FALSE(batchTags.value().contains(999999));
    std::vector<std::string> lookupHashes{docC.sha256Hash, docD.sha256Hash, docC.sha256Hash};
    for (int i = 0; i < 40; ++i)
        lookupHashes.push_back("missing-batch-hash-" + std::to_string(i));
    auto hydrated = fix.repository_->batchGetDocumentsByHash(lookupHashes);
    REQUIRE(hydrated.has_value());
    REQUIRE(hydrated.value().size() == 2);
    CHECK(hydrated.value().at(docC.sha256Hash).id == cId.value());
    CHECK(hydrated.value().at(docD.sha256Hash).id == dId.value());

    // Failed preparation must release the read savepoint and leave the connection reusable.
    int previousSqlLimit = 0;
    REQUIRE(fix.pool_
                ->withConnection([&](Database& db) -> Result<void> {
                    CHECK(sqlite3_get_autocommit(db.rawHandle()) == 1);
                    db.clearStatementCache();
                    previousSqlLimit = sqlite3_limit(db.rawHandle(), SQLITE_LIMIT_SQL_LENGTH, 64);
                    return {};
                })
                .has_value());
    auto prepareFailure = fix.repository_->batchGetDocumentsByHash(lookupHashes);
    REQUIRE(fix.pool_
                ->withConnection([&](Database& db) -> Result<void> {
                    sqlite3_limit(db.rawHandle(), SQLITE_LIMIT_SQL_LENGTH, previousSqlLimit);
                    CHECK(sqlite3_get_autocommit(db.rawHandle()) == 1);
                    return {};
                })
                .has_value());
    CHECK_FALSE(prepareFailure.has_value());
    auto retryHydration = fix.repository_->batchGetDocumentsByHash(lookupHashes);
    REQUIRE(retryHydration.has_value());
    CHECK(retryHydration.value().size() == 2);

    auto allTags = fix.repository_->getAllTags();
    REQUIRE((allTags.has_value()));
    CHECK((allTags.value() == std::vector<std::string>{"alpha", "beta"}));
    auto allTagsCached = fix.repository_->getAllTags();
    REQUIRE((allTagsCached.has_value()));
    CHECK((allTagsCached.value() == allTags.value()));

    setStringMeta(dId.value(), "tag:gamma", "gamma");
    auto allTagsAfterInsert = fix.repository_->getAllTags();
    REQUIRE((allTagsAfterInsert.has_value()));
    CHECK((allTagsAfterInsert.value() == std::vector<std::string>{"alpha", "beta", "gamma"}));

    auto anyAlpha = fix.repository_->findDocumentsByTags({"alpha"}, false);
    REQUIRE((anyAlpha.has_value()));
    REQUIRE((anyAlpha.value().size() == 2));

    auto matchAll = fix.repository_->findDocumentsByTags({"alpha", "beta"}, true);
    REQUIRE((matchAll.has_value()));
    REQUIRE((matchAll.value().size() == 1));
    CHECK((matchAll.value().front().id == cId.value()));

    auto legacyMatch = fix.repository_->findDocumentsByTags({"legacy"}, false);
    REQUIRE((legacyMatch.has_value()));
    REQUIRE((legacyMatch.value().size() == 1));
    CHECK((legacyMatch.value().front().id == dId.value()));

    auto emptyMatch = fix.repository_->findDocumentsByTags({}, true);
    REQUIRE((emptyMatch.has_value()));
    CHECK((emptyMatch.value().empty()));
}

TEST_CASE("MetadataRepository: dual-pool embedding-hash read route isolation",
          "[unit][metadata][repository][dual-pool]") {
    auto dbPath = tempDbPath("metadata_repo_dual_pool_embedding_hash_");

    ConnectionPoolConfig writeCfg;
    writeCfg.minConnections = 1;
    writeCfg.maxConnections = 2;

    ConnectionPoolConfig readCfg;
    readCfg.minConnections = 1;
    readCfg.maxConnections = 2;

    auto writePool = std::make_unique<ConnectionPool>(dbPath.string(), writeCfg);
    auto readPool = std::make_unique<ConnectionPool>(dbPath.string(), readCfg);

    REQUIRE((writePool->initialize().has_value()));
    REQUIRE((readPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *writePool, readPool.get(), MetadataRepository::SchemaBootstrapMode::AssumeReady);

    auto doc = makeDocumentWithPath("/tmp/dual-pool-embed-hash.txt", "dual-pool-embed-hash");
    auto insertResult = repository->insertDocument(doc);
    REQUIRE((insertResult.has_value()));

    auto readOk = repository->hasDocumentEmbeddingByHash("dual-pool-embed-hash");
    REQUIRE((readOk.has_value()));
    REQUIRE_FALSE(readOk.value());

    readPool->shutdown();

    auto readAfterShutdown = repository->hasDocumentEmbeddingByHash("dual-pool-embed-hash");
    REQUIRE_FALSE(readAfterShutdown.has_value());

    repository.reset();
    writePool->shutdown();
    writePool.reset();
    readPool.reset();

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

TEST_CASE("MetadataRepository: reconcile embedding status from vector-backed hashes",
          "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/reconcile_a.txt", "reconcile-hash-a");
    auto docB = makeDocumentWithPath("/tmp/reconcile_b.txt", "reconcile-hash-b");

    auto docAId = fix.repository_->insertDocument(docA);
    auto docBId = fix.repository_->insertDocument(docB);
    REQUIRE((docAId.has_value()));
    REQUIRE((docBId.has_value()));

    REQUIRE(fix.repository_
                ->updateDocumentEmbeddingStatusByHash("reconcile-hash-a", true, "legacy-embed")
                .has_value());
    REQUIRE(fix.repository_
                ->updateDocumentEmbeddingStatusByHash("reconcile-hash-b", true, "legacy-embed")
                .has_value());

    auto beforeA = fix.repository_->hasDocumentEmbeddingByHash("reconcile-hash-a");
    auto beforeB = fix.repository_->hasDocumentEmbeddingByHash("reconcile-hash-b");
    REQUIRE((beforeA.has_value()));
    REQUIRE((beforeB.has_value()));
    CHECK((beforeA.value()));
    CHECK((beforeB.value()));

    REQUIRE(fix.repository_
                ->reconcileDocumentEmbeddingStatusByHashes({"reconcile-hash-a", "missing-hash"})
                .has_value());

    auto afterA = fix.repository_->hasDocumentEmbeddingByHash("reconcile-hash-a");
    auto afterB = fix.repository_->hasDocumentEmbeddingByHash("reconcile-hash-b");
    REQUIRE((afterA.has_value()));
    REQUIRE((afterB.has_value()));
    CHECK((afterA.value()));
    CHECK_FALSE(afterB.value());

    auto stats = fix.repository_->getCorpusStats();
    REQUIRE((stats.has_value()));
    CHECK((stats.value().embeddingCount == 1));
}

TEST_CASE(
    "MetadataRepository: batch embedding status update ignores missing hashes and duplicate inputs",
    "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/batch_embed.txt", "batch-embed-hash");
    REQUIRE((fix.repository_->insertDocument(doc).has_value()));

    REQUIRE(fix.repository_->batchUpdateDocumentEmbeddingStatusByHashes({}, true, "ignored-model")
                .has_value());

    auto warmStats = fix.repository_->getCorpusStats();
    REQUIRE((warmStats.has_value()));
    CHECK((warmStats.value().embeddingCount == 0));

    REQUIRE(
        fix.repository_
            ->batchUpdateDocumentEmbeddingStatusByHashes(
                {"missing-batch-hash", "batch-embed-hash", "batch-embed-hash"}, true, "batch-model")
            .has_value());

    auto hasEmbedding = fix.repository_->hasDocumentEmbeddingByHash("batch-embed-hash");
    REQUIRE((hasEmbedding.has_value()));
    CHECK((hasEmbedding.value()));

    auto missingEmbedding = fix.repository_->hasDocumentEmbeddingByHash("missing-batch-hash");
    REQUIRE((missingEmbedding.has_value()));
    CHECK_FALSE(missingEmbedding.value());

    auto storedStatus = getEmbeddingStatusRow(*fix.pool_, "batch-embed-hash");
    REQUIRE((storedStatus.has_value()));
    REQUIRE((storedStatus.value().has_value()));
    CHECK((storedStatus.value()->hasEmbedding));
    REQUIRE((storedStatus.value()->modelId.has_value()));
    CHECK((storedStatus.value()->modelId.value() == "batch-model"));

    auto afterEnableStats = fix.repository_->getCorpusStats();
    REQUIRE((afterEnableStats.has_value()));
    CHECK((afterEnableStats.value().embeddingCount == 1));

    REQUIRE(fix.repository_
                ->batchUpdateDocumentEmbeddingStatusByHashes(
                    {"batch-embed-hash", "missing-batch-hash", "batch-embed-hash"}, false, "")
                .has_value());

    auto disabledStatus = getEmbeddingStatusRow(*fix.pool_, "batch-embed-hash");
    REQUIRE((disabledStatus.has_value()));
    REQUIRE((disabledStatus.value().has_value()));
    CHECK_FALSE(disabledStatus.value()->hasEmbedding);
    CHECK_FALSE(disabledStatus.value()->modelId.has_value());

    auto afterDisableStats = fix.repository_->getCorpusStats();
    REQUIRE((afterDisableStats.has_value()));
    CHECK((afterDisableStats.value().embeddingCount == 0));
}

TEST_CASE("MetadataRepository: embedding completion commits embedding and repair state together",
          "[unit][metadata][repository][embeddings][completion]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/complete_embed.txt", "complete-embed-hash");
    auto inserted = fix.repository_->insertDocument(doc);
    REQUIRE((inserted.has_value()));

    REQUIRE(
        fix.repository_->batchCompleteDocumentEmbeddingsByHashes({}, "ignored-model").has_value());
    REQUIRE(fix.repository_
                ->batchCompleteDocumentEmbeddingsByHashes(
                    {"missing-complete-hash", "complete-embed-hash", "complete-embed-hash"},
                    "complete-model")
                .has_value());

    auto embedding = getEmbeddingStatusRow(*fix.pool_, "complete-embed-hash");
    REQUIRE((embedding.has_value()));
    REQUIRE((embedding.value().has_value()));
    CHECK(embedding.value()->hasEmbedding);
    REQUIRE((embedding.value()->modelId.has_value()));
    CHECK((embedding.value()->modelId.value() == "complete-model"));

    auto completed = fix.repository_->getDocument(inserted.value());
    REQUIRE((completed.has_value()));
    REQUIRE((completed.value().has_value()));
    CHECK((completed.value()->repairStatus == RepairStatus::Completed));
    CHECK((completed.value()->repairAttempts == 2));

    auto missing = fix.repository_->hasDocumentEmbeddingByHash("missing-complete-hash");
    REQUIRE((missing.has_value()));
    CHECK_FALSE(missing.value());

    auto stats = fix.repository_->getCorpusStats();
    REQUIRE((stats.has_value()));
    CHECK((stats.value().embeddingCount == 1));
}

TEST_CASE("MetadataRepository fences superseded embedding derivations",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "derivation-hash";
    auto inserted =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/derivation.txt", hash));
    REQUIRE(inserted.has_value());
    auto a = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe-a");
    REQUIRE(a.has_value());
    auto b = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe-b");
    REQUIRE(b.has_value());
    auto pendingStates = fix.repository_->batchGetDocumentEmbeddingDerivations({hash, "missing"});
    REQUIRE(pendingStates.has_value());
    REQUIRE(pendingStates.value().size() == 1);
    CHECK_FALSE(pendingStates.value().at(hash).completed);
    CHECK(pendingStates.value().at(hash).token.generation == b.value().generation);
    CHECK(a.value().generation != b.value().generation);
    auto tampered = b.value();
    tampered.recipe = "wrong-recipe";
    auto rejected = fix.repository_->completeDocumentEmbeddingDerivation(tampered, "wrong-model");
    REQUIRE(rejected.has_value());
    CHECK_FALSE(rejected.value());
    auto stale = fix.repository_->completeDocumentEmbeddingDerivation(a.value(), "model-a");
    REQUIRE(stale.has_value());
    CHECK_FALSE(stale.value());
    REQUIRE(fix.repository_->batchCompleteDocumentEmbeddingsByHashes({hash}, "legacy-model")
                .has_value());
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(ready.has_value());
    CHECK_FALSE(ready.value());
    auto current = fix.repository_->completeDocumentEmbeddingDerivation(b.value(), "model-b");
    REQUIRE(current.has_value());
    CHECK(current.value());
    auto duplicate = fix.repository_->completeDocumentEmbeddingDerivation(b.value(), "model-b");
    REQUIRE(duplicate.has_value());
    CHECK_FALSE(duplicate.value());
    auto status = getEmbeddingStatusRow(*fix.pool_, hash);
    REQUIRE(status.has_value());
    REQUIRE(status.value().has_value());
    CHECK(status.value()->modelId == "model-b");
    auto publishedStates = fix.repository_->batchGetDocumentEmbeddingDerivations({hash});
    REQUIRE(publishedStates.has_value());
    CHECK(publishedStates.value().at(hash).completed);
    CHECK(publishedStates.value().at(hash).token.recipe == "recipe-b");
    auto stats = fix.repository_->getCorpusStats();
    REQUIRE(stats.has_value());
    CHECK(stats.value().embeddingCount == 1);
    auto next = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe-b");
    REQUIRE(next.has_value());
    CHECK(next.value().generation != b.value().generation);
    auto invalidated = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(invalidated.has_value());
    CHECK_FALSE(invalidated.value());
    auto nextStats = fix.repository_->getCorpusStats();
    REQUIRE(nextStats.has_value());
    CHECK(nextStats.value().embeddingCount == 0);
}

TEST_CASE("MetadataRepository invalidates derivation tokens on extracted input changes",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "derivation-content-hash";
    auto inserted =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/derivation-content.txt", hash));
    REQUIRE(inserted.has_value());
    auto entry = makeBatchContentEntry(inserted.value(), "Title", "Original text");
    REQUIRE(fix.repository_->batchInsertContentAndIndex({entry}).has_value());
    auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(token.has_value());
    bool changed = true;
    SECTION("changed text") {
        entry.contentText = "New text";
    }
    SECTION("changed method") {
        entry.extractionMethod += "-new";
    }
    SECTION("changed language") {
        entry.language = "fr";
    }
    SECTION("identical content") {
        changed = false;
    }
    REQUIRE(fix.repository_->batchInsertContentAndIndex({entry}).has_value());
    auto completed = fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
    REQUIRE(completed.has_value());
    CHECK(completed.value() == !changed);
    if (changed) {
        REQUIRE(fix.repository_->batchCompleteDocumentEmbeddingsByHashes({hash}, "legacy-model")
                    .has_value());
        auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
        REQUIRE(ready.has_value());
        CHECK_FALSE(ready.value());
    }
}

TEST_CASE("MetadataRepository direct content writes invalidate pending derivations",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "direct-derivation-hash";
    auto id =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/direct-derivation.txt", hash));
    REQUIRE(id.has_value());
    DocumentContent content;
    content.documentId = id.value();
    content.contentText = "original";
    REQUIRE(fix.repository_->insertContent(content).has_value());
    auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(token.has_value());
    SECTION("update") {
        content.contentText = "changed";
        REQUIRE(fix.repository_->updateContent(content).has_value());
    }
    SECTION("delete") {
        REQUIRE(fix.repository_->deleteContent(id.value()).has_value());
    }
    SECTION("delete and reinsert identical content does not revive old token") {
        REQUIRE(fix.repository_->deleteContent(id.value()).has_value());
        REQUIRE(fix.repository_->insertContent(content).has_value());
    }
    auto completed = fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
    REQUIRE(completed.has_value());
    CHECK_FALSE(completed.value());
}

TEST_CASE("MetadataRepository content writes clear completed embedding readiness",
          "[unit][metadata][embeddings][content-readiness]") {
    MetadataRepositoryFixture fix;
    const bool tracked = GENERATE(false, true);
    const std::string hash = "completed-content-hash";
    auto id =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/completed-content.txt", hash));
    REQUIRE(id.has_value());
    DocumentContent content;
    content.documentId = id.value();
    content.contentText = "original";
    REQUIRE(fix.repository_->insertContent(content).has_value());
    if (tracked) {
        auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
        REQUIRE(token.has_value());
        auto complete =
            fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
        REQUIRE(complete.has_value());
        REQUIRE(complete.value());
    } else {
        REQUIRE(
            fix.repository_->updateDocumentEmbeddingStatus(id.value(), true, "model").has_value());
    }
    REQUIRE(fix.repository_->getCachedEmbeddedCount() == 1);
    bool changed = true;
    SECTION("upsert changed text") {
        content.contentText = "replacement";
        REQUIRE(fix.repository_->insertContent(content).has_value());
    }
    SECTION("update changed text") {
        content.contentText = "replacement";
        REQUIRE(fix.repository_->updateContent(content).has_value());
    }
    SECTION("delete") {
        REQUIRE(fix.repository_->deleteContent(id.value()).has_value());
    }
    SECTION("identical update preserves readiness") {
        changed = false;
        REQUIRE(fix.repository_->updateContent(content).has_value());
    }
    SECTION("identical upsert preserves readiness") {
        changed = false;
        REQUIRE(fix.repository_->insertContent(content).has_value());
    }
    auto status = getEmbeddingStatusRow(*fix.pool_, hash);
    REQUIRE(status.has_value());
    REQUIRE(status.value().has_value());
    CHECK(status.value()->hasEmbedding == !changed);
    auto count = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE(count.has_value());
    CHECK(count.value() == (changed ? 0 : 1));
    CHECK(fix.repository_->getCachedEmbeddedCount() == (changed ? 0 : 1));
}

TEST_CASE("Embedding readiness migration upgrades version 38 databases",
          "[unit][metadata][embeddings][content-readiness][migration]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "upgrade-content-hash";
    auto id =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/upgrade-content.txt", hash));
    REQUIRE(id.has_value());
    DocumentContent content;
    content.documentId = id.value();
    content.contentText = "original";
    REQUIRE(fix.repository_->insertContent(content).has_value());
    auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(token.has_value());
    REQUIRE(
        fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model").has_value());
    auto upgraded = fix.pool_->withConnection([&](Database& db) -> Result<void> {
        MigrationManager manager(db);
        manager.registerMigrations(YamsMetadataMigrations::getAllMigrations());
        if (auto rolledBack = manager.rollbackTo(38); !rolledBack) {
            return rolledBack.error();
        }
        // Version 38 invalidates the token but leaves the legacy ready bit set.
        if (auto changed =
                db.execute("UPDATE document_content SET content_text = 'changed before upgrade'");
            !changed) {
            return changed.error();
        }
        return manager.migrateTo(39);
    });
    REQUIRE(upgraded.has_value());
    auto status = getEmbeddingStatusRow(*fix.pool_, hash);
    REQUIRE(status.has_value());
    REQUIRE(status.value().has_value());
    CHECK_FALSE(status.value()->hasEmbedding);

    auto next = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(next.has_value());
    REQUIRE(
        fix.repository_->completeDocumentEmbeddingDerivation(next.value(), "model").has_value());
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute(
                        "UPDATE document_content SET content_text = 'changed after upgrade'");
                })
                .has_value());
    auto invalidated = getEmbeddingStatusRow(*fix.pool_, hash);
    REQUIRE(invalidated.has_value());
    REQUIRE(invalidated.value().has_value());
    CHECK_FALSE(invalidated.value()->hasEmbedding);
}

TEST_CASE("WriteCoordinator preserves derivation identities across completion batches",
          "[unit][metadata][embeddings][derivation][write-coordinator]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "queued-derivation-hash";
    REQUIRE(
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/queued-derivation.txt", hash))
            .has_value());
    auto old = fix.repository_->beginDocumentEmbeddingDerivation(hash, "old-recipe");
    REQUIRE(old.has_value());
    auto current = fix.repository_->beginDocumentEmbeddingDerivation(hash, "new-recipe");
    REQUIRE(current.has_value());
    boost::asio::io_context io;
    auto repoRef =
        std::shared_ptr<MetadataRepository>(fix.repository_.get(), [](MetadataRepository*) {});
    yams::daemon::WriteCoordinator coordinator(io, {}, repoRef);
    auto batch = std::make_unique<yams::daemon::WriteBatch>();
    batch->source = "test/derivation-completion";
    // New completion precedes a late old completion using the same model/grouping key.
    batch->ops.emplace_back(
        yams::daemon::CompleteDocumentEmbeddingsByHashesOp{{}, "model", {current.value()}});
    batch->ops.emplace_back(
        yams::daemon::CompleteDocumentEmbeddingsByHashesOp{{}, "model", {old.value()}});
    coordinator.enqueue(std::move(batch));
    coordinator.start();
    auto runner = std::async(std::launch::async, [&] { io.run(); });
    const auto flushed = coordinator.flush(std::chrono::seconds{5});
    coordinator.shutdown();
    io.stop();
    runner.get();
    REQUIRE(flushed.has_value());
    CHECK(coordinator.getStats().embeddingStatusesUpdated == 1);
    CHECK(coordinator.getStats().repairStatusesUpdated == 1);
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(ready.has_value());
    CHECK(ready.value());
}

TEST_CASE("WriteCoordinator reports derivation SQL failures through flush",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "queued-derivation-error";
    REQUIRE(fix.repository_
                ->insertDocument(makeDocumentWithPath("/tmp/queued-derivation-error.txt", hash))
                .has_value());
    auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(token.has_value());
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute(
                        "CREATE TRIGGER reject_queued_completion BEFORE UPDATE OF repair_status ON "
                        "documents "
                        "BEGIN SELECT RAISE(ABORT, 'queued completion failure'); END");
                })
                .has_value());

    boost::asio::io_context io;
    auto repo =
        std::shared_ptr<MetadataRepository>(fix.repository_.get(), [](MetadataRepository*) {});
    yams::daemon::WriteCoordinator coordinator(io, {}, repo);
    auto batch = std::make_unique<yams::daemon::WriteBatch>();
    batch->source = "test/derivation-error";
    batch->ops.emplace_back(
        yams::daemon::CompleteDocumentEmbeddingsByHashesOp{{}, "model", {token.value()}});
    coordinator.enqueue(std::move(batch));
    coordinator.start();
    auto runner = std::async(std::launch::async, [&] { io.run(); });
    auto flushed = coordinator.flush(std::chrono::seconds{5});
    coordinator.shutdown();
    io.stop();
    runner.get();

    CHECK_FALSE(flushed.has_value());
    CHECK(coordinator.getStats().batchesCommitted == 0);
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(ready.has_value());
    CHECK_FALSE(ready.value());
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) {
                    return db.execute("DROP TRIGGER reject_queued_completion");
                })
                .has_value());
    auto retried = fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
    REQUIRE(retried.has_value());
    CHECK(retried.value());
}

TEST_CASE("MetadataRepository derivation completion rolls back and can be retried",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "derivation-rollback-hash";
    REQUIRE(
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/derivation-rollback.txt", hash))
            .has_value());
    REQUIRE_FALSE(fix.repository_->beginDocumentEmbeddingDerivation(hash, "").has_value());
    REQUIRE_FALSE(
        fix.repository_->beginDocumentEmbeddingDerivation("missing", "recipe").has_value());
    auto token = fix.repository_->beginDocumentEmbeddingDerivation(hash, "recipe");
    REQUIRE(token.has_value());
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute(R"(
            CREATE TRIGGER reject_derivation_repair BEFORE UPDATE OF repair_status ON documents
            BEGIN SELECT RAISE(ABORT, 'test derivation completion rollback'); END
        )");
                })
                .has_value());
    auto failed = fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
    REQUIRE_FALSE(failed.has_value());
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(ready.has_value());
    CHECK_FALSE(ready.value());
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute("DROP TRIGGER reject_derivation_repair");
                })
                .has_value());
    auto retried = fix.repository_->completeDocumentEmbeddingDerivation(token.value(), "model");
    REQUIRE(retried.has_value());
    CHECK(retried.value());
}

TEST_CASE("MetadataRepository: reconcile empty vector-backed hash set clears embedding ownership",
          "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/reconcile_clear_a.txt", "reconcile-clear-hash-a");
    auto docB = makeDocumentWithPath("/tmp/reconcile_clear_b.txt", "reconcile-clear-hash-b");
    REQUIRE((fix.repository_->insertDocument(docA).has_value()));
    REQUIRE((fix.repository_->insertDocument(docB).has_value()));

    REQUIRE(
        fix.repository_
            ->updateDocumentEmbeddingStatusByHash("reconcile-clear-hash-a", true, "legacy-model")
            .has_value());
    REQUIRE(
        fix.repository_
            ->updateDocumentEmbeddingStatusByHash("reconcile-clear-hash-b", true, "legacy-model")
            .has_value());
    REQUIRE((setEmbeddingChunkCount(*fix.pool_, "reconcile-clear-hash-a", 4).has_value()));
    REQUIRE((setEmbeddingChunkCount(*fix.pool_, "reconcile-clear-hash-b", 6).has_value()));

    auto beforeStats = fix.repository_->getCorpusStats();
    REQUIRE((beforeStats.has_value()));
    CHECK((beforeStats.value().embeddingCount == 2));

    REQUIRE((fix.repository_->reconcileDocumentEmbeddingStatusByHashes({}, "").has_value()));

    auto statusA = getEmbeddingStatusRow(*fix.pool_, "reconcile-clear-hash-a");
    auto statusB = getEmbeddingStatusRow(*fix.pool_, "reconcile-clear-hash-b");
    REQUIRE((statusA.has_value()));
    REQUIRE((statusB.has_value()));
    REQUIRE((statusA.value().has_value()));
    REQUIRE((statusB.value().has_value()));
    CHECK_FALSE(statusA.value()->hasEmbedding);
    CHECK_FALSE(statusB.value()->hasEmbedding);
    CHECK_FALSE(statusA.value()->modelId.has_value());
    CHECK_FALSE(statusB.value()->modelId.has_value());
    CHECK((statusA.value()->chunkCount == 0));
    CHECK((statusB.value()->chunkCount == 0));

    auto afterStats = fix.repository_->getCorpusStats();
    REQUIRE((afterStats.has_value()));
    CHECK((afterStats.value().embeddingCount == 0));
}

TEST_CASE("MetadataRepository: reconcile preserves chunk count and omitted model ids",
          "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/reconcile_preserve.txt", "reconcile-preserve-hash");
    REQUIRE((fix.repository_->insertDocument(doc).has_value()));
    REQUIRE(
        fix.repository_
            ->updateDocumentEmbeddingStatusByHash("reconcile-preserve-hash", true, "legacy-model")
            .has_value());
    REQUIRE((setEmbeddingChunkCount(*fix.pool_, "reconcile-preserve-hash", 7).has_value()));

    REQUIRE(
        fix.repository_->reconcileDocumentEmbeddingStatusByHashes({"reconcile-preserve-hash"}, "")
            .has_value());

    auto preserved = getEmbeddingStatusRow(*fix.pool_, "reconcile-preserve-hash");
    REQUIRE((preserved.has_value()));
    REQUIRE((preserved.value().has_value()));
    CHECK((preserved.value()->hasEmbedding));
    REQUIRE((preserved.value()->modelId.has_value()));
    CHECK((preserved.value()->modelId.value() == "legacy-model"));
    CHECK((preserved.value()->chunkCount == 7));

    REQUIRE(fix.repository_
                ->reconcileDocumentEmbeddingStatusByHashes(
                    {"reconcile-preserve-hash", "reconcile-preserve-hash"}, "fresh-model")
                .has_value());

    auto updated = getEmbeddingStatusRow(*fix.pool_, "reconcile-preserve-hash");
    REQUIRE((updated.has_value()));
    REQUIRE((updated.value().has_value()));
    CHECK((updated.value()->hasEmbedding));
    REQUIRE((updated.value()->modelId.has_value()));
    CHECK((updated.value()->modelId.value() == "fresh-model"));
    CHECK((updated.value()->chunkCount == 7));

    auto stats = fix.repository_->getCorpusStats();
    REQUIRE((stats.has_value()));
    CHECK((stats.value().embeddingCount == 1));
}

TEST_CASE("MetadataRepository: vector hash ownership ignores missing documents",
          "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/existing_vector_a.txt", "existing-vector-hash-a");
    auto docB = makeDocumentWithPath("/tmp/existing_vector_b.txt", "existing-vector-hash-b");

    REQUIRE((fix.repository_->insertDocument(docA).has_value()));
    REQUIRE((fix.repository_->insertDocument(docB).has_value()));

    auto existing = fix.repository_->getExistingDocumentHashes(
        {"existing-vector-hash-a", "orphan-vector-hash", "existing-vector-hash-b",
         "existing-vector-hash-a"});

    REQUIRE((existing.has_value()));
    CHECK((existing.value().size() == 2));
    CHECK((existing.value().contains("existing-vector-hash-a")));
    CHECK((existing.value().contains("existing-vector-hash-b")));
    CHECK_FALSE(existing.value().contains("orphan-vector-hash"));
}

TEST_CASE(
    "MetadataRepository: embedding status update by document id tracks counts and model changes",
    "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    auto doc = makeDocumentWithPath("/tmp/embed_by_id.txt", "embed-by-id-hash");
    auto insert = fix.repository_->insertDocument(doc);
    REQUIRE((insert.has_value()));
    const auto docId = insert.value();

    auto initialCount = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((initialCount.has_value()));
    CHECK((initialCount.value() == 0));

    REQUIRE((fix.repository_->updateDocumentEmbeddingStatus(docId, true, "model-a").has_value()));

    auto enabled = getEmbeddingStatusRow(*fix.pool_, "embed-by-id-hash");
    REQUIRE((enabled.has_value()));
    REQUIRE((enabled.value().has_value()));
    CHECK((enabled.value()->hasEmbedding));
    REQUIRE((enabled.value()->modelId.has_value()));
    CHECK((enabled.value()->modelId.value() == "model-a"));

    auto afterEnableCount = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((afterEnableCount.has_value()));
    CHECK((afterEnableCount.value() == 1));

    auto afterEnableStats = fix.repository_->getCorpusStats();
    REQUIRE((afterEnableStats.has_value()));
    CHECK((afterEnableStats.value().embeddingCount == 1));

    REQUIRE((fix.repository_->updateDocumentEmbeddingStatus(docId, true, "model-b").has_value()));

    auto updated = getEmbeddingStatusRow(*fix.pool_, "embed-by-id-hash");
    REQUIRE((updated.has_value()));
    REQUIRE((updated.value().has_value()));
    CHECK((updated.value()->hasEmbedding));
    REQUIRE((updated.value()->modelId.has_value()));
    CHECK((updated.value()->modelId.value() == "model-b"));

    auto afterRetagCount = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((afterRetagCount.has_value()));
    CHECK((afterRetagCount.value() == 1));

    REQUIRE((fix.repository_->updateDocumentEmbeddingStatus(docId, false, "").has_value()));

    auto disabled = getEmbeddingStatusRow(*fix.pool_, "embed-by-id-hash");
    REQUIRE((disabled.has_value()));
    REQUIRE((disabled.value().has_value()));
    CHECK_FALSE(disabled.value()->hasEmbedding);
    CHECK_FALSE(disabled.value()->modelId.has_value());

    auto afterDisableCount = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((afterDisableCount.has_value()));
    CHECK((afterDisableCount.value() == 0));

    auto afterDisableStats = fix.repository_->getCorpusStats();
    REQUIRE((afterDisableStats.has_value()));
    CHECK((afterDisableStats.value().embeddingCount == 0));

    REQUIRE((fix.repository_->updateDocumentEmbeddingStatus(docId, false, "").has_value()));
    auto afterNoopDisable = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((afterNoopDisable.has_value()));
    CHECK((afterNoopDisable.value() == 0));
}

TEST_CASE("MetadataRepository: embedding status update surfaces missing document identifiers",
          "[unit][metadata][repository][embeddings]") {
    MetadataRepositoryFixture fix;

    auto missingById = fix.repository_->updateDocumentEmbeddingStatus(999999, true, "model-x");
    REQUIRE_FALSE((missingById.has_value()));
    CHECK((missingById.error().message.find("Document not found") != std::string::npos));

    auto missingByHash =
        fix.repository_->updateDocumentEmbeddingStatusByHash("missing-embed-hash", true, "model-x");
    REQUIRE_FALSE((missingByHash.has_value()));
    CHECK(
        (missingByHash.error().message.find("Document with hash not found") != std::string::npos));

    auto hasMissing = fix.repository_->hasDocumentEmbeddingByHash("missing-embed-hash");
    REQUIRE((hasMissing.has_value()));
    CHECK_FALSE(hasMissing.value());

    auto embeddedCount = fix.repository_->getEmbeddedDocumentCount();
    REQUIRE((embeddedCount.has_value()));
    CHECK((embeddedCount.value() == 0));
}

TEST_CASE("MetadataRepository: repair status update increments attempts and persists state",
          "[unit][metadata][repository][repair]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/repair-single.txt", "repair-single-hash");
    auto insert = fix.repository_->insertDocument(doc);
    REQUIRE((insert.has_value()));
    const auto docId = insert.value();

    REQUIRE(
        (fix.repository_->updateDocumentRepairStatus("repair-single-hash", RepairStatus::Processing)
             .has_value()));

    auto firstRead = fix.repository_->getDocument(docId);
    REQUIRE((firstRead.has_value()));
    REQUIRE((firstRead.value().has_value()));
    CHECK((firstRead.value()->repairStatus == RepairStatus::Processing));
    CHECK((firstRead.value()->repairAttempts == 1));
    CHECK((firstRead.value()->repairAttemptedAt.time_since_epoch().count() > 0));

    const auto firstAttemptedAt = firstRead.value()->repairAttemptedAt.time_since_epoch().count();

    REQUIRE((fix.repository_->updateDocumentRepairStatus("repair-single-hash", RepairStatus::Failed)
                 .has_value()));

    auto secondRead = fix.repository_->getDocument(docId);
    REQUIRE((secondRead.has_value()));
    REQUIRE((secondRead.value().has_value()));
    CHECK((secondRead.value()->repairStatus == RepairStatus::Failed));
    CHECK((secondRead.value()->repairAttempts == 2));
    CHECK((secondRead.value()->repairAttemptedAt.time_since_epoch().count() >= firstAttemptedAt));
}

TEST_CASE("MetadataRepository: repair status update ignores missing hash",
          "[unit][metadata][repository][repair]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/repair-missing-control.txt", "repair-missing-control");
    auto insert = fix.repository_->insertDocument(doc);
    REQUIRE((insert.has_value()));

    REQUIRE(
        (fix.repository_->updateDocumentRepairStatus("repair-missing-hash", RepairStatus::Completed)
             .has_value()));

    auto control = fix.repository_->getDocument(insert.value());
    REQUIRE((control.has_value()));
    REQUIRE((control.value().has_value()));
    CHECK((control.value()->repairStatus == RepairStatus::Pending));
    CHECK((control.value()->repairAttempts == 0));
    CHECK((control.value()->repairAttemptedAt.time_since_epoch().count() == 0));
}

TEST_CASE("MetadataRepository: batch repair status update skips missing hashes and preserves no-op "
          "empty batch",
          "[unit][metadata][repository][repair]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/repair-batch-a.txt", "repair-batch-hash-a");
    auto docB = makeDocumentWithPath("/tmp/repair-batch-b.txt", "repair-batch-hash-b");
    auto insertA = fix.repository_->insertDocument(docA);
    auto insertB = fix.repository_->insertDocument(docB);
    REQUIRE((insertA.has_value()));
    REQUIRE((insertB.has_value()));

    REQUIRE((fix.repository_->batchUpdateDocumentRepairStatuses({}, RepairStatus::Processing)
                 .has_value()));

    auto beforeA = fix.repository_->getDocument(insertA.value());
    auto beforeB = fix.repository_->getDocument(insertB.value());
    REQUIRE((beforeA.has_value()));
    REQUIRE((beforeA.value().has_value()));
    REQUIRE((beforeB.has_value()));
    REQUIRE((beforeB.value().has_value()));
    CHECK((beforeA.value()->repairAttempts == 0));
    CHECK((beforeB.value()->repairAttempts == 0));

    REQUIRE((fix.repository_
                 ->batchUpdateDocumentRepairStatuses(
                     {"repair-batch-hash-a", "repair-missing-hash", "repair-batch-hash-b"},
                     RepairStatus::Processing)
                 .has_value()));

    auto afterA = fix.repository_->getDocument(insertA.value());
    auto afterB = fix.repository_->getDocument(insertB.value());
    REQUIRE((afterA.has_value()));
    REQUIRE((afterA.value().has_value()));
    REQUIRE((afterB.has_value()));
    REQUIRE((afterB.value().has_value()));
    CHECK((afterA.value()->repairStatus == RepairStatus::Processing));
    CHECK((afterB.value()->repairStatus == RepairStatus::Processing));
    CHECK((afterA.value()->repairAttempts == 1));
    CHECK((afterB.value()->repairAttempts == 1));
    CHECK((afterA.value()->repairAttemptedAt.time_since_epoch().count() > 0));
    CHECK((afterB.value()->repairAttemptedAt.time_since_epoch().count() > 0));
}

TEST_CASE("MetadataRepository: batch repair status update preserves duplicate hash attempt counts",
          "[unit][metadata][repository][repair]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/repair-duplicate.txt", "repair-duplicate-hash");
    auto insert = fix.repository_->insertDocument(doc);
    REQUIRE((insert.has_value()));

    REQUIRE((fix.repository_
                 ->batchUpdateDocumentRepairStatuses(
                     {"repair-duplicate-hash", "repair-duplicate-hash", "repair-duplicate-hash"},
                     RepairStatus::Processing)
                 .has_value()));

    auto after = fix.repository_->getDocument(insert.value());
    REQUIRE((after.has_value()));
    REQUIRE((after.value().has_value()));
    CHECK((after.value()->repairStatus == RepairStatus::Processing));
    CHECK((after.value()->repairAttempts == 3));
    CHECK((after.value()->repairAttemptedAt.time_since_epoch().count() > 0));
}

TEST_CASE("MetadataRepository: batch repair status update does not stack retry loops under lock",
          "[unit][metadata][repository][repair][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_repair_lock_");
    // Other contention tests intentionally increment this process-global advisory counter.
    // Each case owns its observation window and must not depend on Catch2 execution order.
    (void)daemon::TuneAdvisor::getAndResetDbLockErrors();

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc = makeDocumentWithPath("/tmp/repair-lock.txt", "repair-lock-hash");
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    const auto start = std::chrono::steady_clock::now();
    auto result =
        repository->batchUpdateDocumentRepairStatuses({"repair-lock-hash"}, RepairStatus::Failed);
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((result.error().message.find("locked") != std::string::npos));
    CHECK((elapsed < std::chrono::milliseconds(3000)));
    const auto lockErrors = daemon::TuneAdvisor::getAndResetDbLockErrors();
    INFO("reported lock errors: " << lockErrors);
    CHECK((lockErrors == 2));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataRepository: getMetadataValueCounts with filters",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::system_clock::now();
    auto nowSec = std::chrono::floor<std::chrono::seconds>(now);

    // doc1: old document (outside filter range)
    auto doc1 = makeDocumentWithPath("/tmp/a.txt", "hash-a");
    doc1.createdTime = nowSec - std::chrono::hours(48);
    doc1.modifiedTime = doc1.createdTime;
    doc1.indexedTime = doc1.createdTime;
    auto doc1Res = fix.repository_->insertDocument(doc1);
    REQUIRE((doc1Res.has_value()));

    // doc2: recent document (inside filter range)
    auto doc2 = makeDocumentWithPath("/tmp/b.txt", "hash-b");
    doc2.createdTime = nowSec;
    doc2.modifiedTime = nowSec;
    doc2.indexedTime = nowSec;
    auto doc2Res = fix.repository_->insertDocument(doc2);
    REQUIRE((doc2Res.has_value()));

    // doc3: another recent document (inside filter range)
    auto doc3 = makeDocumentWithPath("/tmp/c.txt", "hash-c");
    doc3.createdTime = nowSec;
    doc3.modifiedTime = nowSec;
    doc3.indexedTime = nowSec;
    auto doc3Res = fix.repository_->insertDocument(doc3);
    REQUIRE((doc3Res.has_value()));

    MetadataValue v1;
    v1.value = "PBI-070";
    v1.type = MetadataValueType::String;
    // doc1 gets PBI-070 but is outside time filter
    REQUIRE((fix.repository_->setMetadata(doc1Res.value(), "pbi", v1).has_value()));
    // doc2 gets PBI-070 (inside time filter)
    REQUIRE((fix.repository_->setMetadata(doc2Res.value(), "pbi", v1).has_value()));

    MetadataValue v2;
    v2.value = "PBI-071";
    v2.type = MetadataValueType::String;
    // doc3 gets PBI-071 (inside time filter)
    REQUIRE((fix.repository_->setMetadata(doc3Res.value(), "pbi", v2).has_value()));

    DocumentQueryOptions opts;
    opts.modifiedAfter = std::chrono::duration_cast<std::chrono::seconds>(
                             (nowSec - std::chrono::hours(24)).time_since_epoch())
                             .count();

    auto countsRes = fix.repository_->getMetadataValueCounts({"pbi"}, opts);
    REQUIRE((countsRes.has_value()));

    const auto& map = countsRes.value();
    auto it = map.find("pbi");
    REQUIRE((it != map.end()));

    std::unordered_map<std::string, int64_t> counts;
    for (const auto& row : it->second) {
        counts[row.value] = row.count;
    }

    // Only doc2 (PBI-070) and doc3 (PBI-071) are in the time range
    // doc1 (PBI-070) is filtered out because it's 48 hours old
    REQUIRE((counts.size() == 2));
    REQUIRE((counts["PBI-070"] == 1));
    REQUIRE((counts["PBI-071"] == 1));
}

TEST_CASE("MetadataRepository: metadata value counts materialized view stays consistent",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/materialized_a.txt", "materialized-a");
    auto docB = makeDocumentWithPath("/tmp/materialized_b.txt", "materialized-b");

    auto docARes = fix.repository_->insertDocument(docA);
    auto docBRes = fix.repository_->insertDocument(docB);
    REQUIRE((docARes.has_value()));
    REQUIRE((docBRes.has_value()));

    MetadataValue v1;
    v1.value = "PBI-500";
    v1.type = MetadataValueType::String;
    REQUIRE((fix.repository_->setMetadata(docARes.value(), "pbi", v1).has_value()));
    REQUIRE((fix.repository_->setMetadata(docBRes.value(), "pbi", v1).has_value()));

    DocumentQueryOptions opts;
    auto countsRes = fix.repository_->getMetadataValueCounts({"pbi"}, opts);
    REQUIRE((countsRes.has_value()));
    auto it = countsRes.value().find("pbi");
    REQUIRE((it != countsRes.value().end()));
    REQUIRE((it->second.size() == 1));
    CHECK((it->second.front().value == "PBI-500"));
    CHECK((it->second.front().count == 2));

    MetadataValue v2;
    v2.value = "PBI-600";
    v2.type = MetadataValueType::String;
    REQUIRE((fix.repository_->setMetadata(docBRes.value(), "pbi", v2).has_value()));

    countsRes = fix.repository_->getMetadataValueCounts({"pbi"}, opts);
    REQUIRE((countsRes.has_value()));
    it = countsRes.value().find("pbi");
    REQUIRE((it != countsRes.value().end()));
    std::unordered_map<std::string, int64_t> latest;
    for (const auto& row : it->second) {
        latest[row.value] = row.count;
    }
    REQUIRE((latest.size() == 2));
    CHECK((latest["PBI-500"] == 1));
    CHECK((latest["PBI-600"] == 1));

    auto deleteResult = fix.repository_->deleteDocument(docARes.value());
    REQUIRE((deleteResult.has_value()));

    countsRes = fix.repository_->getMetadataValueCounts({"pbi"}, opts);
    REQUIRE((countsRes.has_value()));
    it = countsRes.value().find("pbi");
    REQUIRE((it != countsRes.value().end()));
    REQUIRE((it->second.size() == 1));
    CHECK((it->second.front().value == "PBI-600"));
    CHECK((it->second.front().count == 1));
}

TEST_CASE("MetadataRepository: semantic duplicate groups round-trip",
          "[unit][metadata][repository][semantic-dedupe]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/semantic-a.txt", "semantic-a");
    auto docB = makeDocumentWithPath("/tmp/semantic-b.txt", "semantic-b");
    auto aId = fix.repository_->insertDocument(docA);
    auto bId = fix.repository_->insertDocument(docB);
    REQUIRE((aId.has_value()));
    REQUIRE((bId.has_value()));

    const auto now =
        std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());

    SemanticDuplicateGroup group;
    group.groupKey = "semantic:test-group";
    group.algorithmVersion = "semantic-dedupe-v1";
    group.status = "suggested";
    group.reviewState = "pending";
    group.canonicalDocumentId = aId.value();
    group.memberCount = 2;
    group.maxPairScore = 0.97;
    group.threshold = 0.92;
    group.evidenceJson = R"({"source":"test"})";
    group.createdAt = now;
    group.updatedAt = now;
    group.lastComputedAt = now;

    auto groupId = fix.repository_->upsertSemanticDuplicateGroup(group);
    REQUIRE((groupId.has_value()));

    SemanticDuplicateGroupMember canonical;
    canonical.documentId = aId.value();
    canonical.role = "canonical";
    canonical.decision = "keep";
    canonical.reason = "keep-newest";
    canonical.createdAt = now;
    canonical.updatedAt = now;

    SemanticDuplicateGroupMember duplicate;
    duplicate.documentId = bId.value();
    duplicate.role = "duplicate";
    duplicate.similarityToCanonical = 0.98;
    duplicate.titleOverlap = 0.75;
    duplicate.pathOverlap = 0.25;
    duplicate.pairScore = 0.93;
    duplicate.decision = "unknown";
    duplicate.createdAt = now;
    duplicate.updatedAt = now;

    REQUIRE(fix.repository_
                ->replaceSemanticDuplicateGroupMembers(groupId.value(), {canonical, duplicate})
                .has_value());

    auto fetched = fix.repository_->getSemanticDuplicateGroupByKey("semantic:test-group");
    REQUIRE((fetched.has_value()));
    REQUIRE((fetched.value().has_value()));
    CHECK((fetched.value()->canonicalDocumentId == aId.value()));
    CHECK((fetched.value()->memberCount == 2));

    auto listed = fix.repository_->listSemanticDuplicateGroups();
    REQUIRE((listed.has_value()));
    REQUIRE_FALSE(listed.value().empty());

    auto byDoc = fix.repository_->getSemanticDuplicateGroupsForDocuments(
        std::array<int64_t, 2>{aId.value(), bId.value()});
    REQUIRE((byDoc.has_value()));
    REQUIRE((byDoc.value().contains(aId.value())));
    REQUIRE((byDoc.value().contains(bId.value())));
    CHECK((byDoc.value().at(aId.value()).group.groupKey == "semantic:test-group"));
    CHECK((byDoc.value().at(aId.value()).members.size() == 2));

    REQUIRE(fix.repository_->updateSemanticDuplicateGroupStatus("semantic:test-group", "applied")
                .has_value());
    auto updated = fix.repository_->getSemanticDuplicateGroupByKey("semantic:test-group");
    REQUIRE((updated.has_value()));
    REQUIRE((updated.value().has_value()));
    CHECK((updated.value()->status == "applied"));
}

TEST_CASE("MetadataRepository: metadata value counts falls back when path FTS join fails",
          "[unit][metadata][repository][fts-fallback]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/notes/todo.md", "fts-fallback-a");
    auto docB = makeDocumentWithPath("/archive/todo-old.md", "fts-fallback-b");

    auto aId = fix.repository_->insertDocument(docA);
    auto bId = fix.repository_->insertDocument(docB);
    REQUIRE((aId.has_value()));
    REQUIRE((bId.has_value()));

    MetadataValue ownerA;
    ownerA.type = MetadataValueType::String;
    ownerA.value = "alice";
    REQUIRE((fix.repository_->setMetadata(aId.value(), "owner", ownerA).has_value()));

    MetadataValue ownerB;
    ownerB.type = MetadataValueType::String;
    ownerB.value = "bob";
    REQUIRE((fix.repository_->setMetadata(bId.value(), "owner", ownerB).has_value()));

    // Force the FTS-join query path to fail while pathFtsAvailable_ remains true in the repository.
    Database db;
    REQUIRE((db.open(fix.dbPath_.string()).has_value()));
    REQUIRE((db.execute("DROP TABLE IF EXISTS documents_path_fts").has_value()));
    db.close();

    DocumentQueryOptions fallbackOpts;
    fallbackOpts.containsFragment = "todo.md";
    fallbackOpts.containsUsesFts = true;

    auto fallbackRes = fix.repository_->getMetadataValueCounts({"owner"}, fallbackOpts);
    REQUIRE((fallbackRes.has_value()));

    DocumentQueryOptions baselineOpts = fallbackOpts;
    baselineOpts.containsUsesFts = false;
    auto baselineRes = fix.repository_->getMetadataValueCounts({"owner"}, baselineOpts);
    REQUIRE((baselineRes.has_value()));

    const auto& fallbackMap = fallbackRes.value();
    const auto& baselineMap = baselineRes.value();
    REQUIRE((fallbackMap.contains("owner")));
    REQUIRE((baselineMap.contains("owner")));
    REQUIRE((fallbackMap.at("owner").size() == baselineMap.at("owner").size()));

    std::unordered_map<std::string, int64_t> fallbackCounts;
    std::unordered_map<std::string, int64_t> baselineCounts;
    for (const auto& row : fallbackMap.at("owner")) {
        fallbackCounts[row.value] = row.count;
    }
    for (const auto& row : baselineMap.at("owner")) {
        baselineCounts[row.value] = row.count;
    }

    CHECK((fallbackCounts == baselineCounts));
    CHECK((fallbackCounts["alice"] == 1));
    CHECK_FALSE(fallbackCounts.contains("bob"));
}

TEST_CASE("MetadataRepository: get document not found", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto result = fix.repository_->getDocument(999999);
    REQUIRE((result.has_value()));
    CHECK_FALSE(result.value().has_value());
}

TEST_CASE("MetadataRepository: get document by hash", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "hash456";
    docInfo.fileName = "hash_test.txt";
    docInfo.fileSize = 2048;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));

    auto findResult = fix.repository_->getDocumentByHash("hash456");
    REQUIRE((findResult.has_value()));
    REQUIRE((findResult.value().has_value()));

    auto foundDoc = findResult.value().value();
    CHECK((foundDoc.sha256Hash == "hash456"));
    CHECK((foundDoc.fileName == "hash_test.txt"));
}

TEST_CASE("MetadataRepository: update document", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "hash789";
    docInfo.fileName = "original.txt";
    docInfo.fileSize = 2048;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    docInfo.id = docId;
    docInfo.fileName = "updated.txt";
    docInfo.fileSize = 4096;
    docInfo.modifiedTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    auto updateResult = fix.repository_->updateDocument(docInfo);
    REQUIRE((updateResult.has_value()));

    auto getResult = fix.repository_->getDocument(docId);
    REQUIRE((getResult.has_value()));
    REQUIRE((getResult.value().has_value()));

    auto updatedDoc = getResult.value().value();
    CHECK((updatedDoc.fileName == "updated.txt"));
    CHECK((updatedDoc.fileSize == 4096));
}

TEST_CASE("MetadataRepository: corpus stats path depth max shrinks after update",
          "[unit][metadata][repository][stats]") {
    MetadataRepositoryFixture fix;

    auto shallow = makeDocumentWithPath("/root.txt", "path-depth-update-shallow");
    auto deep = makeDocumentWithPath("/alpha/beta/gamma/deep.txt", "path-depth-update-deep");
    REQUIRE((deep.pathDepth > shallow.pathDepth));

    auto shallowInsert = fix.repository_->insertDocument(shallow);
    auto deepInsert = fix.repository_->insertDocument(deep);
    REQUIRE((shallowInsert.has_value()));
    REQUIRE((deepInsert.has_value()));

    auto warmStats = fix.repository_->getCorpusStats();
    REQUIRE((warmStats.has_value()));
    CHECK_THAT(warmStats.value().pathDepthMax,
               Catch::Matchers::WithinRel(static_cast<double>(deep.pathDepth), 0.01));

    deep.id = deepInsert.value();
    deep.filePath = "/flattened.txt";
    deep.fileName = "flattened.txt";
    deep.fileExtension = ".txt";
    populatePathDerivedFields(deep);
    deep.modifiedTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    REQUIRE((fix.repository_->updateDocument(deep).has_value()));

    auto afterStats = fix.repository_->getCorpusStats();
    REQUIRE((afterStats.has_value()));
    const auto expectedMaxDepth = static_cast<double>(std::max(shallow.pathDepth, deep.pathDepth));
    CHECK_THAT(afterStats.value().pathDepthMax, Catch::Matchers::WithinRel(expectedMaxDepth, 0.01));
}

TEST_CASE("MetadataRepository: query documents handles exact prefix and suffix",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath(
        "/notes/todo.md", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA");
    auto docB = makeDocumentWithPath(
        "/notes/log.txt", "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB");
    auto docC = makeDocumentWithPath(
        "/reports/summary.pdf", "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC",
        "application/pdf");

    REQUIRE((fix.repository_->insertDocument(docA).has_value()));
    REQUIRE((fix.repository_->insertDocument(docB).has_value()));
    REQUIRE((fix.repository_->insertDocument(docC).has_value()));

    SECTION("Exact path lookup") {
        DocumentQueryOptions exactOpts;
        exactOpts.exactPath = "/notes/todo.md";
        auto exactRes = fix.repository_->queryDocuments(exactOpts);
        REQUIRE((exactRes.has_value()));
        REQUIRE((exactRes.value().size() == 1));
        CHECK((exactRes.value().front().filePath == "/notes/todo.md"));
    }

    SECTION("Directory prefix with subdirectories") {
        DocumentQueryOptions prefixOpts;
        prefixOpts.pathPrefix = "/notes";
        prefixOpts.prefixIsDirectory = true;
        prefixOpts.includeSubdirectories = true;
        auto prefixRes = fix.repository_->queryDocuments(prefixOpts);
        REQUIRE((prefixRes.has_value()));
        std::vector<std::string> notePaths;
        for (const auto& d : prefixRes.value())
            notePaths.push_back(d.filePath);
        std::sort(notePaths.begin(), notePaths.end());
        REQUIRE((notePaths.size() == 2));
        CHECK((notePaths[0] == "/notes/log.txt"));
        CHECK((notePaths[1] == "/notes/todo.md"));
    }

    SECTION("Contains fragment via FTS") {
        DocumentQueryOptions containsOpts;
        containsOpts.containsFragment = "todo.md";
        containsOpts.containsUsesFts = true;
        auto containsRes = fix.repository_->queryDocuments(containsOpts);
        REQUIRE((containsRes.has_value()));
        REQUIRE((containsRes.value().size() == 1));
        CHECK((containsRes.value().front().filePath == "/notes/todo.md"));
    }

    SECTION("Contains fragment via FTS survives status updates and tracks path changes") {
        DocumentQueryOptions todoOpts;
        todoOpts.containsFragment = "todo.md";
        todoOpts.containsUsesFts = true;
        auto initialTodo = fix.repository_->queryDocuments(todoOpts);
        REQUIRE((initialTodo.has_value()));
        REQUIRE((initialTodo.value().size() == 1));
        CHECK((initialTodo.value().front().filePath == "/notes/todo.md"));

        auto docResult = fix.repository_->getDocumentByHash(docA.sha256Hash);
        REQUIRE((docResult.has_value()));
        REQUIRE((docResult.value().has_value()));
        auto updated = docResult.value().value();

        std::vector<ExtractionStatusUpdate> statusUpdates;
        statusUpdates.push_back(
            ExtractionStatusUpdate{updated.id, false, ExtractionStatus::Failed, "expected test"});
        REQUIRE(
            (fix.repository_->batchUpdateDocumentExtractionStatuses(statusUpdates).has_value()));

        auto afterStatusUpdate = fix.repository_->queryDocuments(todoOpts);
        REQUIRE((afterStatusUpdate.has_value()));
        REQUIRE((afterStatusUpdate.value().size() == 1));
        CHECK((afterStatusUpdate.value().front().filePath == "/notes/todo.md"));

        updated.filePath = "/archive/renamed.md";
        updated.fileName = "renamed.md";
        updated.fileExtension = ".md";
        updated.modifiedTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        populatePathDerivedFields(updated);
        REQUIRE((fix.repository_->updateDocument(updated).has_value()));

        auto afterRenameOld = fix.repository_->queryDocuments(todoOpts);
        REQUIRE((afterRenameOld.has_value()));
        CHECK((afterRenameOld.value().empty()));

        DocumentQueryOptions renamedOpts;
        renamedOpts.containsFragment = "renamed.md";
        renamedOpts.containsUsesFts = true;
        auto afterRenameNew = fix.repository_->queryDocuments(renamedOpts);
        REQUIRE((afterRenameNew.has_value()));
        REQUIRE((afterRenameNew.value().size() == 1));
        CHECK((afterRenameNew.value().front().filePath == "/archive/renamed.md"));
    }

    SECTION("Extension filter") {
        DocumentQueryOptions extOpts;
        extOpts.extension = ".txt";
        auto extRes = fix.repository_->queryDocuments(extOpts);
        REQUIRE((extRes.has_value()));
        REQUIRE((extRes.value().size() == 1));
        CHECK((extRes.value().front().filePath == "/notes/log.txt"));
    }
}

TEST_CASE("MetadataRepository: list projection matches queryDocuments for shared filters",
          "[unit][metadata][repository][list-projection][parity]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath(
        "/work/notes/todo.md", "1111111111111111111111111111111111111111111111111111111111111111");
    auto docB = makeDocumentWithPath(
        "/work/notes/log.txt", "2222222222222222222222222222222222222222222222222222222222222222");
    auto docC = makeDocumentWithPath(
        "/work/reports/summary.pdf",
        "3333333333333333333333333333333333333333333333333333333333333333", "application/pdf");
    auto docD =
        makeDocumentWithPath("/archive/notes/todo-old.md",
                             "4444444444444444444444444444444444444444444444444444444444444444");

    docA.fileSize = 10;
    docB.fileSize = 20;
    docC.fileSize = 30;
    docD.fileSize = 40;
    docB.extractionStatus = ExtractionStatus::Failed;
    docB.contentExtracted = false;

    auto aId = fix.repository_->insertDocument(docA);
    auto bId = fix.repository_->insertDocument(docB);
    auto cId = fix.repository_->insertDocument(docC);
    auto dId = fix.repository_->insertDocument(docD);
    REQUIRE((aId.has_value()));
    REQUIRE((bId.has_value()));
    REQUIRE((cId.has_value()));
    REQUIRE((dId.has_value()));

    auto setStringMeta = [&](int64_t docId, const std::string& key, const std::string& value) {
        MetadataValue mv;
        mv.type = MetadataValueType::String;
        mv.value = value;
        REQUIRE((fix.repository_->setMetadata(docId, key, mv).has_value()));
    };

    setStringMeta(aId.value(), "owner", "alice");
    setStringMeta(bId.value(), "owner", "alice");
    setStringMeta(cId.value(), "owner", "bob");
    setStringMeta(aId.value(), "tag:notes", "notes");
    setStringMeta(dId.value(), "tag:notes", "notes");

    auto assertParity = [&](const DocumentQueryOptions& opts, const std::string& label) {
        CAPTURE(label);
        auto docsRes = fix.repository_->queryDocuments(opts);
        auto projRes = fix.repository_->queryDocumentsForListProjection(opts);
        REQUIRE((docsRes.has_value()));
        REQUIRE((projRes.has_value()));

        const auto& docs = docsRes.value();
        const auto& projs = projRes.value();
        REQUIRE((docs.size() == projs.size()));

        for (size_t i = 0; i < docs.size(); ++i) {
            CAPTURE(i);
            CHECK((projs[i].id == docs[i].id));
            CHECK((projs[i].filePath == docs[i].filePath));
            CHECK((projs[i].fileName == docs[i].fileName));
            CHECK((projs[i].fileExtension == docs[i].fileExtension));
            CHECK((projs[i].fileSize == docs[i].fileSize));
            CHECK((projs[i].sha256Hash == docs[i].sha256Hash));
            CHECK((projs[i].mimeType == docs[i].mimeType));
            CHECK((projs[i].createdTime == docs[i].createdTime));
            CHECK((projs[i].modifiedTime == docs[i].modifiedTime));
            CHECK((projs[i].indexedTime == docs[i].indexedTime));
            CHECK((projs[i].extractionStatus == docs[i].extractionStatus));
        }
    };

    SECTION("exact path") {
        DocumentQueryOptions opts;
        opts.exactPath = "/work/notes/todo.md";
        assertParity(opts, "exactPath");
    }

    SECTION("directory prefix") {
        DocumentQueryOptions opts;
        opts.pathPrefix = "/work/notes";
        opts.prefixIsDirectory = true;
        opts.includeSubdirectories = true;
        opts.orderByNameAsc = true;
        assertParity(opts, "pathPrefix");
    }

    SECTION("contains fragment via FTS") {
        DocumentQueryOptions opts;
        opts.containsFragment = "todo";
        opts.containsUsesFts = true;
        opts.orderByNameAsc = true;
        assertParity(opts, "containsFragment");
    }

    SECTION("combined metadata and tag filter") {
        DocumentQueryOptions opts;
        opts.extension = ".md";
        opts.metadataFilters.push_back({"owner", "alice"});
        opts.tags.push_back("notes");
        opts.orderByNameAsc = true;
        assertParity(opts, "metadata+tag");
    }

    SECTION("extraction status plus pagination") {
        DocumentQueryOptions opts;
        opts.pathPrefix = "/work";
        opts.prefixIsDirectory = true;
        opts.orderByNameAsc = true;
        opts.extractionStatuses = {ExtractionStatus::Success, ExtractionStatus::Failed};
        opts.limit = 2;
        opts.offset = 1;
        assertParity(opts, "extractionStatuses+pagination");
    }
}

TEST_CASE("MetadataRepository: delete document", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "hash999";
    docInfo.fileName = "delete_me.txt";
    docInfo.fileSize = 512;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    auto deleteResult = fix.repository_->deleteDocument(docId);
    REQUIRE((deleteResult.has_value()));

    auto getResult = fix.repository_->getDocument(docId);
    REQUIRE((getResult.has_value()));
    CHECK_FALSE(getResult.value().has_value());
}

TEST_CASE("MetadataRepository: corpus stats path depth max shrinks after delete",
          "[unit][metadata][repository][stats]") {
    MetadataRepositoryFixture fix;

    auto shallow = makeDocumentWithPath("/keep.txt", "path-depth-delete-shallow");
    auto deep = makeDocumentWithPath("/one/two/three/four/deep.txt", "path-depth-delete-deep");
    REQUIRE((deep.pathDepth > shallow.pathDepth));

    auto shallowInsert = fix.repository_->insertDocument(shallow);
    auto deepInsert = fix.repository_->insertDocument(deep);
    REQUIRE((shallowInsert.has_value()));
    REQUIRE((deepInsert.has_value()));

    auto warmStats = fix.repository_->getCorpusStats();
    REQUIRE((warmStats.has_value()));
    CHECK_THAT(warmStats.value().pathDepthMax,
               Catch::Matchers::WithinRel(static_cast<double>(deep.pathDepth), 0.01));

    REQUIRE((fix.repository_->deleteDocument(deepInsert.value()).has_value()));

    auto afterStats = fix.repository_->getCorpusStats();
    REQUIRE((afterStats.has_value()));
    CHECK_THAT(afterStats.value().pathDepthMax,
               Catch::Matchers::WithinRel(static_cast<double>(shallow.pathDepth), 0.01));
}

TEST_CASE("MetadataRepository: corpus stats path depth max shrinks after batch delete",
          "[unit][metadata][repository][stats]") {
    MetadataRepositoryFixture fix;

    auto shallow = makeDocumentWithPath("/batch/keep.txt", "path-depth-batch-shallow");
    auto medium = makeDocumentWithPath("/batch/one/two/keep.txt", "path-depth-batch-medium");
    auto deep = makeDocumentWithPath("/batch/one/two/three/four/deep.txt", "path-depth-batch-deep");
    REQUIRE((deep.pathDepth > medium.pathDepth));
    REQUIRE((medium.pathDepth > shallow.pathDepth));

    auto shallowInsert = fix.repository_->insertDocument(shallow);
    auto mediumInsert = fix.repository_->insertDocument(medium);
    auto deepInsert = fix.repository_->insertDocument(deep);
    REQUIRE((shallowInsert.has_value()));
    REQUIRE((mediumInsert.has_value()));
    REQUIRE((deepInsert.has_value()));

    auto warmStats = fix.repository_->getCorpusStats();
    REQUIRE((warmStats.has_value()));
    CHECK_THAT(warmStats.value().pathDepthMax,
               Catch::Matchers::WithinRel(static_cast<double>(deep.pathDepth), 0.01));

    auto deleted = fix.repository_->deleteDocumentsBatch({deepInsert.value(), 999999});
    REQUIRE((deleted.has_value()));
    CHECK((deleted.value() == 1));

    auto afterStats = fix.repository_->getCorpusStats();
    REQUIRE((afterStats.has_value()));
    CHECK_THAT(afterStats.value().pathDepthMax,
               Catch::Matchers::WithinRel(static_cast<double>(medium.pathDepth), 0.01));
}

TEST_CASE("MetadataRepository: set and get metadata", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "metadata_test";
    docInfo.fileName = "meta.txt";
    docInfo.fileSize = 1024;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    MetadataValue value("Test Author");

    auto setResult = fix.repository_->setMetadata(docId, "author", value);
    REQUIRE((setResult.has_value()));

    auto getResult = fix.repository_->getMetadata(docId, "author");
    REQUIRE((getResult.has_value()));
    REQUIRE((getResult.value().has_value()));

    auto retrievedValue = getResult.value().value();
    CHECK((retrievedValue.asString() == "Test Author"));
}

TEST_CASE("MetadataRepository: get all metadata", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "all_metadata_test";
    docInfo.fileName = "meta_all.txt";
    docInfo.fileSize = 1024;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    MetadataValue authorValue("John Doe");
    fix.repository_->setMetadata(docId, "author", authorValue);

    MetadataValue yearValue(int64_t(2024));
    fix.repository_->setMetadata(docId, "year", yearValue);

    MetadataValue ratingValue(4.5);
    fix.repository_->setMetadata(docId, "rating", ratingValue);

    auto getAllResult = fix.repository_->getAllMetadata(docId);
    REQUIRE((getAllResult.has_value()));

    auto metadata = getAllResult.value();
    CHECK((metadata.size() == 3));

    CHECK((metadata["author"].asString() == "John Doe"));
    CHECK((metadata["year"].asInteger() == 2024));
    CHECK_THAT(metadata["rating"].asReal(), Catch::Matchers::WithinRel(4.5, 0.0001));
}

TEST_CASE("MetadataRepository: remove metadata", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "remove_metadata_test";
    docInfo.fileName = "remove.txt";
    docInfo.fileSize = 1024;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    MetadataValue value("Temporary");
    fix.repository_->setMetadata(docId, "temp", value);

    auto getResult = fix.repository_->getMetadata(docId, "temp");
    REQUIRE((getResult.has_value()));
    REQUIRE((getResult.value().has_value()));

    auto removeResult = fix.repository_->removeMetadata(docId, "temp");
    REQUIRE((removeResult.has_value()));

    getResult = fix.repository_->getMetadata(docId, "temp");
    REQUIRE((getResult.has_value()));
    CHECK_FALSE(getResult.value().has_value());
}

TEST_CASE("MetadataRepository: hasFtsEntry reports FTS5 presence correctly",
          "[unit][metadata][repository][fts5]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "fts5_entry_check";
    docInfo.fileName = "fts5check.txt";
    docInfo.fileSize = 100;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    SECTION("returns false before indexing") {
        auto hasFts = fix.repository_->hasFtsEntry(docId);
        REQUIRE((hasFts.has_value()));
        CHECK_FALSE(hasFts.value());
    }

    SECTION("returns true after indexing") {
        auto indexResult = fix.repository_->indexDocumentContent(
            docId, "Test Document", "Searchable content here", "text/plain");
        REQUIRE((indexResult.has_value()));

        auto hasFts = fix.repository_->hasFtsEntry(docId);
        REQUIRE((hasFts.has_value()));
        CHECK((hasFts.value()));
    }

    SECTION("returns false for non-existent document ID") {
        auto hasFts = fix.repository_->hasFtsEntry(999999);
        REQUIRE((hasFts.has_value()));
        CHECK_FALSE(hasFts.value());
    }
}

TEST_CASE("MetadataRepository: indexed count reflects actual FTS rows, not extraction success",
          "[unit][metadata][repository][fts5][counts]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    DocumentInfo doc;
    doc.sha256Hash = "indexed_count_fts_only_hash";
    doc.fileName = "indexed_count_fts_only.txt";
    doc.fileSize = 128;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;

    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    const auto docId = insertResult.value();

    REQUIRE(fix.repository_->updateDocumentExtractionStatus(docId, true, ExtractionStatus::Success)
                .has_value());

    auto indexedWithoutFts = fix.repository_->getIndexedDocumentCount();
    REQUIRE((indexedWithoutFts.has_value()));
    CHECK((indexedWithoutFts.value() == 0));
    CHECK((fix.repository_->getCachedIndexedCount() == 0));
    CHECK((fix.repository_->getCachedExtractedCount() == 1));

    DocumentContent content;
    content.documentId = docId;
    content.contentText = "fts count probe";
    content.contentLength = static_cast<int64_t>(content.contentText.size());
    content.extractionMethod = "test";
    content.language = "en";
    REQUIRE((fix.repository_->insertContent(content).has_value()));
    REQUIRE(fix.repository_
                ->indexDocumentContent(docId, doc.fileName, content.contentText, "text/plain")
                .has_value());

    auto indexedWithFts = fix.repository_->getIndexedDocumentCount();
    REQUIRE((indexedWithFts.has_value()));
    CHECK((indexedWithFts.value() == 1));
    CHECK((fix.repository_->getCachedIndexedCount() == 1));
}

TEST_CASE("MetadataRepository: batch extraction status updates share one transaction",
          "[unit][metadata][repository][extraction]") {
    MetadataRepositoryFixture fix;

    auto pendingDoc = makeDocumentWithPath("/tmp/yams/batch-extraction-pending.txt",
                                           "batch-extraction-pending-hash");
    pendingDoc.contentExtracted = false;
    pendingDoc.extractionStatus = ExtractionStatus::Pending;
    auto pendingId = fix.repository_->insertDocument(pendingDoc);
    REQUIRE((pendingId.has_value()));

    auto extractedDoc = makeDocumentWithPath("/tmp/yams/batch-extraction-extracted.txt",
                                             "batch-extraction-extracted-hash");
    extractedDoc.contentExtracted = true;
    extractedDoc.extractionStatus = ExtractionStatus::Success;
    auto extractedId = fix.repository_->insertDocument(extractedDoc);
    REQUIRE((extractedId.has_value()));

    fix.repository_->initializeCounters();
    CHECK((fix.repository_->getCachedExtractedCount() == 1));

    std::vector<ExtractionStatusUpdate> updates;
    updates.push_back(
        ExtractionStatusUpdate{pendingId.value(), true, ExtractionStatus::Success, std::string{}});
    updates.push_back(ExtractionStatusUpdate{extractedId.value(), false, ExtractionStatus::Failed,
                                             "extraction failed"});
    updates.push_back(
        ExtractionStatusUpdate{999999, true, ExtractionStatus::Success, "missing ignored"});

    auto result = fix.repository_->batchUpdateDocumentExtractionStatuses(updates);
    REQUIRE((result.has_value()));

    auto pendingAfter = fix.repository_->getDocument(pendingId.value());
    REQUIRE((pendingAfter.has_value()));
    REQUIRE((pendingAfter.value().has_value()));
    CHECK((pendingAfter.value()->contentExtracted));
    CHECK((pendingAfter.value()->extractionStatus == ExtractionStatus::Success));
    CHECK((pendingAfter.value()->extractionError.empty()));

    auto extractedAfter = fix.repository_->getDocument(extractedId.value());
    REQUIRE((extractedAfter.has_value()));
    REQUIRE((extractedAfter.value().has_value()));
    CHECK_FALSE(extractedAfter.value()->contentExtracted);
    CHECK((extractedAfter.value()->extractionStatus == ExtractionStatus::Failed));
    CHECK((extractedAfter.value()->extractionError == "extraction failed"));
    CHECK((fix.repository_->getCachedExtractedCount() == 1));
}

TEST_CASE("MetadataRepository: batch extraction status updates use the last update per document",
          "[unit][metadata][repository][extraction]") {
    MetadataRepositoryFixture fix;

    auto pendingDoc =
        makeDocumentWithPath("/tmp/yams/extraction-duplicate-pending.txt", "extract-dup-pending");
    pendingDoc.contentExtracted = false;
    pendingDoc.extractionStatus = ExtractionStatus::Pending;

    auto successDoc =
        makeDocumentWithPath("/tmp/yams/extraction-duplicate-success.txt", "extract-dup-success");
    successDoc.contentExtracted = true;
    successDoc.extractionStatus = ExtractionStatus::Success;

    auto pendingId = fix.repository_->insertDocument(pendingDoc);
    auto successId = fix.repository_->insertDocument(successDoc);
    REQUIRE((pendingId.has_value()));
    REQUIRE((successId.has_value()));

    fix.repository_->initializeCounters();
    CHECK((fix.repository_->getCachedExtractedCount() == 1));

    std::vector<ExtractionStatusUpdate> updates;
    updates.push_back(
        ExtractionStatusUpdate{pendingId.value(), true, ExtractionStatus::Success, std::string{}});
    updates.push_back(ExtractionStatusUpdate{pendingId.value(), false, ExtractionStatus::Failed,
                                             "duplicate failure wins"});
    updates.push_back(
        ExtractionStatusUpdate{successId.value(), false, ExtractionStatus::Failed, "transient"});
    updates.push_back(
        ExtractionStatusUpdate{successId.value(), true, ExtractionStatus::Success, std::string{}});

    REQUIRE((fix.repository_->batchUpdateDocumentExtractionStatuses(updates).has_value()));

    auto pendingAfter = fix.repository_->getDocument(pendingId.value());
    auto successAfter = fix.repository_->getDocument(successId.value());
    REQUIRE((pendingAfter.has_value()));
    REQUIRE((successAfter.has_value()));
    REQUIRE((pendingAfter.value().has_value()));
    REQUIRE((successAfter.value().has_value()));

    CHECK_FALSE(pendingAfter.value()->contentExtracted);
    CHECK((pendingAfter.value()->extractionStatus == ExtractionStatus::Failed));
    CHECK((pendingAfter.value()->extractionError == "duplicate failure wins"));

    CHECK((successAfter.value()->contentExtracted));
    CHECK((successAfter.value()->extractionStatus == ExtractionStatus::Success));
    CHECK((successAfter.value()->extractionError.empty()));
    CHECK((fix.repository_->getCachedExtractedCount() == 1));
}

TEST_CASE(
    "MetadataRepository: extraction status counters handle empty, negative, and missing updates",
    "[unit][metadata][repository][extraction]") {
    MetadataRepositoryFixture fix;

    auto pendingDoc =
        makeDocumentWithPath("/tmp/yams/extraction-count-pending.txt", "extract-count-pending");
    pendingDoc.contentExtracted = false;
    pendingDoc.extractionStatus = ExtractionStatus::Pending;

    auto failedDoc =
        makeDocumentWithPath("/tmp/yams/extraction-count-failed.txt", "extract-count-failed");
    failedDoc.contentExtracted = false;
    failedDoc.extractionStatus = ExtractionStatus::Failed;

    auto successDoc =
        makeDocumentWithPath("/tmp/yams/extraction-count-success.txt", "extract-count-success");
    successDoc.contentExtracted = true;
    successDoc.extractionStatus = ExtractionStatus::Success;

    auto pendingId = fix.repository_->insertDocument(pendingDoc);
    auto failedId = fix.repository_->insertDocument(failedDoc);
    auto successId = fix.repository_->insertDocument(successDoc);
    REQUIRE((pendingId.has_value()));
    REQUIRE((failedId.has_value()));
    REQUIRE((successId.has_value()));

    auto pendingCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Pending);
    auto failedCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Failed);
    auto successCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Success);
    auto skippedCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Skipped);
    REQUIRE((pendingCount.has_value()));
    REQUIRE((failedCount.has_value()));
    REQUIRE((successCount.has_value()));
    REQUIRE((skippedCount.has_value()));
    CHECK((pendingCount.value() == 1));
    CHECK((failedCount.value() == 1));
    CHECK((successCount.value() == 1));
    CHECK((skippedCount.value() == 0));

    fix.repository_->initializeCounters();
    CHECK((fix.repository_->getCachedExtractedCount() == 1));

    REQUIRE((fix.repository_->batchUpdateDocumentExtractionStatuses({}).has_value()));
    CHECK((fix.repository_->getCachedExtractedCount() == 1));

    std::vector<ExtractionStatusUpdate> updates;
    updates.push_back(
        ExtractionStatusUpdate{-7, true, ExtractionStatus::Success, "ignored negative"});
    updates.push_back(
        ExtractionStatusUpdate{999999, true, ExtractionStatus::Success, "ignored missing"});
    updates.push_back(
        ExtractionStatusUpdate{pendingId.value(), true, ExtractionStatus::Success, std::string{}});
    updates.push_back(
        ExtractionStatusUpdate{failedId.value(), false, ExtractionStatus::Failed, "still failed"});

    REQUIRE((fix.repository_->batchUpdateDocumentExtractionStatuses(updates).has_value()));

    auto pendingAfter = fix.repository_->getDocument(pendingId.value());
    auto failedAfter = fix.repository_->getDocument(failedId.value());
    auto successAfter = fix.repository_->getDocument(successId.value());
    REQUIRE((pendingAfter.has_value()));
    REQUIRE((failedAfter.has_value()));
    REQUIRE((successAfter.has_value()));
    REQUIRE((pendingAfter.value().has_value()));
    REQUIRE((failedAfter.value().has_value()));
    REQUIRE((successAfter.value().has_value()));
    CHECK((pendingAfter.value()->contentExtracted));
    CHECK((pendingAfter.value()->extractionStatus == ExtractionStatus::Success));
    CHECK((pendingAfter.value()->extractionError.empty()));
    CHECK_FALSE(failedAfter.value()->contentExtracted);
    CHECK((failedAfter.value()->extractionStatus == ExtractionStatus::Failed));
    CHECK((failedAfter.value()->extractionError == "still failed"));
    CHECK((successAfter.value()->contentExtracted));
    CHECK((fix.repository_->getCachedExtractedCount() == 2));

    auto pendingAfterCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Pending);
    auto failedAfterCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Failed);
    auto successAfterCount =
        fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Success);
    REQUIRE((pendingAfterCount.has_value()));
    REQUIRE((failedAfterCount.has_value()));
    REQUIRE((successAfterCount.has_value()));
    CHECK((pendingAfterCount.value() == 0));
    CHECK((failedAfterCount.value() == 1));
    CHECK((successAfterCount.value() == 2));

    auto stats = fix.repository_->getCorpusStats();
    REQUIRE((stats.has_value()));
    CHECK((stats.value().contentExtractedCount == 2));
}

TEST_CASE("MetadataRepository: single extraction status update does not amplify lock retries",
          "[unit][metadata][repository][extraction][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_single_extraction_lock_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc =
        makeDocumentWithPath("/tmp/single-extraction-lock.txt", "single-extraction-lock-hash");
    doc.contentExtracted = false;
    doc.extractionStatus = ExtractionStatus::Pending;
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    const auto start = std::chrono::steady_clock::now();
    auto result = repository->updateDocumentExtractionStatus(
        insert.value(), true, ExtractionStatus::Success, std::string{});
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((result.error().message.find("locked") != std::string::npos));
    CHECK((elapsed < std::chrono::milliseconds(3000)));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataRepository: batch extraction status update does not amplify lock retries",
          "[unit][metadata][repository][extraction][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_extraction_lock_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc = makeDocumentWithPath("/tmp/extraction-lock.txt", "extraction-lock-hash");
    doc.contentExtracted = false;
    doc.extractionStatus = ExtractionStatus::Pending;
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    const auto start = std::chrono::steady_clock::now();
    auto result = repository->batchUpdateDocumentExtractionStatuses(
        {{insert.value(), true, ExtractionStatus::Success, std::string{}}});
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((result.error().message.find("locked") != std::string::npos));
    CHECK((elapsed < std::chrono::milliseconds(3000)));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataRepository: initializeCounters keeps extracted and indexed counts distinct",
          "[unit][metadata][repository][fts5][counters]") {
    MetadataRepositoryFixture fix;

    DocumentInfo doc;
    doc.sha256Hash = "counter_distinct_hash";
    doc.fileName = "counter_distinct.txt";
    doc.fileSize = 64;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;

    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    const auto docId = insertResult.value();

    REQUIRE(fix.repository_->updateDocumentExtractionStatus(docId, true, ExtractionStatus::Success)
                .has_value());

    fix.repository_->initializeCounters();
    CHECK((fix.repository_->getCachedExtractedCount() == 1));
    CHECK((fix.repository_->getCachedIndexedCount() == 0));

    DocumentContent content;
    content.documentId = docId;
    content.contentText = "counter refresh content";
    content.contentLength = static_cast<int64_t>(content.contentText.size());
    content.extractionMethod = "test";
    content.language = "en";
    REQUIRE((fix.repository_->insertContent(content).has_value()));
    REQUIRE(fix.repository_
                ->indexDocumentContent(docId, doc.fileName, content.contentText, "text/plain")
                .has_value());

    auto exactIndexed = fix.repository_->getIndexedDocumentCount();
    REQUIRE((exactIndexed.has_value()));
    CHECK((exactIndexed.value() == 1));
    CHECK((fix.repository_->getCachedExtractedCount() == 1));
}

TEST_CASE("MetadataRepository: getFts5IndexedRowIdSet returns indexed document IDs",
          "[unit][metadata][repository][fts5]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    // Insert two documents
    DocumentInfo doc1;
    doc1.sha256Hash = "fts5_set_doc1";
    doc1.fileName = "doc1.txt";
    doc1.fileSize = 50;
    doc1.mimeType = "text/plain";
    doc1.createdTime = now;
    doc1.modifiedTime = now;
    auto id1Res = fix.repository_->insertDocument(doc1);
    REQUIRE((id1Res.has_value()));
    auto docId1 = id1Res.value();

    DocumentInfo doc2;
    doc2.sha256Hash = "fts5_set_doc2";
    doc2.fileName = "doc2.txt";
    doc2.fileSize = 60;
    doc2.mimeType = "text/plain";
    doc2.createdTime = now;
    doc2.modifiedTime = now;
    auto id2Res = fix.repository_->insertDocument(doc2);
    REQUIRE((id2Res.has_value()));
    auto docId2 = id2Res.value();

    // Before indexing: set should be empty
    auto setBefore = fix.repository_->getFts5IndexedRowIdSet();
    REQUIRE((setBefore.has_value()));
    CHECK((setBefore.value().count(docId1) == 0));
    CHECK((setBefore.value().count(docId2) == 0));

    // Index only doc1
    auto indexResult = fix.repository_->indexDocumentContent(
        docId1, "Doc One", "First document content", "text/plain");
    REQUIRE((indexResult.has_value()));

    // Set should contain only doc1
    auto setAfter = fix.repository_->getFts5IndexedRowIdSet();
    REQUIRE((setAfter.has_value()));
    CHECK((setAfter.value().count(docId1) == 1));
    CHECK((setAfter.value().count(docId2) == 0));

    // Index doc2
    auto indexResult2 = fix.repository_->indexDocumentContent(
        docId2, "Doc Two", "Second document content", "text/plain");
    REQUIRE((indexResult2.has_value()));

    // Set should contain both
    auto setFinal = fix.repository_->getFts5IndexedRowIdSet();
    REQUIRE((setFinal.has_value()));
    CHECK((setFinal.value().count(docId1) == 1));
    CHECK((setFinal.value().count(docId2) == 1));
}

TEST_CASE("MetadataRepository: search functionality", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "search_test";
    docInfo.fileName = "search.txt";
    docInfo.fileSize = 1024;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    auto indexResult = fix.repository_->indexDocumentContent(
        docId, "Test Document", "This is a test document with searchable content", "text/plain");
    REQUIRE((indexResult.has_value()));

    auto searchResult = fix.repository_->search("test", 10, 0);
    REQUIRE((searchResult.has_value()));

    auto results = searchResult.value();
    CHECK((results.results.size() > 0));
}

TEST_CASE("MetadataRepository: search totalCount honors docIds filter",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto doc1 = makeDocumentWithPath("/tmp/doc_count_1.txt", "search_doc_count_1");
    auto doc2 = makeDocumentWithPath("/tmp/doc_count_2.txt", "search_doc_count_2");

    auto insert1 = fix.repository_->insertDocument(doc1);
    auto insert2 = fix.repository_->insertDocument(doc2);
    REQUIRE((insert1.has_value()));
    REQUIRE((insert2.has_value()));

    auto docId1 = insert1.value();
    auto docId2 = insert2.value();

    REQUIRE(fix.repository_
                ->indexDocumentContent(docId1, "Doc One", "needle token appears in doc one",
                                       "text/plain")
                .has_value());
    REQUIRE(fix.repository_
                ->indexDocumentContent(docId2, "Doc Two", "needle token appears in doc two",
                                       "text/plain")
                .has_value());

    const std::optional<std::vector<int64_t>> scopedDocIds = std::vector<int64_t>{docId1};
    auto searchResult = fix.repository_->search("needle", 1, 0, scopedDocIds);
    REQUIRE((searchResult.has_value()));

    const auto& results = searchResult.value();
    REQUIRE((results.results.size() == 1));
    CHECK((results.results.front().document.id == docId1));
    CHECK((results.totalCount == 1));
}

TEST_CASE("MetadataRepository: fuzzy search returns content matches",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath(
        "/notes/fuzzy_content.txt",
        "DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD");

    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    constexpr char kRareTerm[] = "blorptastic";
    std::string contentText = std::string("The ") + kRareTerm;

    DocumentContent content;
    content.documentId = docId;
    content.contentText = contentText;
    content.contentLength = static_cast<int64_t>(contentText.length());
    content.extractionMethod = "test";
    content.language = "en";
    auto contentInsertResult = fix.repository_->insertContent(content);
    REQUIRE((contentInsertResult.has_value()));

    auto indexResult =
        fix.repository_->indexDocumentContent(docId, doc.fileName, contentText, "text/plain");
    REQUIRE((indexResult.has_value()));

    auto fuzzyResult = fix.repository_->fuzzySearch(kRareTerm, 0.6f, 10);
    REQUIRE((fuzzyResult.has_value()));

    const auto& searchResults = fuzzyResult.value();
    REQUIRE((searchResults.results.size() == 1));
    CHECK((searchResults.results.front().document.id == docId));
}

TEST_CASE("MetadataRepository: fuzzy search honors minSimilarity threshold",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto exactDoc = makeDocumentWithPath(
        "/notes/fuzzy_threshold_exact.txt",
        "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE1");
    auto nearDoc = makeDocumentWithPath(
        "/notes/fuzzy_threshold_near.txt",
        "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE2");

    auto exactInsert = fix.repository_->insertDocument(exactDoc);
    REQUIRE((exactInsert.has_value()));
    auto nearInsert = fix.repository_->insertDocument(nearDoc);
    REQUIRE((nearInsert.has_value()));

    constexpr char kExactTerm[] = "fuzzthresholdtarget";
    constexpr char kNearTerm[] = "fuzzthresholtarget";
    REQUIRE(
        fix.repository_
            ->indexDocumentContent(exactInsert.value(), exactDoc.fileName, kExactTerm, "text/plain")
            .has_value());
    REQUIRE(
        fix.repository_
            ->indexDocumentContent(nearInsert.value(), nearDoc.fileName, kNearTerm, "text/plain")
            .has_value());
    REQUIRE(
        (fix.repository_
             ->tryAddSymSpellTerms({std::pair<std::string, int64_t>{kExactTerm, 1}, {kNearTerm, 1}})
             .has_value()));

    auto lowThreshold = fix.repository_->fuzzySearch(kExactTerm, 0.0f, 10);
    REQUIRE((lowThreshold.has_value()));
    auto highThreshold = fix.repository_->fuzzySearch(kExactTerm, 1.0f, 10);
    REQUIRE((highThreshold.has_value()));

    auto hasDocId = [](const SearchResults& results, int64_t docId) {
        return std::ranges::any_of(results.results, [docId](const SearchResult& result) {
            return result.document.id == docId;
        });
    };

    CHECK((hasDocId(lowThreshold.value(), exactInsert.value())));
    CHECK((hasDocId(lowThreshold.value(), nearInsert.value())));
    CHECK((hasDocId(highThreshold.value(), exactInsert.value())));
    CHECK_FALSE((hasDocId(highThreshold.value(), nearInsert.value())));
    CHECK((lowThreshold.value().results.size() >= highThreshold.value().results.size()));
}

TEST_CASE("MetadataRepository: keyword search does not implicitly run fuzzy fallback",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath(
        "/notes/keyword_no_fuzzy_fallback.txt",
        "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE");

    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    constexpr char kIndexedTerm[] = "blorptastic";
    std::string contentText = std::string("The ") + kIndexedTerm;

    DocumentContent content;
    content.documentId = docId;
    content.contentText = contentText;
    content.contentLength = static_cast<int64_t>(contentText.length());
    content.extractionMethod = "test";
    content.language = "en";
    REQUIRE((fix.repository_->insertContent(content).has_value()));

    REQUIRE(fix.repository_->indexDocumentContent(docId, doc.fileName, contentText, "text/plain")
                .has_value());
    fix.repository_->addSymSpellTerm(kIndexedTerm, 1);

    auto fuzzyResult = fix.repository_->fuzzySearch("blorptastik", 0.6f, 10);
    REQUIRE((fuzzyResult.has_value()));
    REQUIRE((fuzzyResult.value().results.size() == 1));
    CHECK((fuzzyResult.value().results.front().document.id == docId));

    auto keywordResult = fix.repository_->search("blorptastik", 10);
    REQUIRE((keywordResult.has_value()));
    CHECK((keywordResult.value().results.empty()));
    CHECK((keywordResult.value().totalCount == 0));
}

TEST_CASE(
    "MetadataRepository: concurrent SymSpell writes use coordinator without blocking fuzzy reads",
    "[unit][metadata][repository][symspell]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath(
        "/notes/symspell_lock_storm.txt",
        "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEF");
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    const auto docId = insertResult.value();

    const std::string indexedTerm = "symstormtarget";
    const std::string contentText = std::string("The ") + indexedTerm;
    REQUIRE((fix.repository_->indexDocumentContent(docId, doc.fileName, contentText, "text/plain")
                 .has_value()));

    boost::asio::io_context io;
    yams::daemon::WriteCoordinator::Config config;
    config.maxBatchSize = 8;
    config.maxBatchDelayMs = std::chrono::milliseconds{1};
    config.channelCapacity = 512;
    auto repoRef =
        std::shared_ptr<MetadataRepository>(fix.repository_.get(), [](MetadataRepository*) {});
    yams::daemon::WriteCoordinator coordinator(io, {}, repoRef, config);
    coordinator.start();
    std::thread writerLoop([&io] { io.run(); });

    std::atomic<bool> failed{false};
    std::vector<std::thread> workers;
    for (int worker = 0; worker < 4; ++worker) {
        workers.emplace_back([&coordinator, worker, indexedTerm] {
            for (int i = 0; i < 50; ++i) {
                (void)worker;
                (void)i;
                coordinator.enqueue(makeSymSpellBatch("test/symspell_lock_storm", indexedTerm));
            }
        });
    }
    for (int worker = 0; worker < 4; ++worker) {
        workers.emplace_back([&] {
            for (int i = 0; i < 50; ++i) {
                auto result = fix.repository_->fuzzySearch(indexedTerm, 0.6f, 5);
                if (!result || result.value().results.empty()) {
                    failed.store(true, std::memory_order_relaxed);
                    return;
                }
            }
        });
    }

    for (auto& worker : workers) {
        worker.join();
    }

    auto flushResult = coordinator.flush(std::chrono::seconds{10});
    coordinator.shutdown();
    io.stop();
    if (writerLoop.joinable()) {
        writerLoop.join();
    }

    REQUIRE((flushResult.has_value()));
    CHECK_FALSE((failed.load(std::memory_order_relaxed)));
    CHECK((coordinator.getStats().symSpellTermsAdded >= 200));

    auto fuzzyResult = fix.repository_->fuzzySearch("symstormtargat", 0.6f, 10);
    REQUIRE((fuzzyResult.has_value()));
    REQUIRE((fuzzyResult.value().results.size() == 1));
    CHECK((fuzzyResult.value().results.front().document.id == docId));
}

TEST_CASE("MetadataRepository: coordinator coalesces duplicate SymSpell terms preserving "
          "frequency",
          "[unit][metadata][repository][symspell]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath(
        "/notes/symspell_coalesce.txt",
        "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE70");
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    const auto docId = insertResult.value();

    const std::string indexedTerm = "coalescestormtarget";
    const std::string contentText = std::string("The ") + indexedTerm;
    REQUIRE((fix.repository_->indexDocumentContent(docId, doc.fileName, contentText, "text/plain")
                 .has_value()));

    boost::asio::io_context io;
    yams::daemon::WriteCoordinator::Config config;
    config.maxBatchSize = 8;
    config.maxBatchDelayMs = std::chrono::milliseconds{1};
    config.channelCapacity = 512;
    auto repoRef =
        std::shared_ptr<MetadataRepository>(fix.repository_.get(), [](MetadataRepository*) {});
    yams::daemon::WriteCoordinator coordinator(io, {}, repoRef, config);
    coordinator.start();
    std::thread writerLoop([&io] { io.run(); });

    constexpr std::uint64_t kDuplicateBatches = 300;
    for (std::uint64_t i = 0; i < kDuplicateBatches; ++i) {
        coordinator.enqueue(makeSymSpellBatch("test/symspell_coalesce", indexedTerm));
    }

    auto flushResult = coordinator.flush(std::chrono::seconds{10});
    coordinator.shutdown();
    io.stop();
    if (writerLoop.joinable()) {
        writerLoop.join();
    }

    REQUIRE((flushResult.has_value()));
    CHECK((coordinator.getStats().symSpellTermsAdded == kDuplicateBatches));

    auto fuzzyResult = fix.repository_->fuzzySearch("coalescestormtargat", 0.6f, 10);
    REQUIRE((fuzzyResult.has_value()));
    REQUIRE((fuzzyResult.value().results.size() == 1));
    CHECK((fuzzyResult.value().results.front().document.id == docId));
}

TEST_CASE("MetadataRepository: SymSpell terms spanning write chunk boundaries are all indexed",
          "[unit][metadata][repository][symspell]") {
    MetadataRepositoryFixture fix;

    struct Target {
        std::string term;
        std::string typo;
        std::string path;
        std::string hashSuffix;
        int64_t docId = 0;
    };
    std::vector<Target> targets = {
        {"alphachunkzero", "alphachunkzera", "/notes/chunk_alpha.txt", "71", 0},
        {"betachunkmiddle", "betachunkmiddel", "/notes/chunk_beta.txt", "72", 0},
        {"gammachunklast", "gammachunklasd", "/notes/chunk_gamma.txt", "73", 0},
    };

    for (auto& target : targets) {
        auto doc = makeDocumentWithPath(
            target.path,
            std::string("EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE") +
                target.hashSuffix);
        auto insertResult = fix.repository_->insertDocument(doc);
        REQUIRE((insertResult.has_value()));
        target.docId = insertResult.value();
        const std::string contentText = std::string("The ") + target.term;
        REQUIRE((fix.repository_
                     ->indexDocumentContent(target.docId, doc.fileName, contentText, "text/plain")
                     .has_value()));
    }

    constexpr std::size_t kTotalTerms = 520;
    std::vector<std::pair<std::string, int64_t>> termPairs;
    termPairs.reserve(kTotalTerms);
    for (std::size_t i = 0; i < kTotalTerms; ++i) {
        termPairs.emplace_back("fillerchunkterm" + std::to_string(i), 1);
    }
    termPairs[0] = {targets[0].term, 1};
    termPairs[511] = {targets[1].term, 1};
    termPairs[519] = {targets[2].term, 1};

    auto addTermsResult = fix.repository_->tryAddSymSpellTerms(termPairs);
    REQUIRE((addTermsResult.has_value()));

    for (const auto& target : targets) {
        auto fuzzyResult = fix.repository_->fuzzySearch(target.typo, 0.6f, 10);
        REQUIRE((fuzzyResult.has_value()));
        REQUIRE((fuzzyResult.value().results.size() == 1));
        CHECK((fuzzyResult.value().results.front().document.id == target.docId));
    }
}

TEST_CASE("MetadataRepository: search sanitizes snippet UTF-8", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    DocumentInfo docInfo;
    docInfo.sha256Hash = "search_bad_utf8";
    docInfo.fileName = "bad_utf8.txt";
    docInfo.fileSize = 32;
    docInfo.mimeType = "text/plain";
    docInfo.createdTime =
        std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    docInfo.modifiedTime = docInfo.createdTime;

    auto createResult = fix.repository_->insertDocument(docInfo);
    REQUIRE((createResult.has_value()));
    auto docId = createResult.value();

    std::string badContent = "alpha";
    badContent.push_back(static_cast<char>(0xFF));
    badContent += "beta";

    auto indexResult =
        fix.repository_->indexDocumentContent(docId, "Bad UTF-8", badContent, "text/plain");
    REQUIRE((indexResult.has_value()));

    auto searchResult = fix.repository_->search("alpha", 10, 0);
    REQUIRE((searchResult.has_value()));
    const auto& results = searchResult.value();
    REQUIRE_FALSE(results.results.empty());

    bool matchedDocument = false;
    for (const auto& item : results.results) {
        if (item.document.sha256Hash == "search_bad_utf8") {
            matchedDocument = true;
            auto sanitized = common::sanitizeUtf8(item.snippet);
            CHECK((item.snippet == sanitized));
            CHECK((item.snippet.find(static_cast<char>(0xFF)) == std::string::npos));
        }
    }

    CHECK((matchedDocument));
}

TEST_CASE("MetadataRepository: counts and modified since", "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    using clock = std::chrono::system_clock;
    auto now = std::chrono::floor<std::chrono::seconds>(clock::now());

    auto d1 = makeDocumentWithPath("/tmp/a.txt", "H-A");
    d1.extractionStatus = ExtractionStatus::Success;
    d1.contentExtracted = true;
    d1.modifiedTime = now - std::chrono::hours(2);

    auto d2 = makeDocumentWithPath("/tmp/b.txt", "H-B");
    d2.extractionStatus = ExtractionStatus::Pending;
    d2.contentExtracted = false;
    d2.modifiedTime = now - std::chrono::seconds(10);

    auto d3 = makeDocumentWithPath("/tmp/c.txt", "H-C");
    d3.extractionStatus = ExtractionStatus::Failed;
    d3.contentExtracted = true;
    d3.modifiedTime = now - std::chrono::hours(3);

    REQUIRE((fix.repository_->insertDocument(d1).has_value()));
    REQUIRE((fix.repository_->insertDocument(d2).has_value()));
    REQUIRE((fix.repository_->insertDocument(d3).has_value()));

    auto total = fix.repository_->getDocumentCount();
    REQUIRE((total.has_value()));
    CHECK((total.value() >= 3));

    auto extracted = fix.repository_->getContentExtractedDocumentCount();
    REQUIRE((extracted.has_value()));
    CHECK((extracted.value() >= 2));

    auto c_pending = fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Pending);
    auto c_success = fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Success);
    auto c_failed = fix.repository_->getDocumentCountByExtractionStatus(ExtractionStatus::Failed);
    REQUIRE((c_pending.has_value()));
    REQUIRE((c_success.has_value()));
    REQUIRE((c_failed.has_value()));
    CHECK((c_pending.value() >= 1));
    CHECK((c_success.value() >= 1));
    CHECK((c_failed.value() >= 1));

    auto since = clock::now() - std::chrono::seconds(60);
    auto modifiedRes = fix.repository_->findDocumentsModifiedSince(since);
    REQUIRE((modifiedRes.has_value()));
    bool foundRecent = false;
    for (const auto& doc : modifiedRes.value()) {
        if (doc.sha256Hash == d2.sha256Hash) {
            foundRecent = true;
            break;
        }
    }
    CHECK((foundRecent));
}

TEST_CASE("MetadataRepository: path tree upsert creates nodes and counts",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto docInfo = makeDocumentWithPath(
        "/src/example.txt", "DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD");
    auto docInsert = fix.repository_->insertDocument(docInfo);
    REQUIRE((docInsert.has_value()));
    auto docId = docInsert.value();
    docInfo.id = docId;

    auto upsert = fix.repository_->upsertPathTreeForDocument(docInfo, docId, true, {});
    REQUIRE((upsert.has_value()));

    auto lookup = fix.repository_->findPathTreeNode(PathTreeNode::kNullParent, "src");
    REQUIRE((lookup.has_value()));
    REQUIRE((lookup.value().has_value()));
    auto node = lookup.value().value();
    CHECK((node.fullPath == "/src"));
    CHECK((node.docCount == 1));

    auto repeat = fix.repository_->upsertPathTreeForDocument(docInfo, docId, false, {});
    REQUIRE((repeat.has_value()));
    auto afterRepeat = fix.repository_->findPathTreeNode(PathTreeNode::kNullParent, "src");
    REQUIRE((afterRepeat.has_value()));
    REQUIRE((afterRepeat.value().has_value()));
    CHECK((afterRepeat.value()->docCount == 1));

    auto fullLookup = fix.repository_->findPathTreeNodeByFullPath("/src/example.txt");
    REQUIRE((fullLookup.has_value()));
    REQUIRE((fullLookup.value().has_value()));
    CHECK((fullLookup.value()->fullPath == "/src/example.txt"));
}

TEST_CASE("MetadataRepository: path tree centroid accumulates embeddings",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto docInfo = makeDocumentWithPath(
        "/src/lib/foo.cpp", "EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE");
    auto docInsert = fix.repository_->insertDocument(docInfo);
    REQUIRE((docInsert.has_value()));
    auto docId = docInsert.value();
    docInfo.id = docId;

    REQUIRE((fix.repository_->upsertPathTreeForDocument(docInfo, docId, true, {}).has_value()));

    std::vector<float> embedding{1.0F, 2.0F, 3.0F};
    REQUIRE(
        fix.repository_
            ->upsertPathTreeForDocument(docInfo, docId, false,
                                        std::span<const float>(embedding.data(), embedding.size()))
            .has_value());

    auto node = fix.repository_->findPathTreeNode(PathTreeNode::kNullParent, "src");
    REQUIRE((node.has_value()));
    REQUIRE((node.value().has_value()));
    CHECK((node.value()->centroidWeight == 1));
}

TEST_CASE("MetadataRepository: path tree upsert rolls back partial changes on failure",
          "[unit][metadata][repository][path-tree][atomicity]") {
    MetadataRepositoryFixture fix;

    auto docInfo = makeDocumentWithPath(
        "/atomic/lib/file.txt", "ABABABABABABABABABABABABABABABABABABABABABABABABABABABABABABABAB");
    auto docInsert = fix.repository_->insertDocument(docInfo);
    REQUIRE((docInsert.has_value()));
    auto docId = docInsert.value();
    docInfo.id = docId;

    auto installTrigger = fix.pool_->withConnection([&](Database& db) -> Result<void> {
        return db.execute("CREATE TRIGGER abort_path_tree_upsert_leaf_centroid "
                          "BEFORE UPDATE OF centroid_weight ON path_tree_nodes "
                          "WHEN NEW.full_path = '/atomic/lib/file.txt' "
                          "AND NEW.centroid_weight = OLD.centroid_weight + 1 "
                          "BEGIN "
                          "SELECT RAISE(ABORT, 'injected path tree upsert failure'); "
                          "END");
    });
    REQUIRE((installTrigger.has_value()));

    std::vector<float> embedding{1.0F, 2.0F, 3.0F};
    auto upsert = fix.repository_->upsertPathTreeForDocument(
        docInfo, docId, true, std::span<const float>(embedding.data(), embedding.size()));
    REQUIRE_FALSE((upsert.has_value()));

    auto rootNode = fix.repository_->findPathTreeNodeByFullPath("/atomic");
    REQUIRE((rootNode.has_value()));
    CHECK_FALSE((rootNode.value().has_value()));

    auto libNode = fix.repository_->findPathTreeNodeByFullPath("/atomic/lib");
    REQUIRE((libNode.has_value()));
    CHECK_FALSE((libNode.value().has_value()));

    auto leafNode = fix.repository_->findPathTreeNodeByFullPath("/atomic/lib/file.txt");
    REQUIRE((leafNode.has_value()));
    CHECK_FALSE((leafNode.value().has_value()));

    auto assocCount = fix.pool_->withConnection([&](Database& db) -> Result<int64_t> {
        auto stmtResult =
            db.prepare("SELECT COUNT(*) FROM path_tree_node_documents WHERE document_id = ?");
        if (!stmtResult) {
            return stmtResult.error();
        }

        auto stmt = std::move(stmtResult).value();
        if (auto bindDoc = stmt.bind(1, docId); !bindDoc) {
            return bindDoc.error();
        }

        auto stepResult = stmt.step();
        if (!stepResult) {
            return stepResult.error();
        }
        if (!stepResult.value()) {
            return int64_t{0};
        }
        return stmt.getInt64(0);
    });
    REQUIRE((assocCount.has_value()));
    CHECK((assocCount.value() == 0));
}

TEST_CASE("MetadataRepository: remove path tree decrements counts and deletes empty nodes",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto docInfo =
        makeDocumentWithPath("/project/src/lib/util.cpp",
                             "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA");
    auto docInsert = fix.repository_->insertDocument(docInfo);
    REQUIRE((docInsert.has_value()));
    auto docId = docInsert.value();
    docInfo.id = docId;

    std::vector<float> embedding{1.0F, 2.0F, 3.0F};
    auto upsert = fix.repository_->upsertPathTreeForDocument(
        docInfo, docId, true, std::span<const float>(embedding.data(), embedding.size()));
    REQUIRE((upsert.has_value()));

    auto leafNode = fix.repository_->findPathTreeNodeByFullPath("/project/src/lib/util.cpp");
    REQUIRE((leafNode.has_value()));
    REQUIRE((leafNode.value().has_value()));
    CHECK((leafNode.value()->docCount == 1));
    CHECK((leafNode.value()->centroidWeight == 1));

    auto libNode = fix.repository_->findPathTreeNodeByFullPath("/project/src/lib");
    REQUIRE((libNode.has_value()));
    REQUIRE((libNode.value().has_value()));
    CHECK((libNode.value()->docCount == 1));

    auto remove = fix.repository_->removePathTreeForDocument(
        docInfo, docId, std::span<const float>(embedding.data(), embedding.size()));
    REQUIRE((remove.has_value()));

    auto afterRemoveLeaf = fix.repository_->findPathTreeNodeByFullPath("/project/src/lib/util.cpp");
    REQUIRE((afterRemoveLeaf.has_value()));
    CHECK_FALSE(afterRemoveLeaf.value().has_value());

    auto afterRemoveLib = fix.repository_->findPathTreeNodeByFullPath("/project/src/lib");
    REQUIRE((afterRemoveLib.has_value()));
    CHECK_FALSE(afterRemoveLib.value().has_value());
}

TEST_CASE("MetadataRepository: remove path tree recalculates centroid",
          "[unit][metadata][repository]") {
    MetadataRepositoryFixture fix;

    auto doc1 = makeDocumentWithPath(
        "/shared/file1.txt", "1111111111111111111111111111111111111111111111111111111111111111");
    auto doc1Insert = fix.repository_->insertDocument(doc1);
    REQUIRE((doc1Insert.has_value()));
    doc1.id = doc1Insert.value();

    auto doc2 = makeDocumentWithPath(
        "/shared/file2.txt", "2222222222222222222222222222222222222222222222222222222222222222");
    auto doc2Insert = fix.repository_->insertDocument(doc2);
    REQUIRE((doc2Insert.has_value()));
    doc2.id = doc2Insert.value();

    std::vector<float> emb1{1.0F, 0.0F, 0.0F};
    std::vector<float> emb2{0.0F, 1.0F, 0.0F};

    auto upsert1 = fix.repository_->upsertPathTreeForDocument(
        doc1, doc1.id, true, std::span<const float>(emb1.data(), emb1.size()));
    REQUIRE((upsert1.has_value()));

    auto upsert2 = fix.repository_->upsertPathTreeForDocument(
        doc2, doc2.id, true, std::span<const float>(emb2.data(), emb2.size()));
    REQUIRE((upsert2.has_value()));

    auto sharedNode = fix.repository_->findPathTreeNodeByFullPath("/shared");
    REQUIRE((sharedNode.has_value()));
    REQUIRE((sharedNode.value().has_value()));
    CHECK((sharedNode.value()->docCount == 2));
    CHECK((sharedNode.value()->centroidWeight == 2));

    auto remove1 = fix.repository_->removePathTreeForDocument(
        doc1, doc1.id, std::span<const float>(emb1.data(), emb1.size()));
    REQUIRE((remove1.has_value()));

    auto afterRemove = fix.repository_->findPathTreeNodeByFullPath("/shared");
    REQUIRE((afterRemove.has_value()));
    REQUIRE((afterRemove.value().has_value()));
    CHECK((afterRemove.value()->docCount == 1));
    CHECK((afterRemove.value()->centroidWeight == 1));
}

TEST_CASE("MetadataRepository: path tree child listing and repair helpers",
          "[unit][metadata][repository][path-tree]") {
    MetadataRepositoryFixture fix;

    auto missingDoc = makeDocumentWithPath("/repair/missing.txt", "path-tree-missing-hash");
    auto treeDocA = makeDocumentWithPath("/repair/tree/a.txt", "path-tree-a-hash");
    auto treeDocB = makeDocumentWithPath("/repair/tree/b.txt", "path-tree-b-hash");

    auto missingId = fix.repository_->insertDocument(missingDoc);
    auto treeAId = fix.repository_->insertDocument(treeDocA);
    auto treeBId = fix.repository_->insertDocument(treeDocB);
    REQUIRE((missingId.has_value()));
    REQUIRE((treeAId.has_value()));
    REQUIRE((treeBId.has_value()));

    treeDocA.id = treeAId.value();
    treeDocB.id = treeBId.value();
    REQUIRE(
        fix.repository_->upsertPathTreeForDocument(treeDocA, treeDocA.id, true, {}).has_value());
    REQUIRE(
        fix.repository_->upsertPathTreeForDocument(treeDocB, treeDocB.id, true, {}).has_value());

    auto missingCount = fix.repository_->countDocsMissingPathTree();
    REQUIRE((missingCount.has_value()));
    CHECK((missingCount.value() == 1));

    auto missingDocs = fix.repository_->findDocsMissingPathTree(5);
    REQUIRE((missingDocs.has_value()));
    REQUIRE((missingDocs.value().size() == 1));
    CHECK((missingDocs.value().front().sha256Hash == missingDoc.sha256Hash));

    auto rootChildren = fix.repository_->listPathTreeChildren("/", 10);
    REQUIRE((rootChildren.has_value()));
    auto repairIt =
        std::find_if(rootChildren.value().begin(), rootChildren.value().end(),
                     [](const PathTreeNode& node) { return node.fullPath == "/repair"; });
    REQUIRE((repairIt != rootChildren.value().end()));
    CHECK((repairIt->docCount == 2));

    auto repairChildren = fix.repository_->listPathTreeChildren("/repair", 10);
    REQUIRE((repairChildren.has_value()));
    REQUIRE((repairChildren.value().size() == 1));
    CHECK((repairChildren.value().front().fullPath == "/repair/tree"));
    CHECK((repairChildren.value().front().docCount == 2));

    auto limitedChildren = fix.repository_->listPathTreeChildren("/repair/tree", 1);
    REQUIRE((limitedChildren.has_value()));
    REQUIRE((limitedChildren.value().size() == 1));
    CHECK((limitedChildren.value().front().fullPath == "/repair/tree/a.txt"));

    auto missingParentChildren = fix.repository_->listPathTreeChildren("/repair/missing.txt", 10);
    REQUIRE((missingParentChildren.has_value()));
    CHECK((missingParentChildren.value().empty()));

    auto singleConnDbPath = tempDbPath("metadata_repo_single_conn_path_tree_");
    ConnectionPoolConfig singleConnConfig;
    singleConnConfig.minConnections = 1;
    singleConnConfig.maxConnections = 1;
    auto singleConnPool =
        std::make_unique<ConnectionPool>(singleConnDbPath.string(), singleConnConfig);
    REQUIRE((singleConnPool->initialize().has_value()));
    auto singleConnRepo = std::make_unique<MetadataRepository>(
        *singleConnPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    auto singleConnDoc =
        makeDocumentWithPath("/single/connection/child.txt", "single-connection-path-tree-hash");
    auto singleConnInsert = singleConnRepo->insertDocument(singleConnDoc);
    REQUIRE((singleConnInsert.has_value()));
    singleConnDoc.id = singleConnInsert.value();
    REQUIRE((singleConnRepo->upsertPathTreeForDocument(singleConnDoc, singleConnDoc.id, true, {})
                 .has_value()));

    auto singleConnChildren = singleConnRepo->listPathTreeChildren("/single/connection", 10);
    REQUIRE((singleConnChildren.has_value()));
    REQUIRE((singleConnChildren.value().size() == 1));
    CHECK((singleConnChildren.value().front().fullPath == "/single/connection/child.txt"));

    singleConnRepo.reset();
    singleConnPool->shutdown();
    singleConnPool.reset();
    {
        std::error_code ec;
        std::filesystem::remove(singleConnDbPath, ec);
    }

    auto prefixDocs = fix.repository_->findDocumentsByPathTreePrefix("/repair/tree", true, 10);
    REQUIRE((prefixDocs.has_value()));
    std::vector<std::string> hashes;
    for (const auto& doc : prefixDocs.value()) {
        hashes.push_back(doc.sha256Hash);
    }
    CHECK((std::find(hashes.begin(), hashes.end(), treeDocA.sha256Hash) != hashes.end()));
    CHECK((std::find(hashes.begin(), hashes.end(), treeDocB.sha256Hash) != hashes.end()));
    CHECK((std::find(hashes.begin(), hashes.end(), missingDoc.sha256Hash) == hashes.end()));
}

TEST_CASE("MetadataRepository: direct path tree helpers round-trip",
          "[unit][metadata][repository][path-tree][helpers]") {
    MetadataRepositoryFixture fix;

    auto rootNode =
        fix.repository_->insertPathTreeNode(PathTreeNode::kNullParent, "manual", "/manual");
    REQUIRE((rootNode.has_value()));
    CHECK((rootNode.value().fullPath == "/manual"));
    CHECK((rootNode.value().parentId == PathTreeNode::kNullParent));

    auto childNode =
        fix.repository_->insertPathTreeNode(rootNode.value().id, "child", "/manual/child");
    REQUIRE((childNode.has_value()));
    CHECK((childNode.value().parentId == rootNode.value().id));
    CHECK((childNode.value().fullPath == "/manual/child"));

    auto rootLookup = fix.repository_->findPathTreeNode(PathTreeNode::kNullParent, "manual");
    REQUIRE((rootLookup.has_value()));
    REQUIRE((rootLookup.value().has_value()));
    CHECK((rootLookup.value()->id == rootNode.value().id));

    auto childLookup = fix.repository_->findPathTreeNode(rootNode.value().id, "child");
    REQUIRE((childLookup.has_value()));
    REQUIRE((childLookup.value().has_value()));
    CHECK((childLookup.value()->id == childNode.value().id));

    auto missingLookup = fix.repository_->findPathTreeNode(rootNode.value().id, "missing");
    REQUIRE((missingLookup.has_value()));
    CHECK_FALSE((missingLookup.value().has_value()));

    auto emptyPathLookup = fix.repository_->findPathTreeNodeByFullPath("");
    REQUIRE((emptyPathLookup.has_value()));
    CHECK_FALSE((emptyPathLookup.value().has_value()));

    auto doc = makeDocumentWithPath("/manual/document.txt", "path-tree-helper-doc");
    auto docId = fix.repository_->insertDocument(doc);
    REQUIRE((docId.has_value()));

    REQUIRE((fix.repository_->incrementPathTreeDocCount(rootNode.value().id, docId.value())
                 .has_value()));
    REQUIRE((fix.repository_->incrementPathTreeDocCount(rootNode.value().id, docId.value())
                 .has_value()));

    auto rootAfterIncrement = fix.repository_->findPathTreeNodeByFullPath("/manual");
    REQUIRE((rootAfterIncrement.has_value()));
    REQUIRE((rootAfterIncrement.value().has_value()));
    CHECK((rootAfterIncrement.value()->docCount == 1));

    auto rootChildrenNoLimit = fix.repository_->listPathTreeChildren("", 0);
    REQUIRE((rootChildrenNoLimit.has_value()));
    auto manualIt =
        std::find_if(rootChildrenNoLimit.value().begin(), rootChildrenNoLimit.value().end(),
                     [](const PathTreeNode& node) { return node.fullPath == "/manual"; });
    REQUIRE((manualIt != rootChildrenNoLimit.value().end()));
    CHECK((manualIt->docCount == 1));
}

TEST_CASE("MetadataRepository: direct path tree centroid helper handles reset and missing node",
          "[unit][metadata][repository][path-tree][helpers]") {
    MetadataRepositoryFixture fix;

    auto node = fix.repository_->insertPathTreeNode(PathTreeNode::kNullParent, "embed", "/embed");
    REQUIRE((node.has_value()));

    REQUIRE((fix.repository_->accumulatePathTreeCentroid(node.value().id, std::span<const float>{})
                 .has_value()));

    auto afterEmpty = fix.repository_->findPathTreeNodeByFullPath("/embed");
    REQUIRE((afterEmpty.has_value()));
    REQUIRE((afterEmpty.value().has_value()));
    CHECK((afterEmpty.value()->centroidWeight == 0));
    CHECK((afterEmpty.value()->centroid.empty()));

    const std::vector<float> firstEmbedding{1.0F, 2.0F, 3.0F};
    REQUIRE((fix.repository_
                 ->accumulatePathTreeCentroid(
                     node.value().id,
                     std::span<const float>(firstEmbedding.data(), firstEmbedding.size()))
                 .has_value()));

    auto afterFirst = fix.repository_->findPathTreeNodeByFullPath("/embed");
    REQUIRE((afterFirst.has_value()));
    REQUIRE((afterFirst.value().has_value()));
    CHECK((afterFirst.value()->centroidWeight == 1));
    CHECK((afterFirst.value()->centroid == firstEmbedding));

    const std::vector<float> secondEmbedding{4.0F, 5.0F, 6.0F};
    REQUIRE((fix.repository_
                 ->accumulatePathTreeCentroid(
                     node.value().id,
                     std::span<const float>(secondEmbedding.data(), secondEmbedding.size()))
                 .has_value()));

    auto afterSecond = fix.repository_->findPathTreeNodeByFullPath("/embed");
    REQUIRE((afterSecond.has_value()));
    REQUIRE((afterSecond.value().has_value()));
    CHECK((afterSecond.value()->centroidWeight == 2));
    REQUIRE((afterSecond.value()->centroid.size() == 3));
    CHECK_THAT(afterSecond.value()->centroid[0], Catch::Matchers::WithinRel(2.5F, 0.01F));
    CHECK_THAT(afterSecond.value()->centroid[1], Catch::Matchers::WithinRel(3.5F, 0.01F));
    CHECK_THAT(afterSecond.value()->centroid[2], Catch::Matchers::WithinRel(4.5F, 0.01F));

    const std::vector<std::byte> corruptBlob{std::byte{0x01}, std::byte{0x02}, std::byte{0x03}};
    REQUIRE((overwritePathTreeCentroid(*fix.pool_, "/embed", corruptBlob, 5).has_value()));

    const std::vector<float> resetEmbedding{9.0F, 10.0F, 11.0F};
    REQUIRE((fix.repository_
                 ->accumulatePathTreeCentroid(
                     node.value().id,
                     std::span<const float>(resetEmbedding.data(), resetEmbedding.size()))
                 .has_value()));

    auto afterReset = fix.repository_->findPathTreeNodeByFullPath("/embed");
    REQUIRE((afterReset.has_value()));
    REQUIRE((afterReset.value().has_value()));
    CHECK((afterReset.value()->centroidWeight == 1));
    CHECK((afterReset.value()->centroid == resetEmbedding));

    auto missingNodeResult = fix.repository_->accumulatePathTreeCentroid(
        999999, std::span<const float>(resetEmbedding.data(), resetEmbedding.size()));
    REQUIRE_FALSE((missingNodeResult.has_value()));
    CHECK((missingNodeResult.error().message.find("not found") != std::string::npos));
}

TEST_CASE("MetadataRepository: path tree supports relative document paths",
          "[unit][metadata][repository][path-tree][relative]") {
    MetadataRepositoryFixture fix;

    auto storedDoc = makeDocumentWithPath("/backing/relative-source.txt", "path-tree-relative-doc");
    auto docInsert = fix.repository_->insertDocument(storedDoc);
    REQUIRE((docInsert.has_value()));
    storedDoc.id = docInsert.value();

    auto relativePathInfo = storedDoc;
    relativePathInfo.filePath = "relative/tree/file.txt";

    REQUIRE((fix.repository_->upsertPathTreeForDocument(relativePathInfo, storedDoc.id, true, {})
                 .has_value()));

    auto relativeRoot = fix.repository_->findPathTreeNode(PathTreeNode::kNullParent, "relative");
    REQUIRE((relativeRoot.has_value()));
    REQUIRE((relativeRoot.value().has_value()));
    CHECK((relativeRoot.value()->fullPath == "relative"));
    CHECK((relativeRoot.value()->docCount == 1));

    auto relativeChild = fix.repository_->findPathTreeNodeByFullPath("relative/tree");
    REQUIRE((relativeChild.has_value()));
    REQUIRE((relativeChild.value().has_value()));
    CHECK((relativeChild.value()->docCount == 1));

    auto relativeChildren = fix.repository_->listPathTreeChildren("relative", 0);
    REQUIRE((relativeChildren.has_value()));
    REQUIRE((relativeChildren.value().size() == 1));
    CHECK((relativeChildren.value().front().fullPath == "relative/tree"));

    REQUIRE((fix.repository_->removePathTreeForDocument(relativePathInfo, storedDoc.id, {})
                 .has_value()));

    auto removedLeaf = fix.repository_->findPathTreeNodeByFullPath("relative/tree/file.txt");
    REQUIRE((removedLeaf.has_value()));
    CHECK_FALSE((removedLeaf.value().has_value()));

    auto preservedRoot = fix.repository_->findPathTreeNodeByFullPath("relative");
    REQUIRE((preservedRoot.has_value()));
    REQUIRE((preservedRoot.value().has_value()));
    CHECK((preservedRoot.value()->docCount == 0));
}

TEST_CASE("MetadataRepository: tree snapshots and changes round-trip",
          "[unit][metadata][repository][tree-diff]") {
    MetadataRepositoryFixture fix;

    TreeSnapshotRecord base;
    base.snapshotId = "snapshot-base";
    base.rootTreeHash = "root-base";
    base.createdTime = 100;
    base.fileCount = 1;
    base.metadata["directory_path"] = "/repo";
    base.metadata["snapshot_label"] = "base";
    base.metadata["git_commit"] = "abc123";
    base.metadata["git_branch"] = "main";
    base.metadata["git_remote"] = "origin";

    TreeSnapshotRecord target = base;
    target.snapshotId = "snapshot-target";
    target.rootTreeHash = "root-target";
    target.createdTime = 200;
    target.fileCount = 2;
    target.metadata["snapshot_label"] = "target";
    target.metadata["git_commit"] = "def456";

    REQUIRE((fix.repository_->upsertTreeSnapshot(base).has_value()));
    REQUIRE((fix.repository_->upsertTreeSnapshot(target).has_value()));

    auto snapshots = fix.repository_->listTreeSnapshots(10);
    REQUIRE((snapshots.has_value()));
    REQUIRE((snapshots.value().size() >= 2));
    CHECK((snapshots.value().front().snapshotId == "snapshot-target"));
    CHECK((snapshots.value().front().metadata.at("snapshot_label") == "target"));
    CHECK((snapshots.value().front().metadata.at("git_commit") == "def456"));
    CHECK((snapshots.value().front().fileCount == 2));

    auto fetchedTarget = fix.repository_->getTreeSnapshot(target.snapshotId);
    if (!fetchedTarget.has_value()) {
        INFO("getTreeSnapshot error: " << fetchedTarget.error().message);
    }
    REQUIRE((fetchedTarget.has_value()));
    REQUIRE((fetchedTarget.value().has_value()));
    CHECK((fetchedTarget.value()->snapshotId == target.snapshotId));
    CHECK((fetchedTarget.value()->rootTreeHash == target.rootTreeHash));
    CHECK((fetchedTarget.value()->metadata.at("directory_path") == "/repo"));
    CHECK((fetchedTarget.value()->metadata.at("snapshot_label") == "target"));
    CHECK((fetchedTarget.value()->metadata.at("git_commit") == "def456"));
    CHECK((fetchedTarget.value()->fileCount == 2));

    auto missingSnapshot = fix.repository_->getTreeSnapshot("missing-snapshot");
    REQUIRE((missingSnapshot.has_value()));
    CHECK_FALSE((missingSnapshot.value().has_value()));

    TreeDiffDescriptor descriptor;
    descriptor.baseSnapshotId = base.snapshotId;
    descriptor.targetSnapshotId = target.snapshotId;
    descriptor.computedAt = 300;
    descriptor.status = "pending";
    auto diffId = fix.repository_->beginTreeDiff(descriptor);
    REQUIRE((diffId.has_value()));
    REQUIRE((diffId.value() > 0));

    TreeChangeRecord added;
    added.type = TreeChangeType::Added;
    added.newPath = "/repo/src/new.cpp";
    added.newHash = "new-hash";
    added.mode = 0644;
    added.contentDeltaHash = "delta-hash";

    TreeChangeRecord moved;
    moved.type = TreeChangeType::Moved;
    moved.oldPath = "/repo/src/old.cpp";
    moved.newPath = "/repo/include/old.hpp";
    moved.oldHash = "old-hash";
    moved.newHash = "old-hash";

    TreeChangeRecord deletedDirectory;
    deletedDirectory.type = TreeChangeType::Deleted;
    deletedDirectory.oldPath = "/repo/tmp";
    deletedDirectory.isDirectory = true;

    REQUIRE(fix.repository_->appendTreeChanges(diffId.value(), {added, moved, deletedDirectory})
                .has_value());

    TreeDiffQuery allChanges;
    allChanges.baseSnapshotId = base.snapshotId;
    allChanges.targetSnapshotId = target.snapshotId;
    allChanges.limit = 10;
    auto changes = fix.repository_->listTreeChanges(allChanges);
    REQUIRE((changes.has_value()));
    REQUIRE((changes.value().size() == 3));
    CHECK((changes.value()[0].type == TreeChangeType::Added));
    CHECK((changes.value()[1].type == TreeChangeType::Moved));
    CHECK((changes.value()[2].type == TreeChangeType::Deleted));

    TreeDiffQuery addedUnderSrc;
    addedUnderSrc.baseSnapshotId = base.snapshotId;
    addedUnderSrc.targetSnapshotId = target.snapshotId;
    addedUnderSrc.pathPrefix = "/repo/src";
    addedUnderSrc.typeFilter = TreeChangeType::Added;
    auto filtered = fix.repository_->listTreeChanges(addedUnderSrc);
    REQUIRE((filtered.has_value()));
    REQUIRE((filtered.value().size() == 1));
    CHECK((filtered.value().front().newPath == added.newPath));
    REQUIRE((filtered.value().front().mode.has_value()));
    CHECK((filtered.value().front().mode.value() == 0644));
    REQUIRE((filtered.value().front().contentDeltaHash.has_value()));
    CHECK((filtered.value().front().contentDeltaHash.value() == "delta-hash"));

    REQUIRE(fix.repository_->finalizeTreeDiff(diffId.value(), changes.value().size(), "complete")
                .has_value());

    auto verifyStatus = fix.pool_->withConnection([&](Database& db) -> Result<void> {
        auto stmtResult = db.prepare(R"(
            SELECT files_added, files_deleted, files_modified, files_renamed, status
            FROM tree_diffs WHERE diff_id = ?
        )");
        REQUIRE((stmtResult.has_value()));
        auto& stmt = stmtResult.value();
        REQUIRE((stmt.bind(1, diffId.value()).has_value()));
        auto step = stmt.step();
        REQUIRE((step.has_value()));
        REQUIRE((step.value()));
        CHECK((stmt.getInt64(0) == 1));
        CHECK((stmt.getInt64(1) == 0));
        CHECK((stmt.getInt64(2) == 0));
        CHECK((stmt.getInt64(3) == 1));
        CHECK((stmt.getString(4) == "complete"));
        return Result<void>();
    });
    REQUIRE((verifyStatus.has_value()));
}

// --- batchInsertContentAndIndex counter accuracy tests ---

TEST_CASE("batchInsertContentAndIndex: fresh documents increment extracted and indexed counters",
          "[unit][metadata-repo][batch-content][counters]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    // Insert 3 fresh documents (not yet extracted/indexed)
    std::vector<int64_t> docIds;
    for (int i = 0; i < 3; ++i) {
        DocumentInfo doc;
        doc.sha256Hash = "counter_hash_fresh_" + std::to_string(i);
        doc.fileName = "counter_fresh_" + std::to_string(i) + ".txt";
        doc.fileSize = 100;
        doc.mimeType = "text/plain";
        doc.createdTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        doc.modifiedTime = doc.createdTime;
        // Leave contentExtracted=false, extractionStatus=Pending (defaults)
        doc.contentExtracted = false;
        doc.extractionStatus = ExtractionStatus::Pending;
        auto result = fix.repository_->insertDocument(doc);
        REQUIRE((result.has_value()));
        docIds.push_back(result.value());
    }

    auto extractedBefore = fix.repository_->getCachedExtractedCount();
    auto indexedBefore = fix.repository_->getCachedIndexedCount();

    // Batch insert content for all 3
    std::vector<BatchContentEntry> entries;
    for (int i = 0; i < 3; ++i) {
        BatchContentEntry entry;
        entry.documentId = docIds[static_cast<size_t>(i)];
        entry.title = "Title " + std::to_string(i);
        entry.contentText = "Content text " + std::to_string(i);
        entry.mimeType = "text/plain";
        entry.extractionMethod = "test";
        entry.language = "en";
        entry.priorStateKnown = true;
        entry.priorContentExtracted = false;
        entry.priorExtractionStatus = ExtractionStatus::Pending;
        entries.push_back(entry);
    }
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    CHECK((fix.repository_->getCachedExtractedCount() == extractedBefore + 3));
    CHECK((fix.repository_->getCachedIndexedCount() == indexedBefore + 3));

    for (auto docId : docIds) {
        auto docResult = fix.repository_->getDocument(docId);
        REQUIRE((docResult.has_value()));
        REQUIRE((docResult.value().has_value()));
        CHECK((docResult.value()->repairStatus == RepairStatus::Completed));
        CHECK((docResult.value()->repairAttempts == 1));
    }
}

TEST_CASE("batchInsertContentAndIndex persists extracted title metadata in the content transaction",
          "[unit][metadata-repo][batch-content][title]") {
    MetadataRepositoryFixture fix;

    auto inserted = fix.repository_->insertDocument(
        makeDocumentWithPath("/tmp/content-title.txt", "content_title_hash", "text/plain"));
    REQUIRE((inserted.has_value()));

    auto entry = makeBatchContentEntry(inserted.value(), "FTS fallback title", "Content text");
    entry.metadataTitle = "Extracted document title";
    REQUIRE((fix.repository_->batchInsertContentAndIndex({entry}).has_value()));

    auto title = fix.repository_->getMetadata(inserted.value(), "title");
    REQUIRE((title.has_value()));
    REQUIRE((title.value().has_value()));
    CHECK((title.value()->value == "Extracted document title"));
}

TEST_CASE("batchInsertContentAndIndex invalidates embeddings only when extracted input changes",
          "[unit][metadata-repo][batch-content][embedding-freshness]") {
    MetadataRepositoryFixture fix;
    const std::string hash = "content-freshness-hash";
    auto inserted =
        fix.repository_->insertDocument(makeDocumentWithPath("/tmp/content-freshness.txt", hash));
    REQUIRE(inserted.has_value());
    auto entry = makeBatchContentEntry(inserted.value(), "Title", "Original text");
    REQUIRE(fix.repository_->batchInsertContentAndIndex({entry}).has_value());
    REQUIRE(fix.repository_->batchCompleteDocumentEmbeddingsByHashes({hash}, "model").has_value());

    bool unchanged = false;
    bool rejectWrite = false;
    SECTION("identical extraction preserves readiness") {
        unchanged = true;
    }
    SECTION("changed text invalidates readiness") {
        entry.contentText = "Replacement text";
    }
    SECTION("empty replacement invalidates readiness") {
        entry.contentText.clear();
    }
    SECTION("title-only update preserves readiness") {
        entry.title = "New display title";
        unchanged = true;
    }
    SECTION("changed extraction method invalidates readiness") {
        entry.extractionMethod += "-v2";
    }
    SECTION("changed language invalidates readiness") {
        entry.language = "fr";
    }
    SECTION("failed content write rolls back readiness invalidation") {
        entry.contentText = "Rejected replacement";
        rejectWrite = true;
        unchanged = true;
        REQUIRE(fix.pool_
                    ->withConnection([](Database& db) -> Result<void> {
                        return db.execute(R"(
                CREATE TRIGGER reject_content_update BEFORE UPDATE ON document_content
                BEGIN SELECT RAISE(ABORT, 'test content rejection'); END
            )");
                    })
                    .has_value());
    }

    // Duplicate entries must invalidate/count the embedding only once.
    auto written = fix.repository_->batchInsertContentAndIndex({entry, entry});
    REQUIRE(written.has_value() == !rejectWrite);
    if (rejectWrite) {
        auto content = fix.repository_->getContent(inserted.value());
        REQUIRE(content.has_value());
        REQUIRE(content.value().has_value());
        CHECK(content.value()->contentText == "Original text");
    }
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(ready.has_value());
    CHECK(ready.value() == unchanged);
    auto status = getEmbeddingStatusRow(*fix.pool_, hash);
    REQUIRE(status.has_value());
    REQUIRE(status.value().has_value());
    CHECK(status.value()->modelId.has_value() == unchanged);
    auto stats = fix.repository_->getCorpusStats();
    REQUIRE(stats.has_value());
    CHECK(stats.value().embeddingCount == (unchanged ? 1 : 0));
}

TEST_CASE(
    "batchInsertContentAndIndex: duplicate document entries preserve per-entry repair attempts",
    "[unit][metadata-repo][batch-content][status-update][duplicates]") {
    MetadataRepositoryFixture fix;

    DocumentInfo doc;
    doc.sha256Hash = "counter_hash_duplicate_status";
    doc.fileName = "duplicate_status.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    const auto docId = insertResult.value();

    BatchContentEntry first = makeBatchContentEntry(docId, "First title", "First content");
    BatchContentEntry second = makeBatchContentEntry(docId, "Second title", "Second content");

    auto batchResult = fix.repository_->batchInsertContentAndIndex({first, second});
    REQUIRE((batchResult.has_value()));

    auto docResult = fix.repository_->getDocument(docId);
    REQUIRE((docResult.has_value()));
    REQUIRE((docResult.value().has_value()));
    CHECK((docResult.value()->repairStatus == RepairStatus::Completed));
    CHECK((docResult.value()->repairAttempts == 2));

    auto contentResult = fix.repository_->getContent(docId);
    REQUIRE((contentResult.has_value()));
    REQUIRE((contentResult.value().has_value()));
    CHECK((contentResult.value()->contentText == "Second content"));
}

TEST_CASE("batchInsertContentAndIndex: skips stale missing document ids",
          "[unit][metadata-repo][batch-content][counters][stale]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    const auto docsBefore = fix.repository_->getCachedDocumentCount();
    const auto extractedBefore = fix.repository_->getCachedExtractedCount();
    const auto indexedBefore = fix.repository_->getCachedIndexedCount();

    BatchContentEntry stale;
    stale.documentId = 999999;
    stale.title = "stale title";
    stale.contentText = "stale content";
    stale.mimeType = "text/plain";
    stale.extractionMethod = "test";
    stale.language = "en";

    auto batchResult = fix.repository_->batchInsertContentAndIndex({stale});
    REQUIRE((batchResult.has_value()));

    CHECK((fix.repository_->getCachedDocumentCount() == docsBefore));
    CHECK((fix.repository_->getCachedExtractedCount() == extractedBefore));
    CHECK((fix.repository_->getCachedIndexedCount() == indexedBefore));

    auto exactIndexed = fix.repository_->getIndexedDocumentCount();
    REQUIRE((exactIndexed.has_value()));
    CHECK((static_cast<uint64_t>(exactIndexed.value()) == indexedBefore));
}

TEST_CASE("batchInsertContentAndIndex: updates warmed FTS indexed ID cache",
          "[unit][metadata-repo][batch-content][fts-cache]") {
    MetadataRepositoryFixture fix;

    std::vector<int64_t> docIds;
    for (int i = 0; i < 3; ++i) {
        DocumentInfo doc;
        doc.sha256Hash = "fts_cache_batch_hash_" + std::to_string(i);
        doc.fileName = "fts_cache_batch_" + std::to_string(i) + ".txt";
        doc.fileSize = 100;
        doc.mimeType = "text/plain";
        doc.createdTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        doc.modifiedTime = doc.createdTime;
        auto result = fix.repository_->insertDocument(doc);
        REQUIRE((result.has_value()));
        docIds.push_back(result.value());
    }

    auto initialIds = fix.repository_->getAllFts5IndexedDocumentIds();
    REQUIRE((initialIds.has_value()));
    REQUIRE((initialIds.value().empty()));

    std::vector<BatchContentEntry> entries;
    for (int i = 0; i < 3; ++i) {
        BatchContentEntry entry;
        entry.documentId = docIds[static_cast<size_t>(i)];
        entry.title = "FTS Cache Title " + std::to_string(i);
        entry.contentText = "FTS cache content " + std::to_string(i);
        entry.mimeType = "text/plain";
        entry.extractionMethod = "test";
        entry.language = "en";
        entries.push_back(entry);
    }

    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    auto indexedIds = fix.repository_->getAllFts5IndexedDocumentIds();
    REQUIRE((indexedIds.has_value()));

    auto actual = indexedIds.value();
    std::sort(actual.begin(), actual.end());
    std::sort(docIds.begin(), docIds.end());
    CHECK((actual == docIds));
}

TEST_CASE("batchInsertContentAndIndex: already-extracted documents don't double-count",
          "[unit][metadata-repo][batch-content][counters]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    DocumentInfo doc;
    doc.sha256Hash = "counter_hash_already";
    doc.fileName = "counter_already.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    // First batch insert — transitions to extracted+indexed
    std::vector<BatchContentEntry> entries = {makeBatchContentEntry(docId, "Title", "Content")};
    auto batch1 = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batch1.has_value()));

    auto extractedAfterFirst = fix.repository_->getCachedExtractedCount();
    auto indexedAfterFirst = fix.repository_->getCachedIndexedCount();

    // Re-insert same document — counters should not change
    auto batch2 = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batch2.has_value()));

    CHECK((fix.repository_->getCachedExtractedCount() == extractedAfterFirst));
    CHECK((fix.repository_->getCachedIndexedCount() == indexedAfterFirst));
}

TEST_CASE("batchInsertContentAndIndex: stale extraction hints don't double-count",
          "[unit][metadata-repo][batch-content][counters]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    auto inserted = fix.repository_->insertDocument(makeDocumentWithPath(
        "/tmp/stale-extraction-hint.txt", "stale_extraction_hint_hash", "text/plain"));
    REQUIRE((inserted.has_value()));

    auto staleEntry = makeBatchContentEntry(inserted.value(), "Title", "First content");
    staleEntry.priorStateKnown = true;
    staleEntry.priorContentExtracted = false;
    REQUIRE((fix.repository_->batchInsertContentAndIndex({staleEntry}).has_value()));

    const auto extractedAfterFirst = fix.repository_->getCachedExtractedCount();
    const auto indexedAfterFirst = fix.repository_->getCachedIndexedCount();

    // Concurrent post-ingest tasks for duplicate content can carry the same pre-commit hint.
    // The repository must reconcile each transaction against the database state it serializes
    // behind, rather than trusting a hint captured before an earlier transaction committed.
    staleEntry.contentText = "Second content";
    REQUIRE((fix.repository_->batchInsertContentAndIndex({staleEntry}).has_value()));

    CHECK((fix.repository_->getCachedExtractedCount() == extractedAfterFirst));
    CHECK((fix.repository_->getCachedIndexedCount() == indexedAfterFirst));
}

TEST_CASE("batchInsertContentAndIndex: mixed fresh and already-extracted batch",
          "[unit][metadata-repo][batch-content][counters]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    // Insert 2 documents
    DocumentInfo doc1;
    doc1.sha256Hash = "counter_mixed_1";
    doc1.fileName = "mixed_1.txt";
    doc1.fileSize = 100;
    doc1.mimeType = "text/plain";
    doc1.createdTime = now;
    doc1.modifiedTime = now;
    auto id1 = fix.repository_->insertDocument(doc1);
    REQUIRE((id1.has_value()));

    DocumentInfo doc2;
    doc2.sha256Hash = "counter_mixed_2";
    doc2.fileName = "mixed_2.txt";
    doc2.fileSize = 200;
    doc2.mimeType = "text/plain";
    doc2.createdTime = now;
    doc2.modifiedTime = now;
    auto id2 = fix.repository_->insertDocument(doc2);
    REQUIRE((id2.has_value()));

    // Pre-extract doc1 via batchInsertContentAndIndex so it has extraction_status='Success'
    std::vector<BatchContentEntry> preEntries = {
        makeBatchContentEntry(id1.value(), "Title 1", "Content 1")};
    auto preResult = fix.repository_->batchInsertContentAndIndex(preEntries);
    REQUIRE((preResult.has_value()));

    auto extractedBefore = fix.repository_->getCachedExtractedCount();
    auto indexedBefore = fix.repository_->getCachedIndexedCount();

    // Now batch both: doc1 (already extracted) + doc2 (fresh)
    std::vector<BatchContentEntry> entries = {
        makeBatchContentEntry(id1.value(), "Title 1 updated", "Content 1 updated"),
        makeBatchContentEntry(id2.value(), "Title 2", "Content 2")};
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    // Only doc2 (fresh) should increment counters
    CHECK((fix.repository_->getCachedExtractedCount() == extractedBefore + 1));
    CHECK((fix.repository_->getCachedIndexedCount() == indexedBefore + 1));
}

TEST_CASE(
    "batchInsertContentAndIndex: priorStateKnown must not treat extraction success as FTS indexed",
    "[unit][metadata-repo][batch-content][counters][fts]") {
    MetadataRepositoryFixture fix;

    DocumentInfo doc;
    doc.sha256Hash = "counter_hash_missing_fts";
    doc.fileName = "missing_fts.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    BatchContentEntry initialEntry;
    initialEntry.documentId = docId;
    initialEntry.title = "Title";
    initialEntry.contentText = "Content";
    initialEntry.mimeType = "text/plain";
    initialEntry.extractionMethod = "test";
    initialEntry.language = "en";
    REQUIRE((fix.repository_->batchInsertContentAndIndex({initialEntry}).has_value()));

    REQUIRE(fix.pool_
                ->withConnection([&](Database& db) -> Result<void> {
                    auto stmtResult = db.prepare("DELETE FROM documents_fts WHERE rowid = ?");
                    REQUIRE((stmtResult.has_value()));
                    auto& stmt = stmtResult.value();
                    REQUIRE((stmt.bind(1, docId).has_value()));
                    REQUIRE((stmt.execute().has_value()));
                    return {};
                })
                .has_value());

    fix.repository_->initializeCounters();
    auto indexedBefore = fix.repository_->getCachedIndexedCount();
    REQUIRE((indexedBefore == 0));

    BatchContentEntry repairEntry;
    repairEntry.documentId = docId;
    repairEntry.title = "Title";
    repairEntry.contentText = "Content repaired";
    repairEntry.mimeType = "text/plain";
    repairEntry.extractionMethod = "test";
    repairEntry.language = "en";
    repairEntry.priorStateKnown = true;
    repairEntry.priorContentExtracted = true;
    repairEntry.priorExtractionStatus = ExtractionStatus::Success;

    auto batchResult = fix.repository_->batchInsertContentAndIndex({repairEntry});
    REQUIRE((batchResult.has_value()));

    auto indexedIds = fix.repository_->getAllFts5IndexedDocumentIds();
    REQUIRE((indexedIds.has_value()));
    REQUIRE(std::find(indexedIds.value().begin(), indexedIds.value().end(), docId) !=
            indexedIds.value().end());
    CHECK((fix.repository_->getCachedIndexedCount() == indexedBefore + 1));
}

TEST_CASE("batchInsertContentAndIndex: mixed repaired and stale entries stay accurate",
          "[unit][metadata-repo][batch-content][counters][fts]") {
    MetadataRepositoryFixture fix;

    DocumentInfo doc;
    doc.sha256Hash = "counter_hash_mixed_repair_stale";
    doc.fileName = "mixed_repair_stale.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    BatchContentEntry initialEntry;
    initialEntry.documentId = docId;
    initialEntry.title = "Title";
    initialEntry.contentText = "Content";
    initialEntry.mimeType = "text/plain";
    initialEntry.extractionMethod = "test";
    initialEntry.language = "en";
    REQUIRE((fix.repository_->batchInsertContentAndIndex({initialEntry}).has_value()));

    REQUIRE(fix.pool_
                ->withConnection([&](Database& db) -> Result<void> {
                    auto stmtResult = db.prepare("DELETE FROM documents_fts WHERE rowid = ?");
                    REQUIRE((stmtResult.has_value()));
                    auto& stmt = stmtResult.value();
                    REQUIRE((stmt.bind(1, docId).has_value()));
                    REQUIRE((stmt.execute().has_value()));
                    return {};
                })
                .has_value());

    fix.repository_->initializeCounters();
    auto indexedBefore = fix.repository_->getCachedIndexedCount();
    REQUIRE((indexedBefore == 0));

    BatchContentEntry repairEntry;
    repairEntry.documentId = docId;
    repairEntry.title = "Title repaired";
    repairEntry.contentText = "Content repaired";
    repairEntry.mimeType = "text/plain";
    repairEntry.extractionMethod = "test";
    repairEntry.language = "en";
    repairEntry.priorStateKnown = true;
    repairEntry.priorContentExtracted = true;
    repairEntry.priorExtractionStatus = ExtractionStatus::Success;

    BatchContentEntry staleEntry = repairEntry;
    staleEntry.documentId = docId + 999999;
    staleEntry.title = "Stale";
    staleEntry.contentText = "Stale content";

    auto batchResult = fix.repository_->batchInsertContentAndIndex({repairEntry, staleEntry});
    REQUIRE((batchResult.has_value()));

    auto indexedIds = fix.repository_->getAllFts5IndexedDocumentIds();
    REQUIRE((indexedIds.has_value()));
    REQUIRE(std::find(indexedIds.value().begin(), indexedIds.value().end(), docId) !=
            indexedIds.value().end());
    CHECK((fix.repository_->getCachedIndexedCount() == indexedBefore + 1));
}

TEST_CASE("batchInsertContentAndIndex: clears extraction_error",
          "[unit][metadata-repo][batch-content][counters]") {
    MetadataRepositoryFixture fix;
    fix.repository_->initializeCounters();

    // Insert a document with extraction_status=Success but an error set
    DocumentInfo doc;
    doc.sha256Hash = "counter_hash_error_clear";
    doc.fileName = "error_clear.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    doc.modifiedTime = doc.createdTime;
    doc.contentExtracted = true;
    doc.extractionStatus = ExtractionStatus::Success;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));
    auto docId = insertResult.value();

    // Manually set an extraction error via raw SQL
    fix.pool_->withConnection([&](Database& db) -> Result<void> {
        auto stmtResult =
            db.prepare("UPDATE documents SET extraction_error = 'stale error' WHERE id = ?");
        REQUIRE((stmtResult.has_value()));
        auto& stmt = stmtResult.value();
        REQUIRE((stmt.bind(1, docId).has_value()));
        REQUIRE((stmt.execute().has_value()));
        return {};
    });

    // Batch insert should clear the error
    std::vector<BatchContentEntry> entries = {makeBatchContentEntry(docId, "Title", "Content")};
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    // Verify error is cleared
    fix.pool_->withConnection([&](Database& db) -> Result<void> {
        auto stmtResult = db.prepare("SELECT extraction_error FROM documents WHERE id = ?");
        REQUIRE((stmtResult.has_value()));
        auto& stmt = stmtResult.value();
        REQUIRE((stmt.bind(1, docId).has_value()));
        auto hasRow = stmt.step();
        REQUIRE((hasRow.has_value()));
        REQUIRE((hasRow.value()));
        CHECK((stmt.isNull(0)));
        return {};
    });
}

TEST_CASE("batchInsertContentAndIndex: does not amplify lock retries",
          "[unit][metadata-repo][batch-content][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_batch_content_lock_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc = makeDocumentWithPath("/tmp/batch-content-lock.txt", "batch-content-lock-hash");
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    BatchContentEntry entry;
    entry.documentId = insert.value();
    entry.title = "Title";
    entry.contentText = "Locked content write";
    entry.mimeType = "text/plain";
    entry.extractionMethod = "test";
    entry.language = "en";
    entry.priorStateKnown = true;
    entry.priorContentExtracted = false;
    entry.priorExtractionStatus = ExtractionStatus::Pending;

    const auto start = std::chrono::steady_clock::now();
    auto result = repository->batchInsertContentAndIndex({entry});
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((result.error().message.find("locked") != std::string::npos));
    CHECK((elapsed < std::chrono::milliseconds(3000)));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

// --- batchGetDocumentsWithContentPreview tests ---

TEST_CASE("batchGetDocumentsWithContentPreview: documents with content",
          "[metadata-repo][batch-docs-preview]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    // Insert 3 documents with content
    std::vector<int64_t> docIds;
    std::vector<std::string> hashes;
    for (int i = 0; i < 3; ++i) {
        DocumentInfo doc;
        doc.sha256Hash = "preview_hash_" + std::to_string(i);
        doc.fileName = "preview_doc_" + std::to_string(i) + ".txt";
        doc.filePath = "/test/preview_doc_" + std::to_string(i) + ".txt";
        doc.fileSize = 100;
        doc.mimeType = "text/plain";
        doc.createdTime = now;
        doc.modifiedTime = now;
        auto result = fix.repository_->insertDocument(doc);
        REQUIRE((result.has_value()));
        docIds.push_back(result.value());
        hashes.push_back(doc.sha256Hash);
    }

    // Insert content for each document
    std::vector<BatchContentEntry> entries;
    for (int i = 0; i < 3; ++i) {
        entries.push_back(makeBatchContentEntry(docIds[static_cast<size_t>(i)],
                                                "Title " + std::to_string(i),
                                                "Content text for document " + std::to_string(i)));
    }
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    // Fetch with combined method
    auto result = fix.repository_->batchGetDocumentsWithContentPreview(hashes, 500);
    REQUIRE((result.has_value()));

    auto& map = result.value();
    CHECK((map.size() == 3));

    for (int i = 0; i < 3; ++i) {
        std::string hash = "preview_hash_" + std::to_string(i);
        REQUIRE((map.count(hash) == 1));
        auto& [info, preview] = map.at(hash);
        CHECK((info.id == docIds[static_cast<size_t>(i)]));
        CHECK((info.sha256Hash == hash));
        CHECK_FALSE(preview.empty());
        CHECK(
            (preview.find("Content text for document " + std::to_string(i)) != std::string::npos));
    }
}

TEST_CASE("batchGetDocumentsWithContentPreview: documents without content",
          "[metadata-repo][batch-docs-preview]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    DocumentInfo doc;
    doc.sha256Hash = "no_content_hash";
    doc.fileName = "no_content_doc.txt";
    doc.filePath = "/test/no_content_doc.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = now;
    doc.modifiedTime = now;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));

    std::vector<std::string> hashes = {"no_content_hash"};
    auto result = fix.repository_->batchGetDocumentsWithContentPreview(hashes, 500);
    REQUIRE((result.has_value()));

    auto& map = result.value();
    REQUIRE((map.size() == 1));
    auto& [info, preview] = map.at("no_content_hash");
    CHECK((info.id == insertResult.value()));
    CHECK((preview.empty()));
}

TEST_CASE("batchGetDocumentsWithContentPreview: mixed with and without content",
          "[metadata-repo][batch-docs-preview]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());

    // Doc with content
    DocumentInfo doc1;
    doc1.sha256Hash = "mixed_hash_with";
    doc1.fileName = "mixed_with.txt";
    doc1.filePath = "/test/mixed_with.txt";
    doc1.fileSize = 100;
    doc1.mimeType = "text/plain";
    doc1.createdTime = now;
    doc1.modifiedTime = now;
    auto id1 = fix.repository_->insertDocument(doc1);
    REQUIRE((id1.has_value()));

    std::vector<BatchContentEntry> entries = {
        makeBatchContentEntry(id1.value(), "Title", "Has content")};
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    // Doc without content
    DocumentInfo doc2;
    doc2.sha256Hash = "mixed_hash_without";
    doc2.fileName = "mixed_without.txt";
    doc2.filePath = "/test/mixed_without.txt";
    doc2.fileSize = 100;
    doc2.mimeType = "text/plain";
    doc2.createdTime = now;
    doc2.modifiedTime = now;
    auto id2 = fix.repository_->insertDocument(doc2);
    REQUIRE((id2.has_value()));

    std::vector<std::string> hashes = {"mixed_hash_with", "mixed_hash_without"};
    auto result = fix.repository_->batchGetDocumentsWithContentPreview(hashes, 500);
    REQUIRE((result.has_value()));

    auto& map = result.value();
    CHECK((map.size() == 2));
    CHECK_FALSE(map.at("mixed_hash_with").second.empty());
    CHECK((map.at("mixed_hash_without").second.empty()));
}

TEST_CASE("batchGetDocumentsWithContentPreview: empty hash list",
          "[metadata-repo][batch-docs-preview]") {
    MetadataRepositoryFixture fix;

    std::vector<std::string> hashes;
    auto result = fix.repository_->batchGetDocumentsWithContentPreview(hashes, 500);
    REQUIRE((result.has_value()));
    CHECK((result.value().empty()));
}

TEST_CASE("batchGetDocumentsWithContentPreview: preview truncation",
          "[metadata-repo][batch-docs-preview]") {
    MetadataRepositoryFixture fix;

    auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    DocumentInfo doc;
    doc.sha256Hash = "truncate_hash";
    doc.fileName = "truncate_doc.txt";
    doc.filePath = "/test/truncate_doc.txt";
    doc.fileSize = 100;
    doc.mimeType = "text/plain";
    doc.createdTime = now;
    doc.modifiedTime = now;
    auto insertResult = fix.repository_->insertDocument(doc);
    REQUIRE((insertResult.has_value()));

    // Insert large content (10KB)
    std::string largeContent(10000, 'x');
    std::vector<BatchContentEntry> entries = {
        makeBatchContentEntry(insertResult.value(), "Title", largeContent)};
    auto batchResult = fix.repository_->batchInsertContentAndIndex(entries);
    REQUIRE((batchResult.has_value()));

    std::vector<std::string> hashes = {"truncate_hash"};
    auto result = fix.repository_->batchGetDocumentsWithContentPreview(hashes, 200);
    REQUIRE((result.has_value()));

    auto& [info, preview] = result.value().at("truncate_hash");
    CHECK((preview.size() <= 200));
    CHECK_FALSE(preview.empty());
}

TEST_CASE("MetadataRepository: setMetadataBatch preserves typed values and latest write wins",
          "[unit][metadata][repository][batch][metadata-value]") {
    MetadataRepositoryFixture fix;

    auto doc = makeDocumentWithPath("/tmp/typed-metadata.txt", "typed-metadata-hash");
    auto insert = fix.repository_->insertDocument(doc);
    REQUIRE((insert.has_value()));
    const int64_t docId = insert.value();

    const std::vector<uint8_t> blob = {0x00, 0x01, 0x41, 0xFE, 0xFF};
    std::vector<std::tuple<int64_t, std::string, MetadataValue>> entries = {
        {docId, "author", MetadataValue("old")},
        {docId, "author", MetadataValue("new")},
        {docId, "revision", MetadataValue(int64_t{42})},
        {docId, "score", MetadataValue(0.875)},
        {docId, "published", MetadataValue(true)},
        {docId, "payload", MetadataValue::fromBlob(blob)},
    };

    auto setBatch = fix.repository_->setMetadataBatch(entries);
    REQUIRE((setBatch.has_value()));

    auto all = fix.repository_->getAllMetadata(docId);
    REQUIRE((all.has_value()));
    REQUIRE((all.value().size() == 5));

    CHECK((all.value().at("author").type == MetadataValueType::String));
    CHECK((all.value().at("author").asString() == "new"));
    CHECK((all.value().at("revision").type == MetadataValueType::Integer));
    CHECK((all.value().at("revision").asInteger() == 42));
    CHECK((all.value().at("score").type == MetadataValueType::Real));
    CHECK_THAT(all.value().at("score").asReal(), Catch::Matchers::WithinRel(0.875, 0.0001));
    CHECK((all.value().at("published").type == MetadataValueType::Boolean));
    CHECK((all.value().at("published").asBoolean()));
    CHECK((all.value().at("payload").type == MetadataValueType::Blob));

    auto payload = all.value().at("payload").asVariant();
    REQUIRE((std::holds_alternative<std::vector<uint8_t>>(payload)));
    CHECK((std::get<std::vector<uint8_t>>(payload) == blob));

    std::vector<int64_t> ids{docId, 999999};
    auto batch = fix.repository_->getMetadataForDocuments(ids);
    REQUIRE((batch.has_value()));
    REQUIRE((batch.value().contains(docId)));
    CHECK((batch.value().at(docId).at("author").asString() == "new"));
    CHECK_FALSE(batch.value().contains(999999));
}

TEST_CASE("MetadataRepository: setMetadataBatch does not amplify lock retries",
          "[unit][metadata][repository][batch][metadata][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_set_metadata_batch_lock_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc = makeDocumentWithPath("/tmp/metadata-lock.txt", "metadata-lock-hash");
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    const auto start = std::chrono::steady_clock::now();
    auto result = repository->setMetadataBatch(
        {{insert.value(), "title", MetadataValue("locked metadata write")}});
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((result.error().message.find("locked") != std::string::npos));
    CHECK((elapsed < std::chrono::milliseconds(3000)));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataRepository: feedback event write does not wait for a busy pool",
          "[unit][metadata][repository][feedback][contention]") {
    const auto dbPath = tempDbPath("metadata_repo_feedback_pool_busy_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    auto heldConnResult = repoPool->acquire();
    REQUIRE((heldConnResult.has_value()));
    auto heldConn = std::move(heldConnResult).value();

    FeedbackEvent event;
    event.eventId = "feedback-busy-pool-event";
    event.traceId = "feedback-busy-pool-trace";
    event.createdAt = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    event.source = "test";
    event.eventType = "retrieval_served";
    event.payloadJson = "{\"served\":1}";

    const auto start = std::chrono::steady_clock::now();
    auto result = repository->insertFeedbackEvent(event);
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);

    REQUIRE_FALSE((result.has_value()));
    CHECK((elapsed < std::chrono::milliseconds(250)));
    CHECK((result.error().code == ErrorCode::Timeout ||
           result.error().code == ErrorCode::DatabaseError ||
           result.error().message.find("Timeout acquiring connection") != std::string::npos));

    heldConn.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataWriteFacade: fallback path batches writes until flush",
          "[unit][daemon][metadata-write-facade][fallback]") {
    const auto dbPath = tempDbPath("metadata_write_facade_fallback_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);
    auto doc = makeDocumentWithPath("/tmp/facade-fallback.txt", "facade-fallback-hash");
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));
    const auto docId = insert.value();
    REQUIRE((repository->updateDocumentExtractionStatus(docId, true, ExtractionStatus::Success)
                 .has_value()));
    REQUIRE(
        (repository->batchUpdateDocumentRepairStatuses({doc.sha256Hash}, RepairStatus::Completed)
             .has_value()));

    yams::daemon::MetadataWriteFacade facade(nullptr, repository.get());
    facade.setMetadata(docId, "title", MetadataValue(std::string("title-1")));
    facade.setMetadata(docId, "title", MetadataValue(std::string("title-2")));
    facade.updateExtractionStatus(docId, false, ExtractionStatus::Pending,
                                  "reset by fallback facade");
    facade.updateRepairStatus({doc.sha256Hash}, RepairStatus::Processing);
    facade.updateRepairStatus({doc.sha256Hash}, RepairStatus::Pending);

    auto metadataBeforeFlush = repository->getMetadata(docId, "title");
    REQUIRE((metadataBeforeFlush.has_value()));
    CHECK_FALSE((metadataBeforeFlush.value().has_value()));

    auto docBeforeFlush = repository->getDocument(docId);
    REQUIRE((docBeforeFlush.has_value()));
    REQUIRE((docBeforeFlush.value().has_value()));
    CHECK((docBeforeFlush.value()->extractionStatus == ExtractionStatus::Success));
    CHECK((docBeforeFlush.value()->repairStatus == RepairStatus::Completed));

    facade.flush();

    auto metadataAfterFlush = repository->getMetadata(docId, "title");
    REQUIRE((metadataAfterFlush.has_value()));
    REQUIRE((metadataAfterFlush.value().has_value()));
    CHECK((metadataAfterFlush.value()->asString() == "title-2"));

    auto docAfterFlush = repository->getDocument(docId);
    REQUIRE((docAfterFlush.has_value()));
    REQUIRE((docAfterFlush.value().has_value()));
    CHECK((docAfterFlush.value()->extractionStatus == ExtractionStatus::Pending));
    CHECK((docAfterFlush.value()->repairStatus == RepairStatus::Pending));
    CHECK_FALSE((docAfterFlush.value()->contentExtracted));

    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataWriteFacade: fallback flush retains pending writes after a transient lock",
          "[unit][daemon][metadata-write-facade][fallback][contention]") {
    const auto dbPath = tempDbPath("metadata_write_facade_retry_");

    ConnectionPoolConfig repoCfg;
    repoCfg.minConnections = 1;
    repoCfg.maxConnections = 1;
    repoCfg.busyTimeout = std::chrono::milliseconds(10);
    auto repoPool = std::make_unique<ConnectionPool>(dbPath.string(), repoCfg);
    REQUIRE((repoPool->initialize().has_value()));

    auto repository = std::make_unique<MetadataRepository>(
        *repoPool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    ConnectionPoolConfig lockerCfg;
    lockerCfg.minConnections = 1;
    lockerCfg.maxConnections = 1;
    lockerCfg.busyTimeout = std::chrono::milliseconds(10);
    auto lockerPool = std::make_unique<ConnectionPool>(dbPath.string(), lockerCfg);
    REQUIRE((lockerPool->initialize().has_value()));

    auto doc = makeDocumentWithPath("/tmp/facade-fallback-retry.txt", "facade-fallback-retry-hash");
    auto insert = repository->insertDocument(doc);
    REQUIRE((insert.has_value()));
    const auto docId = insert.value();
    REQUIRE((repository->updateDocumentExtractionStatus(docId, true, ExtractionStatus::Success)
                 .has_value()));
    REQUIRE(
        (repository->batchUpdateDocumentRepairStatuses({doc.sha256Hash}, RepairStatus::Completed)
             .has_value()));

    yams::daemon::MetadataWriteFacade facade(nullptr, repository.get());
    facade.setMetadata(docId, "title", MetadataValue(std::string("title-after-retry")));
    facade.updateExtractionStatus(docId, false, ExtractionStatus::Pending,
                                  "retry after lock clears");
    facade.updateRepairStatus({doc.sha256Hash}, RepairStatus::Pending);

    auto heldConn = lockerPool->acquire();
    REQUIRE((heldConn.has_value()));
    REQUIRE(((*heldConn.value())->execute("BEGIN IMMEDIATE").has_value()));

    facade.flush();

    auto metadataDuringLock = repository->getMetadata(docId, "title");
    REQUIRE((metadataDuringLock.has_value()));
    CHECK_FALSE((metadataDuringLock.value().has_value()));

    auto docDuringLock = repository->getDocument(docId);
    REQUIRE((docDuringLock.has_value()));
    REQUIRE((docDuringLock.value().has_value()));
    CHECK((docDuringLock.value()->extractionStatus == ExtractionStatus::Success));
    CHECK((docDuringLock.value()->repairStatus == RepairStatus::Completed));
    CHECK((docDuringLock.value()->contentExtracted));

    REQUIRE(((*heldConn.value())->execute("ROLLBACK").has_value()));

    facade.flush();

    auto metadataAfterRetry = repository->getMetadata(docId, "title");
    REQUIRE((metadataAfterRetry.has_value()));
    REQUIRE((metadataAfterRetry.value().has_value()));
    CHECK((metadataAfterRetry.value()->asString() == "title-after-retry"));

    auto docAfterRetry = repository->getDocument(docId);
    REQUIRE((docAfterRetry.has_value()));
    REQUIRE((docAfterRetry.value().has_value()));
    CHECK((docAfterRetry.value()->extractionStatus == ExtractionStatus::Pending));
    CHECK((docAfterRetry.value()->repairStatus == RepairStatus::Pending));
    CHECK_FALSE((docAfterRetry.value()->contentExtracted));

    lockerPool->shutdown();
    lockerPool.reset();
    repository.reset();
    repoPool->shutdown();
    repoPool.reset();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataRepository: getMetadataForDocuments chunks large batches",
          "[unit][metadata][repository][batch][metadata]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/chunk-a.txt", "chunk-metadata-hash-a");
    auto docB = makeDocumentWithPath("/tmp/chunk-b.txt", "chunk-metadata-hash-b");
    auto docAInsert = fix.repository_->insertDocument(docA);
    auto docBInsert = fix.repository_->insertDocument(docB);
    REQUIRE((docAInsert.has_value()));
    REQUIRE((docBInsert.has_value()));

    const int64_t docAId = docAInsert.value();
    const int64_t docBId = docBInsert.value();
    REQUIRE((fix.repository_->setMetadata(docAId, "author", MetadataValue("alpha")).has_value()));
    REQUIRE(
        (fix.repository_->setMetadata(docBId, "revision", MetadataValue(int64_t{7})).has_value()));

    constexpr std::size_t kLargeBatchSize = 33000;
    std::vector<int64_t> ids(kLargeBatchSize, 999999);
    ids.front() = docAId;
    ids[kLargeBatchSize / 2] = docBId;
    ids.back() = docAId;

    auto batch = fix.repository_->getMetadataForDocuments(ids);
    REQUIRE((batch.has_value()));
    REQUIRE((batch.value().size() == 2));
    REQUIRE((batch.value().contains(docAId)));
    REQUIRE((batch.value().contains(docBId)));
    CHECK((batch.value().at(docAId).at("author").asString() == "alpha"));
    CHECK((batch.value().at(docBId).at("revision").asInteger() == 7));
    CHECK_FALSE(batch.value().contains(999999));
}

TEST_CASE("MetadataRepository: batch document mutations preserve observable behavior",
          "[unit][metadata][repository][batch][document]") {
    MetadataRepositoryFixture fix;

    auto parent = makeDocumentWithPath("/tmp/batch-parent.txt", "batch-parent-hash");
    auto child = makeDocumentWithPath("/tmp/batch-child.txt", "batch-child-hash");
    auto survivor = makeDocumentWithPath("/tmp/batch-survivor.md", "batch-survivor-hash");

    auto parentInsert = fix.repository_->insertDocument(parent);
    auto childInsert = fix.repository_->insertDocument(child);
    auto survivorInsert = fix.repository_->insertDocument(survivor);
    REQUIRE((parentInsert.has_value()));
    REQUIRE((childInsert.has_value()));
    REQUIRE((survivorInsert.has_value()));

    const int64_t parentId = parentInsert.value();
    const int64_t childId = childInsert.value();
    const int64_t survivorId = survivorInsert.value();

    REQUIRE(fix.repository_
                ->insertContent(DocumentContent{parentId, "parent content", 14, "test", "en"})
                .has_value());
    REQUIRE(
        fix.repository_->setMetadata(parentId, "tag:batch", MetadataValue("delete")).has_value());
    REQUIRE(fix.repository_->indexDocumentContent(parentId, "parent", "parent content", "text")
                .has_value());

    DocumentRelationship rel;
    rel.parentId = parentId;
    rel.childId = childId;
    rel.relationshipType = RelationshipType::References;
    rel.createdTime = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    auto relInsert = fix.repository_->insertRelationship(rel);
    REQUIRE((relInsert.has_value()));

    auto mimeUpdate = fix.repository_->updateDocumentsMimeBatch(
        {{childId, "text/markdown"}, {survivorId, "application/json"}, {999999, "ignored/type"}});
    REQUIRE((mimeUpdate.has_value()));
    CHECK((mimeUpdate.value() == 2));

    auto updatedChild = fix.repository_->getDocument(childId);
    REQUIRE((updatedChild.has_value()));
    REQUIRE((updatedChild.value().has_value()));
    CHECK((updatedChild.value()->mimeType == "text/markdown"));

    auto deleted = fix.repository_->deleteDocumentsBatch({parentId, 999999});
    REQUIRE((deleted.has_value()));
    CHECK((deleted.value() == 1));

    auto missingParent = fix.repository_->getDocument(parentId);
    REQUIRE((missingParent.has_value()));
    CHECK_FALSE(missingParent.value().has_value());

    auto removedContent = fix.repository_->getContent(parentId);
    REQUIRE((removedContent.has_value()));
    CHECK_FALSE(removedContent.value().has_value());

    auto removedMetadata = fix.repository_->getAllMetadata(parentId);
    REQUIRE((removedMetadata.has_value()));
    CHECK((removedMetadata.value().empty()));

    auto childRels = fix.repository_->getRelationships(childId);
    REQUIRE((childRels.has_value()));
    CHECK((childRels.value().empty()));

    auto survivorDoc = fix.repository_->getDocument(survivorId);
    REQUIRE((survivorDoc.has_value()));
    REQUIRE((survivorDoc.value().has_value()));
    CHECK((survivorDoc.value()->mimeType == "application/json"));
}

TEST_CASE("MetadataRepository: batch get APIs skip missing inputs and keep keyed results",
          "[unit][metadata][repository][batch][read]") {
    MetadataRepositoryFixture fix;

    auto docA = makeDocumentWithPath("/tmp/batch-a.txt", "batch-get-hash-a");
    auto docB = makeDocumentWithPath("/tmp/batch-b.txt", "batch-get-hash-b");
    auto docC = makeDocumentWithPath("/tmp/batch-c.txt", "batch-get-hash-c");
    auto insertA = fix.repository_->insertDocument(docA);
    auto insertB = fix.repository_->insertDocument(docB);
    auto insertC = fix.repository_->insertDocument(docC);
    REQUIRE((insertA.has_value()));
    REQUIRE((insertB.has_value()));
    REQUIRE((insertC.has_value()));

    REQUIRE(fix.repository_
                ->insertContent(DocumentContent{insertA.value(), "alpha body", 10, "test", "en"})
                .has_value());
    REQUIRE(fix.repository_
                ->insertContent(DocumentContent{insertC.value(), "gamma body", 10, "test", "en"})
                .has_value());

    auto docs = fix.repository_->batchGetDocumentsByHash(
        {"batch-get-hash-a", "missing-hash", "batch-get-hash-c", "batch-get-hash-a"});
    REQUIRE((docs.has_value()));
    CHECK((docs.value().size() == 2));
    CHECK((docs.value().at("batch-get-hash-a").id == insertA.value()));
    CHECK((docs.value().at("batch-get-hash-c").id == insertC.value()));
    CHECK_FALSE(docs.value().contains("missing-hash"));

    auto content = fix.repository_->batchGetContent(
        {insertA.value(), insertB.value(), insertC.value(), 999999, insertA.value()});
    REQUIRE((content.has_value()));
    CHECK((content.value().size() == 2));
    CHECK((content.value().at(insertA.value()).contentText == "alpha body"));
    CHECK((content.value().at(insertC.value()).contentText == "gamma body"));
    CHECK_FALSE(content.value().contains(insertB.value()));
    CHECK_FALSE(content.value().contains(999999));

    auto emptyDocs = fix.repository_->batchGetDocumentsByHash({});
    REQUIRE((emptyDocs.has_value()));
    CHECK((emptyDocs.value().empty()));

    auto emptyContent = fix.repository_->batchGetContent({});
    REQUIRE((emptyContent.has_value()));
    CHECK((emptyContent.value().empty()));
}

TEST_CASE("MetadataRepository: removeFromIndexByHashBatch reports actual FTS removals",
          "[unit][metadata][repository][batch][fts5]") {
    MetadataRepositoryFixture fix;

    auto indexed = makeDocumentWithPath("/tmp/indexed.txt", "batch-indexed-hash");
    auto notIndexed = makeDocumentWithPath("/tmp/not-indexed.txt", "batch-not-indexed-hash");
    auto indexedInsert = fix.repository_->insertDocument(indexed);
    auto notIndexedInsert = fix.repository_->insertDocument(notIndexed);
    REQUIRE((indexedInsert.has_value()));
    REQUIRE((notIndexedInsert.has_value()));

    REQUIRE(fix.repository_
                ->indexDocumentContent(indexedInsert.value(), "indexed", "needle", "text/plain")
                .has_value());

    auto beforeIndexed = fix.repository_->hasFtsEntry(indexedInsert.value());
    auto beforeNotIndexed = fix.repository_->hasFtsEntry(notIndexedInsert.value());
    REQUIRE((beforeIndexed.has_value()));
    REQUIRE((beforeNotIndexed.has_value()));
    REQUIRE((beforeIndexed.value()));
    REQUIRE_FALSE(beforeNotIndexed.value());

    auto removed = fix.repository_->removeFromIndexByHashBatch(
        {"batch-indexed-hash", "batch-not-indexed-hash", "missing-hash"});
    REQUIRE((removed.has_value()));
    CHECK((removed.value() == 1));

    auto afterIndexed = fix.repository_->hasFtsEntry(indexedInsert.value());
    auto afterNotIndexed = fix.repository_->hasFtsEntry(notIndexedInsert.value());
    REQUIRE((afterIndexed.has_value()));
    REQUIRE((afterNotIndexed.has_value()));
    CHECK_FALSE(afterIndexed.value());
    CHECK_FALSE(afterNotIndexed.value());
}

TEST_CASE("MetadataRepository: batchInsertDocumentsWithMetadata coalesces inserts",
          "[metadata][batch][insert]") {
    MetadataRepositoryFixture fix;

    auto makeItem = [](const std::string& path, const std::string& hash, const std::string& tagKey,
                       const std::string& tagVal, bool pathTree) {
        BatchDocumentInsert item;
        item.info = makeDocumentWithPath(path, hash);
        item.tags.emplace_back(tagKey, MetadataValue(tagVal));
        item.updatePathTreeInTransaction = pathTree;
        item.initializePathSeriesInTransaction = true;
        return item;
    };

    std::vector<BatchDocumentInsert> items;
    items.push_back(makeItem("repo/batch/a.txt", "batch-hash-a", "stage", "alpha", true));
    items.push_back(makeItem("repo/batch/b.txt", "batch-hash-b", "stage", "beta", true));
    items.push_back(makeItem("repo/batch/c.txt", "batch-hash-c", "stage", "gamma", true));

    // Attach a snapshot to the middle item.
    TreeSnapshotRecord snap;
    snap.snapshotId = "batch-snap-1";
    snap.rootTreeHash = "batch-root-hash";
    snap.createdTime = 1700000000;
    snap.metadata["directory_path"] = "repo/batch";
    items[1].snapshot = snap;

    auto batchResult = fix.repository_->batchInsertDocumentsWithMetadata(items);
    REQUIRE((batchResult.has_value()));
    const auto outcomes = batchResult.value();
    std::vector<int64_t> ids;
    ids.reserve(outcomes.size());
    std::ranges::transform(outcomes, std::back_inserter(ids),
                           [](const auto& outcome) { return outcome.documentId; });
    REQUIRE((ids.size() == 3));
    CHECK(std::ranges::all_of(outcomes,
                              [](const auto& outcome) { return outcome.insertedNewDocument; }));
    CHECK(std::ranges::all_of(outcomes,
                              [](const auto& outcome) { return outcome.pathSeriesInitialized; }));

    SECTION("ids are valid, distinct, and resolve to the right documents") {
        std::vector<int64_t> sorted = ids;
        std::sort(sorted.begin(), sorted.end());
        CHECK((std::adjacent_find(sorted.begin(), sorted.end()) == sorted.end()));
        for (auto id : ids) {
            CHECK((id > 0));
        }
        const std::vector<std::string> hashes = {"batch-hash-a", "batch-hash-b", "batch-hash-c"};
        for (std::size_t i = 0; i < hashes.size(); ++i) {
            auto doc = fix.repository_->getDocumentByHash(hashes[i]);
            REQUIRE((doc.has_value()));
            REQUIRE((doc.value().has_value()));
            CHECK((doc.value().value().id == ids[i]));
        }
    }

    SECTION("metadata tags are persisted for each document") {
        const std::vector<std::pair<int64_t, std::string>> expected = {
            {ids[0], "alpha"}, {ids[1], "beta"}, {ids[2], "gamma"}};
        for (const auto& [id, val] : expected) {
            auto mv = fix.repository_->getMetadata(id, "stage");
            REQUIRE((mv.has_value()));
            REQUIRE((mv.value().has_value()));
            CHECK((mv.value().value().value == val));
        }
    }

    SECTION("fresh paths initialize canonical path-series metadata") {
        for (std::size_t i = 0; i < ids.size(); ++i) {
            auto version = fix.repository_->getMetadata(ids[i], "version");
            auto latest = fix.repository_->getMetadata(ids[i], "is_latest");
            auto seriesKey = fix.repository_->getMetadata(ids[i], "series_key");
            REQUIRE((version.has_value()));
            REQUIRE((version.value().has_value()));
            REQUIRE((latest.has_value()));
            REQUIRE((latest.value().has_value()));
            REQUIRE((seriesKey.has_value()));
            REQUIRE((seriesKey.value().has_value()));
            CHECK((version.value()->asInteger() == 1));
            CHECK((latest.value()->asBoolean()));
            CHECK((seriesKey.value()->asString() == items[i].info.filePath));
        }
    }

    SECTION("snapshot attached to one item is committed and bound to its document") {
        // All-or-nothing batch committed (batchResult ok), so the tree_snapshots row was written
        // without constraint failure. Verify the in-transaction wiring set ingestDocumentId.
        REQUIRE((items[1].snapshot.has_value()));
        REQUIRE((items[1].snapshot->ingestDocumentId.has_value()));
        CHECK((items[1].snapshot->ingestDocumentId.value() == ids[1]));
    }

    SECTION("existing-hash item in a later batch dedups to the same id") {
        std::vector<BatchDocumentInsert> dupItems;
        dupItems.push_back(makeItem("repo/batch/a.txt", "batch-hash-a", "stage", "alpha", true));
        auto dupResult = fix.repository_->batchInsertDocumentsWithMetadata(dupItems);
        REQUIRE((dupResult.has_value()));
        REQUIRE((dupResult.value().size() == 1));
        CHECK((dupResult.value()[0].documentId == ids[0]));
        CHECK_FALSE(dupResult.value()[0].insertedNewDocument);
        CHECK_FALSE(dupResult.value()[0].pathSeriesInitialized);
    }

    SECTION("a new hash at an existing path remains pending version assignment") {
        std::vector<BatchDocumentInsert> nextVersion;
        nextVersion.push_back(
            makeItem("repo/batch/a.txt", "batch-hash-a-v2", "stage", "delta", true));
        auto nextResult = fix.repository_->batchInsertDocumentsWithMetadata(nextVersion);
        REQUIRE((nextResult.has_value()));
        REQUIRE((nextResult.value().size() == 1));
        CHECK(nextResult.value()[0].insertedNewDocument);
        CHECK_FALSE(nextResult.value()[0].pathSeriesInitialized);

        auto version = fix.repository_->getMetadata(nextResult.value()[0].documentId, "version");
        REQUIRE((version.has_value()));
        CHECK_FALSE(version.value().has_value());
    }

    SECTION("empty batch is a no-op success") {
        std::vector<BatchDocumentInsert> empty;
        auto emptyResult = fix.repository_->batchInsertDocumentsWithMetadata(empty);
        REQUIRE((emptyResult.has_value()));
        CHECK((emptyResult.value().empty()));
    }
}

TEST_CASE("MetadataInsertWriter fallback preserves single-document insert outcome",
          "[metadata][batch][writer][fallback]") {
    auto dbPath = tempDbPath("metadata_insert_writer_fallback_outcome_");
    ConnectionPoolConfig config;
    config.minConnections = 1;
    config.maxConnections = 2;
    auto pool = std::make_shared<ConnectionPool>(dbPath.string(), config);
    REQUIRE((pool->initialize().has_value()));
    auto repo = std::make_shared<MetadataRepository>(
        *pool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);

    BatchDocumentInsert clean;
    clean.info = makeDocumentWithPath("repo/writer/fallback-clean.txt", "writer-fallback-clean");
    clean.initializePathSeriesInTransaction = true;

    BatchDocumentInsert poison;
    poison.info = makeDocumentWithPath("repo/writer/fallback-poison.txt", "writer-fallback-poison");
    TreeSnapshotRecord poisonSnapshot;
    poisonSnapshot.snapshotId = "writer-fallback-poison-snapshot";
    poisonSnapshot.rootTreeHash = "root";
    poisonSnapshot.createdTime = 1;
    poisonSnapshot.fileCount = 1;
    poison.snapshot = poisonSnapshot;
    // No directory_path metadata: tree_snapshots.directory_path is NOT NULL. This poisons the
    // coalesced batch transaction while leaving the clean document valid for per-item fallback.

    {
        MetadataInsertWriter writer(
            repo, MetadataInsertWriter::Options{.maxBatchCount = 2,
                                                .maxDelay = std::chrono::milliseconds{50}});
        auto cleanFuture = writer.submit(clean);
        auto poisonFuture = writer.submit(poison);

        auto cleanResult = cleanFuture.get();
        REQUIRE((cleanResult.has_value()));
        CHECK((cleanResult.value().documentId > 0));
        CHECK(cleanResult.value().insertedNewDocument);
        CHECK(cleanResult.value().pathSeriesInitialized);

        auto poisonResult = poisonFuture.get();
        REQUIRE_FALSE((poisonResult.has_value()));

        REQUIRE(writer.flush());
        const auto metrics = writer.metricsSnapshot();
        CHECK(metrics.failedBatches == 1);
        CHECK(metrics.fallbackItems == 2);
    }

    auto cleanDoc = repo->getDocumentByHash("writer-fallback-clean");
    REQUIRE((cleanDoc.has_value()));
    REQUIRE((cleanDoc.value().has_value()));
    auto version = repo->getMetadata(cleanDoc.value()->id, "version");
    REQUIRE((version.has_value()));
    CHECK((version.value().has_value()));

    auto poisonDoc = repo->getDocumentByHash("writer-fallback-poison");
    REQUIRE((poisonDoc.has_value()));
    CHECK_FALSE((poisonDoc.value().has_value()));

    pool->shutdown();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("MetadataInsertWriter coalesces concurrent submits into correct per-doc ids",
          "[metadata][batch][writer]") {
    auto dbPath = tempDbPath("metadata_insert_writer_test_");
    ConnectionPoolConfig config;
    config.minConnections = 1;
    config.maxConnections = 2;
    auto pool = std::make_shared<ConnectionPool>(dbPath.string(), config);
    REQUIRE((pool->initialize().has_value()));
    auto repo = std::make_shared<MetadataRepository>(
        *pool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);
    resetMetadataInsertPhaseTimings();

    constexpr int kDocs = 64;
    {
        MetadataInsertWriter writer(
            repo, MetadataInsertWriter::Options{.maxBatchCount = kDocs,
                                                .maxDelay = std::chrono::milliseconds{50}});

        // Submit from several threads to exercise concurrent coalescing.
        std::atomic<int> next{0};
        std::mutex futMutex;
        std::vector<std::future<Result<DocumentInsertOutcome>>> collected;
        collected.reserve(kDocs);
        constexpr int kThreads = 8;
        std::vector<std::thread> submitters;
        for (int t = 0; t < kThreads; ++t) {
            submitters.emplace_back([&]() {
                while (true) {
                    int i = next.fetch_add(1);
                    if (i >= kDocs)
                        break;
                    auto fut = writer.submit(makeIndexedBatchInsert(i));
                    std::lock_guard<std::mutex> lk(futMutex);
                    collected.push_back(std::move(fut));
                }
            });
        }
        for (auto& th : submitters)
            th.join();

        std::vector<int64_t> ids;
        ids.reserve(kDocs);
        for (auto& f : collected) {
            auto r = f.get();
            REQUIRE((r.has_value()));
            CHECK((r.value().documentId > 0));
            CHECK(r.value().insertedNewDocument);
            ids.push_back(r.value().documentId);
        }
        REQUIRE((static_cast<int>(ids.size()) == kDocs));
        std::sort(ids.begin(), ids.end());
        CHECK((std::adjacent_find(ids.begin(), ids.end()) == ids.end()));

        REQUIRE(writer.flush());
        const auto metrics = writer.metricsSnapshot();
        CHECK(metrics.submittedItems == kDocs);
        CHECK(metrics.completedItems == kDocs);
        CHECK(metrics.batchItems == kDocs);
        CHECK(metrics.batches > 0);
        CHECK(metrics.batches < kDocs);
        CHECK(metrics.maxBatchSize > 1);
        CHECK(metrics.queueWaitSamples == kDocs);
        CHECK(metrics.batchApplySamples == metrics.batches);
        CHECK(metrics.failedBatches == 0);
        CHECK(metrics.fallbackItems == 0);

        const auto phaseTimings = getMetadataInsertPhaseTimingsSnapshot();
        const auto transaction = phaseTimings.find("transaction_total");
        REQUIRE(transaction != phaseTimings.end());
        CHECK(transaction->second.calls == metrics.batches);
        CHECK(transaction->second.totalUs >= phaseTimings.at("commit").totalUs);

        writer.resetMetrics();
        const auto reset = writer.metricsSnapshot();
        CHECK(reset.submittedItems == 0);
        CHECK(reset.completedItems == 0);
        CHECK(reset.batches == 0);
        CHECK(reset.batchItems == 0);
    }

    // All documents are durable after the writer drains.
    for (int i = 0; i < kDocs; ++i) {
        auto doc = repo->getDocumentByHash("writer-hash-" + std::to_string(i));
        REQUIRE((doc.has_value()));
        CHECK((doc.value().has_value()));
    }

    pool->shutdown();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE("ContentIndexWriter commits content-index chunks from concurrent submits",
          "[metadata][batch][writer][content]") {
    auto dbPath = tempDbPath("content_index_writer_test_");
    ConnectionPoolConfig config;
    config.minConnections = 1;
    config.maxConnections = 2;
    auto pool = std::make_shared<ConnectionPool>(dbPath.string(), config);
    REQUIRE((pool->initialize().has_value()));
    auto repo = std::make_shared<MetadataRepository>(
        *pool, nullptr, MetadataRepository::SchemaBootstrapMode::AssumeReady);
    repo->initializeCounters();

    constexpr int kDocs = 32;
    std::vector<int64_t> docIds;
    for (int i = 0; i < kDocs; ++i) {
        DocumentInfo doc;
        doc.sha256Hash = "ciw-hash-" + std::to_string(i);
        doc.fileName = "ciw-" + std::to_string(i) + ".txt";
        doc.fileSize = 100;
        doc.mimeType = "text/plain";
        doc.createdTime =
            std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
        doc.modifiedTime = doc.createdTime;
        doc.contentExtracted = false;
        doc.extractionStatus = ExtractionStatus::Pending;
        auto r = repo->insertDocument(doc);
        REQUIRE((r.has_value()));
        docIds.push_back(r.value());
    }

    {
        ContentIndexWriter writer(repo);
        std::atomic<int> next{0};
        std::mutex futMutex;
        std::vector<std::future<Result<void>>> collected;
        std::vector<std::thread> submitters;
        for (int t = 0; t < 6; ++t) {
            submitters.emplace_back([&]() {
                while (true) {
                    int i = next.fetch_add(1);
                    if (i >= kDocs)
                        break;
                    std::vector<BatchContentEntry> chunk{makeBatchContentEntry(
                        docIds[static_cast<size_t>(i)], "Title " + std::to_string(i),
                        "Content body " + std::to_string(i))};
                    auto fut = writer.submit(std::move(chunk));
                    std::lock_guard<std::mutex> lk(futMutex);
                    collected.push_back(std::move(fut));
                }
            });
        }
        for (auto& th : submitters)
            th.join();
        for (auto& f : collected) {
            auto r = f.get();
            CHECK((r.has_value()));
        }
    }

    // Every document is now FTS-indexed after the writer drains.
    for (int i = 0; i < kDocs; ++i) {
        auto fts = repo->hasFtsEntry(docIds[static_cast<size_t>(i)]);
        REQUIRE((fts.has_value()));
        CHECK((fts.value()));
    }

    pool->shutdown();
    std::error_code ec;
    std::filesystem::remove(dbPath, ec);
}

TEST_CASE_METHOD(MetadataRepositoryFixture,
                 "MetadataRepository publishes its knowledge graph store concurrently",
                 "[metadata][repository][concurrency]") {
    auto kgStoreResult = makeSqliteKnowledgeGraphStore(*pool_);
    REQUIRE((kgStoreResult.has_value()));
    std::shared_ptr<KnowledgeGraphStore> kgStore = std::move(kgStoreResult.value());

    std::atomic<bool> start{false};
    std::atomic<bool> stop{false};
    std::atomic<bool> sawPublishedStore{false};
    std::atomic<bool> sawInvalidStore{false};

    std::thread reader([&]() {
        while (!start.load(std::memory_order_acquire)) {
            std::this_thread::yield();
        }
        while (!stop.load(std::memory_order_acquire)) {
            const auto snapshot = repository_->getKnowledgeGraphStore();
            if (snapshot) {
                if (snapshot != kgStore) {
                    sawInvalidStore.store(true, std::memory_order_release);
                }
                sawPublishedStore.store(true, std::memory_order_release);
            }
        }
    });

    start.store(true, std::memory_order_release);
    for (int i = 0; i < 1000; ++i) {
        repository_->setKnowledgeGraphStore(kgStore);
        repository_->setKnowledgeGraphStore(nullptr);
    }
    repository_->setKnowledgeGraphStore(kgStore);
    while (!sawPublishedStore.load(std::memory_order_acquire)) {
        std::this_thread::yield();
    }
    stop.store(true, std::memory_order_release);
    reader.join();

    CHECK_FALSE(sawInvalidStore.load(std::memory_order_acquire));
    CHECK((repository_->getKnowledgeGraphStore() == kgStore));
}
// NOLINTEND(bugprone-chained-comparison)

TEST_CASE("MetadataRepository batch begin mints one derivation per document",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    std::vector<std::string> hashes;
    for (int i = 0; i < 3; ++i) {
        const auto hash = "batch-derivation-" + std::to_string(i);
        auto inserted = fix.repository_->insertDocument(
            makeDocumentWithPath("/tmp/batch-derivation-" + std::to_string(i) + ".txt", hash));
        REQUIRE(inserted.has_value());
        hashes.push_back(hash);
    }
    // The first document is already embedded; minting must reset its readiness.
    REQUIRE(fix.repository_->batchCompleteDocumentEmbeddingsByHashes({hashes[0]}, "model-old")
                .has_value());
    hashes.push_back("batch-derivation-missing");

    auto tokens = fix.repository_->batchBeginDocumentEmbeddingDerivations(hashes, "recipe");
    REQUIRE(tokens.has_value());
    REQUIRE(tokens.value().size() == 3);
    for (std::size_t i = 0; i < 3; ++i) {
        CHECK(tokens.value()[i].hash == hashes[i]);
        CHECK(tokens.value()[i].recipe == "recipe");
        CHECK_FALSE(tokens.value()[i].generation.empty());
    }
    auto ready = fix.repository_->hasDocumentEmbeddingByHash(hashes[0]);
    REQUIRE(ready.has_value());
    CHECK_FALSE(ready.value());
    auto stats = fix.repository_->getCorpusStats();
    REQUIRE(stats.has_value());
    CHECK(stats.value().embeddingCount == 0);

    auto states = fix.repository_->batchGetDocumentEmbeddingDerivations(hashes);
    REQUIRE(states.has_value());
    CHECK(states.value().size() == 3);
    for (const auto& [hash, state] : states.value()) {
        CHECK_FALSE(state.completed);
    }
}

TEST_CASE("MetadataRepository batch complete publishes only current derivations",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::string current = "batch-complete-current";
    const std::string superseded = "batch-complete-superseded";
    const std::string changed = "batch-complete-changed";
    for (const auto& hash : {current, superseded, changed}) {
        auto inserted =
            fix.repository_->insertDocument(makeDocumentWithPath("/tmp/" + hash + ".txt", hash));
        REQUIRE(inserted.has_value());
        if (hash == changed) {
            auto entry = makeBatchContentEntry(inserted.value(), "Title", "Original text");
            REQUIRE(fix.repository_->batchInsertContentAndIndex({entry}).has_value());
        }
    }
    auto tokens = fix.repository_->batchBeginDocumentEmbeddingDerivations(
        {current, superseded, changed}, "recipe");
    REQUIRE(tokens.has_value());
    REQUIRE(tokens.value().size() == 3);

    // A newer attempt supersedes the second token; new content invalidates the third.
    REQUIRE(fix.repository_->beginDocumentEmbeddingDerivation(superseded, "recipe").has_value());
    {
        auto doc = fix.repository_->getDocumentByHash(changed);
        REQUIRE(doc.has_value());
        REQUIRE(doc.value().has_value());
        auto entry = makeBatchContentEntry(doc.value()->id, "Title", "New text");
        REQUIRE(fix.repository_->batchInsertContentAndIndex({entry}).has_value());
    }

    auto completed =
        fix.repository_->batchCompleteDocumentEmbeddingDerivations(tokens.value(), "model-new");
    REQUIRE(completed.has_value());
    CHECK(completed.value() == 1);

    auto ready = fix.repository_->hasDocumentEmbeddingByHash(current);
    REQUIRE(ready.has_value());
    CHECK(ready.value());
    for (const auto& hash : {superseded, changed}) {
        auto notReady = fix.repository_->hasDocumentEmbeddingByHash(hash);
        REQUIRE(notReady.has_value());
        CHECK_FALSE(notReady.value());
    }
    auto stats = fix.repository_->getCorpusStats();
    REQUIRE(stats.has_value());
    CHECK(stats.value().embeddingCount == 1);

    // Completing the same tokens again is a no-op.
    auto again =
        fix.repository_->batchCompleteDocumentEmbeddingDerivations(tokens.value(), "model-new");
    REQUIRE(again.has_value());
    CHECK(again.value() == 0);
}

TEST_CASE("MetadataRepository batch derivation failure rolls back the whole batch",
          "[unit][metadata][embeddings][derivation]") {
    MetadataRepositoryFixture fix;
    const std::vector<std::string> hashes{"batch-rollback-first", "batch-rollback-second"};
    for (const auto& hash : hashes) {
        REQUIRE(fix.repository_->insertDocument(makeDocumentWithPath("/tmp/" + hash, hash))
                    .has_value());
    }
    auto tokens = fix.repository_->batchBeginDocumentEmbeddingDerivations(hashes, "recipe");
    REQUIRE(tokens.has_value());
    REQUIRE(tokens.value().size() == 2);
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute(R"(
                    CREATE TRIGGER reject_second_batch_completion
                    BEFORE UPDATE OF repair_status ON documents
                    WHEN NEW.sha256_hash = 'batch-rollback-second'
                    BEGIN SELECT RAISE(ABORT, 'injected batch completion failure'); END
                )");
                })
                .has_value());
    REQUIRE_FALSE(
        fix.repository_->batchCompleteDocumentEmbeddingDerivations(tokens.value(), "model")
            .has_value());
    auto states = fix.repository_->batchGetDocumentEmbeddingDerivations(hashes);
    REQUIRE(states.has_value());
    REQUIRE(states.value().size() == 2);
    for (const auto& hash : hashes) {
        CHECK_FALSE(states.value().at(hash).completed);
        auto ready = fix.repository_->hasDocumentEmbeddingByHash(hash);
        REQUIRE(ready.has_value());
        CHECK_FALSE(ready.value());
    }
    REQUIRE(fix.pool_
                ->withConnection([](Database& db) -> Result<void> {
                    return db.execute("DROP TRIGGER reject_second_batch_completion");
                })
                .has_value());
    auto retried =
        fix.repository_->batchCompleteDocumentEmbeddingDerivations(tokens.value(), "model");
    REQUIRE(retried.has_value());
    CHECK(retried.value() == 2);
}

TEST_CASE("MetadataRepository reconcile keeps an open derivation unembedded",
          "[unit][metadata][embeddings][derivation]") {
    // Vector rows can exist for text that was superseded while its embed job ran. The
    // startup reconcile must not flip such a document back to embedded.
    MetadataRepositoryFixture fix;
    const std::string hash = "reconcile-open-derivation";
    REQUIRE(fix.repository_->insertDocument(makeDocumentWithPath("/tmp/reconcile-open.txt", hash))
                .has_value());
    auto tokens = fix.repository_->batchBeginDocumentEmbeddingDerivations({hash}, "recipe");
    REQUIRE(tokens.has_value());
    REQUIRE(tokens.value().size() == 1);

    REQUIRE(fix.repository_->reconcileDocumentEmbeddingStatusByHashes({hash}, "model").has_value());
    auto open = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(open.has_value());
    CHECK_FALSE(open.value());

    REQUIRE(fix.repository_->batchCompleteDocumentEmbeddingDerivations(tokens.value(), "model")
                .has_value());
    REQUIRE(fix.repository_->reconcileDocumentEmbeddingStatusByHashes({hash}, "model").has_value());
    auto completed = fix.repository_->hasDocumentEmbeddingByHash(hash);
    REQUIRE(completed.has_value());
    CHECK(completed.value());
}