import { TradovateClient } from './tradovate-class'; import { TopstepXClient } from './topstepx-class'; import { DXTradeClient } from './dxtrade-class'; import { VestClient } from './vest-class'; import type { FirmPlatform } from './master-protocol'; import { errorMessage, isClientReady, type PlatformClient } from './platform-client'; import { getTradableFirms, startCatalogSync } from './firm-catalog'; import { firmInstruments, isFuturesFirm, prewarmMarket } from './instruments'; import { housekeepingRoot } from './instrument-specs'; import { resolveFirmAssignments } from './account-type-resolver'; import { resolveContracts } from './contract-resolver'; import { startReporter, sendReport } from './reporter'; import { startBackupSync } from './backup-sync'; import { applyBrokerHttpDefaults, applyBrokerProxy } from './outbound-proxy'; import axios from './http-client'; import { startMasterLink } from './master-link'; import { startUpdaterWatchdog } from './updater-watchdog'; import { resumeSchedulerIfPersisted, startClaimWatcher } from './auto-trade'; import { superviseQuotes, SUPERVISE_INTERVAL_MS, type QuoteClient } from './live-quotes'; // Use global to persist the client pool across HMR reloads in dev mode. // Keyed by the master's firm id, which is a uuid string. // // It is also what keeps the pool single in production: instrumentation.ts // boots it from its own server bundle, and the route handlers can each hold a // separate instance of this module. Registry state therefore lives here, never // in module-level variables. const g = global as typeof globalThis & { __tradovateClients?: Map; __tradovateClientsInitialized?: boolean; __tradovateClientCreds?: Map; __contractResolverTimer?: ReturnType; __contractResolveRetryTimer?: ReturnType; __contractResolveRetries?: number; __contractResolvedRoots?: Set; __clientSupervisorTimer?: ReturnType; __clientSupervision?: Map; __quoteSupervisorTimer?: ReturnType; }; function ensureMap(): Map { if (!g.__tradovateClients) { g.__tradovateClients = new Map(); } return g.__tradovateClients; } /** Tracks which credentials each live client was built with, so a password * change can be noticed and the client rebuilt. */ function credsMap(): Map { if (!g.__tradovateClientCreds) g.__tradovateClientCreds = new Map(); return g.__tradovateClientCreds; } /** What a client was built from — a change to any of it rebuilds the client. * DXTrade's server URL and login domain are the firm's (master), not the * machine's, but they decide the login just as much. */ function credsSignature(firm: { platform: string; username: string; password: string; apiKey: string; serverUrl?: string; loginDomain?: string }): string { return `${firm.platform} ${firm.username} ${firm.password} ${firm.apiKey} ${firm.serverUrl ?? ''} ${firm.loginDomain ?? ''}`; } export function initClient( id: string, firmName: string, platform: FirmPlatform, creds: { username: string; password: string; apiKey: string; serverUrl?: string; loginDomain?: string }, ): PlatformClient { const map = ensureMap(); const onSyncComplete = async () => { const c = map.get(id); if (!c) return; console.log(`[${firmName}] sync complete — ${c.accountList.length} account(s)`); // Every sync is a chance to bind accounts whose funding history has // finally arrived. The first pass can already resolve untraded accounts // from their exact cash balance; traded ones wait for the separate // reports fetch that produces fund transactions. try { const counts = resolveFirmAssignments(id, c); console.log( `[resolver] ${firmName}: ${counts.assigned} assigned, ` + `${counts.unassigned} unassigned, ${counts.manual} manual`, ); } catch (err) { console.error(`[resolver] ${firmName} failed`, err); } }; let client: PlatformClient; switch (platform) { case 'topstepx': client = new TopstepXClient(creds.username, creds.apiKey, onSyncComplete); break; case 'dxtrade': client = new DXTradeClient({ firmId: id, firmName, serverUrl: creds.serverUrl ?? '', loginDomain: creds.loginDomain ?? '', username: creds.username, password: creds.password, }, onSyncComplete); break; case 'tradovate': client = new TradovateClient(creds.username, creds.password, onSyncComplete); break; case 'vest': // The saved "API key" is the Vest login's wallet: whose session, // relayed by the extension, this client trades. client = new VestClient({ firmId: id, firmName, wallet: creds.apiKey }, onSyncComplete); break; default: { const unknown: never = platform; throw new Error(`no client for platform ${String(unknown)}`); } } map.set(id, client); credsMap().set(id, credsSignature({ platform, ...creds })); return client; } export function removeClient(id: string): void { const client = ensureMap().get(id); // A throwing disconnect() must still take the client out of the pool: // left in, half torn down, it would never reconnect — and the supervisor // rebuilding it would fail the same way every time. try { client?.disconnect(); } catch (err) { console.error(`[clients] disconnect failed for firm ${id}: ${errorMessage(err)}`); } ensureMap().delete(id); credsMap().delete(id); } /** Disconnect all clients and clear the pool so they are recreated on next getClients() call. */ export function resetClients(): void { const map = g.__tradovateClients; if (map) { for (const client of map.values()) { try { client.disconnect(); } catch { /* ignore */ } } } g.__tradovateClients = undefined; g.__tradovateClientCreds = undefined; g.__tradovateClientsInitialized = false; } /** * Brings the client pool in line with the catalog and the stored credentials. * * Called at boot, whenever credentials are saved, and after the catalog * changes — a firm can appear, disappear, or have its password corrected long * after startup, none of which the old one-shot initialization could handle. */ export function reconcileClients(): void { const map = ensureMap(); const firms = getTradableFirms(); const wanted = new Set(firms.map((f) => f.id)); // Firms removed on the master, or whose credentials were cleared. for (const id of [...map.keys()]) { if (!wanted.has(id)) { console.log(`[clients] dropping client for firm ${id} — no longer tradable`); removeClient(id); } } for (const firm of firms) { const signature = credsSignature(firm); const existing = map.get(firm.id); if (existing) { if (credsMap().get(firm.id) === signature) continue; console.log(`[clients] credentials changed for ${firm.name} — reconnecting`); removeClient(firm.id); } console.log(`[clients] starting ${firm.platform} client for ${firm.name}`); initClient(firm.id, firm.name, firm.platform, firm); } } export function getClients(): Map { const map = ensureMap(); // `next build` imports route modules to collect page data. Nothing there // should call this, but a build that did must never dial the brokers or // resume the scheduler — so hand back the empty pool and leave the flag // unset, as if nothing had happened. if (process.env.NEXT_PHASE === 'phase-production-build') return map; if (!g.__tradovateClientsInitialized) { g.__tradovateClientsInitialized = true; // Payouts at target are claimed whether or not the scheduler runs — // started first, so a later init step that throws can't leave it off. startClaimWatcher(); try { // Before anything dials out: a timeout on every broker HTTP call, // and the operator's proxy (if any) under the broker clients. // Master traffic stays direct by design. (http-client applies // both when it loads, too; these are idempotent.) applyBrokerHttpDefaults(axios.defaults); applyBrokerProxy(axios.defaults); // Pull the catalog before anything else — without it there are no // firms to start clients for. Subsequent changes re-reconcile and // re-resolve, so a new account type picks up accounts that were // sitting Unassigned without waiting for a Tradovate sync. startCatalogSync(() => { reconcileClients(); // A newly listed root needs its front-month contract. if (resolveRootsHere().some((r) => !g.__contractResolvedRoots?.has(r))) triggerContractResolve(); for (const [firmId, client] of ensureMap()) { try { resolveFirmAssignments(firmId, client); } catch (err) { console.error(`[resolver] re-resolve failed for ${firmId}`, err); } } }); reconcileClients(); console.log(`[clients] Initialized ${map.size} platform client(s)`); // Auto-resolve contracts after clients have time to authenticate, // and measure every future's movement for the Auto draw. setTimeout(() => triggerContractResolve(), 15_000); prewarmMarket(); // Schedule daily resolve at midnight scheduleDailyResolve(); // Rebuild any client that has been wedged for too long. startClientSupervisor(); // Live quotes off one futures client, for the spread every entry // pays and the Auto draw. startQuoteSupervisor(); // Start master dashboard reporter startReporter(); // Bring this machine's auto-updater back if its PM2 cron stops firing. startUpdaterWatchdog(); // Push settings+credentials backups to the master, so a dead VPS // can be replaced and this identity adopted. startBackupSync(); // ...and the socket it prefers to report over, which is also how // the master triggers trades and AutoBuyer runs here. startMasterLink(() => { void sendReport(); }); // Pick the auto-trade schedule back up if it was running before restart resumeSchedulerIfPersisted(); } catch (err) { console.error('[clients] Failed to initialize clients', err); } } return map; } // ── Contract auto-resolve ──────────────────────────────────────────────────── // With no authenticated Tradovate client yet — a cold catalog, a slow login, // broker maintenance at boot — keep checking once a minute for half an hour // instead of giving up until midnight. Missing it leaves the contract cache // cold: the settings page shows nothing, TopstepX loses the resolver's // front-month names, and the first trade pays for the lookups itself. const CONTRACT_RESOLVE_RETRY_MS = 60_000; const CONTRACT_RESOLVE_MAX_RETRIES = 30; /** Every root a futures firm here lists, plus each one's drain micro. */ function resolveRootsHere(): string[] { const roots = new Set(); for (const firm of getTradableFirms().filter(isFuturesFirm)) { const listed = firmInstruments(firm); for (const r of listed) roots.add(r); const micro = housekeepingRoot(listed); if (micro) roots.add(micro); } return [...roots]; } function triggerContractResolve(): void { // A midnight run supersedes a pending retry; either way only one chain. if (g.__contractResolveRetryTimer) clearTimeout(g.__contractResolveRetryTimer); g.__contractResolveRetryTimer = undefined; const map = ensureMap(); // Find first client with a valid access token (the Yahoo-driven resolver // confirms contract ids against Tradovate; TopstepX clients resolve their // own contracts but still reuse the resolver's front-month names). // A token alone isn't enough: a client whose session died keeps its // expired one, and resolving with that fails with nothing left to retry. let accessToken: string | null = null; for (const [, c] of map) { // eslint-disable-next-line @typescript-eslint/no-explicit-any const token = (c as any).accessInfo?.accessToken; if (!token) continue; let authenticated = true; try { authenticated = c.health().authenticated; } catch { /* no health: trust the token */ } if (authenticated) { accessToken = token; break; } } if (!accessToken) { const retries = (g.__contractResolveRetries ?? 0) + 1; if (retries > CONTRACT_RESOLVE_MAX_RETRIES) { console.log( `[contract-resolver] No authenticated Tradovate client after ${CONTRACT_RESOLVE_MAX_RETRIES} retries` + ' — giving up until the midnight resolve', ); g.__contractResolveRetries = 0; return; } if (retries === 1) { console.log('[contract-resolver] No authenticated Tradovate client yet — retrying every 60s'); } g.__contractResolveRetries = retries; g.__contractResolveRetryTimer = setTimeout(() => triggerContractResolve(), CONTRACT_RESOLVE_RETRY_MS); return; } g.__contractResolveRetries = 0; // Runs from timers: a throw here (the instruments read) would escape as // an uncaught exception. try { const symbols = resolveRootsHere(); g.__contractResolvedRoots = new Set(symbols); console.log(`[contract-resolver] Resolving ${symbols.length} instruments...`); resolveContracts(symbols, accessToken) .then((results) => { const rolled = Object.entries(results).filter(([, v]) => v?.alternative); console.log(`[contract-resolver] Done — ${rolled.length} contract(s) rolled via price match`); }) .catch((err) => console.error(`[contract-resolver] Resolve failed: ${errorMessage(err)}`)); } catch (err) { console.error(`[contract-resolver] Resolve failed: ${errorMessage(err)}`); } } function scheduleDailyResolve(): void { // Clear any existing timer (HMR safety) if (g.__contractResolverTimer) clearInterval(g.__contractResolverTimer); // Check every minute if it's midnight (00:00) g.__contractResolverTimer = setInterval(() => { const now = new Date(); if (now.getHours() === 0 && now.getMinutes() === 0) { console.log('[contract-resolver] Midnight resolve triggered'); // Its own half hour of retries, not what's left of a chain it cuts off. g.__contractResolveRetries = 0; triggerContractResolve(); prewarmMarket(); } }, 60_000); } // ── Client supervision ─────────────────────────────────────────────────────── /** * Rebuilds a client that has been unable to trade for a long time. * * The clients reconnect themselves, but only along the paths their authors * foresaw. A re-login that fails in an unexpected way, an await that never * settles, a socket that dies without a close event — any of those used to * leave a client wedged for the life of the process, and because the * scheduler waits on every firm, one wedged firm stopped them all. After a * weekend of broker maintenance that meant a pm2 restart. A fresh client is * exactly what the restart gave it, so the registry hands one out itself: * same firm, same credentials, new instance. * * "Unready" is isClientReady() saying no — the same rule the scheduler and * /api/health use: not synced, or no good sync in the last 5 minutes. * * Deliberately slow — 20 minutes unready before a rebuild, and at most one * rebuild per firm per 20 minutes. The clients' own retries get the first go, * and a broker that is really down, or rate-limiting logins, must not be * hammered with new sessions. Some clients are never replaced while unready * (see rebuildHold): one whose own login retry is pending, and a TopstepX * one holding a position. */ const SUPERVISOR_INTERVAL_MS = 60_000; const REBUILD_AFTER_UNREADY_MS = 20 * 60_000; const REBUILD_COOLDOWN_MS = 20 * 60_000; /** Clients that can't log in until someone acts — a rebuild changes nothing, * and fixing the credentials (or the firm's settings) rebuilds them through * reconcileClients anyway. TopstepX with an API key but no username never * tries to log in; DXTrade with no server URL can't, and after refused * credentials it stops trying so as not to lock the account; Vest waits on * the extension to relay a signed-in tab's session. */ function needsAttention(platform: FirmPlatform, client: PlatformClient): boolean { if (platform === 'topstepx') return /username missing/i.test(client.lastFetchErrors?.['login'] ?? ''); if (platform === 'dxtrade') return (client as DXTradeClient).needsAttention === true; if (platform === 'vest') return (client as VestClient).needsAttention === true; return false; } interface SupervisionEntry { /** The instance being judged. A different one (rebuilt here, or replaced * by reconcileClients after a credentials change) starts a fresh clock. */ client: PlatformClient; /** When it was first seen unready (0 = ready at the last check). */ unhealthySince: number; lastRebuildAt: number; /** Last time a due rebuild was held back and said so (log rate limit). */ holdLoggedAt: number; } function supervisionMap(): Map { if (!g.__clientSupervision) g.__clientSupervision = new Map(); return g.__clientSupervision; } /** Keep the live quote feed on a futures client, subscribed to every root * this machine may trade (lib/live-quotes.ts). */ function superviseLiveQuotes(): void { // Runs from a timer: nothing may escape. try { const map = ensureMap(); const candidates = getTradableFirms().filter(isFuturesFirm).flatMap((firm) => { const client = map.get(firm.id); return client ? [{ firmId: firm.id, firmName: firm.name, platform: firm.platform, client: client as unknown as QuoteClient }] : []; }); superviseQuotes(candidates, resolveRootsHere()) .catch((err) => console.error(`[quotes] supervision failed: ${errorMessage(err)}`)); } catch (err) { console.error(`[quotes] supervision failed: ${errorMessage(err)}`); } } function startQuoteSupervisor(): void { if (g.__quoteSupervisorTimer) clearInterval(g.__quoteSupervisorTimer); g.__quoteSupervisorTimer = setInterval(superviseLiveQuotes, SUPERVISE_INTERVAL_MS); // The clients log in within seconds; the first look shouldn't wait 30. setTimeout(superviseLiveQuotes, 10_000); } function startClientSupervisor(): void { // HMR-safe, and safe when resetClients() re-runs the bootstrap. if (g.__clientSupervisorTimer) clearInterval(g.__clientSupervisorTimer); g.__clientSupervisorTimer = setInterval(superviseClients, SUPERVISOR_INTERVAL_MS); } /** Why a client isn't ready, short enough for one log line. Never the token. */ function describeUnready(client: PlatformClient, now: number): string { const parts: string[] = []; try { const health = client.health(); parts.push(`transport ${health.transport}`); if (health.lastLoginError) parts.push(`login error: ${health.lastLoginError.slice(0, 160)}`); } catch { parts.push('health unavailable'); } parts.push(client.lastSyncAt > 0 ? `last good sync ${Math.round((now - client.lastSyncAt) / 60_000)}m ago` : 'never synced'); return parts.join(', '); } /** * Why replacing this client now would do more harm than good, or null. * * - Its login keeps failing and its own retry is already scheduled. Nothing * in the instance is broken — the broker is down, refusing the * credentials, or rate-limiting — and a fresh instance would fail the same * way while starting its backoff over at seconds: a burst of logins after * every rebuild instead of one every few minutes, which is how a login * rate-limit turns into a captcha lockout. * - TopstepX with a position open. The gateway has no OCO, so the client * itself cancels the surviving exit leg when the other fills; a fresh * instance knows nothing of those pairs, and the survivor would be left * working at the broker after the position closes — an order that can open * a position nobody asked for. It waits until the account is flat. */ function rebuildHold(platform: string, client: PlatformClient, now: number): string | null { try { const health = client.health(); if (health.consecutiveLoginFailures > 0 && health.nextReconnectAt > now) { return `its login retry is due in ${Math.max(1, Math.round((health.nextReconnectAt - now) / 60_000))}m` + ` (${health.consecutiveLoginFailures} failure(s) in a row)`; } } catch { /* no health: judge it on the rest */ } if ((platform === 'topstepx' || platform === 'dxtrade') && Object.keys(client.positions ?? {}).length > 0) { return 'it holds an open position (its exit pairs are tracked only inside the client)'; } return null; } function superviseClients(): void { // Runs inside setInterval: nothing may escape, or the process logs an // uncaught exception and the supervisor keeps firing into the same fault. try { const map = ensureMap(); const book = supervisionMap(); const now = Date.now(); // A firm that left the registry takes its bookkeeping with it. for (const id of [...book.keys()]) { if (!map.has(id)) book.delete(id); } let firms: ReturnType | null = null; for (const [id, client] of [...map]) { try { let entry = book.get(id); if (!entry) { entry = { client, unhealthySince: 0, lastRebuildAt: 0, holdLoggedAt: 0 }; book.set(id, entry); } else if (entry.client !== client) { entry.client = client; entry.unhealthySince = 0; } if (isClientReady(client, now)) { entry.unhealthySince = 0; continue; } if (!entry.unhealthySince) { entry.unhealthySince = now; continue; } if (now - entry.unhealthySince < REBUILD_AFTER_UNREADY_MS) continue; if (now - entry.lastRebuildAt < REBUILD_COOLDOWN_MS) continue; // Only read when a rebuild is actually due — it hits the DB. firms ??= getTradableFirms(); const firm = firms.find((f) => f.id === id); if (!firm) continue; // no longer tradable: reconcileClients drops it if (needsAttention(firm.platform, client)) continue; const minutes = Math.round((now - entry.unhealthySince) / 60_000); const hold = rebuildHold(firm.platform, client, now); if (hold) { if (now - entry.holdLoggedAt >= REBUILD_COOLDOWN_MS) { entry.holdLoggedAt = now; console.log(`[clients] ${firm.name} unhealthy for ${minutes}m — not rebuilding while ${hold}`); } continue; } console.log( `[clients] ${firm.name} unhealthy for ${minutes}m (${describeUnready(client, now)})` + ' — rebuilding client', ); entry.lastRebuildAt = now; removeClient(id); entry.client = initClient(firm.id, firm.name, firm.platform, firm); entry.unhealthySince = now; } catch (err) { console.error(`[clients] supervisor check failed for firm ${id}: ${errorMessage(err)}`); } } } catch (err) { console.error(`[clients] supervisor tick failed: ${errorMessage(err)}`); } }