test_replication_new_features.cpp

/**
 * ThemisDB Replication New Feature Tests
 *
 * Covers:
 *   1. ReplicationObserver — lag snapshots, topology, bottleneck detection,
 *      health score
 *   2. ThreeWayMergeResolver — basic merge, all-same winner
 *   3. FieldLevelMergeResolver — UNION / INTERSECT / LEFT_BIAS / RIGHT_BIAS
 *   4. ReplicationEventStream — subscribe, emit via listener interface,
 *      historical query, RAII unsubscribe
 *   5. ReplicationPolicy — define, assign, validate, violations
 *   6. ReplicationSlot / ReplicationSlotManager — create, pause, resume,
 *      advance, drop, lag, persistence
 */

#include <gtest/gtest.h>

#include "replication/replication_manager.h"
#include "replication/multi_master_replication.h"
#include "replication/observability.h"
#include "replication/conflict_resolution.h"
#include "replication/event_stream.h"
#include "replication/policy.h"
#include "replication/replication_slot.h"

#include <chrono>
#include <filesystem>
#include <future>
#include <memory>
#include <string>
#include <thread>
#include <vector>

using namespace themisdb::replication;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

static ReplicationConfig makeConfig(
    const std::string& wal_dir = "/tmp/themis_newfeature_wal")
{
    ReplicationConfig cfg;
    cfg.enabled                      = true;
    cfg.mode                         = ReplicationMode::ASYNC;
    cfg.heartbeat_interval_ms        = 100;
    cfg.election_timeout_min_ms      = 150;
    cfg.election_timeout_max_ms      = 300;
    cfg.batch_size                   = 100;
    cfg.batch_timeout_ms             = 50;
    cfg.wal_directory                = wal_dir;
    cfg.failure_detection_timeout_ms = 1000;
    cfg.degraded_lag_threshold_ms    = 5000;
    cfg.min_sync_replicas            = 1;
    cfg.wal_sync_on_commit           = false;
    cfg.enable_leader_lease          = false;
    return cfg;
}

