// Task 1 — RelationManager: the declaration registry for referential
// integrity relations. No enforcement, no LMDB index yet (later tasks).
//
// Modeled on tests/test_view_manager_paging.cpp: relations are keyed by
// their project-qualified name in a `_relations` system collection, and
// loadFromStore() must page explicitly since Query::limit defaults to 100
// and limit=0 returns nothing (not everything) — the same trap that has
// already shipped as a bug in ViewManager, PolicyManager and
// CollectionConfigManager.

#include <iostream>
#include <string>

#include <nlohmann/json.hpp>

#include "document.hpp"
#include "memory_store.hpp"
#include "relations/relation_manager.hpp"

using namespace smartbotic::database;

namespace {

int g_pass = 0;
int g_fail = 0;

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

struct Fixture {
    MemoryStore store;
    Fixture() : store(MemoryStore::Config{}) {
        store.start();
    }
    ~Fixture() { store.stop(); }
};

void test_relations_are_project_scoped_and_survive_reload() {
    Fixture f;                       // MemoryStore, started
    RelationManager rm(f.store);
    rm.loadFromStore();

    RelationInfo a;
    a.name = "default:exec_wf";
    a.child = "default:executions";
    a.childField = "workflowId";
    a.parent = "default:workflows";
    std::string err;
    check(rm.createRelation(a, err), "created in default");

    RelationInfo b = a;              // SAME bare name, different project
    b.name = "acme:exec_wf";
    b.child = "acme:executions";
    b.parent = "acme:workflows";
    check(rm.createRelation(b, err), "the same name in another project is allowed");

    check(rm.listRelations("default").size() == 1, "listing is project-filtered");
    check(rm.listRelations().size() == 2, "empty project lists everything");

    RelationManager fresh(f.store);   // restart
    fresh.loadFromStore();
    check(fresh.getRelation("default:exec_wf").has_value(), "survives reload");
    check(fresh.getRelation("acme:exec_wf").has_value(), "both survive");
}

void test_cross_project_relation_is_refused() {
    Fixture f;
    RelationManager rm(f.store);
    RelationInfo r;
    r.name = "default:bad";
    r.child = "default:executions";
    r.childField = "workflowId";
    r.parent = "acme:workflows";      // different env - no txn spans two
    std::string err;
    check(!rm.createRelation(r, err), "a cross-project relation is refused");
    check(err.find("project") != std::string::npos, "and says why");
}

void test_more_than_one_page_of_relations_loads() {
    Fixture f;
    RelationManager rm(f.store);
    for (int i = 0; i < 250; ++i) {          // Query::limit defaults to 100
        RelationInfo r;
        char buf[32];
        std::snprintf(buf, sizeof(buf), "default:r%03d", i);
        r.name = buf;
        r.child = "default:c";
        r.childField = "p";
        r.parent = "default:p";
        std::string err;
        rm.createRelation(r, err);
    }
    RelationManager fresh(f.store);
    fresh.loadFromStore();
    check(fresh.listRelations().size() == 250,
          "all 250 load - a bare Query would stop at 100, as it did for views, "
          "policies and collection configs");
}

// v2.11.0 T13 round 2 (review finding 3) — createRelation() refuses a
// cross-project declaration (test_cross_project_relation_is_refused,
// above), but loadFromStore() is the OTHER way a RelationInfo enters the
// cache and did not re-check it. A legacy or hand-written `_relations`
// document naming a cross-project parent must be skipped at load time too -
// arming it would resolve the bare parent name inside the CHILD's own
// project env (validate_on_write/RelationRef only ever resolve `parent`
// against the child's project), silently checking the wrong collection.
void test_load_skips_a_cross_project_relation_written_by_hand() {
    Fixture f;

    // Bypass createRelation()'s own guard entirely - write the raw document
    // straight into `_relations`, the way a legacy record or a hand-edited
    // one would exist on disk. `parent` names a DIFFERENT project than
    // `child`, which createRelation() would refuse today.
    nlohmann::json bad = {
        {"name", "default:bad_cross"},
        {"child", "default:executions"},
        {"child_field", "workflowId"},
        {"parent", "acme:workflows"},
        {"on_delete", "restrict"},
        {"validate_on_write", true},
        {"created_at", 0},
        {"updated_at", 0},
    };
    Document d;
    d.id = "default:bad_cross";
    d.collection = RelationManager::SYSTEM_COLLECTION;
    d.set_data(bad);
    f.store.insert(RelationManager::SYSTEM_COLLECTION, d);

    // A well-formed, same-project relation alongside it, to confirm one bad
    // record does not stop the rest of the load.
    RelationInfo good;
    good.name = "default:exec_wf";
    good.child = "default:executions";
    good.childField = "ownerId";
    good.parent = "default:users";
    {
        RelationManager rm(f.store);
        rm.loadFromStore();
        std::string err;
        check(rm.createRelation(good, err), "the well-formed sibling declares fine");
    }

    RelationManager fresh(f.store);
    fresh.loadFromStore();
    check(!fresh.getRelation("default:bad_cross").has_value(),
          "the cross-project relation was skipped, not armed with the wrong parent");
    check(fresh.getRelation("default:exec_wf").has_value(),
          "the well-formed sibling still loaded - one bad record did not stop the rest");
}

}  // namespace


