sc-dev-deploy/test/pullGate.js

// Phase 1 PULL peering gate (the reverse of mixedTopologyGate's push/promote).
// mixedTopologyGate proves dev -> t1 PROMOTE (push to /api/ingest). This gate
// proves the PULL path (/admin/dev-deploy/pull -> the peer's signed
// /api/journal) across the same multi-tenant boundary, which nothing else
// covers: e2e.js exercises pull only single-instance (SQLite MAIN <-> MAIN),
// never across the Postgres tenant boundary or tenant<->tenant.
//
// Topology: STANDALONE dev (SQLite MAIN :3000) + tenants t1, t2 on the
// multi-tenant Postgres "prod" (:3002). We verify:
//   1. REVERSE PULL (mixed topology): t1 PULLS dev's journal -> dev's ops land
//      in t1 (applied >= 1), proving the host-bound HMAC works on the GET pull
//      path (signedFetch targetHost = dev's host), not just the push path.
//   2. ISOLATION ON PULL: t2's journal is UNCHANGED by t1 pulling from dev --
//      a pull into one tenant never leaks into a sibling tenant.
//   3. IDEMPOTENT RE-PULL: pulling again with the advanced inbound anchor
//      returns "nothing to pull" (no duplicate apply).
//   4. TENANT<->TENANT PULL: t2 PULLS t1's journal (same Postgres server, two
//      tenants) -> t1's OWN ops land in t2 while dev is untouched. apiJournal
//      serves only first-party ops (source_env_id == self), so t2 receives
//      t1's ops, NOT the dev ops t1 itself pulled in step 1 (no transitive
//      relay / no loops).
//
// Run: node test/pullGate.js   (MAIN :3000 + PG :3002 up, dev-deploy installed
// on MAIN and per-tenant on t1/t2). Self-skips if either is down.

const http = require("http");
const net  = require("net");

const DEV       = { port: 3000, host: "localhost:3000", base: "http://localhost:3000" };
const PROD_PORT = 3002;
const ADMIN_PW  = "AdminP@ss1";
const RUN       = Date.now();
const DEV_PROBE = "pull_dev_" + RUN;
const T1_PROBE  = "pull_t1_" + RUN;

const thost = (t) => t + ".localhost.localdomain:" + PROD_PORT;
const jars  = { dev: {}, t1: {}, t2: {} };
let pass = 0;
let fail = 0;


const ok = (cond, msg) => {
    if (cond) {
        pass++;
        console.log("  PASS  " + msg);
    } else {
        fail++;
        console.log("  FAIL  " + msg);
    }
};


const portOpen = (port) => {
    return new Promise((resolve) => {
        const s = net.connect(port, "127.0.0.1");
        const done = (up) => { s.destroy(); resolve(up); };
        s.setTimeout(1000);
        s.on("connect", () => done(true));
        s.on("timeout", () => done(false));
        s.on("error", () => done(false));
    });
};


const HOSTOF = (inst) => (inst === "dev" ? DEV.host : thost(inst));
const PORTOF = (inst) => (inst === "dev" ? DEV.port : PROD_PORT);
const ADMIN  = (inst) => (inst === "dev" ? "admin@local" : "admin@" + inst + ".local");
const BASEOF = (inst) => (inst === "dev" ? DEV.base : "http://" + thost(inst));


const storeCookies = (inst, headers) => {
    const sc = headers["set-cookie"];
    if (!sc) {
        return;
    }
    for (const line of sc) {
        const pair = line.split(";")[0];
        const eq = pair.indexOf("=");
        if (eq > 0) {
            jars[inst][pair.slice(0, eq).trim()] = pair.slice(eq + 1).trim();
        }
    }
};


const request = (inst, method, path, opts) => {
    const options = opts || {};
    return new Promise((resolve, reject) => {
        const headers = Object.assign({ Host: options.host || HOSTOF(inst) }, options.headers || {});
        const jar = jars[inst];
        if (jar && Object.keys(jar).length > 0) {
            headers["Cookie"] = Object.keys(jar).map((k) => k + "=" + jar[k]).join("; ");
        }
        let data = null;
        if (options.body) {
            data = new URLSearchParams(options.body).toString();
            headers["Content-Type"] = "application/x-www-form-urlencoded";
            headers["Content-Length"] = Buffer.byteLength(data);
        }
        const r = http.request({ host: "127.0.0.1", port: PORTOF(inst), method: method, path: path, headers: headers }, (resp) => {
            storeCookies(inst, resp.headers);
            let body = "";
            resp.on("data", (c) => { body += c; });
            resp.on("end", () => resolve({ status: resp.statusCode, headers: resp.headers, body: body }));
        });
        r.on("error", reject);
        if (data !== null) {
            r.write(data);
        }
        r.end();
    });
};