struct TempWALDir {
    std::string path = {};
    explicit TempWALDir(const std::string& p) {
        const auto ticks = std::chrono::high_resolution_clock::now()
                               .time_since_epoch()
                               .count();
        const auto tid_hash = std::hash<std::thread::id>{}(std::this_thread::get_id());
        const std::string base = std::filesystem::path(p).filename().string();
        path = (std::filesystem::temp_directory_path() /
                (base + "_" + std::to_string(ticks) + "_" + std::to_string(tid_hash)))
                   .string();

        std::error_code ec = {};
        std::filesystem::remove_all(path, ec);
        std::filesystem::create_directories(path, ec);
    }
    ~TempWALDir() {
        for (int i = 0; i < 5; ++i) {
            std::error_code ec = {};
            std::filesystem::remove_all(path, ec);
            if (!ec) {
                break;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    }
};

static MMWriteEntry makeMMEntry(
    const std::string& doc_id,
    const std::string& data,
    uint64_t hlc_physical,
    uint32_t hlc_logical = 0)
{
    MMWriteEntry e;
    e.write_id     = doc_id + "_" + std::to_string(hlc_physical);
    e.document_id  = doc_id;
    e.collection   = "test_col";
    e.operation    = "UPDATE";
    e.data         = data;
    e.hlc.physical = hlc_physical;
    e.hlc.logical  = hlc_logical;
    e.hlc.node_id  = "node_" + std::to_string(hlc_physical % 3);
    return e;
}

// ============================================================================
// 1. ReplicationObserver
// ============================================================================

class ObservabilityTest : public ::testing::Test {
protected:
    void SetUp() override {
        wal_dir_.reset(new TempWALDir("/tmp/themis_obs_test"));
        auto cfg = makeConfig(wal_dir_->path);
        mgr_ = std::make_shared<ReplicationManager>(cfg);
        ASSERT_TRUE(mgr_->initialize());
    }

    void TearDown() override {
        mgr_->shutdown();
    }

    std::unique_ptr<TempWALDir>         wal_dir_;
    std::shared_ptr<ReplicationManager> mgr_;
};

TEST_F(ObservabilityTest, LagSnapshotsEmptyWhenNoReplicas) {
    ReplicationObserver observer(mgr_);
    auto snaps = observer.getLagSnapshots();
    // A single-node cluster has no replicas besides the leader itself
    EXPECT_TRUE(snaps.empty());
}

TEST_F(ObservabilityTest, TopologyContainsLocalNode) {
    ReplicationObserver observer(mgr_);
    auto topo = observer.getTopology();
    // The leader should appear in the topology
    bool found_leader = false;
    for (const auto& n : topo) {
        if (n.role == ReplicationRole::LEADER) { found_leader = true; break; }
    }
    // On a single-node cluster getTopology may return 0 nodes (no replicas
    // registered) — that is acceptable; verify it does not crash.
    (void)found_leader;
    SUCCEED();
}

TEST_F(ObservabilityTest, HealthScoreReturns100ForEmptyCluster) {
    ReplicationObserver observer(mgr_);
    auto score = observer.calculateHealthScore();
    EXPECT_EQ(score.overall_score, 100);
    EXPECT_EQ(score.lag_score, 100);
    EXPECT_EQ(score.availability_score, 100);
    EXPECT_EQ(score.throughput_score, 100);
}

TEST_F(ObservabilityTest, DetectBottlenecksReturnsEmptyForHealthyCluster) {
    ReplicationObserver observer(mgr_);
    auto bottlenecks = observer.detectBottlenecks();
    EXPECT_TRUE(bottlenecks.empty());
}

TEST_F(ObservabilityTest, CustomCriticalLagThresholdApplied) {
    // Lower the critical threshold to 0 ms so even a "fresh" replica qualifies
    ReplicationObserver::ObserverConfig cfg;
    cfg.critical_lag_threshold_ms = 0;
    ReplicationObserver observer(mgr_, cfg);
    // Still should not crash; snaps may be empty or populated depending on
    // replica count.
    auto snaps = observer.getLagSnapshots();
    (void)snaps;
    SUCCEED();
}

// ============================================================================
// 2. ThreeWayMergeResolver
// ============================================================================

class ThreeWayMergeTest : public ::testing::Test {};

TEST_F(ThreeWayMergeTest, SingleWriteReturnedUnchanged) {
    ThreeWayMergeResolver resolver;
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.collection = "col";
    ctx.document_id = "doc1";

    auto entry = makeMMEntry("doc1", R"({"a":"1","b":"2"})", 1000);
    auto result = resolver.resolve("doc1", {entry}, ctx);
    EXPECT_EQ(result.data, entry.data);
}

TEST_F(ThreeWayMergeTest, MergesNonConflictingFields) {
    ThreeWayMergeResolver resolver;
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.collection  = "col";
    ctx.document_id = "doc2";

    // base:  {a:1}
    // left:  {a:1, b:2}   (added b)
    // right: {a:1, c:3}   (added c)
    // expected merge: {a:1, b:2, c:3} (both additions preserved)
    VectorClock vc_base, vc_left, vc_right;
    vc_base.increment("n1");

    vc_left = vc_base;
    vc_left.increment("n2");

    vc_right = vc_base;
    vc_right.increment("n3");

    auto base  = makeMMEntry("doc2", R"({"a":"1"})", 100);
    base.vector_clock = vc_base;

    auto left  = makeMMEntry("doc2", R"({"a":"1","b":"2"})", 200);
    left.vector_clock = vc_left;

    auto right = makeMMEntry("doc2", R"({"a":"1","c":"3"})", 300);
    right.vector_clock = vc_right;

    auto result = resolver.resolve("doc2", {base, left, right}, ctx);

    // Result must contain all three keys
    EXPECT_NE(result.data.find('"' + std::string("a") + '"'), std::string::npos);
    EXPECT_NE(result.data.find('"' + std::string("b") + '"'), std::string::npos);
    EXPECT_NE(result.data.find('"' + std::string("c") + '"'), std::string::npos);
}

TEST_F(ThreeWayMergeTest, StrategyNameIsThreeWayMerge) {
    ThreeWayMergeResolver resolver;
    EXPECT_EQ(resolver.strategyName(), "THREE_WAY_MERGE");
}

TEST_F(ThreeWayMergeTest, EmptyConflictSetFailsClosed) {
    ThreeWayMergeResolver resolver;
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.collection = "col";
    ctx.document_id = "doc_empty";

    EXPECT_THROW(
        resolver.resolve("doc_empty", {}, ctx),
        std::invalid_argument);
}

TEST_F(ThreeWayMergeTest, WinnerCarriesMergedCausalMetadata) {
    ThreeWayMergeResolver resolver;
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.collection = "col";
    ctx.document_id = "doc_meta";

    VectorClock vc_base;
    vc_base.increment("n1");
    VectorClock vc_left = vc_base;
    vc_left.increment("n2");
    VectorClock vc_right = vc_base;
    vc_right.increment("n3");

    auto base = makeMMEntry("doc_meta", R"({"a":"1"})", 100);
    base.vector_clock = vc_base;
    base.dependencies = {"dep_base"};

    auto left = makeMMEntry("doc_meta", R"({"a":"1","b":"2"})", 200);
    left.vector_clock = vc_left;
    left.dependencies = {"dep_left"};

    auto right = makeMMEntry("doc_meta", R"({"a":"1","c":"3"})", 300);
    right.vector_clock = vc_right;
    right.dependencies = {"dep_right"};

    auto result = resolver.resolve("doc_meta", {base, left, right}, ctx);

    EXPECT_EQ(result.hlc.physical, 300u);
    EXPECT_GE(result.vector_clock.get("n1"), 1u);
    EXPECT_GE(result.vector_clock.get("n2"), 1u);
    EXPECT_GE(result.vector_clock.get("n3"), 1u);
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), "dep_base"), result.dependencies.end());
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), "dep_left"), result.dependencies.end());
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), "dep_right"), result.dependencies.end());
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), left.write_id), result.dependencies.end());
    EXPECT_EQ(std::find(result.dependencies.begin(), result.dependencies.end(), result.write_id), result.dependencies.end());
    EXPECT_FALSE(result.checksum.empty());
}