// =========================================================================
// v2.11.0 final review, finding 8 — the bare-vs-qualified bug class, made
// unrepresentable at the RelationManager boundary rather than spot-fixed at
// the caller.
//
// The live bug: armRelationsForChild() looked relations up under the caller's
// RAW string while boot arming used the canonical "<project>:<collection>",
// and set_relations() keys the storage map by the BARE name either way. So a
// raw-gRPC caller naming "executions" instead of "default:executions" found
// zero relations, and set_relations(bare, {}) then ERASED the entry the
// canonical arming had filled - disarming both the reverse index and
// validate_on_write for the life of the process, logged only as "re-armed 0
// relation(s)", and silently repaired by a restart.
//
// This is the third occurrence of the class (v2.4.2 lost every view for two
// releases, v2.4.5 gave every collection a phantom twin), so the fix is at the
// one funnel every entry point already goes through.
void test_bare_and_qualified_names_are_the_same_relation() {
    Fixture f;
    RelationManager rm(f.store);
    rm.loadFromStore();

    // Declared with an UNQUALIFIED name and unqualified endpoints, exactly as a
    // raw-gRPC caller (or a hand-written migration) would.
    RelationInfo bare;
    bare.name = "exec_wf";
    bare.child = "executions";
    bare.childField = "workflowId";
    bare.parent = "workflows";
    std::string err;
    check(rm.createRelation(bare, err), "a bare-named relation is accepted");

    // It is STORED canonically, so everything downstream sees one form.
    auto viaBare = rm.getRelation("exec_wf");
    check(viaBare.has_value(), "found under the bare name the caller used");
    check(viaBare && viaBare->name == "default:exec_wf",
          "but it reports its CANONICAL name - the declaration was normalised, "
          "not stored verbatim");
    check(viaBare && viaBare->child == "default:executions", "child canonicalised too");
    check(viaBare && viaBare->parent == "default:workflows", "parent canonicalised too");

    auto viaQualified = rm.getRelation("default:exec_wf");
    check(viaQualified.has_value(), "and found under the qualified name as well");

    // A second declaration under the OTHER spelling is a DUPLICATE, not a new
    // relation. Before the fix this created a second entry that armed the same
    // bare collection and clobbered the first.
    RelationInfo qualified = bare;
    qualified.name = "default:exec_wf";
    qualified.child = "default:executions";
    qualified.parent = "default:workflows";
    check(!rm.createRelation(qualified, err),
          "declaring the qualified spelling of an existing bare relation is refused "
          "as a duplicate");
    check(err.find("already exists") != std::string::npos, "and says why");
    check(rm.listRelations().size() == 1, "still exactly one relation, not two");

    // THE LOOKUP THAT WAS BROKEN: armRelationsForChild's, and Delete's.
    check(rm.relationsWithChild("executions").size() == 1,
          "relationsWithChild finds it under the BARE collection name - this is "
          "the lookup that returned zero and caused the disarm");
    check(rm.relationsWithChild("default:executions").size() == 1,
          "and under the qualified one");
    check(rm.relationsWithParent("workflows").size() == 1,
          "relationsWithParent likewise - a bare name here would report 'nobody's "
          "parent' and permit every delete");
    check(rm.relationsWithParent("default:workflows").size() == 1, "and qualified");

    // Cross-project isolation is not weakened by canonicalisation.
    check(rm.relationsWithChild("acme:executions").empty(),
          "another project's same-named collection still matches nothing");

    // Dropping by either spelling works, and actually removes the stored row.
    check(rm.dropRelation("exec_wf", err), "droppable by the bare name");
    check(!rm.getRelation("default:exec_wf").has_value(), "gone from the cache");
    RelationManager reloaded(f.store);
    reloaded.loadFromStore();
    check(reloaded.listRelations().empty(),
          "and gone from the store - dropping by the bare name really removed the "
          "canonical document, it did not just clear the cache");
}

