autofirmer-expanded/lib/firm-catalog.ts

/**
 * The firm catalog — firms and their account types, owned by the master
 * dashboard and pulled from its public GET /api/firms/catalog.
 *
 * This machine holds a cache of the last complete, valid response. That cache
 * is the whole point: lib/auto-trade.ts needs every account's profit target and
 * size on each 60s tick, so if those numbers only ever came over the network, a
 * master reboot or a flaky link would stop trading with positions open. The
 * cache is written only by a successful sync and never edited locally, so it
 * cannot drift the way the old per-machine account_configs did.
 *
 * Stored as one JSON blob in `settings` rather than mirrored into tables: the
 * catalog is only ever read whole, or asked "give me firm X's account types".
 *
 * (Unrelated to the `FIRMS` array in lib/automations.ts, which describes
 * prop-firm websites the AutoBuyer clicks through. Different notion of "firm";
 * the two are not linked yet.)
 */

import { isFirmPlatform } from './master-protocol';
import { sanitizeInstruments } from './instrument-specs';
import type {
    CatalogAccountType, CatalogFirm, FirmCatalogPayload, FirmPlatform, SymbolMap, SymbolMapCandidate,
} from './master-protocol';
import {
    getSetting,
    setSetting,
    listCredentials,
    getCredentials,
    rekeyCredentials,
} from './db';

const CATALOG_PATH = '/api/firms/catalog';
const POLL_INTERVAL_MS = 60_000;
const FETCH_TIMEOUT_MS = 10_000;

// Until the very first successful sync there is nothing to trade from, so back
// off quickly rather than sitting on the full poll interval.
const COLD_RETRY_MIN_MS = 10_000;
const COLD_RETRY_MAX_MS = 60_000;

interface CatalogCache {
    payload: FirmCatalogPayload | null;
    /** Index over every firm's account types, for the id lookup. */
    typesById: Map<string, CatalogAccountType>;
}

const g = global as typeof globalThis & {
    __firmCatalog?: CatalogCache;
    __firmCatalogTimer?: ReturnType<typeof setTimeout>;
    __firmCatalogColdDelay?: number;
};

function indexPayload(payload: FirmCatalogPayload | null): CatalogCache {
    const typesById = new Map<string, CatalogAccountType>();
    for (const firm of payload?.firms ?? []) {
        for (const type of firm.accountTypes) typesById.set(type.id, type);
    }
    return { payload, typesById };
}

function cache(): CatalogCache {
    if (g.__firmCatalog) return g.__firmCatalog;

    let payload: FirmCatalogPayload | null = null;
    const raw = getSetting('catalog_payload');
    if (raw) {
        try {
            payload = validatePayload(JSON.parse(raw));
        } catch {
            console.error('[catalog] cached payload is unreadable — ignoring it');
        }
    }
    g.__firmCatalog = indexPayload(payload);
    return g.__firmCatalog;
}

export function getCachedCatalog(): CatalogFirm[] {
    return cache().payload?.firms ?? [];
}

export function getCachedFirm(firmId: string): CatalogFirm | undefined {
    return getCachedCatalog().find((f) => f.id === firmId);
}

export function getAccountTypeById(typeId: string): CatalogAccountType | undefined {
    return cache().typesById.get(typeId);
}

export function getCatalogMeta(): {
    version: number;
    updatedAt: string | null;
    lastSyncAt: string | null;
    lastError: string | null;
    firmCount: number;
} {
    return {
        version: Number(getSetting('catalog_version') ?? 0),
        updatedAt: getSetting('catalog_updated_at'),
        lastSyncAt: getSetting('catalog_last_sync_at'),
        lastError: getSetting('catalog_last_error'),
        firmCount: getCachedCatalog().length,
    };
}

/** Whether stored credentials are what the firm's platform logs in with — a
 *  TopstepX API key, or a username + password (Tradovate, DXTrade). */
export function hasPlatformCredentials(
    platform: FirmPlatform,
    creds: { username: string; password: string; apiKey: string } | null | undefined,
): boolean {
    if (!creds) return false;
    switch (platform) {
        case 'topstepx': return !!creds.apiKey;
        case 'tradovate':
        case 'dxtrade': return !!creds.username && !!creds.password;
    }
}