// ============================================================================
// 3. FieldLevelMergeResolver
// ============================================================================

class FieldLevelMergeTest : public ::testing::Test {};

TEST_F(FieldLevelMergeTest, UnionIncludesAllFields) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::UNION);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc3";

    auto a = makeMMEntry("doc3", R"({"x":"1","y":"2"})", 100);
    auto b = makeMMEntry("doc3", R"({"y":"3","z":"4"})", 200);

    auto result = resolver.resolve("doc3", {a, b}, ctx);
    // All three keys must appear: x, y, z
    EXPECT_NE(result.data.find("\"x\""), std::string::npos);
    EXPECT_NE(result.data.find("\"y\""), std::string::npos);
    EXPECT_NE(result.data.find("\"z\""), std::string::npos);
}

TEST_F(FieldLevelMergeTest, IntersectOnlyCommonFields) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::INTERSECT);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc4";

    auto a = makeMMEntry("doc4", R"({"x":"1","y":"2"})", 100);
    auto b = makeMMEntry("doc4", R"({"y":"3","z":"4"})", 200);

    auto result = resolver.resolve("doc4", {a, b}, ctx);
    // Only "y" is in both writes
    EXPECT_NE(result.data.find("\"y\""), std::string::npos);
    // "x" and "z" should NOT appear
    EXPECT_EQ(result.data.find("\"x\""), std::string::npos);
    EXPECT_EQ(result.data.find("\"z\""), std::string::npos);
}

TEST_F(FieldLevelMergeTest, LeftBiasPreferFirstEntry) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::LEFT_BIAS);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc5";

    auto a = makeMMEntry("doc5", R"({"k":"left_value"})", 100);
    auto b = makeMMEntry("doc5", R"({"k":"right_value"})", 200);

    auto result = resolver.resolve("doc5", {a, b}, ctx);
    EXPECT_NE(result.data.find("left_value"), std::string::npos);
}

TEST_F(FieldLevelMergeTest, RightBiasPreferLastEntry) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::RIGHT_BIAS);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc6";

    auto a = makeMMEntry("doc6", R"({"k":"left_value"})", 100);
    auto b = makeMMEntry("doc6", R"({"k":"right_value"})", 200);

    auto result = resolver.resolve("doc6", {a, b}, ctx);
    EXPECT_NE(result.data.find("right_value"), std::string::npos);
}

TEST_F(FieldLevelMergeTest, StrategyNamesAreCorrect) {
    EXPECT_EQ(FieldLevelMergeResolver(FieldLevelMergeResolver::MergeStrategy::UNION).strategyName(),
              "FIELD_MERGE_UNION");
    EXPECT_EQ(FieldLevelMergeResolver(FieldLevelMergeResolver::MergeStrategy::INTERSECT).strategyName(),
              "FIELD_MERGE_INTERSECT");
    EXPECT_EQ(FieldLevelMergeResolver(FieldLevelMergeResolver::MergeStrategy::LEFT_BIAS).strategyName(),
              "FIELD_MERGE_LEFT_BIAS");
    EXPECT_EQ(FieldLevelMergeResolver(FieldLevelMergeResolver::MergeStrategy::RIGHT_BIAS).strategyName(),
              "FIELD_MERGE_RIGHT_BIAS");
}