// finding 8, the migration half: a record already stored under a non-canonical
// document id is re-keyed on load, in the store as well as in the cache.
// Without the store half, dropRelation() (which removes by canonical name)
// would leave the row behind and the relation would come back on the next boot.
void test_load_rekeys_a_legacy_bare_named_record() {
    Fixture f;
    // Hand-write a record the way a pre-fix createRelation would have stored it.
    Document d;
    d.id = "legacy_rel";
    d.set_data(nlohmann::json{
        {"name", "legacy_rel"},
        {"child", "executions"},
        {"child_field", "workflowId"},
        {"parent", "workflows"},
        {"on_delete", "restrict"},
        {"validate_on_write", false}});
    f.store.createCollection(RelationManager::SYSTEM_COLLECTION, CollectionOptions{});
    f.store.upsert(RelationManager::SYSTEM_COLLECTION, d);

    RelationManager rm(f.store);
    rm.loadFromStore();
    auto got = rm.getRelation("default:legacy_rel");
    check(got.has_value(), "the legacy record loaded");
    check(got && got->name == "default:legacy_rel", "under its canonical name");

    // The STORE was re-keyed, not just the cache.
    check(!f.store.get(RelationManager::SYSTEM_COLLECTION, "legacy_rel").has_value(),
          "the old bare-keyed document is gone");
    check(f.store.get(RelationManager::SYSTEM_COLLECTION, "default:legacy_rel").has_value(),
          "and a canonically-keyed one exists - so dropRelation(), which removes "
          "by the canonical name, can actually remove it");

    std::string err;
    check(rm.dropRelation("default:legacy_rel", err), "and it drops");
    RelationManager reloaded(f.store);
    reloaded.loadFromStore();
    check(reloaded.listRelations().empty(), "staying dropped across a reload");
}

// finding 9 — an unrecognised on_delete must be REFUSED, not coerced to
// Restrict. There used to be two copies of the parser (one in
// database_grpc_impl.cpp, one here) and both coerced silently. That failed safe
// while cascade/set_null were inert; T12 made them destructive in the other
// direction, so an operator who typed "Cascade" was told the relation was
// created and believed cascade was armed while restrict was.
void test_on_delete_is_validated_not_coerced() {
    check(parseOnDelete("restrict").has_value(), "restrict parses");
    check(parseOnDelete("cascade") == OnDelete::Cascade, "cascade parses");
    check(parseOnDelete("set_null") == OnDelete::SetNull, "set_null parses");
    check(parseOnDelete("no_action") == OnDelete::NoAction, "no_action parses");

    // The typo cases that used to become Restrict silently.
    check(!parseOnDelete("Cascade").has_value(), "'Cascade' is REFUSED, not coerced");
    check(!parseOnDelete("CASCADE").has_value(), "'CASCADE' is refused");
    check(!parseOnDelete("cascde").has_value(), "a misspelling is refused");
    check(!parseOnDelete("setnull").has_value(), "'setnull' is refused");
    check(!parseOnDelete("").has_value(),
          "and the empty string is refused HERE - the RPC layer, not the parser, "
          "is what maps absent to the documented default");

    // Round-trips through the one renderer, so the two directions cannot drift.
    for (auto v : {OnDelete::Restrict, OnDelete::Cascade, OnDelete::SetNull,
                   OnDelete::NoAction}) {
        check(parseOnDelete(onDeleteToString(v)) == v,
              "onDeleteToString round-trips through parseOnDelete");
    }

    // A persisted record carrying a bad value falls back to Restrict (the safe
    // direction: over-restricting refuses deletes, it never performs an
    // unintended destructive one) rather than being dropped, which would remove
    // protection entirely.
    Fixture f;
    Document d;
    d.id = "default:bad_od";
    d.set_data(nlohmann::json{
        {"name", "default:bad_od"},
        {"child", "default:executions"},
        {"child_field", "workflowId"},
        {"parent", "default:workflows"},
        {"on_delete", "Cascade"}});
    f.store.createCollection(RelationManager::SYSTEM_COLLECTION, CollectionOptions{});
    f.store.upsert(RelationManager::SYSTEM_COLLECTION, d);

    RelationManager rm(f.store);
    rm.loadFromStore();
    auto got = rm.getRelation("default:bad_od");
    check(got.has_value(),
          "a persisted record with a bad on_delete is still LOADED - dropping it "
          "would silently remove protection");
    check(got && got->onDelete == OnDelete::Restrict,
          "and it falls back to restrict, the non-destructive direction");
}

int main() {
    std::cout << "=== test_relation_manager ===\n";
    test_relations_are_project_scoped_and_survive_reload();
    test_cross_project_relation_is_refused();
    test_more_than_one_page_of_relations_loads();
    test_load_skips_a_cross_project_relation_written_by_hand();
    test_bare_and_qualified_names_are_the_same_relation();
    test_load_rekeys_a_legacy_bare_named_record();
    test_on_delete_is_validated_not_coerced();
    std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
    return g_fail == 0 ? 0 : 1;
}