autofirmer-expanded/lib/master-link.ts

/**
 * Control link to the master dashboard.
 *
 * We dial out, not the other way round: this machine sits on a home network
 * behind NAT and the master is the only end with a public address. Once the
 * socket is up it carries traffic both ways — reports go up it instead of the
 * 30s HTTP POST, and trade / AutoBuyer commands come back down it.
 *
 * The link is optional. Without `master_dashboard_url` and `instance_name`
 * nothing starts, and if the master is unreachable the reporter falls back to
 * the HTTP endpoint it has always used.
 *
 * Neither end authenticates the other. Whatever is at `master_dashboard_url`
 * can place orders on every account this instance holds, so point it only at a
 * master you control, and prefer https so it cannot be redirected in transit.
 */

import { getSetting, setSetting, getRunnerHeartbeat, RUNNER_TIMEOUT_MS } from './db';
import { runTrade, startScheduler, stopScheduler, getSchedulerStatus } from './auto-trade';
import { FIRMS, automationKey } from './automations';
import { CME_ROOTS, CME_SPECS } from './instrument-specs';
import { futuresRootsHere } from './instruments';
import { queueRun } from './queue-run';
import { pushBackup } from './backup-sync';
import { getBuildVersion } from './version';
import { updaterStatus } from './updater-status';
import { runUpdateNow } from './updater-watchdog';
import { directWebSocket } from './outbound-proxy';
import type {
    Command,
    InstanceCapabilities,
    InstanceMessage,
    MasterMessage,
} from './master-protocol';

const WS_PATH = '/ws/instance';
const RECONNECT_MIN_MS = 2_000;
const RECONNECT_MAX_MS = 60_000;

interface LinkState {
    socket: WebSocket | null;
    /** The master_dashboard_url the socket was opened for. */
    url: string | null;
    /** Only true once the master has answered the hello. */
    ready: boolean;
    attempt: number;
    reconnectTimer: ReturnType<typeof setTimeout> | null;
    stopped: boolean;
}

// HMR-safe: a dev reload must not leave a second socket dialling.
const _g = globalThis as typeof globalThis & { __masterLink?: LinkState };

function state(): LinkState {
    if (!_g.__masterLink) {
        _g.__masterLink = { socket: null, url: null, ready: false, attempt: 0, reconnectTimer: null, stopped: false };
    }
    return _g.__masterLink;
}

// ── capabilities ─────────────────────────────────────────────────────────────

/** What this instance can be asked to do, as of right now. */
export function collectCapabilities(): InstanceCapabilities {
    // The roots this machine's futures firms list (the master's per-firm
    // instrument lists); none listed yet means offer everything we can price.
    const listed = futuresRootsHere();

    // Same test the AutoBuyer page uses: `at` is epoch ms from the last ping.
    const heartbeat = getRunnerHeartbeat();
    const runnerOnline = heartbeat !== null && Date.now() - heartbeat.at < RUNNER_TIMEOUT_MS;

    return {
        symbols: listed.length > 0 ? listed : [...CME_ROOTS],
        automations: FIRMS.flatMap((firm) =>
            firm.automations
                .filter((a) => a.steps.length > 0)
                .map((a) => ({
                    key: automationKey(firm.id, a.id),
                    label: a.label,
                    firm: firm.label,
                    inputs: (a.inputs ?? []).map((i) => ({
                        id: i.id,
                        label: i.label,
                        min: i.min,
                        max: i.max,
                        default: i.default,
                    })),
                }))
        ),
        runnerOnline,
        updateNow: true,
    };
}

// ── commands ─────────────────────────────────────────────────────────────────

/**
 * Runs one command and returns whatever the master should show the operator.
 * Throwing is fine — the caller turns it into a failed ack, which surfaces as
 * a red banner on the dashboard rather than silence.
 */
async function execute(command: Command): Promise<unknown> {
    switch (command.kind) {
        case 'trade': {
            if (command.symbol !== 'Auto' && !CME_SPECS[command.symbol]) {
                throw new Error(`Unknown symbol: ${command.symbol}`);
            }
            // Same order as POST /api/trade: (re)start the scheduler first,
            // then fire once now — the scheduler picks up accounts that were
            // busy. Starting it first means a batch that throws or runs long
            // can never leave it unstarted. It can't double-fire against this
            // batch: the tick waits for flat positions, skips while this batch
            // holds the trade mutex, and never picks an account in flight.
            startScheduler(command.action, command.symbol, command.stopAfterAll ?? false);
            const results = await runTrade(command.action, command.symbol);

            const filled = results
                .flatMap((r) => r.results)
                .filter((r) => r.status === 'filled').length;
            return { filled, scheduler: getSchedulerStatus() };
        }

        case 'stop-trading':
            stopScheduler();
            return { stopped: true, scheduler: getSchedulerStatus() };

        case 'autobuyer-run':
            return queueRun(command.automationId, command.inputs ?? {});

        case 'report-now':
            return { reported: true };

        case 'backup-now':
            return pushBackup(true);

        case 'update-now':
            return runUpdateNow();

        default: {
            // A newer master may know commands this build does not. Say so
            // rather than failing silently.
            const unknown = command as { kind?: string };
            throw new Error(`Unsupported command: ${unknown.kind ?? 'unnamed'}`);
        }
    }
}

// ── link ─────────────────────────────────────────────────────────────────────

/** `https://dash.example.com` → `wss://dash.example.com/ws/instance` */
function toWsUrl(masterUrl: string): string | null {
    try {
        const url = new URL(masterUrl);
        url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
        url.pathname = WS_PATH;
        url.search = '';
        url.hash = '';
        return url.toString();
    } catch {
        return null;
    }
}

