// 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 <filesystem>
#include <fstream>
#include <unistd.h>

#include "document.hpp"
#include "memory_store.hpp"
#include "migrations/migration_runner.hpp"
#include "relations/relation_manager.hpp"
#include "views/view_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");
}


// =========================================================================
// v2.11.0 close-out — the `create_relation` MIGRATION OP.
//
// Declaring schema in migration files is how consumers ship views
// (shadowman-cpp: /opt/shadowman/share/shadowman/migrations/json, callerai:
// /etc/callerai/migrations), and until this op existed they could not declare a
// relation at all. Modelled on create_view: same file shape, same idempotency,
// and it goes through RelationManager::createRelation so the same-project rule
// and the on_delete validation apply rather than being bypassed.
// =========================================================================

std::filesystem::path makeMigrationDir(const std::string& tag) {
    auto dir = std::filesystem::temp_directory_path() /
               ("mig-relation-" + tag + "-" + std::to_string(::getpid()));
    std::filesystem::remove_all(dir);
    std::filesystem::create_directories(dir);
    return dir;
}

void writeMigration(const std::filesystem::path& dir, const std::string& file,
                    const std::string& body) {
    std::ofstream out(dir / file);
    out << body;
}

void test_create_relation_migration_op_declares_and_is_idempotent() {
    auto dir = makeMigrationDir("basic");
    writeMigration(dir, "001_relations.json", R"({
      "version": "001",
      "name": "declare_exec_wf",
      "operations": [
        {"type": "create_collection", "collection": "workflows"},
        {"type": "create_collection", "collection": "executions"},
        {"type": "create_relation",
         "name": "exec_wf",
         "child": "executions",
         "child_field": "workflowId",
         "parent": "workflows",
         "on_delete": "cascade",
         "validate_on_write": true}
      ]
    })");

    Fixture f;
    ViewManager vm(f.store);
    RelationManager rm(f.store);
    MigrationRunner::Config cfg;
    cfg.directory = dir;

    {
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(runner.runMigrations(), "the migration ran");
    }

    // Bare names in a migration file qualify to `default:`, exactly as every
    // other collection name in a migration file does.
    auto got = rm.getRelation("default:exec_wf");
    check(got.has_value(), "the create_relation op DECLARED the relation");
    if (got) {
        check(got->child == "default:executions", "child is project-qualified");
        check(got->childField == "workflowId", "child_field carried through");
        check(got->parent == "default:workflows", "parent is project-qualified");
        check(got->onDelete == OnDelete::Cascade, "on_delete was parsed, not defaulted");
        check(got->validateOnWrite, "validate_on_write carried through");
    }

    // Idempotency has TWO layers and both matter. The runner skips an
    // already-applied migration file, so re-running is a no-op at that level;
    // the op itself must ALSO tolerate "already exists", which is what a
    // consumer re-shipping the same declaration under a new version number
    // hits.
    {
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(runner.runMigrations(), "re-running the same migrations still succeeds");
    }
    writeMigration(dir, "002_again.json", R"({
      "version": "002",
      "name": "declare_exec_wf_again",
      "operations": [
        {"type": "create_relation",
         "name": "exec_wf", "child": "executions",
         "child_field": "workflowId", "parent": "workflows",
         "on_delete": "cascade"}
      ]
    })");
    {
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(runner.runMigrations(),
              "a SECOND migration re-declaring the same relation succeeds - "
              "'already exists' is not a failure on replay");
    }
    check(rm.listRelations("default").size() == 1,
          "and it did not duplicate the declaration");

    std::filesystem::remove_all(dir);
}