TEST_F(FieldLevelMergeTest, EmptyConflictSetFailsClosed) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::UNION);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc_empty";

    EXPECT_THROW(
        resolver.resolve("doc_empty", {}, ctx),
        std::invalid_argument);
}

TEST_F(FieldLevelMergeTest, WinnerCarriesMergedCausalMetadata) {
    FieldLevelMergeResolver resolver(FieldLevelMergeResolver::MergeStrategy::UNION);
    AdvancedConflictResolver::ResolutionContext ctx;
    ctx.document_id = "doc_meta_field";

    VectorClock vc_a;
    vc_a.increment("na");
    VectorClock vc_b = vc_a;
    vc_b.increment("nb");

    auto a = makeMMEntry("doc_meta_field", R"({"k1":"v1"})", 100);
    a.vector_clock = vc_a;
    a.dependencies = {"dep_a"};

    auto b = makeMMEntry("doc_meta_field", R"({"k2":"v2"})", 250);
    b.vector_clock = vc_b;
    b.dependencies = {"dep_b"};

    auto result = resolver.resolve("doc_meta_field", {a, b}, ctx);

    EXPECT_EQ(result.hlc.physical, 250u);
    EXPECT_GE(result.vector_clock.get("na"), 1u);
    EXPECT_GE(result.vector_clock.get("nb"), 1u);
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), "dep_a"), result.dependencies.end());
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), "dep_b"), result.dependencies.end());
    EXPECT_NE(std::find(result.dependencies.begin(), result.dependencies.end(), a.write_id), result.dependencies.end());
    EXPECT_FALSE(result.checksum.empty());
}

// ============================================================================
// 4. ReplicationEventStream
// ============================================================================

class EventStreamTest : public ::testing::Test {
protected:
    void SetUp() override {
        stream_ = std::make_shared<ReplicationEventStream>();
    }
    std::shared_ptr<ReplicationEventStream> stream_;
};

TEST_F(EventStreamTest, SubscribeAndReceiveEvent) {
    int calls = 0;
    auto handle = stream_->subscribeAll(
        [&calls](const ReplicationEventStream::Event&) { ++calls; });

    // Simulate a leader election event via IReplicationListener
    stream_->onLeaderElected("node1");
    EXPECT_EQ(calls, 1);
}

TEST_F(EventStreamTest, FilteredSubscriptionOnlyReceivesMatchingType) {
    int failover_calls = 0;
    int lag_calls      = 0;

    auto h1 = stream_->subscribe(
        ReplicationEventStream::EventType::FAILOVER_COMPLETED,
        [&failover_calls](const ReplicationEventStream::Event&) { ++failover_calls; });

    auto h2 = stream_->subscribe(
        ReplicationEventStream::EventType::LAG_WARNING,
        [&lag_calls](const ReplicationEventStream::Event&) { ++lag_calls; });

    stream_->onFailoverCompleted("node2", true);
    stream_->onReplicationLagWarning(5000);

    EXPECT_EQ(failover_calls, 1);
    EXPECT_EQ(lag_calls,      1);
}

TEST_F(EventStreamTest, RAIIHandleUnsubscribesOnDestruction) {
    int calls = 0;
    {
        auto handle = stream_->subscribeAll(
            [&calls](const ReplicationEventStream::Event&) { ++calls; });
        stream_->onLeaderElected("node1");
        EXPECT_EQ(calls, 1);
        // handle destroyed here — should unsubscribe
    }
    stream_->onLeaderElected("node2");
    EXPECT_EQ(calls, 1); // still 1, not 2
}

TEST_F(EventStreamTest, HistoricalQueryReturnsBufferedEvents) {
    const auto before = std::chrono::system_clock::now();
    stream_->onLeaderElected("nodeA");
    stream_->onFailoverCompleted("nodeB", true);
    const auto after = std::chrono::system_clock::now() + std::chrono::seconds(1);

    auto events = stream_->getEvents(before, after);
    EXPECT_GE(events.size(), 2u);
}

TEST_F(EventStreamTest, HistoricalQueryFilterByType) {
    const auto before = std::chrono::system_clock::now();
    stream_->onLeaderElected("nodeA");
    stream_->onReplicationLagWarning(3000);
    const auto after = std::chrono::system_clock::now() + std::chrono::seconds(1);

    auto events = stream_->getEvents(
        before, after,
        ReplicationEventStream::EventType::LEADER_ELECTED);
    EXPECT_EQ(events.size(), 1u);
    EXPECT_EQ(events[0].type, ReplicationEventStream::EventType::LEADER_ELECTED);
}

