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, sanitizeInstrumentLimits, DEFAULT_MAX_ENTRY_SPREAD } from './instrument-specs';
import { normalizeAccountType } from './kept-rules';
import type {
    CatalogAccountType, CatalogFirm, CatalogTradingSettings, FirmCatalogPayload, FirmPlatform, KeptAccountRules, 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;

/** The version of validatePayload below, stamped on the payload it caches.
 *  Bump it whenever the parser starts keeping a field an older one dropped:
 *  a cache from before then lacks the field however often it is re-read, and
 *  `?since=` would answer "unchanged" until the master next changes, so the
 *  first sync on a newer parser fetches the whole catalog.
 *  2: kept rule sets (keptAccountRules) and the firm's maxConcurrentAccounts.
 *  Stamped only on a payload from a master that sends keptAccountRules: one
 *  from an older master can't vouch for kept sets it never sent. */
export const CATALOG_PARSER_VERSION = 2;

/** A Date header further off than this is somebody else's clock, not the master's. */
const MAX_MASTER_OFFSET_MS = 365 * 24 * 60 * 60 * 1000;

interface CatalogCache {
    payload: FirmCatalogPayload | null;
    /** Index over every firm's account types, for the id lookup. */
    typesById: Map<string, CatalogAccountType>;
    /** Each live account type's kept rule sets, by closedAt (oldest window
     *  first). Kept ids are deliberately not in typesById: nothing can be
     *  pinned to one. */
    keptByLiveId: Map<string, KeptAccountRules[]>;
}

const g = global as typeof globalThis & {
    __firmCatalog?: CatalogCache;
    __firmCatalogTimer?: ReturnType<typeof setTimeout>;
    __firmCatalogColdDelay?: number;
    /** The master's clock minus this one, from the last sync's Date header. */
    __masterOffsetMs?: number;
};

function indexPayload(payload: FirmCatalogPayload | null): CatalogCache {
    const typesById = new Map<string, CatalogAccountType>();
    const keptByLiveId = new Map<string, KeptAccountRules[]>();
    for (const firm of payload?.firms ?? []) {
        for (const type of firm.accountTypes) typesById.set(type.id, type);
        for (const kept of firm.keptAccountRules ?? []) {
            const list = keptByLiveId.get(kept.keptFrom);
            if (list) list.push(kept);
            else keptByLiveId.set(kept.keptFrom, [kept]);
        }
    }
    for (const list of keptByLiveId.values()) list.sort((a, b) => Date.parse(a.closedAt) - Date.parse(b.closedAt));
    return { payload, typesById, keptByLiveId };
}

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

    let payload: FirmCatalogPayload | null = null;
    const raw = getSetting('catalog_payload');
    if (raw) {
        try {
            const stored = JSON.parse(raw);
            payload = validatePayload(stored);
            // Whatever parsed the blob when it was stored, not this parser: a
            // field an older build dropped on the way in can't come back from
            // re-reading the blob.
            const parser = (stored as { parser?: unknown }).parser;
            if (typeof parser === 'number') payload.parser = parser;
            else delete payload.parser;
        } 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);
}

/** The kept rule sets of one live account type, oldest window first ([] when
 *  it keeps none). lib/account-type-resolver.ts picks an account's by its
 *  rules date. */
export function getKeptRulesFor(liveTypeId: string): KeptAccountRules[] {
    return cache().keptByLiveId.get(liveTypeId) ?? [];
}

/** The kept rule set ids loaded for a firm, for the report — null while the
 *  cached catalog came from a parser older than CATALOG_PARSER_VERSION, which
 *  can't be trusted to hold them (the master then reads this machine as not
 *  applying kept rules yet). */
export function getKeptRuleIdsForFirm(firmId: string): string[] | null {
    const payload = cache().payload;
    if (!payload || (payload.parser ?? 0) < CATALOG_PARSER_VERSION) return null;
    return (payload.firms.find((f) => f.id === firmId)?.keptAccountRules ?? []).map((k) => k.id);
}

/** The master's clock minus this machine's, as of the last good sync (null
 *  before one this process). Dates new accounts on the master's clock when
 *  this one runs behind (lib/kept-rules.ts stampRulesDate). */
export function getMasterOffsetMs(): number | null {
    return g.__masterOffsetMs ?? null;
}

/** When the cached catalog last changed on the master (its payload's
 *  updatedAt); null with no cache. No account first seen after a change can
 *  be dated before it. */
export function getCatalogUpdatedAt(): string | null {
    return cache().payload?.updatedAt ?? null;
}

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>;
    __catalogKeptSkipWarned?: 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 isDateString(value: unknown): value is string {
    return typeof value === 'string' && Number.isFinite(Date.parse(value));
}

/**
 * A firm's kept rule sets, each as strict as an account type: a malformed
 * entry throws and the whole sync is rejected. One that names no live type
 * of this firm is skipped instead (warned once) — no account can map to it.
 */
