// SPDX-License-Identifier: AGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

// The fixes of the 2026-10-22 code review: the broker, the splitter, the
// connection, the preflight, the calibration, the sampler, the resume, the
// template. The SQL helper is this binary run as `sql`; app_ready, the
// database calls and the methods of Run are replaced through the Run's
// seams.

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
import std;
import Crafter.Review;
namespace fs = std::filesystem;
using namespace Crafter::Review;

#include "../Common/FakePg.hpp"
using namespace Testing;

namespace {
    std::int32_t Failures = 0;
    const std::string Fn(Plan::FunctionalLens);

    void Check(std::string_view name, bool cond, std::string_view detail = {}) {
        std::println("{}{}{}", cond ? "  ok   " : "  FAIL ", name, !cond && !detail.empty() ? std::format("  ({})", detail) : std::string());
        if (!cond) ++Failures;
    }

    void Write(const fs::path& path, std::string_view text) {
        fs::create_directories(path.parent_path());
        std::ofstream(path, std::ios::binary) << text;
    }

    std::string Show(const std::vector<std::string>& lines) {
        Json out = Json::array();
        for (const std::string& line : lines) out.push_back(line);
        return Dyn::ReprValue(out);
    }

    // A server on 127.0.0.1 that resets every connection it takes (SO_LINGER
    // 0, then close) until it goes. A real reset races the startup message,
    // so either it does not go out or its answer never comes.
    class Resetter {
    public:
        Resetter() {
            listener = socket(AF_INET, SOCK_STREAM, 0);
            sockaddr_in address {};
            address.sin_family = AF_INET;
            address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
            bind(listener, reinterpret_cast<sockaddr*>(&address), sizeof(address));
            listen(listener, 8);
            socklen_t length = sizeof(address);
            getsockname(listener, reinterpret_cast<sockaddr*>(&address), &length);
            port = ntohs(address.sin_port);
            thread = std::thread([this] {
                while (true) {
                    const std::int32_t c = accept(listener, nullptr, nullptr);
                    if (c < 0) break;
                    const linger hard {1, 0};
                    setsockopt(c, SOL_SOCKET, SO_LINGER, &hard, sizeof(hard));
                    close(c);
                }
            });
        }

        ~Resetter() {
            // A shut listening socket ends the accept that waits on it.
            shutdown(listener, SHUT_RDWR);
            thread.join();
            close(listener);
        }

        std::uint16_t port = 0;

    private:
        std::int32_t listener = -1;
        std::thread thread;
    };

    bool Reset(std::string_view text) {
        return text.contains("Broken pipe") || text.contains("Connection reset");
    }

