/**
 * Unit tests for v1.7.0 T4+T5+T6: chunked eviction, priority, quiesce, hot-write floor.
 *
 * Exercises MemoryStore eviction directly — no server.
 *
 * Test cases:
 *   1. Pressure levels reported correctly as memory grows
 *   2. hot_write_floor_ms protects recently-written docs
 *   3. Low-priority collection evicted more than High
 *   4. Eviction chunk size honored (doesn't drop everything at once)
 */

#include "../service/src/memory_store.hpp"
#include "../service/src/document.hpp"

#include <cassert>
#include <chrono>
#include <iostream>
#include <nlohmann/json.hpp>
#include <string>
#include <thread>
#include <unistd.h>

using namespace smartbotic::database;

namespace {

// Shared pass/fail counters. New assertions use check() rather than assert()
// so a failure names what it wanted and the binary keeps running to report
// everything, instead of aborting on the first one. (The older tests in this
// file still use assert(); tests/CMakeLists.txt applies -UNDEBUG so those are
// live in every build type - see the v2.4.3 note there.)
int g_pass = 0;
int g_fail = 0;

void check(bool cond, const char* msg) {
    if (cond) { ++g_pass; }
    else { ++g_fail; std::cerr << "FAIL: " << msg << "\n"; }
}

MemoryStore::Config evictionTestConfig(uint64_t maxMb = 1, uint32_t chunkSize = 100) {
    MemoryStore::Config cfg;
    cfg.nodeId = "test";
    cfg.maxMemoryBytes = maxMb * 1024 * 1024;
    cfg.evictionChunkSize = chunkSize;
    cfg.evictionChunkPauseMs = 1;        // fast for tests
    cfg.evictionCheckIntervalMs = 50;    // wake often
    cfg.hotWriteFloorMs = 100;           // short floor for tests
    cfg.memorySoftPercent = 60;
    cfg.memoryHardPercent = 80;
    cfg.memoryEmergencyPercent = 95;
    cfg.evictionTargetPercent = 50;
    return cfg;
}

Document makeDoc(const std::string& id, size_t bytes = 1024) {
    Document d;
    d.id = id;
    std::string pad(bytes, 'x');
    d.set_data(nlohmann::json{{"value", pad}});
    return d;
}

void fillStore(MemoryStore& store, const std::string& collection, int n, size_t docBytes = 1024) {
    for (int i = 0; i < n; ++i) {
        store.insert(collection, makeDoc("d" + std::to_string(i), docBytes));
    }
}

} // anonymous namespace

void test_pressure_levels() {
    auto cfg = evictionTestConfig(1);  // 1 MB cap
    cfg.memorySoftPercent = 50;
    cfg.memoryHardPercent = 75;
    cfg.memoryEmergencyPercent = 90;
    MemoryStore store(cfg);
    store.start();
    store.createCollection("docs", CollectionOptions{});

    assert(store.pressure() == MemoryPressure::Normal);

    // Fill to trigger some pressure. Exact bytes/doc depend on overhead.
    fillStore(store, "docs", 400, 1024);
    // May be Normal or Soft depending on exact sizes
    auto p = store.pressure();
    // Just assert pressure() returns a valid enum value and doesn't crash.
    assert(p == MemoryPressure::Normal || p == MemoryPressure::Soft ||
           p == MemoryPressure::Hard    || p == MemoryPressure::Emergency);

    store.stop();
    std::cout << "PASS: pressure level reported correctly as memory grows (level="
              << memoryPressureToString(p) << ")\n";
}

void test_hot_write_floor_protects_recent_writes() {
    auto cfg = evictionTestConfig(1);
    cfg.hotWriteFloorMs = 60000;  // very high — nothing in test is "old"
    cfg.evictionCheckIntervalMs = 20;
    cfg.memorySoftPercent = 10;   // force eviction immediately
    cfg.memoryHardPercent = 20;
    MemoryStore store(cfg);
    store.start();
    store.createCollection("docs", CollectionOptions{});

    // Fill above threshold — eviction should trigger
    fillStore(store, "docs", 500, 2048);

    // Wait for the eviction loop to tick at least twice
    std::this_thread::sleep_for(std::chrono::milliseconds(200));

    // With hot-write floor 60s, all recently-inserted docs should be
    // protected and live count should NOT drop significantly
    auto stats = store.getMemoryStatsSnapshot();
    uint64_t liveDocs = 0;
    for (const auto& c : stats.collections) liveDocs += c.documentCount;

    // Allow SOME eviction (could be non-hot-write edge case), but not most
    assert(liveDocs >= 450);  // at least 90% preserved by hot-write floor

    store.stop();
    std::cout << "PASS: hot-write floor protects recently-written docs ("
              << liveDocs << "/500 preserved)\n";
}