function parseKeptRules(
    firm: { id: string; name: string },
    raw: unknown[],
    accountTypes: CatalogAccountType[],
    tradesEveryDay: boolean,
): KeptAccountRules[] {
    const liveIds = new Set(accountTypes.map((t) => t.id));
    const kept: KeptAccountRules[] = [];
    raw.forEach((k: unknown, j: number) => {
        const where = `kept rules ${j} of ${firm.name}`;
        const type = normalizeAccountType(k, { tradesEveryDay, where });
        const entry = k as Record<string, unknown>;
        if (typeof entry.keptFrom !== 'string') throw new Error(`${where} is missing keptFrom`);
        if (entry.openedAt !== null && !isDateString(entry.openedAt)) throw new Error(`${where} has bad openedAt`);
        if (!isDateString(entry.closedAt)) throw new Error(`${where} has bad closedAt`);
        if (!liveIds.has(entry.keptFrom) || type.firmId !== firm.id) {
            const warned = (g2.__catalogKeptSkipWarned ??= new Set());
            if (!warned.has(type.id)) {
                warned.add(type.id);
                console.warn(`[catalog] skipping kept rules ${type.id} of "${firm.name}": it names no account type of this firm (keptFrom ${entry.keptFrom}, firmId ${type.firmId})`);
            }
            return;
        }
        kept.push({ ...type, keptFrom: entry.keptFrom, openedAt: entry.openedAt, closedAt: entry.closedAt });
    });
    return kept;
}

/** 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) =>
            normalizeAccountType(t, { tradesEveryDay, where: `account type ${j}` }));
        // Only when the master sent the list: absent = a master from before
        // kept rules, [] = every account on the current rules.
        const keptAccountRules = Array.isArray(firm.keptAccountRules)
            ? parseKeptRules({ id: firm.id, name: firm.name }, firm.keptAccountRules, accountTypes, tradesEveryDay)
            : undefined;

        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) } : {}),
            ...(platform !== 'dxtrade' ? { instrumentLimits: sanitizeInstrumentLimits(firm.instrumentLimits) } : {}),
            bannedSymbols: Array.isArray(firm.bannedSymbols) ? (firm.bannedSymbols as string[]) : [],
            accountTypes,
            ...(keptAccountRules ? { keptAccountRules } : {}),
            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),
            // Lenient: absent or malformed = no firm cap.
            maxConcurrentAccounts: Math.floor(nonNegative(firm.maxConcurrentAccounts)),
            updatedAt: typeof firm.updatedAt === 'string' ? firm.updatedAt : new Date(0).toISOString(),
        });
    });

    const trading = parseTradingSettings(body.trading);
    // A kept-aware master sends keptAccountRules on every firm (even []). A
    // payload without it is a master from before kept rules: cached without
    // a parser version, so the cache isn't trusted to hold kept sets — every
    // sync fetches in full (no `since`), and the first one after a kept-aware
    // master is back brings them, though its catalog version never moved.
    const keptAware = body.firms.every((f: unknown) =>
        typeof f === 'object' && f !== null && Array.isArray((f as Record<string, unknown>).keptAccountRules));
    return {
        version: body.version, updatedAt: body.updatedAt, firms, ...(trading ? { trading } : {}),
        ...(keptAware ? { parser: CATALOG_PARSER_VERSION } : {}),
    };
}

/** The master's trading settings; undefined when it sent none (or nothing
 *  usable), so the defaults apply. */
function parseTradingSettings(raw: unknown): CatalogTradingSettings | undefined {
    if (typeof raw !== 'object' || raw === null) return undefined;
    const max = (raw as Record<string, unknown>).maxEntrySpreadDollars;
    return typeof max === 'number' && Number.isFinite(max) && max >= 0 ? { maxEntrySpreadDollars: max } : undefined;
}

/** What the master's Settings page says, or the defaults from before it. */
export function getTradingSettings(): CatalogTradingSettings {
    return cache().payload?.trading ?? { maxEntrySpreadDollars: DEFAULT_MAX_ENTRY_SPREAD };
}

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

/** The master's clock off a response's Date header (whole seconds, read on
 *  arrival — close enough to date accounts by). An unreadable header, or one
 *  a year off, leaves the last offset standing. */
function recordMasterClock(header: string | null): void {
    const at = header ? Date.parse(header) : NaN;
    if (!Number.isFinite(at)) return;
    const offset = at - Date.now();
    if (Math.abs(offset) > MAX_MASTER_OFFSET_MS) return;
    g.__masterOffsetMs = offset;
}

/** 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. Nor for a cache an older parser wrote:
    // it lacks what that parser dropped, and 'unchanged' would keep it so.
    const hasCache = getCachedCatalog().length > 0;
    const source = url.origin;
    const cacheCurrent = (cache().payload?.parser ?? 0) >= CATALOG_PARSER_VERSION;
    if (hasCache && cacheCurrent && 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}`);
        recordMasterClock(res.headers.get('date'));
        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,
        // with the kept rule sets that went with them (whose keptFrom named
        // types this payload lacks, so none of its own survived parsing). An
        // empty kept list on its own is never suspicious: it is every account
        // on the current rules.
        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;
                if (before.keptAccountRules) firm.keptAccountRules = before.keptAccountRules;
                else delete firm.keptAccountRules;
                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;
}