electric-cherry/src/lib/server/pipeline/beta-runner.ts

import { db } from '../db/index.js';
import { emit } from './bus.js';
import type { BetaStage } from './bus.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
import { komodoRestart, komodoInspectHealthy } from '../integrations/komodo.js';
import { scanForMatchEnd } from '../fleet/match-log.js';

interface BetaServerActiveRow {
    beta_id: number;
    server: string;
    stage: BetaStage;
    healthy_at: string | null;
    wait_for_health: number;
    wait_for_match: number;
}

function nowIso(): string {
    return new Date().toISOString();
}

function integrationsOn(): boolean {
    return komodoConfigured() && rconConfigured();
}

const setStage = () =>
    db().prepare(
        `UPDATE beta_servers
         SET stage = ?, error = ?, updated_at = ?,
             restarted_at = CASE WHEN ? = 'awaiting-health' OR ? = 'ready' THEN COALESCE(restarted_at, ?) ELSE restarted_at END,
             healthy_at   = CASE WHEN ? = 'awaiting-match' OR (? = 'ready' AND healthy_at IS NULL) THEN COALESCE(healthy_at, ?) ELSE healthy_at END,
             match_seen_at = CASE WHEN ? = 'ready' AND match_seen_at IS NULL THEN ? ELSE match_seen_at END
         WHERE beta_id = ? AND server = ?`
    );

function transition(
    betaId: number,
    server: string,
    stage: BetaStage,
    error: string | null = null
): void {
    const t = nowIso();
    setStage().run(
        stage,
        error,
        t,
        stage,
        stage,
        t,
        stage,
        stage,
        t,
        stage,
        t,
        betaId,
        server
    );
    emit({ type: 'beta-server-stage', betaId, server, stage, error });
}

// Bump beta.state to ready-promote when every one of its beta_servers is ready.
// A single failed row is enough to keep the beta in 'running' (operator retries).
function reconcileBetaState(betaId: number): void {
    const rows = db()
        .prepare(`SELECT stage FROM beta_servers WHERE beta_id = ?`)
        .all(betaId) as Array<{ stage: BetaStage }>;
    if (rows.length === 0) return;
    if (rows.every((r) => r.stage === 'ready')) {
        const upd = db()
            .prepare(`UPDATE betas SET state = 'ready-promote' WHERE id = ? AND state = 'running'`)
            .run(betaId);
        if (upd.changes > 0) {
            // The version that just cleared its beta gate is now considered
            // tested. This survives promote (tested → promoted is derived from
            // scripts.current_version_id, not stored) and reject (rollbackBeta
            // demotes back to test-failed).
            const beta = db()
                .prepare(`SELECT script_version_id FROM betas WHERE id = ?`)
                .get(betaId) as { script_version_id: number } | undefined;
            if (beta) {
                db()
                    .prepare(`UPDATE script_versions SET state = 'tested' WHERE id = ?`)
                    .run(beta.script_version_id);
            }
        }
        emit({ type: 'beta-state', betaId, state: 'ready-promote' });
    }
}

export async function betaTick(): Promise<{ processed: number }> {
    if (!integrationsOn()) return { processed: 0 };

    const rows = db()
        .prepare(
            `SELECT bs.beta_id AS beta_id, bs.server AS server, bs.stage AS stage,
                    bs.healthy_at AS healthy_at,
                    b.wait_for_health AS wait_for_health, b.wait_for_match AS wait_for_match
             FROM beta_servers bs
             JOIN betas b ON b.id = bs.beta_id
             WHERE b.state = 'running'
               AND bs.stage IN ('awaiting-empty','restarting','awaiting-health','awaiting-match')`
        )
        .all() as BetaServerActiveRow[];

    if (rows.length === 0) return { processed: 0 };

    const touchedBetas = new Set<number>();
    for (const row of rows) {
        try {
            await advance(row);
        } catch (err) {
            transition(row.beta_id, row.server, 'failed', `runner: ${(err as Error).message}`);
        }
        touchedBetas.add(row.beta_id);
    }
    for (const id of touchedBetas) reconcileBetaState(id);

    return { processed: rows.length };
}

async function advance(row: BetaServerActiveRow): Promise<void> {
    switch (row.stage) {
        case 'awaiting-empty': {
            const players = await rconGetPlayerCount(row.server);
            if (players === null) return; // hold — RCON unreachable, try next tick
            if (players > 0) return;      // hold — still populated
            transition(row.beta_id, row.server, 'restarting');
            // Fall through to restart on the same tick for lower latency.
            await advance({ ...row, stage: 'restarting' });
            return;
        }
        case 'restarting': {
            const r = await komodoRestart(row.server);
            if (!r.ok) {
                transition(row.beta_id, row.server, 'failed', `komodo: ${r.message}`);
                return;
            }
            const next = row.wait_for_health ? 'awaiting-health' : row.wait_for_match ? 'awaiting-match' : 'ready';
            transition(row.beta_id, row.server, next);
            return;
        }
        case 'awaiting-health': {
            const h = await komodoInspectHealthy(row.server);
            if (!h.ok) {
                // Record the failure reason (including any snippet) so the
                // beta detail UI can surface why we're stuck.
                const snippet = h.snippet ? ` snippet=${h.snippet.slice(0, 400)}` : '';
                setStage().run(
                    'awaiting-health',
                    `unhealthy: ${h.reason}${snippet}`,
                    nowIso(),
                    'awaiting-health',
                    'awaiting-health',
                    nowIso(),
                    'awaiting-health',
                    'awaiting-health',
                    nowIso(),
                    'awaiting-health',
                    nowIso(),
                    row.beta_id,
                    row.server
                );
                return;
            }
            const rcon = await rconStatusResult(row.server);
            if (!rcon.ok) return; // container reports up but game not answering yet
            const next = row.wait_for_match ? 'awaiting-match' : 'ready';
            transition(row.beta_id, row.server, next);
            return;
        }
        case 'awaiting-match': {
            const since = row.healthy_at ?? nowIso();
            const result = await scanForMatchEnd(row.server, since);
            if (!result.ended) return;
            transition(row.beta_id, row.server, 'ready');
            return;
        }
    }
}