// Priority bias: with two equally-sized collections, eviction must take from
// the Low-priority one before the High-priority one.
//
// This test USED to call store.start() before filling, then sleep 500ms and
// compare survivor counts. That raced its own fill loop against the 20ms
// eviction tick, and the outcome depended on how fast inserts happened to be:
//
//   - Optimised build: all 800 inserts land inside the first 20ms tick, so
//     eviction sees both collections at 400 and the bias assertion holds.
//   - Unoptimised build (a plain `cmake -B build -G Ninja`, i.e. no
//     CMAKE_BUILD_TYPE, which is what CLAUDE.md documents): only ~290 inserts
//     land in 20ms, so the first tick fires while "important" is still filling
//     and "archive" is empty or absent. Eviction then evicts only from
//     "important", latches the v2.4.3 50%-per-episode drain cap against that
//     partial resident set, and stays PAUSED for the rest of the episode.
//     "archive" is filled afterwards and never evicted at all, so
//     archiveLive (400) > importantLive — a guaranteed failure that says
//     nothing about the priority logic.
//
// Measured 10 failures in 12 runs unoptimised, 0 in 12 optimised, at the
// relations merge-base as well as at branch HEAD. The product was never wrong;
// the test's precondition was unstated.
//
// So the precondition is now established BEFORE the eviction thread exists:
// fill both collections with the store stopped, assert the resident set is
// exactly what the comparison assumes, and only then start eviction. The
// settle wait polls until the live count stops moving instead of trusting a
// fixed 500ms, and a timeout fails loudly rather than asserting on a
// half-evicted store.
void test_priority_low_evicted_first() {
    // 4 MB cap, target 25%: the fill lands well above the hard threshold so
    // eviction has real work, and there are survivors left to compare.
    auto cfg = evictionTestConfig(4);
    cfg.hotWriteFloorMs = 0;           // disable hot-write protection
    cfg.evictionCheckIntervalMs = 20;
    cfg.memorySoftPercent = 30;        // early trickle
    cfg.memoryHardPercent = 50;        // hit hard with our fill
    cfg.memoryEmergencyPercent = 90;
    cfg.evictionTargetPercent = 25;    // clear down to 25% - leaves headroom
    cfg.evictionChunkSize = 50;        // small chunks, biased selection has effect
    cfg.maxEvictionPassesPerTrigger = 10;
    MemoryStore store(cfg);

    // Two collections: one HIGH priority, one LOW
    CollectionOptions highOpts;
    highOpts.memoryPriority = MemoryPriority::High;
    store.createCollection("important", highOpts);

    CollectionOptions lowOpts;
    lowOpts.memoryPriority = MemoryPriority::Low;
    store.createCollection("archive", lowOpts);

    // NOTE: store.start() has deliberately NOT been called yet. Nothing in
    // insert() needs the eviction or expiration thread, so the fill below runs
    // with no concurrent evictor and the resident set is fully determined.
    const uint64_t kPerCollection = 400;
    fillStore(store, "important", static_cast<int>(kPerCollection), 2048);
    fillStore(store, "archive", static_cast<int>(kPerCollection), 2048);

    auto liveCounts = [&]() {
        uint64_t important = 0, archive = 0;
        for (const auto& c : store.getMemoryStatsSnapshot().collections) {
            if (c.collection == "important") important = c.documentCount;
            if (c.collection == "archive")   archive = c.documentCount;
        }
        return std::pair<uint64_t, uint64_t>{important, archive};
    };

    // Precondition, asserted rather than assumed: both collections are whole,
    // and the store is over the hard threshold so eviction is guaranteed to
    // run. If a future change makes the fill cheap enough not to trip
    // pressure, this fails loudly instead of the test passing vacuously.
    {
        auto [important0, archive0] = liveCounts();
        check(important0 == kPerCollection,
              "precondition: 'important' fully resident before eviction starts");
        check(archive0 == kPerCollection,
              "precondition: 'archive' fully resident before eviction starts");
        const MemoryPressure p0 = store.pressure();
        check(p0 == MemoryPressure::Hard || p0 == MemoryPressure::Emergency,
              "precondition: fill must put the store under hard/emergency pressure");
    }

    store.start();

    // Settle: poll until the live count has stopped moving. Eviction pauses
    // itself once it hits the drain cap or reaches the target, so this
    // converges quickly; the deadline exists only so a hang reports rather
    // than comparing a half-evicted store.
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    uint64_t lastTotal = kPerCollection * 2;
    int stableTicks = 0;
    bool evictionRan = false;
    bool settled = false;
    while (std::chrono::steady_clock::now() < deadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(25));
        auto [important, archive] = liveCounts();
        const uint64_t total = important + archive;
        if (total < kPerCollection * 2) evictionRan = true;
        if (evictionRan && total == lastTotal) {
            if (++stableTicks >= 8) { settled = true; break; }  // ~200ms of no change
        } else {
            stableTicks = 0;
        }
        lastTotal = total;
    }
    check(evictionRan, "eviction must run at all given the fill exceeds the cap");
    check(settled, "eviction must settle within the deadline");

    auto [importantLive, archiveLive] = liveCounts();

    // LOW priority should be evicted at least as much as HIGH.
    check(archiveLive <= importantLive,
          "low-priority collection must not outlive the high-priority one");
    // Some eviction MUST have happened given our fill vs cap.
    check(importantLive < kPerCollection || archiveLive < kPerCollection,
          "some eviction must have happened");
    // Expect strict bias once any eviction has run.
    check(archiveLive < importantLive,
          "low-priority collection must be evicted strictly more than high");

    store.stop();
    std::cout << "PASS: priority=Low collection evicted more than priority=High "
              << "(archive=" << archiveLive << ", important=" << importantLive << ")\n";
}

