/** * 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, getInstruments, getRunnerHeartbeat, RUNNER_TIMEOUT_MS } from './db'; import { runTrade, startScheduler, stopScheduler, getSchedulerStatus } from './auto-trade'; import { FIRMS, automationKey } from './automations'; import { POINT_VALUES } from './trading-logic'; import { queueRun } from './queue-run'; import { getBuildVersion } from './version'; 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; /** Only true once the master has answered the hello. */ ready: boolean; attempt: number; reconnectTimer: ReturnType | 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, 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 { // Only symbols the operator has enabled here. An empty instruments table // means nothing has been configured yet, so offer everything we can price. const enabled = getInstruments().filter((i) => i.enabled).map((i) => i.symbol); // 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: enabled.length > 0 ? enabled : Object.keys(POINT_VALUES), 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, }; } // ── 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 { switch (command.kind) { case 'trade': { if (command.symbol !== 'Auto' && !POINT_VALUES[command.symbol]) { throw new Error(`Unknown symbol: ${command.symbol}`); } // Same order as POST /api/trade: fire once now, then leave the // scheduler running to pick up accounts that were busy. const results = await runTrade(command.action, command.symbol); startScheduler(command.action, command.symbol, command.stopAfterAll ?? false); const filled = results .flatMap((r: any) => r.results ?? []) .filter((r: any) => 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 }; 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; } } /** True when the report went up the socket, so the caller can skip the POST. */ export function pushReport(firms: unknown[], scheduler: unknown): boolean { return send({ type: 'report', firms: firms as never, scheduler: scheduler as never, capabilities: collectCapabilities(), version: getBuildVersion() ?? undefined, }); } 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 = new WebSocket(url); } catch (err: any) { scheduleReconnect(err?.message ?? 'could not open socket'); return; } s.socket = socket; 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'); }