    std::int32_t Checks() {
        std::println("the 2026-10-22 code review: the broker, the splitter, the connection, the preflight, the calibration, the sampler, the resume, the template");
        Platform::SetMode(Platform::Mode::Native);
        SetHome(fs::current_path());
        const fs::path base = fs::temp_directory_path() / std::format("crafter-review-fixes-{}", Platform::PublishedPid());
        fs::remove_all(base);
        // finding 8: a SQL-standard routine body stays whole; ROLLBACK [WORK |
        // TRANSACTION] TO goes through.
        const std::string atomic = "create function f(x int) returns int language sql begin atomic select case when x > 0 then 1 else 0 end; select 2; end; "
                                   "create or replace procedure p() language sql begin atomic insert into t values (1); end; select 3; "
                                   "create function g() returns int language plpgsql as $$ begin return 1; end $$; select (case when true then 1 end); begin";
        const std::vector<std::string> parts = PgWire::SplitStatements(atomic);
        Check("split: a SQL-standard routine body (BEGIN ATOMIC ... END, a CASE ... END inside) stays one statement, CREATE OR REPLACE PROCEDURE too; "
              "a plpgsql $$ body, a CASE outside a routine and a bare BEGIN cut as before (finding 8)",
              parts == std::vector<std::string>{"create function f(x int) returns int language sql begin atomic select case when x > 0 then 1 else 0 end; select 2; end",
                                                "create or replace procedure p() language sql begin atomic insert into t values (1); end", "select 3",
                                                "create function g() returns int language plpgsql as $$ begin return 1; end $$", "select (case when true then 1 end)", "begin"},
              Show(parts));
        Check("split: the words are counted per statement - a BEGIN after a CREATE TABLE is no routine body; an E-string after a word is still an E-string",
              PgWire::SplitStatements(R"(create table t (a int); begin; select E'it\'s; here'; select 1)")
                  == std::vector<std::string>{"create table t (a int)", "begin", R"(select E'it\'s; here')", "select 1"});
        const std::vector<std::string> okRb = {"rollback to s", "rollback to savepoint s", "rollback work to s", "ROLLBACK TRANSACTION TO SAVEPOINT s"};
        const std::vector<std::string> noRb = {"rollback", "rollback work", "rollback transaction", "rollback and chain", "rollback prepared 'x'", "rollback work and no chain"};
        std::vector<std::string> verdicts;
        for (const auto& list : {okRb, noRb}) {
            for (const std::string& s : list) verdicts.push_back(std::format("{}: {}", s, PgWire::Refused(s).value_or("None")));
        }
        Check("refused: ROLLBACK [WORK | TRANSACTION] TO [SAVEPOINT] goes through and every other ROLLBACK is refused; the BEGIN ATOMIC routine is not refused",
              std::ranges::all_of(okRb, [](const std::string& s) { return !PgWire::Refused(s); }) && std::ranges::all_of(noRb, [](const std::string& s) { return PgWire::Refused(s).has_value(); })
                  && !PgWire::Refused(parts.at(0)),
              Show(verdicts));

        // finding 7: nothing but a DbError leaves a connection, whatever
        // breaks during the startup message.
        {
            const Resetter rst;
            const std::string url0 = std::format("postgresql://u:p@127.0.0.1:{}/d", rst.port);
            std::optional<std::string> raised;
            try {
                PgWire::Connection connection(url0);
            } catch (const PgWire::DbError& error) {
                raised = error.what();
            }
            std::vector<std::string> out;
            const std::int32_t code = PgWire::RunSql(url0, "select 1", [&](const std::string& line) { out.push_back(line); },
                                                     [](std::string_view u) { return PgWire::Connect(u, 1, [](double) {}); });
            const auto probe0 = PgWire::Probe(url0);
            Check("connection: a reset while the startup message goes out is a DbError, never an OSError - the helper says NO DATABASE (exit 2), the probe that it does not answer (finding 7)",
                  raised && Reset(*raised) && code == PgWire::ExitNoDb && !out.empty() && out[0].starts_with(PgWire::NoDatabase) && !probe0.first,
                  std::format("{} | {} | {}", raised.value_or("None"), Show(out), probe0.second));
        }
        {
            const Resetter srv;
            std::optional<std::string> reset;
            try {
                PgWire::Connect(std::format("postgresql://u:p@127.0.0.1:{}/d", srv.port), 1, [](double) {});
            } catch (const PgWire::DbError& error) {
                reset = error.what();
            }
            Check("connection: a server that resets the connection at once is a DbError too", reset.has_value(), reset.value_or("None"));
        }

        // finding 6: the broker - the URL stays in the driver, the helper
        // reaches the database through it.
        {
            FakePg pg;
            const std::string url = std::format("postgresql://u:pw@127.0.0.1:{}/db", pg.port);
            PgWire::Broker br;
            br.Start();
            const std::string addr = br.Address(url);
            std::map<std::string, std::string> envb = Platform::Environment();
            envb.erase(std::string(PgWire::BrokerEnv));
            envb.erase(std::string(PgWire::UrlEnv));
            auto sh = [&](std::vector<std::string> args, const std::optional<std::string>& broker, std::string_view stdinText = {}) {
                args.insert(args.begin(), {Platform::SelfExe().string(), "sql"});
                std::map<std::string, std::string> env = envb;
                if (broker) env[std::string(PgWire::BrokerEnv)] = *broker;
                return Process::Run(args, std::nullopt, std::chrono::seconds(60), stdinText, env);
            };
            const Process::Captured r1 = sh({}, addr, "select 1;\n");
            const std::vector<std::pair<std::string, std::string>> served = pg.Served();
            const Process::Captured r2 = sh({"select 1"}, std::format("{}#{}", addr.substr(0, addr.rfind('#')), std::string(32, '0')));
            const Process::Captured r3 = sh({"select 1"}, std::nullopt);
            const Process::Captured r4 = sh({}, addr, "commit;\n");
            br.Close();
            const Process::Captured r5 = sh({"select 1"}, addr);
            auto has = [&](const std::pair<std::string, std::string>& entry) { return std::ranges::find(served, entry) != served.end(); };
            Check("broker: the helper's SQL runs through the driver - connected, the rows, rolled back, exit 0 - with only the broker's address and a token in the pass's environment (finding 6)",
                  r1.returnCode == 0 && r1.out.contains(PgWire::Answered) && r1.out.contains("\n x\n") && r1.out.contains("-- rolled back") && !addr.contains("pw") && has({"password", "pw"})
                      && has({"Q", "ROLLBACK"}),
                  r1.out + r1.err);
            Check("broker: a foreign token, no broker variable and a broker that is gone are NO DATABASE (exit 2); a refused statement is refused through it as before (exit 1, nothing connects)",
                  r2.returnCode == 2 && r2.out.contains("no database by this token") && r3.returnCode == 2 && r3.out.contains("this pass has no database") && r4.returnCode == 1
                      && r4.out.contains("refused, nothing ran") && r5.returnCode == 2 && r5.out.contains("does not answer"),
                  Show({r2.out, r3.out, r4.out, r5.out}));
            PgWire::Broker bp({}, [&](const std::string& u) { return std::pair<bool, std::string>(u == url, "db on 127.0.0.1 as u"); });
            bp.Start();
            const std::string probeAddr = bp.Address(url);
            const Process::Captured r6 = sh({"--probe"}, probeAddr);
            const std::size_t colon = probeAddr.rfind(':');
            const auto port = static_cast<std::uint16_t>(std::stoi(probeAddr.substr(colon + 1, probeAddr.rfind('#') - colon - 1)));
            // No newline: the call was cut off.
            const std::string cut = BrokerCall(port, R"({"token": "x", "sql": "select 1"})");
            bp.Close();
            Check("broker: --probe goes through it too; a call that does not arrive whole runs nothing (exit 64)",
                  r6.returnCode == 0 && r6.out.starts_with(PgWire::Answered) && cut.contains("did not arrive whole") && Dyn::RStrip(cut).ends_with(std::format("{}{}", PgWire::ExitMark, PgWire::ExitUsage)),
                  Dyn::ReprValue(Json::array({r6.out, cut})));
        }

        // finding 4: the calibration is a record, never a dependency.
        {
            const fs::path td = base / "calibration";
            const fs::path blocker = td / "file";
            Write(blocker, "x");
            // Its directory is a file.
            const bool wrote = Memory::Calibration(blocker / "memory.json", td, {{"none", 210}}).Record("none", 500.0);
            const fs::path mp = td / "memory.json";
            Memory::Calibration cal(mp, td, {{"none", 210}, {"browser", 840}});
            const std::string me = fs::canonical(td).string();
            Json listed = Json::object();
            listed[me] = Json::array({5, 6});
            Json mixed = Json::object();
            mixed[me] = {{"browser", "x"}, {"none", Json::array({true, -3, "a", 300})}};
            Json shapes = Json::array();
            for (const std::string& content : {std::string("[1, 2]"), DumpSpaced(listed), DumpSpaced(mixed), std::string("{not json"), std::string("\xff\xfe\x00", 3)}) {
                Write(mp, content);
                shapes.push_back(Json::array({cal.CostOf("browser").megabytes, cal.CostOf("none").megabytes}));
            }
            const bool repaired = cal.Record("browser", 700.0);
            Check("calibration: a peak that cannot be written is dropped, never raised - a finished pass is never thrown away for bookkeeping (finding 4)", !wrote);
            Check("calibration: a damaged file, or a part of the wrong shape, reads as absent - the measured defaults, the sound peaks kept - and the next record repairs it",
                  shapes == Json::parse("[[840, 210], [840, 210], [840, 300], [840, 210], [840, 210]]") && repaired && cal.CostOf("browser").megabytes == 700.0, Dyn::ReprValue(shapes));
        }

        // finding 11: the sampler - one process scan a sweep, a pause that
        // grows with the sweep.
        {
            const std::map<std::int64_t, std::vector<std::int64_t>> kids = {{1, {2, 3}}, {2, {4}}, {9, {10}}};
            std::vector<std::int64_t> tree = Platform::Tree(1, kids);
            std::ranges::sort(tree);
            const auto ppids = Platform::PpidMap();
            const auto mine = ppids.find(getppid());
            Check("sampler: a tree from one ppid map (no per-thread children files); the pause is the interval, or three times a sweep that took longer (finding 11)",
                  tree == std::vector<std::int64_t>{1, 2, 3, 4} && Platform::Tree(9, kids) == std::vector<std::int64_t>{9, 10} && Memory::Sampler::Pause(3.0, 0.2) == 3.0
                      && Memory::Sampler::Pause(3.0, 2.5) == 7.5 && mine != ppids.end() && std::ranges::find(mine->second, getpid()) != mine->second.end());
            std::atomic<std::int32_t> scans = 0;
            Memory::Sampler sm(3600, [](std::int64_t, const std::map<std::int64_t, std::vector<std::int64_t>>&) { return 1.0; }, [&] {
                ++scans;
                return std::map<std::int64_t, std::vector<std::int64_t>>();
            });
            for (std::int64_t i = 0; i < 5; ++i) sm.Watch(std::to_string(i), i);
            const std::int32_t launched = scans;
            sm.Sample();
            Check("sampler: one scan per sweep whatever the number of passes, and none on a launch", launched == 0 && scans == 1 && sm.Done("3") == 1.0);
        }

        // findings 1, 5, 9, 12: the preflight.
        {
            const fs::path td = base / "preflight";
            Run::Run g;
            std::vector<std::string> logs;
            g.repo = td;
            g.log = [&](const std::string& line) { logs.push_back(line); };
            g.cfg = {{"functional", {{"start", "make run"}, {"readyPath", "/"}, {"exclude", Json::array()}}},
                     {"supabase", {{"dbUrl", "postgresql://postgres:postgres@127.0.0.1:54322/postgres"}, {"start", nullptr}}},
                     {"slots", {{"dir", ".local-stack/slots"}, {"start", "up --slot {n}"}, {"stop", nullptr}, {"memMb", 1500}}}};
            g.a.dryRun = false;
            g.a.harness = "claude";
            g.a.base = "";
            g.slotProblems.clear();
            g.loginFixture = std::nullopt;
            g.trees = {td.string()};
            Write(td / "l" / "u.json", R"({"email": "a@x.test", "password": "p"})");
            auto slot = [](std::int32_t n, bool db = true, std::string_view user = "postgres", bool isSlot = true) {
                return Json {{"id", std::format("slot {}", n)}, {"appUrl", std::format("http://localhost:2{}180", n)}, {"apiUrls", Json::array()}, {"login", "l/u.json"}, {"slot", isSlot},
                             {"dbUrl", db ? Json(std::format("postgresql://{}:postgres@127.0.0.1:2{}122/postgres", user, n)) : Json(nullptr)}};
            };
            auto pre = [&](const std::vector<std::string>& lenses, const Json& envs, const std::vector<Json>& stacks, bool slotMode = true) -> std::pair<std::string, std::string> {
                g.environments = envs;
                g.stacks = std::make_unique<Pool::Stacks>(stacks);
                g.slotMode = slotMode;
                try {
                    g.StackPreflight(lenses);
                    return {"ok", logs.empty() ? std::string() : logs.back()};
                } catch (const Run::Preflight& e) {
                    return {"3", std::format("preflight: {}\n", e.what())};
                }
            };
            std::vector<std::pair<std::string, std::string>> queried;
            std::int32_t migCalls = 0;
            const Run::QueryResult::second_type users = {{"x"}};
            g.appReady = [](std::string_view u, std::string_view) { return std::pair<bool, std::string>(true, std::format("HTTP 200 from {}", u)); };
            g.probe = [](std::string_view) { return std::pair<bool, std::string>(true, "postgres on 127.0.0.1 - PostgreSQL 17.6"); };
            g.query = [&](std::string_view u, std::string_view sql) {
                queried.emplace_back(u, sql);
                return Run::QueryResult({"?"}, users);
            };
            g.bypassesRls = [](std::string_view u) { return std::optional<bool>(!u.contains("admin")); };
            g.migrationsAtHead = [&] {
                ++migCalls;
                return Run::MigrationsHead();
            };
            const Json bs = {{Fn, "browser+supabase"}, {"repo-bug-prover.md", "supabase"}};
            const Json browser = {{Fn, "browser"}};
            const auto nodbApp = pre({Fn}, bs, {slot(1), slot(2, false)});
            const auto nodbFirst = pre({"repo-bug-prover.md"}, bs, {slot(1, false), slot(2)});
            const auto nodbBrowser = pre({Fn}, browser, {slot(1), slot(2, false)});
            queried.clear();
            const auto implicit = pre({Fn}, browser, {slot(1, true, "postgres", false)}, false);
            const auto implicitQ = queried;
            const auto implicitDb = pre({Fn}, bs, {slot(1, true, "postgres", false)}, false);
            const bool implicitDbQ = std::ranges::any_of(queried, [](const auto& q) { return q.second.contains("auth.users"); });
            const auto mixed = pre({Fn}, bs, {slot(1), slot(2, true, "admin")});
            const auto mixedRole = g.dbRole;
            const auto mixedBypass = g.dbBypass;
            g.migCache.clear();
            migCalls = 0;
            const auto three = pre({Fn}, bs, {slot(1), slot(2), slot(3)});
            const auto threeRole = g.dbRole;
            const auto threeChecked = g.dbChecked;
            const std::int32_t threeMigs = migCalls;
            const auto one = pre({"repo-bug-prover.md"}, bs, {slot(1), slot(2)});
            const auto oneChecked = g.dbChecked;
            Check("preflight: a stack without a database is refused for a lens that would run SQL there - any slot for browser+supabase, the first for supabase alone - "
                  "and is fine for a lens without the database (finding 1)",
                  nodbApp.first == "3" && nodbApp.second.contains("no database for slot 2 (no dbUrl") && nodbFirst.first == "3" && nodbFirst.second.contains("no database for slot 1 (no dbUrl")
                      && nodbBrowser.first == "ok",
                  Show({nodbApp.second, nodbFirst.second, nodbBrowser.second}));
            Check("preflight: the config's one stack, with a browser lens alone, checks no login in a database nobody named (the default URL may be another project's); "
                  "with a lens that uses its database it does (finding 5)",
                  implicit.first == "ok" && std::ranges::none_of(implicitQ, [](const auto& q) { return q.second.contains("auth.users"); }) && implicitDb.first == "ok" && implicitDbQ,
                  Show({implicit.second, implicitDb.second}));
            Check("preflight: the role the block names is the stacks' own, none when they differ; 'each of the 3 stacks' only when 3 were probed, 'it answers' for the one "
                  "a lens with the database alone uses (finding 9)",
                  mixed.first == "ok" && !mixedRole && !mixedBypass && three.first == "ok" && threeRole == "postgres" && threeChecked
                      && threeChecked->starts_with("the database of each of the 3 stacks answers") && one.first == "ok" && oneChecked && oneChecked->starts_with("it answers ("),
                  Show({mixedRole.value_or("None"), threeChecked.value_or("None"), oneChecked.value_or("None")}));
            Check("preflight: the HEAD side of the migrations check is worked out once for all the stacks (finding 12)", threeMigs == 1, std::to_string(threeMigs));
        }

        // finding 10: a gate whose every lens resumes needs no stack.
        {
            Run::Run g;
            std::vector<std::vector<std::string>> calls;
            std::vector<std::string> logs;
            g.log = [&](const std::string& line) { logs.push_back(line); };
            g.resumable = [](const std::string& name, const std::string&, std::string_view) { return std::optional<Json>(Json {{"lens", name}, {"verdict", "SEND-READY"}}); };
            g.instrumentPreflight = [&](std::span<const std::string> lfs) { calls.emplace_back(lfs.begin(), lfs.end()); };
            const std::vector<Json> res = g.RunGate("Gate 2 — Functional", {Fn});
            const auto resumedCalls = calls;
            g.resumable = [](const std::string&, const std::string&, std::string_view) { return std::optional<Json>(); };
            g.instrumentPreflight = [&](std::span<const std::string> lfs) {
                calls.emplace_back(lfs.begin(), lfs.end());
                throw std::runtime_error("stop here");
            };
            try {
                g.RunGate("Gate 2 — Functional", {Fn});
            } catch (const std::runtime_error&) {
            }
            Check("run_gate: a gate whose every lens resumes from the checkpoint runs no preflight - nothing will touch the stack; a lens that runs gets it (finding 10)",
                  resumedCalls.empty() && res.size() == 1 && Dyn::Equal(res[0], {{"lens", "repo-functional"}, {"verdict", "SEND-READY"}}) && calls == std::vector<std::vector<std::string>>{{Fn}},
                  std::format("{} resumed call(s), {} call(s)", resumedCalls.size(), calls.size()));
        }

        // finding 13: the block states :PgWire's own limits. They are
        // constexpr and the block formats them, so the check pins the text it
        // formats and today's values.
        {
            const std::string b1 = Prompt::DbBlock(Json::object(), "c", std::string("postgres"));
            const std::string limits =
                std::format("statement_timeout {} s and lock_timeout {} s per statement; at most {} rows", PgWire::StatementTimeoutS, PgWire::LockTimeoutS, PgWire::MaxRows);
            const std::string differs = Prompt::DbBlock(Json::object(), "c", std::nullopt);
            Check("db block: the limits are pgwire's own constants, so the prompt and the helper cannot drift apart (finding 13); a role that differs between stacks is not named",
                  b1.contains("statement_timeout 20 s and lock_timeout 3 s per statement; at most 100 rows") && b1.contains(limits) && differs.contains("differs between the stacks")
                      && !differs.contains("`None`"));
        }

        // finding 2: the template loads the way a consumer meets it.
        {
            const fs::path td = base / "template";
            fs::create_directories(td / ".review-gates");
            fs::copy_file(fs::current_path() / "process" / "templates" / "config.json", td / ".review-gates" / "config.json");
            const Config::Loaded loaded = Config::Load(td);
            Check("template: process/templates/config.json, copied as the README says, loads with no warning and no problem - unknown keys included (finding 2; the three kinds as modes with every lens of their kind, 2026-11-01)",
                  loaded.problems.empty() && loaded.config.at("_warnings").empty(), Show(loaded.problems));
        }
        fs::remove_all(base);
        return Failures ? 1 : 0;
    }
}

int main(int argc, char** argv) {
    const std::vector<std::string_view> args(argv, argv + argc);
    // The helper a lens runs, for the broker checks.
    if (args.size() > 1 && args[1] == "sql") return PgWire::SqlMain(std::span(args).subspan(2));
    try {
        return Checks();
    } catch (const std::exception& error) {
        std::println(std::cerr, "review fixes: {}", error.what());
        return 1;
    }
}