TEST_F(EventStreamTest, RingBufferDropsOldestWhenFull) {
    ReplicationEventStream::StreamConfig cfg;
    cfg.max_history_events  = 3;
    cfg.drop_oldest_on_full = true;
    auto small_stream = std::make_shared<ReplicationEventStream>(cfg);

    for (int i = 0; i < 5; ++i) {
        small_stream->onLeaderElected("node" + std::to_string(i));
    }
    EXPECT_EQ(small_stream->bufferedEventCount(), 3u);
}

TEST_F(EventStreamTest, WALEntryAppliedEmitsWriteReplicatedEvent) {
    int calls = 0;
    auto h = stream_->subscribe(
        ReplicationEventStream::EventType::WRITE_REPLICATED,
        [&calls](const ReplicationEventStream::Event&) { ++calls; });

    WALEntry entry;
    entry.sequence_number = 1;
    entry.collection      = "orders";
    entry.operation       = "INSERT";
    stream_->onWALEntryApplied(entry);

    EXPECT_EQ(calls, 1);
}

// ============================================================================
// 5. ReplicationPolicy
// ============================================================================

class ReplicationPolicyTest : public ::testing::Test {
protected:
    void SetUp() override {
        wal_dir_.reset(new TempWALDir("/tmp/themis_policy_test"));
        auto cfg = makeConfig(wal_dir_->path);
        mgr_ = std::make_shared<ReplicationManager>(cfg);
        ASSERT_TRUE(mgr_->initialize());
        policy_mgr_ = std::make_unique<ReplicationPolicy>(mgr_);
    }
    void TearDown() override { mgr_->shutdown(); }

    std::unique_ptr<TempWALDir>         wal_dir_;
    std::shared_ptr<ReplicationManager> mgr_;
    std::unique_ptr<ReplicationPolicy>  policy_mgr_;
};

TEST_F(ReplicationPolicyTest, DefineThenGetPolicy) {
    ReplicationPolicy::Policy p;
    p.name             = "test_policy";
    p.desired_replicas = 3;
    p.mode             = ReplicationMode::SYNC;
    policy_mgr_->definePolicy("test_policy", p);

    // Assign to a collection
    EXPECT_TRUE(policy_mgr_->assignPolicy("orders", "test_policy"));

    auto retrieved = policy_mgr_->getPolicy("orders");
    EXPECT_EQ(retrieved.desired_replicas, 3u);
    EXPECT_EQ(retrieved.mode, ReplicationMode::SYNC);
}

TEST_F(ReplicationPolicyTest, AssignNonExistentPolicyReturnsFalse) {
    EXPECT_FALSE(policy_mgr_->assignPolicy("orders", "nonexistent"));
}

TEST_F(ReplicationPolicyTest, GetPolicyForUnassignedCollectionReturnsDefault) {
    auto p = policy_mgr_->getPolicy("unknown_collection");
    EXPECT_EQ(p.name, "__default__");
}

TEST_F(ReplicationPolicyTest, RemovePolicyReturnsTrueIfExists) {
    ReplicationPolicy::Policy p;
    p.name = "temp_policy";
    policy_mgr_->definePolicy("temp_policy", p);
    EXPECT_TRUE(policy_mgr_->removePolicy("temp_policy"));
    EXPECT_FALSE(policy_mgr_->removePolicy("temp_policy")); // already removed
}

TEST_F(ReplicationPolicyTest, ListPoliciesAfterDefine) {
    ReplicationPolicy::Policy p;
    p.name = "p1";
    policy_mgr_->definePolicy("p1", p);
    p.name = "p2";
    policy_mgr_->definePolicy("p2", p);

    auto names = policy_mgr_->listPolicies();
    EXPECT_GE(names.size(), 2u);
}

TEST_F(ReplicationPolicyTest, ValidatePolicyViolatesMinReplicasOnSingleNode) {
    ReplicationPolicy::Policy p;
    p.name         = "strict";
    p.min_replicas = 3; // requires 3 replicas; single-node cluster has 0

    auto v = policy_mgr_->validatePolicy(p);
    EXPECT_FALSE(v.is_valid);
    EXPECT_FALSE(v.violations.empty());
}

TEST_F(ReplicationPolicyTest, ValidatePolicySucceedsForLenientPolicy) {
    ReplicationPolicy::Policy p;
    p.name         = "lenient";
    p.min_replicas = 0;
    p.write_quorum = 0;
    p.min_datacenters = 0;
    p.mode         = ReplicationMode::ASYNC;

    auto v = policy_mgr_->validatePolicy(p);
    // With 0 requirements and ASYNC mode, validation should pass on any cluster
    EXPECT_TRUE(v.is_valid);
}

