autofirmer-expanded/lib/clients.ts

import { TradovateClient } from './tradovate-class';
import { TopstepXClient } from './topstepx-class';
import { DXTradeClient } from './dxtrade-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 { startMasterLink } from './master-link';
import { resumeSchedulerIfPersisted } 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<string, PlatformClient>;
    __tradovateClientsInitialized?: boolean;
    __tradovateClientCreds?: Map<string, string>;
    __contractResolverTimer?: ReturnType<typeof setInterval>;
    __contractResolveRetryTimer?: ReturnType<typeof setTimeout>;
    __contractResolveRetries?: number;
    __contractResolvedRoots?: Set<string>;
    __clientSupervisorTimer?: ReturnType<typeof setInterval>;
    __clientSupervision?: Map<string, SupervisionEntry>;
    __quoteSupervisorTimer?: ReturnType<typeof setInterval>;
};

function ensureMap(): Map<string, PlatformClient> {
    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<string, string> {
    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;
        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<string, PlatformClient> {
    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;
        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. (outbound-proxy applies
            // both when it loads, too; these are idempotent.)
            applyBrokerHttpDefaults();
            applyBrokerProxy();

            // 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();

            // 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<string>();
    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. */
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;
    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<string, SupervisionEntry> {
    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<typeof getTradableFirms> | 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)}`);
    }
}