// v2.11.0 T2 — relation reverse index tests.
//
// Storage-only: this exercises LmdbDocumentStore::relation_index_{add,remove,
// child_count,children} directly. No document-write hooks (Task 3), no
// enforcement (Task 4) are involved.
//
// test_index_created_at_runtime_is_visible_after_reopen is THE regression
// test for this task: since v2.8.1 the read path serves only from the primed
// dbi cache, so a sub-db created at runtime is invisible to every later read
// in the process unless cacheCommittedDbi() runs after commit. Removing that
// call from relation_index_add must fail this test.

#include <algorithm>
#include <atomic>
#include <cstdio>
#include <filesystem>
#include <iostream>
#include <string>
#include <unistd.h>
#include <vector>

#include <nlohmann/json.hpp>

#include "document.hpp"
#include "storage/document_store_lmdb.hpp"
#include "storage/lmdb_env.hpp"

namespace fs = std::filesystem;

using smartbotic::database::Document;
using smartbotic::db::storage::LmdbDocumentStore;
using smartbotic::db::storage::LmdbEnv;
using smartbotic::db::storage::LmdbEnvOpts;

namespace {

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";
    }
}

std::string make_tmpdir(const char* tag) {
    static std::atomic<int> counter{0};
    std::string path = "/tmp/relidx-test-" + std::to_string(::getpid()) + "-" +
                       std::to_string(counter.fetch_add(1)) + "-" + tag;
    std::error_code ec;
    fs::remove_all(path, ec);
    return path;
}

struct TmpEnv {
    std::string path;
    LmdbEnv env;
    explicit TmpEnv(const char* tag)
        : path(make_tmpdir(tag)),
          env(LmdbEnvOpts{path, 64ULL << 20, 256, 126, false}) {}
    ~TmpEnv() {
        std::error_code ec;
        fs::remove_all(path, ec);
    }
    TmpEnv(const TmpEnv&) = delete;
    TmpEnv& operator=(const TmpEnv&) = delete;
};

// -------------------------------------------------------------------------

// DUPSORT shape, per the re-validated design.
void test_children_of_a_parent_are_a_dup_set() {
    TmpEnv t("relidx");
    LmdbDocumentStore store(t.env);

    check(store.relation_index_add("r1", "wf-1", "exec-a"), "added a child");
    check(store.relation_index_add("r1", "wf-1", "exec-b"), "and another");
    check(store.relation_index_add("r1", "wf-2", "exec-c"), "under a second parent");

    // THE operation restrict and DescribeDelete need: a count without reading
    // the children. mdb_cursor_count makes it O(1)-ish.
    check(store.relation_index_child_count("r1", "wf-1") == 2, "two children of wf-1");
    check(store.relation_index_child_count("r1", "wf-2") == 1, "one child of wf-2");
    check(store.relation_index_child_count("r1", "wf-none") == 0,
          "an unreferenced parent has none, and that is not an error");

    auto kids = store.relation_index_children("r1", "wf-1", 10);
    std::sort(kids.begin(), kids.end());
    check(kids == std::vector<std::string>{"exec-a", "exec-b"}, "children listed");

    // Removing one pair must not remove the sibling.
    check(store.relation_index_remove("r1", "wf-1", "exec-a"), "removed one pair");
    check(store.relation_index_child_count("r1", "wf-1") == 1, "the sibling survives");

    // Idempotent: re-adding the same pair is a no-op, not a duplicate.
    store.relation_index_add("r1", "wf-1", "exec-b");
    check(store.relation_index_child_count("r1", "wf-1") == 1, "no duplicate posting");

    // A distinct relation is a distinct sub-db: same parent id, no crosstalk.
    check(store.relation_index_child_count("r2", "wf-1") == 0,
          "a different relation's index is independent");

    // Removing a pair that was never present is reported, not thrown.
    check(!store.relation_index_remove("r1", "wf-1", "exec-does-not-exist"),
          "removing an absent pair returns false rather than throwing");
    check(!store.relation_index_remove("r1", "wf-none", "exec-a"),
          "removing under an absent parent returns false rather than throwing");
}