/** Firms this machine can actually trade: in the catalog AND holding the
 *  credentials their platform needs. Everything that starts a platform client
 *  goes through here. */
export function getTradableFirms(): (CatalogFirm & { username: string; password: string; apiKey: string })[] {
    const result: (CatalogFirm & { username: string; password: string; apiKey: string })[] = [];
    for (const firm of getCachedCatalog()) {
        const creds = getCredentials(firm.id);
        if (creds && hasPlatformCredentials(firm.platform, creds)) result.push({ ...firm, ...creds });
    }
    return result;
}

// ── Sync ─────────────────────────────────────────────────────────────────────

const g2 = global as typeof globalThis & { __catalogUnknownPlatformWarned?: Set<string> };

/** Once per firm per process — the catalog is re-validated every 60s. */
function warnUnknownPlatform(id: string, name: string, platform: unknown): void {
    const warned = (g2.__catalogUnknownPlatformWarned ??= new Set());
    const key = `${id}|${String(platform)}`;
    if (warned.has(key)) return;
    warned.add(key);
    console.warn(`[catalog] skipping "${name}": platform ${JSON.stringify(platform)} is unknown to this build — update this instance`);
}

function nonNegative(value: unknown): number {
    return typeof value === 'number' && Number.isFinite(value) && value > 0 ? value : 0;
}

/** Drops anything malformed rather than failing the sync: a bad symbol-map
 *  entry only means that candidate isn't traded. */
function sanitizeSymbolMap(raw: unknown): SymbolMap {
    const map: SymbolMap = {};
    if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) return map;
    for (const [root, list] of Object.entries(raw as Record<string, unknown>)) {
        if (!Array.isArray(list)) continue;
        const candidates = list.flatMap((c): SymbolMapCandidate[] => {
            if (typeof c !== 'object' || c === null) return [];
            const cand = c as Record<string, unknown>;
            if (typeof cand.symbol !== 'string' || !cand.symbol || cand.symbol.includes('|')) return [];
            return [{ symbol: cand.symbol }];
        });
        if (candidates.length > 0) map[root.toUpperCase()] = candidates;
    }
    return map;
}

function isStages(value: unknown): value is { profit: number; consistency: number; minTradingDays: number }[] {
    return Array.isArray(value) && value.every((s) =>
        typeof s === 'object' && s !== null
        && typeof (s as Record<string, unknown>).profit === 'number'
        && typeof (s as Record<string, unknown>).consistency === 'number'
        && typeof (s as Record<string, unknown>).minTradingDays === 'number');
}

/** Rejects anything that isn't a complete catalog. A half-parsed payload would
 *  silently unassign accounts, so it is better to keep the old cache. */