const csrfOf    = (h) => { const m = h.match(/name="_csrf" value="([^"]+)"/); return m ? m[1] : ""; };
const envIdOf   = (h) => { const m = h.match(/env_id<\/strong> is <code>([0-9a-f-]{36})<\/code>/); return m ? m[1] : ""; };
const opCountOf = (h) => { const m = h.match(/<th>Ops recorded<\/th><td>(\d+)<\/td>/); return m ? parseInt(m[1], 10) : null; };
const secretOf  = (h) => { const m = h.match(/<p class="secret">([0-9a-f]+)<\/p>/); return m ? m[1] : ""; };


// pull redirects 302 with ?msg=... (success) or ?err=... (failure); the message
// reads "pulled N ops (A applied, E errors, C conflicts)" or "nothing to pull".
const redirectMsg = (res) => {
    const loc = res.headers.location || "";
    const m = loc.match(/[?&](?:msg|err)=([^&]+)/);
    return m ? decodeURIComponent(m[1]) : "";
};


const pulledCounts = (msg) => {
    const m = msg.match(/pulled (\d+) ops \((\d+) applied, (\d+) errors, (\d+) conflicts\)/);
    return m ? { n: +m[1], applied: +m[2], errors: +m[3], conflicts: +m[4] } : null;
};


const peerIdFor = (html, envId) => {
    for (const row of html.split("<tr>")) {
        if (row.indexOf("<code>" + envId + "</code>") >= 0) {
            const m = row.match(/name="peer_id" value="(\d+)"/);
            if (m) {
                return m[1];
            }
        }
    }
    return null;
};


// Tracked-tables admin page renders one row per tracked table:
//   <td>NAME</td><td><code>UUID8</code></td><td>LOCAL_ID</td>...
// Map a probe table's NAME to its LOCAL_ID so cleanup can drop it (the source
// copy AND any copy materialized by a pull) without tracking create locations.
const localIdByName = async (inst, name) => {
    const html = (await request(inst, "GET", "/admin/dev-deploy/tables")).body;
    const re = new RegExp("<td>" + name + "</td>\\s*<td><code>[0-9a-f]{8}</code></td>\\s*<td>(\\d+)</td>");
    const m = html.match(re);
    return m ? m[1] : null;
};


const authed = async (inst) => /true/.test((await request(inst, "GET", "/auth/authenticated")).body);


const bootstrap = async (inst) => {
    const lp = await request(inst, "GET", "/auth/login");
    await request(inst, "POST", "/auth/login", { body: { email: ADMIN(inst), password: ADMIN_PW, _csrf: csrfOf(lp.body) } });
    if (await authed(inst)) {
        return true;
    }
    const cp = await request(inst, "GET", "/auth/create_first_user");
    const cc = csrfOf(cp.body);
    if (cc) {
        await request(inst, "POST", "/auth/create_first_user", { body: { email: ADMIN(inst), password: ADMIN_PW, default_language: "en", _csrf: cc } });
    }
    return await authed(inst);
};


const ddCsrf = async (inst) => csrfOf((await request(inst, "GET", "/admin/dev-deploy/peers")).body);


const clearPeer = async (inst, envId) => {
    const pid = peerIdFor((await request(inst, "GET", "/admin/dev-deploy/peers")).body, envId);
    if (pid) {
        await request(inst, "POST", "/admin/dev-deploy/peers/delete", { body: { peer_id: pid, _csrf: await ddCsrf(inst) } });
    }
};


// Returns the issued shared secret when the peer is added WITHOUT one (the
// adding side generates it); returns "" when an existing secret is supplied.
const addPeer = async (inst, envId, label, baseUrl, secret) => {
    const body = { env_id: envId, label: label, base_url: baseUrl, _csrf: await ddCsrf(inst) };
    if (secret) {
        body.existing_secret = secret;
    }
    const res = await request(inst, "POST", "/admin/dev-deploy/peers/add", { body: body });
    return secretOf(res.body);
};