// ============================================================================
// 6. ReplicationSlot / ReplicationSlotManager
// ============================================================================

class ReplicationSlotTest : public ::testing::Test {
protected:
    void SetUp() override {
        wal_dir_.reset(new TempWALDir("/tmp/themis_slot_test_wal"));
        auto cfg = makeConfig(wal_dir_->path);
        wal_mgr_ = std::make_shared<WALManager>(cfg);
    }

    std::unique_ptr<TempWALDir>      wal_dir_;
    std::shared_ptr<WALManager>      wal_mgr_;
};

TEST_F(ReplicationSlotTest, CreateSlotIsActive) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto slot = manager.createSlot("slot1", "physical", "node2");
    ASSERT_NE(slot, nullptr);
    EXPECT_EQ(slot->name(),   "slot1");
    EXPECT_EQ(slot->status(), ReplicationSlot::SlotStatus::ACTIVE);
}

TEST_F(ReplicationSlotTest, DuplicateSlotNameReturnsNull) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    manager.createSlot("dup_slot");
    auto slot2 = manager.createSlot("dup_slot");
    EXPECT_EQ(slot2, nullptr);
}

TEST_F(ReplicationSlotTest, PauseAndResume) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto slot = manager.createSlot("slot_pr");
    ASSERT_NE(slot, nullptr);

    EXPECT_TRUE(slot->pause());
    EXPECT_EQ(slot->status(), ReplicationSlot::SlotStatus::PAUSED);
    EXPECT_FALSE(slot->pause()); // already paused

    EXPECT_TRUE(slot->resume());
    EXPECT_EQ(slot->status(), ReplicationSlot::SlotStatus::ACTIVE);
    EXPECT_FALSE(slot->resume()); // already active
}

TEST_F(ReplicationSlotTest, AdvanceUpdatesLsn) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto slot = manager.createSlot("slot_adv");
    ASSERT_NE(slot, nullptr);

    EXPECT_TRUE(slot->advance(42));
    EXPECT_EQ(slot->state().confirmed_lsn, 42u);

    // Advancing to lower value should fail
    EXPECT_FALSE(slot->advance(10));
    EXPECT_EQ(slot->state().confirmed_lsn, 42u);
}

TEST_F(ReplicationSlotTest, AdvanceOnPausedSlotFails) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto slot = manager.createSlot("slot_adv_paused");
    ASSERT_NE(slot, nullptr);
    slot->pause();

    EXPECT_FALSE(slot->advance(100));
}

TEST_F(ReplicationSlotTest, DropSlot) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    manager.createSlot("slot_drop");
    EXPECT_TRUE(manager.dropSlot("slot_drop"));
    EXPECT_EQ(manager.getSlot("slot_drop"), nullptr);
    EXPECT_FALSE(manager.dropSlot("slot_drop")); // already dropped
}

TEST_F(ReplicationSlotTest, LagCalculation) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto slot = manager.createSlot("slot_lag");
    ASSERT_NE(slot, nullptr);

    // Append some WAL entries to create lag
    WALEntry entry;
    entry.operation   = "INSERT";
    entry.collection  = "test";
    entry.document_id = "doc1";
    entry.data        = R"({"v":1})";
    wal_mgr_->append(entry);
    wal_mgr_->append(entry);
    wal_mgr_->append(entry);

    // Slot is at lsn=0; leader is at 3 → lag = 3
    EXPECT_GE(slot->lag(), 3u);

    // Advance slot; lag should decrease
    slot->advance(2);
    // After advancing to 2, leader is at 3, so lag should be 1
    EXPECT_LE(slot->lag(), 2u);
}

TEST_F(ReplicationSlotTest, ListSlotsReturnsAllRegistered) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    manager.createSlot("s1");
    manager.createSlot("s2");
    manager.createSlot("s3");

    auto states = manager.listSlots();
    EXPECT_EQ(states.size(), 3u);
}

TEST_F(ReplicationSlotTest, MinConfirmedLsnAcrossSlots) {
    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;
    ReplicationSlotManager manager(cfg, wal_mgr_);

    auto s1 = manager.createSlot("mc1");
    auto s2 = manager.createSlot("mc2");

    s1->advance(100);
    s2->advance(50);

    EXPECT_EQ(manager.minConfirmedLsn(), 50u);
}