function send(message: InstanceMessage): boolean {
    const s = state();
    if (!s.socket || !s.ready || s.socket.readyState !== WebSocket.OPEN) return false;
    try {
        s.socket.send(JSON.stringify(message));
        return true;
    } catch {
        return false;
    }
}

/** A changed master_dashboard_url drops the socket, and the reconnect dials
 *  the new master — otherwise reports and commands would stay with the old
 *  one until a restart, while the catalog followed the setting. */
function followMasterUrl(): void {
    const s = state();
    if (!s.socket || s.url === getSetting('master_dashboard_url')) return;
    console.log('[master-link] master_dashboard_url changed — reconnecting');
    try { s.socket.close(1000, 'master changed'); } catch { /* the close handler reconnects */ }
}

/** True when the report went up the socket, so the caller can skip the POST. */
export function pushReport(firms: unknown[], scheduler: unknown): boolean {
    followMasterUrl();
    return send({
        type: 'report',
        firms: firms as never,
        scheduler: scheduler as never,
        capabilities: collectCapabilities(),
        version: getBuildVersion() ?? undefined,
        updater: updaterStatus(),
    });
}

export function isLinked(): boolean {
    return state().ready;
}

function scheduleReconnect(reason: string) {
    const s = state();
    if (s.stopped || s.reconnectTimer) return;

    // Exponential backoff with jitter: a master restart brings every instance
    // back at once, and they should not all land on the same millisecond.
    const base = Math.min(RECONNECT_MAX_MS, RECONNECT_MIN_MS * 2 ** s.attempt);
    const delay = Math.round(base * (0.5 + Math.random() / 2));
    s.attempt++;

    if (s.attempt <= 3 || s.attempt % 10 === 0) {
        console.log(`[master-link] ${reason} — retrying in ${Math.round(delay / 1000)}s`);
    }

    s.reconnectTimer = setTimeout(() => {
        s.reconnectTimer = null;
        connect();
    }, delay);
}

function connect() {
    const s = state();
    if (s.stopped || s.socket) return;

    const masterUrl = getSetting('master_dashboard_url');
    const instance = getSetting('instance_name');
    if (!masterUrl || !instance) {
        // Not configured — check again later in case Settings gets filled in.
        scheduleReconnect('not configured');
        return;
    }

    const url = toWsUrl(masterUrl);
    if (!url) {
        scheduleReconnect(`master_dashboard_url is not a URL: ${masterUrl}`);
        return;
    }

    let socket: WebSocket;
    try {
        socket = directWebSocket(url);
    } catch (err: any) {
        scheduleReconnect(err?.message ?? 'could not open socket');
        return;
    }
    s.socket = socket;
    s.url = masterUrl;
    s.ready = false;

    socket.addEventListener('open', () => {
        // The stored uuid, once the master has issued one — it, not the name,
        // is who this instance is, so renaming instance_name keeps the same
        // person (and their accounting history) on the master.
        const instanceId = getSetting('instance_id');
        socket.send(JSON.stringify({
            type: 'hello',
            instance,
            id: instanceId || undefined,
            capabilities: collectCapabilities(),
            version: getBuildVersion() ?? undefined,
        } satisfies InstanceMessage));
    });

    socket.addEventListener('message', (event) => {
        let msg: MasterMessage;
        try {
            msg = JSON.parse(String(event.data));
        } catch {
            return;
        }

        if (msg.type === 'welcome') {
            s.ready = true;
            s.attempt = 0;
            if (msg.personId && msg.personId !== getSetting('instance_id')) {
                setSetting('instance_id', msg.personId);
            }
            console.log(`[master-link] connected to ${url} as "${instance}"`);
            return;
        }

        if (msg.type !== 'command') return;

        const { id, command } = msg;
        console.log(`[master-link] command: ${command.kind}`);

        // Deliberately not awaited: a trade walks every account and can take a
        // while, and the socket must stay free for a Stop in the meantime.
        void (async () => {
            try {
                const result = await execute(command);
                send({ type: 'ack', id, ok: true, result });
            } catch (err: any) {
                console.error(`[master-link] command ${command.kind} failed:`, err?.message ?? err);
                send({ type: 'ack', id, ok: false, error: err?.message ?? String(err) });
            }
            // Whatever just happened, the dashboard's numbers are now stale.
            reportSoon();
        })();
    });

    const dropped = (reason: string) => {
        if (s.socket !== socket) return; // superseded
        s.socket = null;
        s.ready = false;
        scheduleReconnect(reason);
    };

    socket.addEventListener('close', (event) => {
        dropped(event.reason ? `closed: ${event.reason}` : 'connection closed');
    });

    socket.addEventListener('error', () => {
        // 'close' always follows, and carries the better message of the two.
        if (s.socket === socket && s.ready) console.warn('[master-link] socket error');
    });
}

/** Set by startMasterLink so a command can trigger a report without a cycle. */
let reportSoon: () => void = () => { };

export function startMasterLink(onReportRequested: () => void) {
    if (typeof WebSocket === 'undefined') {
        console.warn('[master-link] no global WebSocket in this Node build — control link disabled');
        return;
    }

    reportSoon = onReportRequested;

    const s = state();
    s.stopped = false;

    // Idempotent on purpose. getClients() is lazy and starts this link, so the
    // first command that reaches it re-enters here from *inside* a command the
    // live socket is waiting to ack. Tearing the socket down there loses the
    // ack, and the master reports a trade that actually executed as a failure —
    // which invites pressing the button a second time. Already dialling or
    // connected means there is nothing to do.
    if (s.socket || s.reconnectTimer) return;

    connect();
    console.log('[master-link] started');
}