// Create a table via Saltcorn's own POST /table (journaled by dev-deploy as a
// create_table op authored by THIS instance). The _csrf comes from dev-deploy's
// self-rendered page: a freshly created themeless tenant 500s on /table/new
// (sendWrap with no layout), but POST /table only mutates + redirects, so it
// succeeds. Returns true on the 302 -> /table/<id> success.
const createTable = async (inst, name) => {
    const csrf = await ddCsrf(inst);
    const res  = await request(inst, "POST", "/table", { body: { name: name, _csrf: csrf } });
    return res.status >= 300 && res.status < 400 && /\/table\/\d+/.test(res.headers.location || "");
};


const dropTableByName = async (inst, name) => {
    const id = await localIdByName(inst, name);
    if (id) {
        await request(inst, "POST", "/table/delete/" + id, { body: { _csrf: await ddCsrf(inst) } });
    }
};


const opCount = async (inst) => opCountOf((await request(inst, "GET", "/admin/dev-deploy/")).body);


const doPull = async (inst, peerId) => {
    return await request(inst, "POST", "/admin/dev-deploy/pull", { body: { peer_id: peerId, _csrf: await ddCsrf(inst) } });
};


const run = async () => {
    ok(await bootstrap("dev"), "dev (standalone MAIN :3000) admin session");
    ok(await bootstrap("t1"),  "t1 admin session");
    ok(await bootstrap("t2"),  "t2 admin session");

    const devEnv = envIdOf((await request("dev", "GET", "/admin/dev-deploy/peers")).body);
    const t1Env  = envIdOf((await request("t1",  "GET", "/admin/dev-deploy/peers")).body);
    const t2Env  = envIdOf((await request("t2",  "GET", "/admin/dev-deploy/peers")).body);
    ok(/^[0-9a-f-]{36}$/.test(devEnv), "dev env_id (" + devEnv + ")");
    ok(t1Env && t2Env && t1Env !== t2Env && t1Env !== devEnv, "dev/t1/t2 have distinct env_ids");

    // Fresh pairings (idempotent reruns): a new peer_id => a fresh inbound anchor
    // => the pull starts from the beginning of the source journal.
    await clearPeer("dev", t1Env);
    await clearPeer("t1", devEnv);
    await clearPeer("t1", t2Env);
    await clearPeer("t2", t1Env);

    // Pair dev <-> t1 so t1 can pull from dev: t1 needs dev as a peer (base_url +
    // secret to initiate); dev needs t1 as a peer (to AUTH t1's signed pull, which
    // peerAuth looks up by t1's env_id). dev issues the secret.
    const secretDevT1 = await addPeer("dev", t1Env, "prod-t1", BASEOF("t1"));
    ok(/^[0-9a-f]{64}$/.test(secretDevT1), "dev<->t1 paired (shared secret issued)");
    await addPeer("t1", devEnv, "dev", BASEOF("dev"), secretDevT1);

    // Give dev a brand-new op to serve: a uniquely named table guarantees an op
    // t1 has never seen (status "applied", not "already_applied"), so the assert
    // holds on every rerun regardless of accumulated history.
    ok(await createTable("dev", DEV_PROBE), "dev created a probe table to seed its journal");

    const t1Before = await opCount("t1");
    const t2Before = await opCount("t2");

    // (1) REVERSE PULL: t1 pulls dev's journal.
    const t1DevPeer = peerIdFor((await request("t1", "GET", "/admin/dev-deploy/peers")).body, devEnv);
    ok(!!t1DevPeer, "t1 has a peer_id for dev (" + t1DevPeer + ")");
    const pull1   = await doPull("t1", t1DevPeer);
    const pull1Msg = redirectMsg(pull1);
    const c1      = pulledCounts(pull1Msg);
    ok(pull1.status === 302, "t1 pull from dev returned 302 (" + pull1.status + ")");
    ok(c1 !== null, "t1 pull reported a pulled-ops summary (\"" + pull1Msg + "\")");
    ok(c1 && c1.applied >= 1, "t1 APPLIED >= 1 op pulled from dev (applied=" + (c1 ? c1.applied : "?") + ")");
    ok(c1 && c1.errors === 0 && c1.conflicts === 0, "t1 pull had no errors/conflicts");

    const t1After1 = await opCount("t1");
    const t2After1 = await opCount("t2");
    // (1) lands in t1.
    ok(t1After1 > t1Before, "t1 journal grew from the pull (" + t1Before + " -> " + t1After1 + ")");
    // (2) ISOLATION ON PULL: t2 untouched.
    ok(t2After1 === t2Before, "t2 journal UNCHANGED by t1 pulling from dev (" + t2Before + " -> " + t2After1 + ")");

    // (3) IDEMPOTENT RE-PULL: anchor advanced, nothing new to fetch.
    const pull2    = await doPull("t1", t1DevPeer);
    const pull2Msg = redirectMsg(pull2);
    ok(/nothing to pull/.test(pull2Msg), "re-pull from dev is idempotent (\"" + pull2Msg + "\")");
    const t1After2 = await opCount("t1");
    ok(t1After2 === t1After1, "t1 journal UNCHANGED by the idempotent re-pull (" + t1After1 + " -> " + t1After2 + ")");

    // (4) TENANT<->TENANT PULL: pair t1<->t2 (t1 issues), seed t1 with its own op,
    // then t2 pulls t1. t1's apiJournal serves only t1-authored ops, so t2 gets
    // t1's probe -- NOT the dev ops t1 pulled in step 1 (no transitive relay).
    const secretT1T2 = await addPeer("t1", t2Env, "prod-t2", BASEOF("t2"));
    ok(/^[0-9a-f]{64}$/.test(secretT1T2), "t1<->t2 paired (shared secret issued)");
    await addPeer("t2", t1Env, "prod-t1", BASEOF("t1"), secretT1T2);

    ok(await createTable("t1", T1_PROBE), "t1 created a probe table to seed its own journal");

    const devBefore4 = await opCount("dev");
    const t2Before4  = await opCount("t2");
    const t2T1Peer   = peerIdFor((await request("t2", "GET", "/admin/dev-deploy/peers")).body, t1Env);
    ok(!!t2T1Peer, "t2 has a peer_id for t1 (" + t2T1Peer + ")");
    const pull3    = await doPull("t2", t2T1Peer);
    const pull3Msg = redirectMsg(pull3);
    const c3       = pulledCounts(pull3Msg);
    ok(c3 && c3.applied >= 1, "t2 APPLIED >= 1 op pulled from t1 (tenant<->tenant; applied=" + (c3 ? c3.applied : "?") + ")");
    ok(c3 && c3.errors === 0 && c3.conflicts === 0, "t2 pull from t1 had no errors/conflicts");

    const t2After4  = await opCount("t2");
    const devAfter4 = await opCount("dev");
    ok(t2After4 > t2Before4, "t2 journal grew from pulling t1 (" + t2Before4 + " -> " + t2After4 + ")");
    ok(devAfter4 === devBefore4, "dev journal UNCHANGED by the t2<->t1 pull (" + devBefore4 + " -> " + devAfter4 + ")");

    // Cleanup (best-effort): drop each probe ONLY on its SOURCE instance (dev
    // authored DEV_PROBE, t1 authored T1_PROBE), never on the puller. A drop is a
    // first-party op too: dropping on the source makes the journal create->drop
    // LINEAR, so the next run's pull replays it in order and cleans up the pulled
    // copy with no conflict. Dropping the PULLED copy instead would author a
    // competing local op, and the next run (replaying the source's create) would
    // legitimately conflict with it -- a self-inflicted divergence, not a bug.
    await dropTableByName("dev", DEV_PROBE);
    await dropTableByName("t1",  T1_PROBE);
    await clearPeer("dev", t1Env);
    await clearPeer("t1", devEnv);
    await clearPeer("t1", t2Env);
    await clearPeer("t2", t1Env);

    console.log("\nRESULT: " + pass + " passed, " + fail + " failed");
    process.exit(fail ? 1 : 0);
};


const main = async () => {
    if (!(await portOpen(DEV.port))) {
        console.log("SKIP: standalone dev not reachable on 127.0.0.1:" + DEV.port);
        process.exit(0);
    }
    if (!(await portOpen(PROD_PORT))) {
        console.log("SKIP: Postgres multi-tenant prod not reachable on 127.0.0.1:" + PROD_PORT);
        process.exit(0);
    }
    await run();
};


main().catch((e) => {
    console.error("PULL GATE ERROR:", e);
    process.exit(2);
});