TEST_F(ReplicationSlotTest, StatePersistenceRoundTrip) {
    const std::string slot_dir = wal_dir_->path + "/slots";

    ReplicationSlotManager::ManagerConfig cfg;
    cfg.wal_directory = wal_dir_->path;

    {
        ReplicationSlotManager manager(cfg, wal_mgr_);
        auto slot = manager.createSlot("persist_slot");
        ASSERT_NE(slot, nullptr);
        slot->advance(77);
        slot->pause();
    }

    // Reload persisted slots
    ReplicationSlotManager manager2(cfg, wal_mgr_);
    manager2.loadPersistedSlots();

    auto slot = manager2.getSlot("persist_slot");
    ASSERT_NE(slot, nullptr);
    EXPECT_EQ(slot->state().confirmed_lsn, 77u);
    EXPECT_EQ(slot->status(), ReplicationSlot::SlotStatus::PAUSED);
}

// ============================================================================
// 7. ParallelReplicationWorker (v1.6.0)
// ============================================================================

namespace {

// Monotonically increasing sequence number so every WALEntry is unique.
static std::atomic<uint64_t> g_seq{1};

WALEntry makeWALEntry(const std::string& doc_id,
                      const std::string& collection = "col",
                      const std::string& op = "INSERT") {
    WALEntry e;
    e.sequence_number = g_seq.fetch_add(1);
    e.term            = 1;
    e.operation       = op;
    e.collection      = collection;
    e.document_id     = doc_id;
    e.data            = R"({"v":1})";
    return e;
}

} // anonymous namespace

class ParallelReplicationWorkerTest : public ::testing::Test {
protected:
    // All config fields are set explicitly so the tests are not sensitive
    // to future default-value changes.
    ParallelReplicationWorker::ParallelConfig defaultConfig() {
        ParallelReplicationWorker::ParallelConfig cfg;
        cfg.worker_threads          = 4;
        cfg.queue_size              = 10000;
        cfg.use_dependency_tracking = true;
        cfg.group_transactions      = true;
        return cfg;
    }
};

TEST_F(ParallelReplicationWorkerTest, ConstructAndDestruct) {
    // Should start and stop cleanly
    ParallelReplicationWorker worker(defaultConfig());
}

TEST_F(ParallelReplicationWorkerTest, SubmitSingleEntryAndSync) {
    ParallelReplicationWorker worker(defaultConfig());

    worker.submit(makeWALEntry("doc1"));
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 1u);
}

TEST_F(ParallelReplicationWorkerTest, SubmitMultipleIndependentEntries) {
    ParallelReplicationWorker worker(defaultConfig());

    for (int i = 0; i < 20; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 20u);
}

TEST_F(ParallelReplicationWorkerTest, DependencyTrackingDetectsConflicts) {
    ParallelReplicationWorker worker(defaultConfig());

    // Three writes to the same document — all should be serialized
    worker.submit(makeWALEntry("shared_doc", "col", "INSERT"));
    worker.submit(makeWALEntry("shared_doc", "col", "UPDATE"));
    worker.submit(makeWALEntry("shared_doc", "col", "DELETE"));
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 3u);
    // The second and third writes each depend on the previous one
    EXPECT_EQ(stats.dependencies_detected, 2u);
}

TEST_F(ParallelReplicationWorkerTest, NoDependencyTrackingSkipsDeps) {
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.use_dependency_tracking = false;
    ParallelReplicationWorker worker(cfg);

    // Same document — without tracking no dependencies should be recorded
    worker.submit(makeWALEntry("same_doc", "col", "INSERT"));
    worker.submit(makeWALEntry("same_doc", "col", "UPDATE"));
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 2u);
    EXPECT_EQ(stats.dependencies_detected, 0u);
}

TEST_F(ParallelReplicationWorkerTest, MixedDocumentsPartialDependencies) {
    ParallelReplicationWorker worker(defaultConfig());

    // 4 writes: 2 to "a", 2 to "b" — 2 dependencies total (one per doc chain)
    worker.submit(makeWALEntry("a", "col", "INSERT"));
    worker.submit(makeWALEntry("b", "col", "INSERT"));
    worker.submit(makeWALEntry("a", "col", "UPDATE"));
    worker.submit(makeWALEntry("b", "col", "UPDATE"));
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 4u);
    EXPECT_EQ(stats.dependencies_detected, 2u);
}

TEST_F(ParallelReplicationWorkerTest, StatsParallelismFactorNonNegative) {
    ParallelReplicationWorker worker(defaultConfig());

    worker.submit(makeWALEntry("d1"));
    worker.submit(makeWALEntry("d2"));
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_GE(stats.parallelism_factor, 0.0);
}