void test_eviction_chunk_size_honored() {
    auto cfg = evictionTestConfig(1);
    cfg.evictionChunkSize = 50;      // small chunk
    cfg.maxEvictionPassesPerTrigger = 1;  // exactly one chunk per tick
    cfg.hotWriteFloorMs = 0;          // disable
    cfg.evictionCheckIntervalMs = 100;
    cfg.memorySoftPercent = 10;
    cfg.memoryHardPercent = 20;
    MemoryStore store(cfg);
    store.start();
    store.createCollection("docs", CollectionOptions{});

    fillStore(store, "docs", 500, 2048);

    // After one eviction tick, at most chunkSize docs should have been evicted
    auto before = store.getMemoryStatsSnapshot();
    uint64_t beforeLive = 0;
    for (const auto& c : before.collections) beforeLive += c.documentCount;

    std::this_thread::sleep_for(std::chrono::milliseconds(120));  // ~1 tick

    auto after = store.getMemoryStatsSnapshot();
    uint64_t afterLive = 0;
    for (const auto& c : after.collections) afterLive += c.documentCount;

    uint64_t evicted = (beforeLive > afterLive) ? (beforeLive - afterLive) : 0;
    // Should evict approximately chunkSize per tick, allow some slack.
    // 0 is possible if timing went weird, but > chunkSize*3 is definitely wrong.
    assert(evicted <= cfg.evictionChunkSize * 3);

    store.stop();
    std::cout << "PASS: eviction honors chunk size (evicted " << evicted
              << " docs in ~1 tick, chunkSize=" << cfg.evictionChunkSize << ")\n";
}

// v2.4.3 — eviction must never drain the store.
//
// Regression for the production incident: the memory estimate stopped
// tracking what eviction could free, so `current <= targetBytes` was never
// satisfied and the loop trimmed one chunk per tick for 30 minutes until 5996
// of ~5990 docs were gone. Reads then fell back to an empty MemoryStore and
// every collection reported 0 docs, which looked exactly like data loss.
//
// Here the target is made unreachable on purpose (evictionTargetPercent = 0,
// so targetBytes = 0 and no amount of eviction satisfies it). Before the drain
// cap this emptied the store; now it must stop at the cap.
void test_eviction_never_drains_the_store() {
    auto cfg = evictionTestConfig(1);
    cfg.evictionTargetPercent = 0;        // unreachable target
    cfg.memorySoftPercent = 1;            // stay under pressure permanently
    cfg.memoryHardPercent = 2;
    cfg.memoryEmergencyPercent = 99;      // don't trip admission control
    cfg.hotWriteFloorMs = 0;              // everything is evictable
    cfg.evictionChunkSize = 50;
    cfg.evictionCheckIntervalMs = 20;
    cfg.evictionMaxEpisodePercent = 50;   // cap under test
    MemoryStore store(cfg);
    store.start();
    store.createCollection("docs", CollectionOptions{});

    const uint64_t seeded = 400;
    fillStore(store, "docs", seeded, 2048);

    auto liveDocs = [&] {
        uint64_t n = 0;
        for (const auto& c : store.getMemoryStatsSnapshot().collections) {
            n += c.documentCount;
        }
        return n;
    };

    const uint64_t before = liveDocs();
    assert(before > 0);

    // Give the eviction thread many ticks — far more than it would need to
    // empty the store one 50-doc chunk at a time.
    std::this_thread::sleep_for(std::chrono::milliseconds(1200));

    const uint64_t after = liveDocs();
    const uint64_t evicted = (before > after) ? (before - after) : 0;

    // The cap is 50% of the episode's starting set. Allow one chunk of
    // overshoot, since the check runs per tick rather than per doc.
    const uint64_t cap = before / 2 + cfg.evictionChunkSize;
    assert(evicted <= cap);
    assert(after > 0);  // the store must never be drained

    store.stop();
    std::cout << "PASS: eviction drain cap held (" << evicted << " of " << before
              << " evicted, " << after << " docs still resident)\n";
}

int main() {
    test_pressure_levels();
    test_hot_write_floor_protects_recent_writes();
    test_priority_low_evicted_first();
    test_eviction_chunk_size_honored();
    test_eviction_never_drains_the_store();
    if (g_fail != 0) {
        std::cerr << "\n" << g_fail << " check(s) FAILED (" << g_pass << " passed)\n";
        return 1;
    }
    std::cout << "\nAll eviction tests PASSED! (" << g_pass << " checks)\n";
    return 0;
}