// The failure this repo is most likely to reproduce. Since v2.8.1 reads serve
// only from the primed dbi cache, so a sub-db created at runtime is invisible
// unless registered after commit - and relations would silently not enforce.
void test_index_created_at_runtime_is_visible_after_reopen() {
    const std::string path = make_tmpdir("relidx-visible");
    const LmdbEnvOpts opts{path, 64ULL << 20, 256, 126, false};
    {
        LmdbEnv env(opts);
        LmdbDocumentStore store(env);
        store.relation_index_add("r1", "wf-1", "exec-a");
        // Same process, no reopen: must be visible immediately.
        check(store.relation_index_child_count("r1", "wf-1") == 1,
              "visible in the process that created it - this is what "
              "cacheCommittedDbi() buys");
    }
    LmdbEnv env2(opts);
    LmdbDocumentStore store2(env2);      // priming runs in the constructor
    check(store2.relation_index_child_count("r1", "wf-1") == 1,
          "and after a restart, via prime_dbi_cache()");

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

// v2.11.0 T7 — declaring a relation over a collection that ALREADY has rows
// must index those pre-existing rows, not just ones written afterwards.
// This is the exact bug the task brief calls out: without a bootstrap scan,
// declaring a relation over populated data leaves every existing child
// invisible to enforcement while looking like it worked.
void test_build_relation_index_covers_pre_existing_rows() {
    TmpEnv t("relbuild-basic");
    LmdbDocumentStore store(t.env);

    // Rows exist BEFORE the relation is ever declared - no set_relations()
    // call has happened, so put() below does not touch the reverse index at
    // all. This is exactly "declare a relation on a populated collection".
    auto put = [&](const std::string& id, const nlohmann::json& data) {
        Document d; d.id = id; d.collection = "executions"; d.set_data(data);
        store.put("executions", id, d);
    };
    put("e1", {{"workflowId", "wf-1"}});
    put("e2", {{"workflowId", "wf-1"}});
    put("e3", {{"workflowId", "wf-2"}});
    put("e4", {{"other", 1}});                 // no reference - must not count
    put("e5", {{"workflowId", nullptr}});      // null - must not count

    check(store.relation_index_child_count("exec_wf", "wf-1") == 0,
          "before the bootstrap scan, the reverse index knows nothing - this "
          "is the bug: a declared relation would silently protect nothing");

    const uint64_t indexed = store.build_relation_index("exec_wf", "executions", "workflowId");
    check(indexed == 3, "3 rows had a resolvable reference (e4/e5 excluded)");
    check(store.relation_index_child_count("exec_wf", "wf-1") == 2,
          "both pre-existing children of wf-1 are now indexed");
    check(store.relation_index_child_count("exec_wf", "wf-2") == 1,
          "and wf-2's child");

    auto kids = store.relation_index_children("exec_wf", "wf-1", 10);
    std::sort(kids.begin(), kids.end());
    check(kids == std::vector<std::string>{"e1", "e2"}, "correct children listed");

    // Idempotent: re-running the scan (e.g. re-declaring the relation) must
    // not double the postings, the same MDB_NODUPDATA guarantee build_index
    // already relies on.
    const uint64_t reindexed = store.build_relation_index("exec_wf", "executions", "workflowId");
    check(reindexed == 3, "the re-scan still visits the same 3 rows");
    check(store.relation_index_child_count("exec_wf", "wf-1") == 2,
          "no duplicate postings from re-running the scan");
    check(store.relation_index_child_count("exec_wf", "wf-2") == 1,
          "no duplicate postings on wf-2 either");

    check(store.relation_index_exists("exec_wf"),
          "the sub-db now exists - relation_index_exists distinguishes this "
          "from 'declared but never built'");
    check(!store.relation_index_exists("never_declared"),
          "an unrelated relation's sub-db was never created");
}

// An array-valued child field must contribute one posting per element, same
// as the write-path maintainRelations() - the bootstrap scan must not
// diverge from ongoing maintenance.
void test_build_relation_index_handles_array_valued_field() {
    TmpEnv t("relbuild-array");
    LmdbDocumentStore store(t.env);

    Document n; n.id = "n1"; n.collection = "nodes";
    n.set_data({{"config", {{"credentialIds", {"c1", "c2"}}}}});
    store.put("nodes", "n1", n);

    const uint64_t indexed =
        store.build_relation_index("node_creds", "nodes", "config.credentialIds");
    check(indexed == 1, "one row contributed (it has two postings, one row)");
    check(store.relation_index_child_count("node_creds", "c1") == 1, "array element 1");
    check(store.relation_index_child_count("node_creds", "c2") == 1, "array element 2");
}

// A relation over a collection that does not exist yet (or is empty) must
// not throw, and must not fabricate a sub-db that then confuses
// relation_index_exists.
void test_build_relation_index_over_absent_collection() {
    TmpEnv t("relbuild-absent");
    LmdbDocumentStore store(t.env);

    const uint64_t indexed = store.build_relation_index("exec_wf", "executions", "workflowId");
    check(indexed == 0, "nothing to index in a collection that was never written");
}

// v2.11.0 T7 — check_relation_dangling: a genuinely dangling reference (a
// child row referencing a parent id that does not exist) must be reported,
// with the correct count and sample child ids; a live reference must not be
// flagged.
void test_check_relation_dangling_finds_missing_parents() {
    TmpEnv t("relcheck-basic");
    LmdbDocumentStore store(t.env);

    auto putParent = [&](const std::string& id) {
        Document d; d.id = id; d.collection = "workflows"; d.set_data({{"name", id}});
        store.put("workflows", id, d);
    };
    auto putChild = [&](const std::string& id, const std::string& wf) {
        Document d; d.id = id; d.collection = "executions";
        d.set_data({{"workflowId", wf}});
        store.put("executions", id, d);
    };

    // wf-1 exists and is referenced - not dangling.
    putParent("wf-1");
    putChild("e1", "wf-1");
    putChild("e2", "wf-1");

    // wf-missing is referenced but was NEVER written as a parent - dangling.
    putChild("e3", "wf-missing");
    putChild("e4", "wf-missing");
    putChild("e5", "wf-missing");

    store.build_relation_index("exec_wf", "executions", "workflowId");

    auto result = store.check_relation_dangling("exec_wf", "workflows", 100);
    check(result.total == 1, "exactly one dangling parent id");
    check(result.entries.size() == 1, "and it is reported (well under the cap)");
    if (!result.entries.empty()) {
        const auto& d = result.entries[0];
        check(d.parentId == "wf-missing", "names the missing parent");
        check(d.childCount == 3, "counts all three referencing children");
        check(d.sampleChildIds.size() == 3, "samples all three (under the 5-sample cap)");
        std::vector<std::string> sorted = d.sampleChildIds;
        std::sort(sorted.begin(), sorted.end());
        check(sorted == std::vector<std::string>{"e3", "e4", "e5"}, "correct sample ids");
    }

    // check_relation_dangling must mutate nothing - the index and parent
    // collection are exactly as they were.
    check(store.relation_index_child_count("exec_wf", "wf-missing") == 3,
          "the check did not remove or alter the dangling postings");
    check(store.count("workflows") == 1, "the check did not create a phantom parent row");
}

// A relation with no dangling references reports zero, and never invents
// one for a value that legitimately exists.
void test_check_relation_dangling_clean_relation() {
    TmpEnv t("relcheck-clean");
    LmdbDocumentStore store(t.env);

    Document p; p.id = "wf-1"; p.collection = "workflows"; p.set_data({{"name", "wf-1"}});
    store.put("workflows", "wf-1", p);
    Document c; c.id = "e1"; c.collection = "executions";
    c.set_data({{"workflowId", "wf-1"}});
    store.put("executions", "e1", c);

    store.build_relation_index("exec_wf", "executions", "workflowId");

    auto result = store.check_relation_dangling("exec_wf", "workflows", 100);
    check(result.total == 0, "no dangling references");
    check(result.entries.empty(), "nothing reported");
}

// A relation that was declared but never built (no build_relation_index /
// relation_index_add call at all) has no sub-db - check_relation_dangling
// must report "nothing to say" rather than treating an absent index as
// "everything is dangling".
void test_check_relation_dangling_never_built_reports_nothing() {
    TmpEnv t("relcheck-unbuilt");
    LmdbDocumentStore store(t.env);

    Document c; c.id = "e1"; c.collection = "executions";
    c.set_data({{"workflowId", "wf-1"}});
    store.put("executions", "e1", c);   // no relation declared/built at all

    auto result = store.check_relation_dangling("exec_wf", "workflows", 100);
    check(result.total == 0, "an unbuilt index has nothing to report - not a false positive");
    check(result.entries.empty(), "nothing reported");
}

// The `maxResults` cap bounds the returned samples but `total` must still be
// exact - a caller must never be able to mistake a capped list for the
// complete one.
void test_check_relation_dangling_respects_cap_but_total_is_exact() {
    TmpEnv t("relcheck-capped");
    LmdbDocumentStore store(t.env);

    for (int i = 0; i < 5; ++i) {
        Document c; c.id = "e" + std::to_string(i); c.collection = "executions";
        c.set_data({{"workflowId", "wf-missing-" + std::to_string(i)}});
        store.put("executions", "e" + std::to_string(i), c);
    }
    store.build_relation_index("exec_wf", "executions", "workflowId");

    auto result = store.check_relation_dangling("exec_wf", "workflows", 2);
    check(result.total == 5, "total counts every dangling parent, not just the capped sample");
    check(result.entries.size() == 2, "entries is capped at maxResults");
}

}  // namespace

int main() {
    std::cout << "=== test_relation_index ===\n";
    test_children_of_a_parent_are_a_dup_set();
    test_index_created_at_runtime_is_visible_after_reopen();
    test_build_relation_index_covers_pre_existing_rows();
    test_build_relation_index_handles_array_valued_field();
    test_build_relation_index_over_absent_collection();
    test_check_relation_dangling_finds_missing_parents();
    test_check_relation_dangling_clean_relation();
    test_check_relation_dangling_never_built_reports_nothing();
    test_check_relation_dangling_respects_cap_but_total_is_exact();

    std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
    return g_fail == 0 ? 0 : 1;
}