void test_create_relation_migration_op_validates() {
    // A typo'd on_delete must FAIL the migration, not silently arm restrict.
    // The reason it matters more here than at the RPC: a typo in a file that
    // ships in a deb would otherwise be wrong on every install, forever.
    {
        auto dir = makeMigrationDir("typo");
        writeMigration(dir, "001_typo.json", R"({
          "version": "001", "name": "typo",
          "operations": [
            {"type": "create_relation", "name": "bad_rel", "child": "executions",
             "child_field": "workflowId", "parent": "workflows",
             "on_delete": "Cascade"}
          ]
        })");
        Fixture f;
        ViewManager vm(f.store);
        RelationManager rm(f.store);
        MigrationRunner::Config cfg;
        cfg.directory = dir;
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(!runner.runMigrations(), "an unrecognised on_delete FAILS the migration");
        check(!rm.getRelation("default:bad_rel").has_value(),
              "and nothing was declared - not coerced to restrict");
        std::filesystem::remove_all(dir);
    }

    // A cross-project declaration must be refused by RelationManager, which is
    // the whole point of routing through createRelation rather than writing the
    // `_relations` record directly.
    {
        auto dir = makeMigrationDir("xproj");
        writeMigration(dir, "001_xproj.json", R"({
          "version": "001", "name": "xproj",
          "operations": [
            {"type": "create_relation", "name": "a:rel", "child": "a:executions",
             "child_field": "workflowId", "parent": "b:workflows"}
          ]
        })");
        Fixture f;
        ViewManager vm(f.store);
        RelationManager rm(f.store);
        MigrationRunner::Config cfg;
        cfg.directory = dir;
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(!runner.runMigrations(),
              "a cross-project relation is refused through the migration op too");
        check(!rm.getRelation("a:rel").has_value(), "and nothing was declared");
        std::filesystem::remove_all(dir);
    }

    // Missing required fields fail rather than declaring a half-relation.
    {
        auto dir = makeMigrationDir("missing");
        writeMigration(dir, "001_missing.json", R"({
          "version": "001", "name": "missing",
          "operations": [
            {"type": "create_relation", "name": "half_rel", "child": "executions"}
          ]
        })");
        Fixture f;
        ViewManager vm(f.store);
        RelationManager rm(f.store);
        MigrationRunner::Config cfg;
        cfg.directory = dir;
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(!runner.runMigrations(), "a create_relation missing child_field/parent fails");
        check(!rm.getRelation("default:half_rel").has_value(), "and declares nothing");
        std::filesystem::remove_all(dir);
    }

    // Absent on_delete means the documented default, and is NOT an error.
    {
        auto dir = makeMigrationDir("default-od");
        writeMigration(dir, "001_default.json", R"({
          "version": "001", "name": "defaulted",
          "operations": [
            {"type": "create_relation", "name": "def_rel", "child": "executions",
             "child_field": "workflowId", "parent": "workflows"}
          ]
        })");
        Fixture f;
        ViewManager vm(f.store);
        RelationManager rm(f.store);
        MigrationRunner::Config cfg;
        cfg.directory = dir;
        MigrationRunner runner(f.store, vm, rm, cfg);
        check(runner.runMigrations(), "an absent on_delete is accepted");
        auto got = rm.getRelation("default:def_rel");
        check(got.has_value(), "and the relation is declared");
        check(got && got->onDelete == OnDelete::Restrict, "with restrict, the documented default");
        check(got && !got->validateOnWrite, "and validate_on_write defaulting to false");
        std::filesystem::remove_all(dir);
    }
}


// v2.11.1 — an operator typed an UNQUALIFIED relation name with a qualified
// child and parent and was told "relation, child and parent must be in one
// project", which named the two arguments that were correct. A bare name
// resolves to `default` like every other bare name, so it could never match a
// child living anywhere else - and since a cross-project relation is impossible,
// there is exactly one project the name could have meant.
void test_unqualified_relation_name_adopts_its_child_project() {
    Fixture f;
    RelationManager rm(f.store);

    RelationInfo r;
    r.name = "exec_wf";                    // bare - would resolve to `default`
    r.child = "acme:executions";
    r.childField = "workflowId";
    r.parent = "acme:workflows";
    std::string err;
    check(rm.createRelation(r, err), "a bare name with an acme child is accepted: " + err);
    check(rm.getRelation("acme:exec_wf").has_value(),
          "and it is stored in acme, not default");
    check(!rm.getRelation("default:exec_wf").has_value(),
          "nothing was created in default");
    check(rm.listRelations("acme").size() == 1, "acme lists it");
    check(rm.listRelations("default").empty(), "default does not");

    // Reload, because adoption must happen in the persisted record and not only
    // in the in-memory cache - otherwise a restart would lose the relation.
    RelationManager fresh(f.store);
    fresh.loadFromStore();
    check(fresh.getRelation("acme:exec_wf").has_value(), "survives a reload in acme");
}

// An EXPLICITLY qualified name that disagrees is a real statement of intent that
// cannot be honoured, so it stays refused - and now the message names the
// argument that is actually wrong plus the project it resolved to.
void test_explicitly_wrong_project_on_the_name_is_still_refused_and_says_which() {
    Fixture f;
    RelationManager rm(f.store);

    RelationInfo r;
    r.name = "default:exec_wf";            // deliberately the wrong project
    r.child = "acme:executions";
    r.childField = "workflowId";
    r.parent = "acme:workflows";
    std::string err;
    check(!rm.createRelation(r, err), "a qualified name in the wrong project is refused");
    check(err.find("relation name") != std::string::npos,
          "the message blames the NAME, not the child or parent: " + err);
    check(err.find("acme:exec_wf") != std::string::npos,
          "and spells out the name that would have worked: " + err);
    check(!rm.getRelation("acme:exec_wf").has_value(), "nothing was created");
}

// The child/parent mismatch message must name both sides and their projects. It
// previously listed all three arguments and said which project none of them was
// in, which is what sent an operator looking at the wrong argument.
void test_child_parent_mismatch_names_both_sides() {
    Fixture f;
    RelationManager rm(f.store);
    RelationInfo r;
    r.name = "acme:bad";
    r.child = "acme:executions";
    r.childField = "workflowId";
    r.parent = "other:workflows";
    std::string err;
    check(!rm.createRelation(r, err), "refused");
    check(err.find("acme:executions") != std::string::npos &&
          err.find("other:workflows") != std::string::npos,
          "both offending arguments are named: " + err);
    check(err.find("'acme'") != std::string::npos && err.find("'other'") != std::string::npos,
          "with the projects they resolved to: " + err);
}

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();
    test_create_relation_migration_op_declares_and_is_idempotent();
    test_create_relation_migration_op_validates();
    test_unqualified_relation_name_adopts_its_child_project();
    test_explicitly_wrong_project_on_the_name_is_still_refused_and_says_which();
    test_child_parent_mismatch_names_both_sides();
    std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
    return g_fail == 0 ? 0 : 1;
}