function validatePayload(raw: unknown): FirmCatalogPayload {
    if (typeof raw !== 'object' || raw === null) throw new Error('payload is not an object');
    const body = raw as Record<string, unknown>;
    if (typeof body.version !== 'number') throw new Error('missing version');
    if (typeof body.updatedAt !== 'string') throw new Error('missing updatedAt');
    if (!Array.isArray(body.firms)) throw new Error('missing firms');

    const firms: CatalogFirm[] = [];
    body.firms.forEach((f: unknown, i: number) => {
        if (typeof f !== 'object' || f === null) throw new Error(`firm ${i} is not an object`);
        const firm = f as Record<string, unknown>;
        if (typeof firm.id !== 'string' || typeof firm.name !== 'string') {
            throw new Error(`firm ${i} is missing id or name`);
        }
        // An old master sends no platform: that is Tradovate. A platform this
        // build doesn't know is NOT — coercing it would start the wrong client
        // with another platform's credentials. Skip the firm (before its
        // account types can fail the whole sync); it comes back once this
        // build knows the platform.
        const platform = firm.platform === undefined ? 'tradovate' : firm.platform;
        if (!isFirmPlatform(platform)) {
            warnUnknownPlatform(firm.id, firm.name, platform);
            return;
        }
        // Percent commission and every-day trading are DXtrade settings;
        // futures firms keep measured per-contract fees and the CME calendar.
        const dx = platform === 'dxtrade';
        const tradesEveryDay = dx && firm.tradesEveryDay === true;
        if (!Array.isArray(firm.accountTypes)) throw new Error(`firm ${firm.name} is missing accountTypes`);

        const accountTypes: CatalogAccountType[] = firm.accountTypes.map((t: unknown, j: number) => {
            if (typeof t !== 'object' || t === null) throw new Error(`account type ${j} is not an object`);
            const type = t as Record<string, unknown>;
            for (const key of ['id', 'firmId', 'label', 'prefix', 'updatedAt']) {
                if (typeof type[key] !== 'string') throw new Error(`account type ${j} is missing ${key}`);
            }
            for (const key of [
                'accountSize', 'profitTarget', 'consistency', 'minDayPnL',
                'minTradingDays', 'maxLoss', 'maxPositionSize',
            ]) {
                if (typeof type[key] !== 'number') throw new Error(`account type ${j} is missing ${key}`);
            }
            if (!isStages(type.withdrawalStages)) throw new Error(`account type ${j} has bad withdrawalStages`);
            // Lenient like platform: a master from before the split toggle
            // doesn't send the field, and those account types all split. Same
            // for CME holiday combining, where the old behavior was "off".
            return {
                ...type,
                splitEnabled: type.splitEnabled !== false,
                // An every-day firm has no CME calendar to fold holidays by.
                combineCmeHolidays: type.combineCmeHolidays === true && !tradesEveryDay,
                // Lenient: a master from before the min-legs field means "no minimum".
                minLegs: typeof type.minLegs === 'number' ? type.minLegs : 0,
                // Lenient: the obfuscation stops are opt-in, so absent means off.
                firstDayStop: type.firstDayStop === true,
                minDayStop: type.minDayStop === true,
                // Lenient: no hard daily limit, no notional cap.
                dailyLossLimit: nonNegative(type.dailyLossLimit),
                // The firm's calendar, stamped on every type so the cfg-based
                // paths (projections, completion, the dashboard) all see it.
                tradesEveryDay,
            } as unknown as CatalogAccountType;
        });

        firms.push({
            id: firm.id,
            name: firm.name,
            platform,
            // The master's list when it sends one; absent from an older
            // master, whose bannedSymbols / symbolMap then apply.
            ...(Array.isArray(firm.instruments) ? { instruments: sanitizeInstruments(platform, firm.instruments) } : {}),
            bannedSymbols: Array.isArray(firm.bannedSymbols) ? (firm.bannedSymbols as string[]) : [],
            accountTypes,
            serverUrl: typeof firm.serverUrl === 'string' ? firm.serverUrl : '',
            loginDomain: typeof firm.loginDomain === 'string' ? firm.loginDomain : '',
            commissionPct: dx ? nonNegative(firm.commissionPct) : 0,
            tradesEveryDay,
            symbolMap: sanitizeSymbolMap(firm.symbolMap),
            targetHoldHours: nonNegative(firm.targetHoldHours),
            financingPctPerDay: nonNegative(firm.financingPctPerDay),
            updatedAt: typeof firm.updatedAt === 'string' ? firm.updatedAt : new Date(0).toISOString(),
        });
    });

    return { version: body.version, updatedAt: body.updatedAt, firms };
}

function catalogUrl(): URL | null {
    const base = getSetting('master_dashboard_url');
    if (!base) return null;
    try {
        // new URL, never string concatenation: the stored value routinely has a
        // trailing slash, which would produce //api/firms/catalog.
        return new URL(CATALOG_PATH, base);
    } catch {
        return null;
    }
}

function recordError(message: string): void {
    setSetting('catalog_last_error', message);
    setSetting('catalog_last_sync_at', new Date().toISOString());
}

/** If the master's database was rebuilt, its firms come back with fresh ids and
 *  every credential row orphans. Match the survivors up by name (once) rather
 *  than making someone retype every password. */
function rekeyCredentialsByName(firms: CatalogFirm[]): void {
    const byName = new Map(firms.map((f) => [f.name.toLowerCase(), f.id]));
    const knownIds = new Set(firms.map((f) => f.id));

    for (const row of listCredentials()) {
        if (knownIds.has(row.firm_id)) continue;
        const match = byName.get(row.firm_name.toLowerCase());
        if (!match) continue;
        console.warn(`[catalog] re-keying credentials for "${row.firm_name}": ${row.firm_id} -> ${match}`);
        rekeyCredentials(row.firm_id, match);
    }
}

