#include <catch2/catch_test_macros.hpp>

#include <yams/daemon/components/db_recovery.h>
#include <yams/daemon/components/db_salvage.h>
#include <yams/daemon/components/RepairService.h>
#include <yams/metadata/database.h>

#include <sqlite3.h>
#include <algorithm>
#include <chrono>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <sstream>

#include "../../common/sqlite_corruption.h"

namespace fs = std::filesystem;

namespace {

fs::path makeScratchDir(const std::string& prefix) {
    auto base = fs::temp_directory_path() /
                (prefix + "_" +
                 std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
    fs::create_directories(base);
    return base;
}

const char* kDocumentsTableDdl = R"(
    CREATE TABLE documents (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        file_path TEXT NOT NULL,
        file_name TEXT NOT NULL,
        file_extension TEXT,
        file_size INTEGER NOT NULL,
        sha256_hash TEXT UNIQUE NOT NULL,
        mime_type TEXT,
        created_time INTEGER,
        modified_time INTEGER,
        indexed_time INTEGER,
        content_extracted BOOLEAN DEFAULT 0,
        extraction_status TEXT DEFAULT 'pending',
        extraction_error TEXT,
        path_prefix TEXT,
        reverse_path TEXT,
        path_hash TEXT,
        parent_hash TEXT,
        path_depth INTEGER DEFAULT 0,
        repair_status TEXT DEFAULT 'pending',
        repair_attempted_at INTEGER,
        repair_attempts INTEGER DEFAULT 0
    )
)";

const char* kDocumentsInsertSql = R"(
    INSERT INTO documents (
        file_path, file_name, file_extension, file_size, sha256_hash,
        mime_type, created_time, modified_time, indexed_time, content_extracted,
        extraction_status, extraction_error, path_prefix, reverse_path, path_hash,
        parent_hash, path_depth, repair_status, repair_attempted_at, repair_attempts
    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
)";

void createAndPopulateDb(const fs::path& dbPath, int docCount,
                         const std::vector<std::string>& hashes) {
    yams::metadata::Database db;
    auto open = db.open(dbPath.string(), yams::metadata::ConnectionMode::Create);
    REQUIRE(open);
    REQUIRE(db.execute(kDocumentsTableDdl));

    for (int i = 0; i < docCount; ++i) {
        const auto& hash = hashes[static_cast<size_t>(i)];
        auto stmtResult = db.prepare(kDocumentsInsertSql);
        REQUIRE(stmtResult);
        auto stmt = std::move(stmtResult).value();

        std::string filePath = "/test/file_" + std::to_string(i) + ".txt";
        std::string fileName = "file_" + std::to_string(i) + ".txt";
        std::string ext = "txt";
        int64_t fileSize = static_cast<int64_t>(1024) * (i + 1);

        REQUIRE(stmt.bind(1, filePath));
        REQUIRE(stmt.bind(2, fileName));
        REQUIRE(stmt.bind(3, ext));
        REQUIRE(stmt.bind(4, fileSize));
        REQUIRE(stmt.bind(5, hash));
        REQUIRE(stmt.bind(6, "text/plain"));
        REQUIRE(stmt.bind(7, static_cast<int64_t>(1000 + i)));
        REQUIRE(stmt.bind(8, static_cast<int64_t>(2000 + i)));
        REQUIRE(stmt.bind(9, static_cast<int64_t>(3000 + i)));
        REQUIRE(stmt.bind(10, 0));
        REQUIRE(stmt.bind(11, std::string("pending")));
        REQUIRE(stmt.bind(12, std::string("")));
        REQUIRE(stmt.bind(13, std::string("/test")));
        REQUIRE(stmt.bind(14, std::string("txt.0_elif/tset/")));
        REQUIRE(stmt.bind(15, std::string("path_hash_") + std::to_string(i)));
        REQUIRE(stmt.bind(16, std::string("parent_hash_") + std::to_string(i)));
        REQUIRE(stmt.bind(17, static_cast<int64_t>(1)));
        REQUIRE(stmt.bind(18, std::string("pending")));
        REQUIRE(stmt.bind(19, static_cast<int64_t>(0)));
        REQUIRE(stmt.bind(20, static_cast<int64_t>(0)));

        auto execResult = stmt.execute();
        REQUIRE(execResult);
    }

    db.close();
}

std::string getJournalMode(const fs::path& dbPath) {
    sqlite3* raw = nullptr;
    int rc = sqlite3_open_v2(dbPath.string().c_str(), &raw, SQLITE_OPEN_READWRITE, nullptr);
    REQUIRE((rc == SQLITE_OK));

    sqlite3_stmt* stmt = nullptr;
    rc = sqlite3_prepare_v2(raw, "PRAGMA journal_mode", -1, &stmt, nullptr);
    REQUIRE((rc == SQLITE_OK));

    std::string mode;
    if (sqlite3_step(stmt) == SQLITE_ROW) {
        const auto* text = reinterpret_cast<const char*>(sqlite3_column_text(stmt, 0));
        if (text != nullptr) {
            mode = text;
        }
    }
    sqlite3_finalize(stmt);
    sqlite3_close(raw);
    return mode;
}

int countDocuments(const fs::path& dbPath) {
    sqlite3* raw = nullptr;
    int rc = sqlite3_open_v2(dbPath.string().c_str(), &raw, SQLITE_OPEN_READONLY, nullptr);
    REQUIRE((rc == SQLITE_OK));

    sqlite3_stmt* stmt = nullptr;
    rc = sqlite3_prepare_v2(raw, "SELECT COUNT(*) FROM documents", -1, &stmt, nullptr);
    REQUIRE((rc == SQLITE_OK));

    int count = 0;
    if (sqlite3_step(stmt) == SQLITE_ROW) {
        count = sqlite3_column_int(stmt, 0);
    }
    sqlite3_finalize(stmt);
    sqlite3_close(raw);
    return count;
}

std::vector<std::string> makeTestHashes(int count) {
    std::vector<std::string> hashes;
    hashes.reserve(static_cast<size_t>(count));
    for (int i = 0; i < count; ++i) {
        std::ostringstream os;
        os << std::hex << std::setfill('0') << std::setw(64) << (1000000 + i);
        hashes.push_back(os.str());
    }
    return hashes;
}

} // namespace

