/packages/happy-server/deploy/integration-tests/test-rpc-dead-daemon.mjs

import tweetnacl from "tweetnacl";
import { io } from "socket.io-client";
import { Buffer } from "buffer";
import { execSync, spawn } from "child_process";

const SERVER = process.env.SERVER_URL || "http://127.0.0.1:3000";
const base64 = (buf) => Buffer.from(buf).toString("base64");

async function getToken() {
    const keyPair = tweetnacl.sign.keyPair();
    const challenge = tweetnacl.randomBytes(32);
    const signature = tweetnacl.sign.detached(challenge, keyPair.secretKey);
    const res = await fetch(`${SERVER}/v1/auth`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
            publicKey: base64(keyPair.publicKey),
            challenge: base64(challenge),
            signature: base64(signature),
        }),
    });
    if (!res.ok) throw new Error(`Auth failed: ${res.status}`);
    return (await res.json()).token;
}

function getPods() {
    return execSync("kubectl get pods -l app=handy-server -o jsonpath='{.items[*].metadata.name}'")
        .toString().replace(/'/g, "").trim().split(/\s+/);
}

function findSocketPod(socketId, pods) {
    for (const pod of pods) {
        const logs = execSync(`kubectl logs ${pod} --tail=500 2>/dev/null`).toString();
        if (logs.includes(socketId)) return pod;
    }
    return null;
}

function waitForPodReady(label, timeoutSec = 60) {
    const deadline = Date.now() + timeoutSec * 1000;
    while (Date.now() < deadline) {
        try {
            const out = execSync(
                `kubectl get pods -l ${label} -o jsonpath='{.items[*].status.conditions[?(@.type=="Ready")].status}'`
            ).toString().replace(/'/g, "").trim();
            const statuses = out.split(/\s+/);
            if (statuses.length > 0 && statuses.every(s => s === "True")) return true;
        } catch {}
        execSync("sleep 2");
    }
    return false;
}

function connectSocket(token, opts = {}, serverUrl = SERVER) {
    const socket = io(serverUrl, {
        path: "/v1/updates",
        auth: { token, ...opts },
        transports: ["websocket"],
        reconnection: false,
    });
    return new Promise((resolve, reject) => {
        const timeout = setTimeout(() => reject(new Error("connect timeout")), 10000);
        socket.on("connect", () => { clearTimeout(timeout); resolve(socket); });
        socket.on("connect_error", (err) => { clearTimeout(timeout); reject(err); });
    });
}

// Port-forward to a specific pod on a given local port. Returns a cleanup function.
function portForwardPod(pod, localPort) {
    const child = spawn("kubectl", ["port-forward", pod, `${localPort}:3005`], {
        stdio: "pipe",
        detached: true,
    });
    child.unref();
    return { url: `http://127.0.0.1:${localPort}`, kill: () => { try { process.kill(-child.pid); } catch {} } };
}

// ---------------------------------------------------------------------------
// Test 1: Baseline — normal RPC works
// ---------------------------------------------------------------------------
async function testBaseline(token) {
    const sessionId = `baseline-${Date.now()}`;
    const METHOD = `${sessionId}:echo`;

    const daemon = await connectSocket(token, { clientType: "session-scoped", sessionId });
    daemon.emit("rpc-register", { method: METHOD });
    daemon.on("rpc-request", (data, callback) => callback(data.params));
    await new Promise(r => setTimeout(r, 500));

    const caller = await connectSocket(token, { clientType: "user-scoped" });
    const t0 = Date.now();
    const result = await caller.timeout(10000).emitWithAck("rpc-call", {
        method: METHOD,
        params: "hello",
    });
    const elapsed = Date.now() - t0;

    daemon.disconnect();
    caller.disconnect();

    if (!result.ok || result.result !== "hello") {
        console.log(`  FAIL: unexpected result: ${JSON.stringify(result)}`);
        return false;
    }
    console.log(`  RPC succeeded in ${elapsed}ms`);
    return true;
}

// ---------------------------------------------------------------------------
// Test 2: Clean client disconnect (daemon.disconnect()) — NOT a pod kill
//         This is the easy case. Socket.IO processes it immediately.
// ---------------------------------------------------------------------------
async function testCleanDisconnect(token) {
    const sessionId = `clean-${Date.now()}`;
    const METHOD = `${sessionId}:slowecho`;

    const daemon = await connectSocket(token, { clientType: "session-scoped", sessionId });
    daemon.emit("rpc-register", { method: METHOD });
    daemon.on("rpc-request", () => {}); // never respond
    await new Promise(r => setTimeout(r, 500));

    const caller = await connectSocket(token, { clientType: "user-scoped" });
    const t0 = Date.now();
    const rpcPromise = caller.timeout(35000).emitWithAck("rpc-call", {
        method: METHOD,
        params: "should-never-arrive",
    });

    await new Promise(r => setTimeout(r, 1000));
    daemon.disconnect();
    console.log(`  Daemon cleanly disconnected after ${Date.now() - t0}ms`);

    const result = await rpcPromise;
    const elapsed = Date.now() - t0;
    caller.disconnect();

    console.log(`  RPC failed after ${elapsed}ms: "${result.error}"`);
    console.log(`  Detection mechanism: ${result.error === "RPC target disconnected" ? "PRESENCE POLL" : "EMIT_WITH_ACK / OTHER"}`);

    if (!result.ok && elapsed < 15000) {
        console.log(`  Fast failure (${(elapsed / 1000).toFixed(1)}s) — but this is the EASY case (clean disconnect)`);
        return true;
    }
    console.log(`  FAIL: took ${elapsed}ms or unexpected result: ${JSON.stringify(result)}`);
    return false;
}

// ---------------------------------------------------------------------------
// Test 3: HARD KILL — delete the pod the daemon is on
//         Connect daemon and caller to DIFFERENT pods via separate
//         port-forwards. Then kill the daemon's pod. The caller's pod
//         must detect the dead daemon cross-replica.
// ---------------------------------------------------------------------------
async function testPodKill(token) {
    const pods = getPods();
    if (pods.length < 2) {
        console.log(`  SKIP: need >= 2 pods for pod-kill test (have ${pods.length})`);
        return true;
    }

    const daemonPod = pods[0];
    const callerPod = pods[1];
    console.log(`  Daemon pod: ${daemonPod}`);
    console.log(`  Caller pod: ${callerPod}`);

    // Port-forward each pod on a separate local port
    const daemonPf = portForwardPod(daemonPod, 4001);
    const callerPf = portForwardPod(callerPod, 4002);
    await new Promise(r => setTimeout(r, 2000)); // let port-forwards establish

    try {
        const sessionId = `podkill-${Date.now()}`;
        const METHOD = `${sessionId}:slowecho`;

        // Connect daemon directly to its pod
        const daemon = await connectSocket(token, { clientType: "session-scoped", sessionId }, daemonPf.url);
        daemon.emit("rpc-register", { method: METHOD });
        daemon.on("rpc-request", () => {}); // never respond
        console.log(`  Daemon socket: ${daemon.id} → ${daemonPod}`);
        await new Promise(r => setTimeout(r, 1000)); // let registration propagate

        // Connect caller directly to the OTHER pod
        const caller = await connectSocket(token, { clientType: "user-scoped" }, callerPf.url);
        console.log(`  Caller socket: ${caller.id} → ${callerPod}`);
        console.log(`  Cross-replica: YES (guaranteed)`);

        // Start the RPC call on the caller's pod — server must find daemon cross-replica
        const t0 = Date.now();
        const rpcPromise = caller.timeout(45000).emitWithAck("rpc-call", {
            method: METHOD,
            params: "should-never-arrive",
        }).catch(err => ({ ok: false, error: err.message }));

        // Kill the daemon's pod (hard kill, no graceful shutdown)
        await new Promise(r => setTimeout(r, 2000)); // let RPC get in-flight
        console.log(`  KILLING pod ${daemonPod} (force, grace-period=0)...`);
        try {
            execSync(`kubectl delete pod ${daemonPod} --force --grace-period=0 2>&1`);
        } catch (e) {
            console.log(`  kubectl delete output: ${e.stdout?.toString() || e.message}`);
        }
        const killTime = Date.now() - t0;
        console.log(`  Pod killed at ${killTime}ms`);

        // Wait for the RPC to resolve
        const result = await rpcPromise;
        const elapsed = Date.now() - t0;
        const detectionTime = elapsed - killTime;
        try { caller.disconnect(); } catch {}

        console.log(`  RPC resolved after ${elapsed}ms (${detectionTime}ms after pod kill)`);
        console.log(`  Result: ok=${result.ok}, error="${result.error}"`);

        if (result.error === "RPC target disconnected") {
            console.log(`  Detection: PRESENCE POLL caught it in ${(detectionTime / 1000).toFixed(1)}s`);
        } else if (result.error === "RPC method not available") {
            console.log(`  Detection: GRACE WINDOW exhausted — room went empty after pod death`);
        } else if (result.error?.includes("timeout") || elapsed >= 29000) {
            console.log(`  Detection: FULL TIMEOUT — presence poll did NOT help (${(elapsed / 1000).toFixed(1)}s)`);
        } else {
            console.log(`  Detection: UNKNOWN mechanism — error: "${result.error}"`);
        }

        // Wait for replacement pod
        console.log(`  Waiting for replacement pod...`);
        waitForPodReady("app=handy-server", 90);
        await new Promise(r => setTimeout(r, 3000));

        if (!result.ok && elapsed < 20000) {
            console.log(`  Pod-kill detected in ${(detectionTime / 1000).toFixed(1)}s (< 20s threshold)`);
            return true;
        }
        if (!result.ok && elapsed >= 20000) {
            console.log(`  SLOW: ${(elapsed / 1000).toFixed(1)}s — poll may not be helping much`);
            return false;
        }
        return false;
    } finally {
        daemonPf.kill();
        callerPf.kill();
    }
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------
async function main() {
    console.log("=== RPC DEAD-DAEMON DETECTION TEST ===\n");

    const health = await fetch(`${SERVER}/health`).then(r => r.json());
    const pods = getPods();
    console.log(`Health: ${health.status}`);
    console.log(`Pods: ${pods.join(", ")} (${pods.length} replicas)\n`);

    const token = await getToken();
    let allPassed = true;

    // Test 1: baseline
    console.log("Test 1: Baseline RPC ...");
    const t1 = await testBaseline(token);
    console.log(t1 ? "  PASS\n" : "  FAIL\n");
    allPassed = allPassed && t1;

    // Test 2: clean disconnect (easy case)
    console.log("Test 2: Clean client disconnect ...");
    const t2 = await testCleanDisconnect(token);
    console.log(t2 ? "  PASS\n" : "  FAIL\n");
    allPassed = allPassed && t2;

    // Test 3: THE REAL TEST — kill the pod
    console.log("Test 3: Pod kill (hard) ...");
    const t3 = await testPodKill(token);
    console.log(t3 ? "  PASS\n" : "  FAIL\n");
    allPassed = allPassed && t3;

    console.log("========================================");
    console.log("           VERDICT");
    console.log("========================================");
    console.log(allPassed ? "ALL PASSED" : "FAILURES DETECTED");
    process.exit(allPassed ? 0 : 1);
}

main().catch((err) => {
    console.error("Test failed:", err.message);
    process.exit(1);
});