TEST_F(ParallelReplicationWorkerTest, GetStatsBeforeSubmitReturnsZero) {
    ParallelReplicationWorker worker(defaultConfig());
    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied,       0u);
    EXPECT_EQ(stats.dependencies_detected, 0u);
    EXPECT_EQ(stats.parallel_batches,      0u);
    EXPECT_DOUBLE_EQ(stats.parallelism_factor, 0.0);
}

TEST_F(ParallelReplicationWorkerTest, SingleThreadedWorker) {
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.worker_threads = 1;
    ParallelReplicationWorker worker(cfg);

    for (int i = 0; i < 10; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    EXPECT_EQ(worker.getStats().entries_applied, 10u);
}

TEST_F(ParallelReplicationWorkerTest, SixteenThreadWorker) {
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.worker_threads = 16;
    ParallelReplicationWorker worker(cfg);

    for (int i = 0; i < 100; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    EXPECT_EQ(worker.getStats().entries_applied, 100u);
}

TEST_F(ParallelReplicationWorkerTest, SyncOnEmptyQueueReturnsImmediately) {
    ParallelReplicationWorker worker(defaultConfig());
    // Verify that sync() on an empty queue returns within a tight deadline.
    auto fut = std::async(std::launch::async, [&worker] { worker.sync(); });
    EXPECT_EQ(fut.wait_for(std::chrono::seconds(5)), std::future_status::ready);
    EXPECT_EQ(worker.getStats().entries_applied, 0u);
}

TEST_F(ParallelReplicationWorkerTest, GroupTransactionsEnabled) {
    // With group_transactions=true entries queued together should be drained in
    // a single worker iteration, yielding parallel_batches strictly less than
    // entries_applied.  We sleep briefly so the worker finishes its current
    // wait, then submit all entries in one rapid burst before the next cycle.
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.group_transactions = true;
    cfg.worker_threads     = 1;  // One worker makes batching deterministic
    ParallelReplicationWorker worker(cfg);

    // Let the worker settle into its 5 ms wait; the burst below (~µs) will
    // complete before the next wake-up, ensuring all entries land in one batch.
    std::this_thread::sleep_for(std::chrono::milliseconds(6));

    constexpr int kEntries = 20;
    for (int i = 0; i < kEntries; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, static_cast<uint64_t>(kEntries));
    // Under heavy scheduler jitter a one-thread worker may still process each
    // queued entry individually; the key invariant is no over-counting.
    EXPECT_LE(stats.parallel_batches, static_cast<uint64_t>(kEntries));
    EXPECT_GE(stats.parallel_batches, 1u);
}

TEST_F(ParallelReplicationWorkerTest, GroupTransactionsDisabled) {
    // With group_transactions=false each entry is processed in its own
    // worker iteration; parallel_batches equals entries_applied.
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.group_transactions = false;
    cfg.worker_threads     = 1;  // Single worker guarantees sequential batches
    ParallelReplicationWorker worker(cfg);

    constexpr int kEntries = 5;
    for (int i = 0; i < kEntries; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, static_cast<uint64_t>(kEntries));
    EXPECT_EQ(stats.parallel_batches, static_cast<uint64_t>(kEntries));
}

TEST_F(ParallelReplicationWorkerTest, AverageLatencyPopulatedAfterWork) {
    ParallelReplicationWorker worker(defaultConfig());

    for (int i = 0; i < 10; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i)));
    }
    worker.sync();

    auto stats = worker.getStats();
    EXPECT_EQ(stats.entries_applied, 10u);
    // Latency must be non-negative; a valid timing gives > 0 µs.
    EXPECT_GE(stats.average_latency_us, 0u);
}

TEST_F(ParallelReplicationWorkerTest, AverageLatencyZeroBeforeSubmit) {
    ParallelReplicationWorker worker(defaultConfig());
    auto stats = worker.getStats();
    EXPECT_EQ(stats.average_latency_us, 0u);
}

TEST_F(ParallelReplicationWorkerTest, LargeSubmitBatch) {
    ParallelReplicationWorker::ParallelConfig cfg = defaultConfig();
    cfg.worker_threads = 8;
    cfg.queue_size     = 50000;
    ParallelReplicationWorker worker(cfg);

    constexpr int kEntries = 1000;
    for (int i = 0; i < kEntries; ++i) {
        worker.submit(makeWALEntry("doc" + std::to_string(i % 50)));
    }
    worker.sync();

    auto stats = worker.getStats();
    // All entries must be applied (queue is large enough to hold them all)
    EXPECT_EQ(stats.entries_applied, static_cast<uint64_t>(kEntries));
}