TEST_CASE("salvageFromCorruptDb returns empty result when corrupt DB is missing",
          "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_missing");
    auto freshPath = dir / "fresh.db";

    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    auto corruptPath = dir / "nonexistent.db";
    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);
    REQUIRE((res.value().documentsSalvaged == 0));
    REQUIRE_FALSE(res.value().diagnostics.empty());

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb copies documents via ATTACH", "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_attach");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(5);

    createAndPopulateDb(corruptPath, 5, hashes);
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);
    REQUIRE((res.value().documentsSalvaged == 5));
    REQUIRE((res.value().documentsFailed == 0));

    REQUIRE((countDocuments(freshPath) == 5));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb preserves fresh DB journaling mode", "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_journal");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(2);

    createAndPopulateDb(corruptPath, 2, hashes);
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        REQUIRE(db.execute("PRAGMA journal_mode = WAL"));
        db.close();
    }

    REQUIRE((getJournalMode(freshPath) == "wal"));
    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);
    CHECK((res.value().documentsSalvaged == 2));
    CHECK((getJournalMode(freshPath) == "wal"));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb handles large document counts", "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_large");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(100);

    createAndPopulateDb(corruptPath, 100, hashes);

    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);
    REQUIRE((res.value().documentsSalvaged == 100));
    REQUIRE((res.value().documentsFailed == 0));

    REQUIRE((countDocuments(freshPath) == 100));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb skips duplicate hashes", "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_dupes");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(3);

    createAndPopulateDb(corruptPath, 3, hashes);

    // Pre-populate fresh DB with one of the same hashes
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));

        auto stmtResult = db.prepare(kDocumentsInsertSql);
        REQUIRE(stmtResult);
        auto stmt = std::move(stmtResult).value();
        REQUIRE(stmt.bind(1, std::string("/test/dup.txt")));
        REQUIRE(stmt.bind(2, std::string("dup.txt")));
        REQUIRE(stmt.bind(3, std::string("txt")));
        REQUIRE(stmt.bind(4, static_cast<int64_t>(100)));
        REQUIRE(stmt.bind(5, hashes[0]));
        REQUIRE(stmt.bind(6, std::string("text/plain")));
        REQUIRE(stmt.bind(7, static_cast<int64_t>(1)));
        REQUIRE(stmt.bind(8, static_cast<int64_t>(2)));
        REQUIRE(stmt.bind(9, static_cast<int64_t>(3)));
        REQUIRE(stmt.bind(10, 0));
        REQUIRE(stmt.bind(11, std::string("pending")));
        REQUIRE(stmt.bind(12, std::string("")));
        REQUIRE(stmt.bind(13, std::string("/")));
        REQUIRE(stmt.bind(14, std::string("")));
        REQUIRE(stmt.bind(15, std::string("")));
        REQUIRE(stmt.bind(16, std::string("")));
        REQUIRE(stmt.bind(17, static_cast<int64_t>(0)));
        REQUIRE(stmt.bind(18, std::string("pending")));
        REQUIRE(stmt.bind(19, static_cast<int64_t>(0)));
        REQUIRE(stmt.bind(20, static_cast<int64_t>(0)));
        REQUIRE(stmt.execute());
        db.close();
    }

    // The duplicate hash should be skipped (INSERT OR IGNORE)
    // So 2 new docs + 1 existing = 3 total
    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);

    REQUIRE((countDocuments(freshPath) == 3));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb fails cleanly when fresh DB is missing",
          "[unit][daemon][db_salvage]") {
    auto dir = makeScratchDir("yams_salvage_fresh_missing");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(1);

    createAndPopulateDb(corruptPath, 1, hashes);

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE_FALSE(res);

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("recovery sentinel cleanup is independent of salvage need",
          "[unit][daemon][db_salvage][startup_refactor]") {
    auto dir = makeScratchDir("yams_recovery_sentinel_cleanup");
    auto dbPath = dir / "yams.db";
    auto hashes = makeTestHashes(2);
    createAndPopulateDb(dbPath, 2, hashes);

    const auto sentinelA = dir / "yams.db.recovered-20260101T000000Z";
    const auto sentinelB = dir / "yams.db.recovered-20260101T000001Z";
    const auto unrelated = dir / "other.db.recovered-20260101T000000Z";
    {
        std::ofstream(sentinelA) << "recovered";
        std::ofstream(sentinelB) << "recovered";
        std::ofstream(unrelated) << "keep";
    }

    const auto qc = yams::daemon::quickCheckSalvageNeeded(dir, dbPath);
    REQUIRE_FALSE(qc.needsSalvage);
    REQUIRE((qc.corruptDbCount == 0));

    const auto cleanup = yams::daemon::removeRecoverySentinels(dbPath);
    REQUIRE(cleanup.errors.empty());
    REQUIRE((cleanup.removed.size() == 2));
    REQUIRE_FALSE(fs::exists(sentinelA));
    REQUIRE_FALSE(fs::exists(sentinelB));
    REQUIRE(fs::exists(unrelated));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("quick salvage check tracks unreadable corrupt DBs without requesting cleanup",
          "[unit][daemon][db_salvage][startup_refactor]") {
    auto dir = makeScratchDir("yams_salvage_unreadable_corrupt");
    auto dbPath = dir / "yams.db";
    auto hashes = makeTestHashes(1);
    createAndPopulateDb(dbPath, 1, hashes);

    const auto unreadableCorrupt = dir / "yams.db.corrupt-bad";
    {
        std::ofstream(unreadableCorrupt, std::ios::binary) << "not a sqlite database";
    }

    const auto qc = yams::daemon::quickCheckSalvageNeeded(dir, dbPath);
    REQUIRE((qc.corruptDbCount == 1));
    REQUIRE((qc.unreadableCorruptDbCount == 1));
    REQUIRE_FALSE(qc.needsSalvage);
    REQUIRE((qc.maxCorruptCount == 0));
    REQUIRE(fs::exists(unreadableCorrupt));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB cleanup removes confirmed artifacts but keeps unreadable evidence",
          "[unit][daemon][db_salvage][startup_refactor][retention]") {
    auto dir = makeScratchDir("yams_corrupt_cleanup_safe");
    auto livePath = dir / "yams.db";
    auto readableCorrupt = dir / "yams.db.corrupt-readable";
    auto unreadableCorrupt = dir / "yams.db.corrupt-unreadable";
    auto readableWal = fs::path(readableCorrupt.string() + "-wal");
    auto readableShm = fs::path(readableCorrupt.string() + "-shm");
    auto hashes = makeTestHashes(1);

    createAndPopulateDb(livePath, 1, hashes);
    createAndPopulateDb(readableCorrupt, 1, hashes);
    {
        std::ofstream(readableWal) << "wal";
        std::ofstream(readableShm) << "shm";
        std::ofstream(unreadableCorrupt, std::ios::binary) << "not sqlite";
    }

    const auto cleanup = yams::daemon::removeSalvagedCorruptDbs(livePath, std::chrono::seconds{0});
    REQUIRE(cleanup.errors.empty());
    REQUIRE((cleanup.removed.size() == 1));
    REQUIRE((cleanup.retained.size() == 1));
    CHECK((cleanup.retained.front().path.filename() == unreadableCorrupt.filename()));
    REQUIRE((cleanup.removed.front().filename() == readableCorrupt.filename()));
    REQUIRE_FALSE(fs::exists(readableCorrupt));
    REQUIRE_FALSE(fs::exists(readableWal));
    REQUIRE_FALSE(fs::exists(readableShm));
    REQUIRE(fs::exists(unreadableCorrupt));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB cleanup keeps readable artifacts whose documents never reached the live DB",
          "[unit][daemon][db_salvage][retention]") {
    auto dir = makeScratchDir("yams_corrupt_cleanup_unsalvaged");
    auto livePath = dir / "yams.db";
    auto corruptPath = dir / "yams.db.corrupt-20200101T000000Z";
    auto hashes = makeTestHashes(3);

    // The live DB only has the first document: salvage of the other two never happened
    // (or failed), so the corrupt DB is still the only copy of them.
    createAndPopulateDb(livePath, 1, hashes);
    createAndPopulateDb(corruptPath, 3, hashes);

    const auto cleanup = yams::daemon::removeSalvagedCorruptDbs(livePath, std::chrono::seconds{0});
    CHECK(cleanup.removed.empty());
    REQUIRE((cleanup.retained.size() == 1));
    CHECK((cleanup.retained.front().reason == "documents missing from live DB"));
    REQUIRE(fs::exists(corruptPath));
    CHECK((countDocuments(corruptPath) == 3));

    auto confirmed = yams::daemon::corruptDbSalvageConfirmed(corruptPath, livePath);
    REQUIRE(confirmed);
    CHECK_FALSE(confirmed.value());

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB cleanup honours the retention window even when salvage is confirmed",
          "[unit][daemon][db_salvage][retention]") {
    using namespace std::chrono;
    auto dir = makeScratchDir("yams_corrupt_cleanup_retention");
    auto livePath = dir / "yams.db";
    auto corruptPath = dir / "yams.db.corrupt-20260301T120000Z";
    auto hashes = makeTestHashes(2);
    createAndPopulateDb(livePath, 2, hashes);
    createAndPopulateDb(corruptPath, 2, hashes);

    const auto quarantinedAt = sys_days{2026y / March / 1} + 12h;
    const auto retention = duration_cast<seconds>(hours{24 * 7});

    SECTION("young artifact is kept") {
        const auto cleanup =
            yams::daemon::removeSalvagedCorruptDbs(livePath, retention, quarantinedAt + 6 * 24h);
        CHECK(cleanup.removed.empty());
        REQUIRE((cleanup.retained.size() == 1));
        CHECK((cleanup.retained.front().reason == "within retention window"));
        CHECK(fs::exists(corruptPath));
    }
    SECTION("aged, confirmed artifact is removed") {
        const auto cleanup =
            yams::daemon::removeSalvagedCorruptDbs(livePath, retention, quarantinedAt + 8 * 24h);
        REQUIRE((cleanup.removed.size() == 1));
        CHECK(cleanup.retained.empty());
        CHECK_FALSE(fs::exists(corruptPath));
    }

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB cleanup falls back to mtime when the name has no timestamp",
          "[unit][daemon][db_salvage][retention]") {
    using namespace std::chrono;
    auto dir = makeScratchDir("yams_corrupt_cleanup_mtime");
    auto livePath = dir / "yams.db";
    auto corruptPath = dir / "yams.db.corrupt-manual-copy";
    auto hashes = makeTestHashes(1);
    createAndPopulateDb(livePath, 1, hashes);
    createAndPopulateDb(corruptPath, 1, hashes);
    const auto retention = duration_cast<seconds>(hours{24 * 7});

    auto cleanup = yams::daemon::removeSalvagedCorruptDbs(livePath, retention);
    CHECK(cleanup.removed.empty());
    CHECK(fs::exists(corruptPath));

    fs::last_write_time(corruptPath, fs::file_time_type::clock::now() - hours{24 * 10});
    cleanup = yams::daemon::removeSalvagedCorruptDbs(livePath, retention);
    CHECK((cleanup.removed.size() == 1));
    CHECK_FALSE(fs::exists(corruptPath));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB cleanup keeps everything when the live DB is missing",
          "[unit][daemon][db_salvage][retention]") {
    auto dir = makeScratchDir("yams_corrupt_cleanup_no_live");
    auto corruptPath = dir / "yams.db.corrupt-20200101T000000Z";
    createAndPopulateDb(corruptPath, 1, makeTestHashes(1));

    const auto cleanup =
        yams::daemon::removeSalvagedCorruptDbs(dir / "yams.db", std::chrono::seconds{0});
    CHECK(cleanup.removed.empty());
    CHECK((cleanup.retained.size() == 1));
    CHECK(fs::exists(corruptPath));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("corrupt DB artifact names are recognised in one place",
          "[unit][daemon][db_salvage][retention]") {
    using yams::daemon::isCorruptDbFileName;
    CHECK((yams::daemon::corruptDbPrefix() == "yams.db.corrupt-"));
    CHECK(isCorruptDbFileName("yams.db.corrupt-20260101T000000Z"));
    CHECK(isCorruptDbFileName("yams.db.corrupt-20260101T000000Z.1"));
    CHECK_FALSE(isCorruptDbFileName("yams.db.corrupt-20260101T000000Z-wal"));
    CHECK_FALSE(isCorruptDbFileName("yams.db.corrupt-20260101T000000Z-shm"));
    CHECK_FALSE(isCorruptDbFileName("yams.db.corrupt-"));
    CHECK_FALSE(isCorruptDbFileName("vectors.db.corrupt-20260101T000000Z"));
    CHECK(isCorruptDbFileName("vectors.db.corrupt-20260101T000000Z", "vectors.db"));

    auto dir = makeScratchDir("yams_corrupt_list");
    for (const auto* name : {"yams.db", "yams.db.corrupt-a", "yams.db.corrupt-a-wal",
                             "yams.db-wal.quarantine-20260101T000000Z", "vectors.db.corrupt-b"}) {
        std::ofstream(dir / name) << "x";
    }
    const auto listed = yams::daemon::listCorruptDbs(dir);
    REQUIRE((listed.size() == 1));
    CHECK((listed.front().filename() == "yams.db.corrupt-a"));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("RepairService maintenance removes only aged, confirmed corrupt DBs",
          "[unit][daemon][db_salvage][retention][repair]") {
    auto dir = makeScratchDir("yams_repair_recovery_artifacts");
    auto livePath = dir / "yams.db";
    auto hashes = makeTestHashes(3);
    createAndPopulateDb(livePath, 2, hashes);

    const auto oldConfirmed = dir / "yams.db.corrupt-20200101T000000Z";
    const auto oldUnconfirmed = dir / "yams.db.corrupt-20200102T000000Z";
    createAndPopulateDb(oldConfirmed, 2, hashes);
    createAndPopulateDb(oldUnconfirmed, 3, hashes);
    // Quarantine time in the future: always inside the retention window.
    auto youngCorrupt = dir / "yams.db.corrupt-21000101T000000Z";
    createAndPopulateDb(youngCorrupt, 2, hashes);

    yams::daemon::RepairService::Config cfg;
    cfg.dataDir = dir;
    yams::daemon::RepairService service(
        yams::daemon::RepairServiceContext{}, nullptr, [] { return std::size_t{0}; }, cfg);
    const auto stats = service.runRecoveryArtifactMaintenance();

    CHECK((stats.corruptDbsRemoved == 1));
    CHECK((stats.corruptDbsRetained == 2));
    CHECK_FALSE(fs::exists(oldConfirmed));
    CHECK(fs::exists(oldUnconfirmed));
    CHECK(fs::exists(youngCorrupt));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

namespace {

void createAndPopulateWideDb(const fs::path& dbPath, int docCount,
                             const std::vector<std::string>& hashes) {
    yams::metadata::Database db;
    REQUIRE(db.open(dbPath.string(), yams::metadata::ConnectionMode::Create));
    REQUIRE(db.execute(kDocumentsTableDdl));
    REQUIRE(db.execute("BEGIN IMMEDIATE"));
    for (int i = 0; i < docCount; ++i) {
        auto stmtResult = db.prepare(kDocumentsInsertSql);
        REQUIRE(stmtResult);
        auto stmt = std::move(stmtResult).value();
        std::string filePath =
            "/test/" + std::string(1200, 'a' + (i % 26)) + "_" + std::to_string(i) + ".txt";
        REQUIRE(stmt.bind(1, filePath));
        REQUIRE(stmt.bind(2, "file_" + std::to_string(i) + ".txt"));
        REQUIRE(stmt.bind(3, std::string("txt")));
        REQUIRE(stmt.bind(4, static_cast<int64_t>(1024)));
        REQUIRE(stmt.bind(5, hashes[static_cast<size_t>(i)]));
        REQUIRE(stmt.bind(6, std::string("text/plain")));
        REQUIRE(stmt.bind(7, static_cast<int64_t>(1000 + i)));
        REQUIRE(stmt.bind(8, static_cast<int64_t>(2000 + i)));
        REQUIRE(stmt.bind(9, static_cast<int64_t>(3000 + i)));
        REQUIRE(stmt.bind(10, 0));
        REQUIRE(stmt.bind(11, std::string("pending")));
        REQUIRE(stmt.bind(12, std::string("")));
        REQUIRE(stmt.bind(13, std::string("/test")));
        REQUIRE(stmt.bind(14, std::string("")));
        REQUIRE(stmt.bind(15, std::string("path_hash_") + std::to_string(i)));
        REQUIRE(stmt.bind(16, std::string("")));
        REQUIRE(stmt.bind(17, static_cast<int64_t>(1)));
        REQUIRE(stmt.bind(18, std::string("pending")));
        REQUIRE(stmt.bind(19, static_cast<int64_t>(0)));
        REQUIRE(stmt.bind(20, static_cast<int64_t>(0)));
        REQUIRE(stmt.execute());
    }
    REQUIRE(db.execute("COMMIT"));
    REQUIRE(db.execute("PRAGMA wal_checkpoint(TRUNCATE)"));
    db.close();
}

bool hasFallbackDiagnostic(const yams::daemon::DbSalvageResult& result) {
    return std::any_of(result.diagnostics.begin(), result.diagnostics.end(),
                       [](const std::string& d) {
                           return d.find("Falling back to row-by-row salvage") != std::string::npos;
                       });
}

} // namespace

TEST_CASE("salvageFromCorruptDb fails cleanly on truncated DB and preserves fresh DB",
          "[unit][daemon][db_salvage][corruption]") {
    auto dir = makeScratchDir("yams_salvage_truncated");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(200);

    createAndPopulateWideDb(corruptPath, 200, hashes);
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    yams::test::truncateFile(corruptPath, 0.6);

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE_FALSE(res);

    yams::metadata::Database fresh;
    REQUIRE(fresh.open(freshPath.string(), yams::metadata::ConnectionMode::ReadWrite));
    REQUIRE(fresh.checkIntegrity());
    fresh.close();
    CHECK((countDocuments(freshPath) == 0));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvageFromCorruptDb row-by-row fallback survives mid-file page corruption",
          "[unit][daemon][db_salvage][corruption]") {
    auto dir = makeScratchDir("yams_salvage_pagesmash");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(200);

    createAndPopulateWideDb(corruptPath, 200, hashes);
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    const auto pageCount = yams::test::sqlitePageCount(corruptPath);
    REQUIRE(pageCount > 20);
    for (std::uint32_t page = pageCount / 2; page < (pageCount / 2) + 3; ++page) {
        yams::test::corruptPage(corruptPath, page);
    }

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);
    CHECK(hasFallbackDiagnostic(res.value()));
    CHECK((res.value().documentsSalvaged > 0));
    CHECK((res.value().documentsSalvaged < 200));
    CHECK((countDocuments(freshPath) == static_cast<int>(res.value().documentsSalvaged)));

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("salvaged fresh DB passes integrity check", "[unit][daemon][db_salvage][corruption]") {
    auto dir = makeScratchDir("yams_salvage_clean_output");
    auto corruptPath = dir / "corrupt.db";
    auto freshPath = dir / "fresh.db";
    auto hashes = makeTestHashes(200);

    createAndPopulateWideDb(corruptPath, 200, hashes);
    {
        yams::metadata::Database db;
        REQUIRE(db.open(freshPath.string(), yams::metadata::ConnectionMode::Create));
        REQUIRE(db.execute(kDocumentsTableDdl));
        db.close();
    }

    const auto pageCount = yams::test::sqlitePageCount(corruptPath);
    REQUIRE(pageCount > 20);
    yams::test::corruptPage(corruptPath, pageCount / 2);

    auto res = yams::daemon::salvageFromCorruptDb(corruptPath, freshPath);
    REQUIRE(res);

    yams::metadata::Database fresh;
    REQUIRE(fresh.open(freshPath.string(), yams::metadata::ConnectionMode::ReadWrite));
    REQUIRE(fresh.checkIntegrity());
    fresh.close();

    std::error_code ec;
    fs::remove_all(dir, ec);
}

TEST_CASE("RepairService maintenance sweeps stale clean-shutdown stamp leftovers",
          "[unit][daemon][db_salvage][repair][stamp_sweep]") {
    auto dir = makeScratchDir("yams_repair_stamp_sweep");
    auto livePath = dir / "yams.db";
    createAndPopulateDb(livePath, 1, makeTestHashes(1));
    const fs::path staleTmp(livePath.string() + ".integrity.tmp");
    std::ofstream(staleTmp) << "version=1\n";
    fs::last_write_time(staleTmp, fs::file_time_type::clock::now() - std::chrono::hours{2});

    yams::daemon::RepairService::Config cfg;
    cfg.dataDir = dir;
    yams::daemon::RepairService service(
        yams::daemon::RepairServiceContext{}, nullptr, [] { return std::size_t{0}; }, cfg);
    const auto stats = service.runRecoveryArtifactMaintenance();

    CHECK((stats.stampArtifactsRemoved == 1));
    CHECK_FALSE(fs::exists(staleTmp));

    std::error_code ec;
    fs::remove_all(dir, ec);
}