export async function syncCatalog(): Promise<{ ok: boolean; changed: boolean; version: number; error?: string }> {
    const version = Number(getSetting('catalog_version') ?? 0);
    const url = catalogUrl();
    if (!url) {
        const error = 'master_dashboard_url is not set';
        recordError(error);
        return { ok: false, changed: false, version, error };
    }

    // Only the URL is required — deliberately not instance_name, unlike the
    // reporter. A freshly imaged machine has to be able to pull its catalog
    // before anyone has got round to naming it.
    // `since` only against the master the cache came from: two masters'
    // version numbers can collide, and an 'unchanged' from the wrong one
    // would keep the other's catalog.
    const hasCache = getCachedCatalog().length > 0;
    const source = url.origin;
    if (hasCache && getSetting('catalog_source_url') === source) url.searchParams.set('since', String(version));

    try {
        const res = await fetch(url, { signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) });
        if (!res.ok) throw new Error(`master returned ${res.status}`);
        const body = await res.json();

        if (body && typeof body === 'object' && (body as Record<string, unknown>).unchanged === true) {
            setSetting('catalog_source_url', source);
            setSetting('catalog_last_sync_at', new Date().toISOString());
            setSetting('catalog_last_error', '');
            return { ok: true, changed: false, version };
        }

        const payload = validatePayload(body);

        // A master pointed at the wrong FIRMS_DB, or restored from an empty
        // one, serves {"firms":[]} as a perfectly valid 200. Mirroring that
        // would stop trading on every machine at once, so keep what we have.
        if (payload.firms.length === 0 && hasCache) {
            const error = 'master returned an empty catalog — ignored';
            console.error(`[catalog] ${error}`);
            recordError(error);
            return { ok: false, changed: false, version, error };
        }

        // Same reasoning one level down: a firm losing every account type
        // unassigns all of its accounts. Keep the previous types and flag it.
        const previous = new Map(getCachedCatalog().map((f) => [f.id, f]));
        const suspicious: string[] = [];
        for (const firm of payload.firms) {
            const before = previous.get(firm.id);
            if (firm.accountTypes.length === 0 && before && before.accountTypes.length > 0) {
                firm.accountTypes = before.accountTypes;
                suspicious.push(firm.name);
            }
        }

        setSetting('catalog_payload', JSON.stringify(payload));
        setSetting('catalog_version', String(payload.version));
        setSetting('catalog_source_url', source);
        setSetting('catalog_updated_at', payload.updatedAt);
        setSetting('catalog_last_sync_at', new Date().toISOString());
        setSetting('catalog_last_error', suspicious.length > 0
            ? `kept cached account types for ${suspicious.join(', ')} — master sent none`
            : '');
        g.__firmCatalog = indexPayload(payload);

        rekeyCredentialsByName(payload.firms);

        console.log(`[catalog] synced v${payload.version} — ${payload.firms.length} firm(s)`);
        return { ok: true, changed: true, version: payload.version };
    } catch (err) {
        const error = err instanceof Error ? err.message : String(err);
        console.error('[catalog] sync failed:', error);
        recordError(error);
        return { ok: false, changed: false, version, error };
    }
}

export function startCatalogSync(onChange?: () => void): void {
    stopCatalogSync();

    const tick = async () => {
        const result = await syncCatalog();
        if (result.changed) {
            try {
                onChange?.();
            } catch (err) {
                console.error('[catalog] onChange handler failed', err);
            }
        }

        let delay = POLL_INTERVAL_MS;
        if (getCachedCatalog().length === 0) {
            // Never synced: retry hard, with backoff, until something lands.
            const current = g.__firmCatalogColdDelay ?? COLD_RETRY_MIN_MS;
            delay = current;
            g.__firmCatalogColdDelay = Math.min(current * 2, COLD_RETRY_MAX_MS);
        } else {
            g.__firmCatalogColdDelay = COLD_RETRY_MIN_MS;
            // Jitter so a fleet restarting together doesn't poll in lockstep.
            delay = POLL_INTERVAL_MS + Math.floor(Math.random() * 10_000);
        }

        g.__firmCatalogTimer = setTimeout(() => void tick(), delay);
    };

    void tick();
}

export function stopCatalogSync(): void {
    if (g.__firmCatalogTimer) clearTimeout(g.__firmCatalogTimer);
    g.__firmCatalogTimer = undefined;
}