Backups to master, replacement-VPS restore, broker proxy, multi-instance
Four features: - Backup: lib/backup-sync.ts pushes the full settings table and firm credentials (the deliberate exception to "credentials never leave the machine") to the master's bearer-gated /api/instances/backup — every 5 minutes when the content hash changed, within seconds of a settings or credential edit, and on the new backup-now command. Enabled by the master_backup_token setting. - Restore: a dead VPS's replacement pulls the backup and adopts its uuid/name, so roster, payouts and card links carry on. Pre-boot via seed-settings.js --restore-from (setup-linux.sh prompts for it) or the Settings page's restore picker (lib/backup-restore.ts). Stop the old machine first; the master 409s a live-name rename. - Proxy: the proxy_url setting routes BROKER traffic only — TopstepX REST + SignalR hub, Tradovate REST + websocket, Yahoo quotes — via lib/outbound-proxy.ts (axios agents, undici dispatcher, ws agent). The master link, reporter, catalog and backup push stay direct, so a dead proxy cannot take down the control channel. Restart to apply. - Multi-instance: AUTOFIRMER_DB moves the SQLite file so one checkout runs N instances; scripts/instances.json (per-device, gitignored) drives start-all/update-check, and add-instance.sh / remove-instance.sh manage the extra PM2 processes. Absent registry = single-instance behavior, unchanged. Instance 1 keeps port 3000 for the clicker and extension. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
97947bffde
commit
8154f25bb9
@@ -0,0 +1,126 @@
|
||||
/**
|
||||
* Restores a backup pulled from the master onto THIS machine — the other half
|
||||
* of lib/backup-sync.ts, for replacing a dead VPS.
|
||||
*
|
||||
* Applying a backup overwrites this machine's settings, broker credentials,
|
||||
* instruments and assignments, and — the point of the exercise — adopts the
|
||||
* backed-up instance's uuid and name, so the master's roster, payouts and card
|
||||
* links carry straight on. The master already handles adoption: a hello
|
||||
* presenting a known uuid lands on that person's roster row.
|
||||
*
|
||||
* Run it before first boot via scripts/seed-settings.js --restore-from (which
|
||||
* duplicates this apply logic — keep them in sync), or from the settings UI
|
||||
* via POST /api/backup/restore. Either way the old machine must be stopped
|
||||
* first: two live instances presenting one uuid fight over the socket.
|
||||
*/
|
||||
|
||||
import {
|
||||
setSetting,
|
||||
getSetting,
|
||||
setCredentials,
|
||||
setInstrumentEnabled,
|
||||
setAssignment,
|
||||
runInTransaction,
|
||||
} from './db';
|
||||
import type { BackupPayload } from './backup-sync';
|
||||
|
||||
/** Mirror of the master's listing rows (lib/backups.ts BackupMeta). */
|
||||
export interface BackupMeta {
|
||||
id: string;
|
||||
name: string;
|
||||
savedAt: string;
|
||||
sizeBytes: number;
|
||||
}
|
||||
|
||||
const FETCH_TIMEOUT_MS = 15_000;
|
||||
|
||||
// Runtime bookkeeping describing the machine the backup was taken on, not
|
||||
// this one. Kept in sync with the skip-set in scripts/seed-settings.js.
|
||||
const RESTORE_SKIP_KEYS = new Set([
|
||||
'runner_heartbeat',
|
||||
'catalog_last_sync_at',
|
||||
'catalog_last_error',
|
||||
'backup_last_push_at',
|
||||
'backup_last_error',
|
||||
]);
|
||||
|
||||
function authedFetch(masterUrl: string, token: string, path: string) {
|
||||
const url = new URL(path, masterUrl);
|
||||
return fetch(url, {
|
||||
headers: { authorization: `Bearer ${token}` },
|
||||
signal: AbortSignal.timeout(FETCH_TIMEOUT_MS),
|
||||
});
|
||||
}
|
||||
|
||||
export async function fetchBackupList(masterUrl: string, token: string): Promise<BackupMeta[]> {
|
||||
const res = await authedFetch(masterUrl, token, '/api/instances/backup');
|
||||
if (!res.ok) {
|
||||
const detail = await res.json().then((b) => b?.error).catch(() => '');
|
||||
throw new Error(`master returned ${res.status}${detail ? ` — ${detail}` : ''}`);
|
||||
}
|
||||
const body = await res.json();
|
||||
if (!Array.isArray(body?.backups)) throw new Error('unrecognized backup list');
|
||||
return body.backups as BackupMeta[];
|
||||
}
|
||||
|
||||
export async function fetchBackup(masterUrl: string, token: string, id: string): Promise<BackupPayload> {
|
||||
const res = await authedFetch(masterUrl, token, `/api/instances/backup/${encodeURIComponent(id)}`);
|
||||
if (!res.ok) {
|
||||
const detail = await res.json().then((b) => b?.error).catch(() => '');
|
||||
throw new Error(`master returned ${res.status}${detail ? ` — ${detail}` : ''}`);
|
||||
}
|
||||
const payload = await res.json() as BackupPayload;
|
||||
if (payload?.v !== 1 || typeof payload.settings !== 'object' || payload.settings === null) {
|
||||
throw new Error('unrecognized backup payload');
|
||||
}
|
||||
return payload;
|
||||
}
|
||||
|
||||
export interface RestoreCounts {
|
||||
settings: number;
|
||||
credentials: number;
|
||||
instruments: number;
|
||||
assignments: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Overwrites this machine's configuration with the backup, including
|
||||
* instance_id and instance_name — the identity adoption. Callers restart the
|
||||
* process afterwards: live platform clients, the master link's hello and any
|
||||
* cached settings all predate the restore.
|
||||
*/
|
||||
export function applyBackup(payload: BackupPayload, opts: { keepMasterUrl?: boolean } = {}): RestoreCounts {
|
||||
// Defaults to keeping this machine's master url: the machine is talking to
|
||||
// that master right now, while an old backup may point at a dead host.
|
||||
const keepMasterUrl = opts.keepMasterUrl ?? true;
|
||||
const ownMasterUrl = getSetting('master_dashboard_url');
|
||||
const ownToken = getSetting('master_backup_token');
|
||||
|
||||
const counts: RestoreCounts = { settings: 0, credentials: 0, instruments: 0, assignments: 0 };
|
||||
|
||||
runInTransaction(() => {
|
||||
for (const [key, value] of Object.entries(payload.settings)) {
|
||||
if (RESTORE_SKIP_KEYS.has(key)) continue;
|
||||
if (typeof value !== 'string') continue;
|
||||
setSetting(key, value);
|
||||
counts.settings++;
|
||||
}
|
||||
if (keepMasterUrl && ownMasterUrl) setSetting('master_dashboard_url', ownMasterUrl);
|
||||
if (ownToken) setSetting('master_backup_token', ownToken);
|
||||
|
||||
for (const c of payload.credentials ?? []) {
|
||||
setCredentials(c.firm_id, c.firm_name, c.username, c.password, c.api_key ?? '');
|
||||
counts.credentials++;
|
||||
}
|
||||
for (const i of payload.instruments ?? []) {
|
||||
setInstrumentEnabled(i.symbol, i.enabled);
|
||||
counts.instruments++;
|
||||
}
|
||||
for (const a of payload.assignments ?? []) {
|
||||
setAssignment(a.account_id, a.firm_id, a.account_type_id ?? null, a.manual === 1);
|
||||
counts.assignments++;
|
||||
}
|
||||
});
|
||||
|
||||
return counts;
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
/**
|
||||
* Pushes this machine's full configuration — settings, broker credentials,
|
||||
* instruments, manual assignments — to the master, so a dead VPS can be
|
||||
* replaced by restoring the backup (lib/backup-restore.ts) and adopting this
|
||||
* instance's uuid.
|
||||
*
|
||||
* This is the deliberate exception to "credentials never leave the machine":
|
||||
* the payload goes over HTTPS to the bearer-gated /api/instances/backup and is
|
||||
* AES-encrypted at rest on the master. It never rides the unauthenticated
|
||||
* control socket.
|
||||
*
|
||||
* Push policy: on bootstrap, then every POLL_INTERVAL, but only when the
|
||||
* content actually changed (hash comparison) — plus an immediate debounced
|
||||
* push whenever settings or credentials are edited (scheduleBackupPush, called
|
||||
* from the mutation routes).
|
||||
*/
|
||||
|
||||
import { createHash } from 'node:crypto';
|
||||
import {
|
||||
getSetting,
|
||||
setSetting,
|
||||
listSettings,
|
||||
listCredentials,
|
||||
getInstruments,
|
||||
listAllAssignments,
|
||||
type FirmCredentialRow,
|
||||
type AssignmentRow,
|
||||
} from './db';
|
||||
|
||||
export interface BackupPayload {
|
||||
v: 1;
|
||||
savedAt: string;
|
||||
settings: Record<string, string>;
|
||||
credentials: FirmCredentialRow[];
|
||||
instruments: { symbol: string; enabled: boolean }[];
|
||||
assignments: AssignmentRow[];
|
||||
}
|
||||
|
||||
const BACKUP_PATH = '/api/instances/backup';
|
||||
const POLL_INTERVAL_MS = 5 * 60_000;
|
||||
const DEBOUNCE_MS = 5_000;
|
||||
const FETCH_TIMEOUT_MS = 10_000;
|
||||
|
||||
const g = globalThis as typeof globalThis & {
|
||||
__backupSyncTimer?: ReturnType<typeof setTimeout>;
|
||||
__backupDebounceTimer?: ReturnType<typeof setTimeout>;
|
||||
__backupLastHash?: string;
|
||||
};
|
||||
|
||||
// Bookkeeping that describes this process run, not the configuration worth
|
||||
// restoring. Excluded from the hash so a heartbeat doesn't look like a change
|
||||
// worth pushing. (The restore side has its own skip-set; these still travel in
|
||||
// the payload so a restore debugger can see them.)
|
||||
const VOLATILE_KEYS = new Set([
|
||||
'runner_heartbeat',
|
||||
'catalog_last_sync_at',
|
||||
'catalog_last_error',
|
||||
'backup_last_push_at',
|
||||
'backup_last_error',
|
||||
]);
|
||||
|
||||
export function collectBackup(): BackupPayload {
|
||||
return {
|
||||
v: 1,
|
||||
savedAt: new Date().toISOString(),
|
||||
settings: listSettings(),
|
||||
credentials: listCredentials(),
|
||||
instruments: getInstruments().map((i) => ({ symbol: i.symbol, enabled: Boolean(i.enabled) })),
|
||||
assignments: listAllAssignments(),
|
||||
};
|
||||
}
|
||||
|
||||
/** Stable digest of everything restorable, so unchanged config never pushes. */
|
||||
function contentHash(payload: BackupPayload): string {
|
||||
const settings = Object.entries(payload.settings)
|
||||
.filter(([k]) => !VOLATILE_KEYS.has(k))
|
||||
.sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
|
||||
return createHash('sha256')
|
||||
.update(JSON.stringify({
|
||||
settings,
|
||||
credentials: payload.credentials,
|
||||
instruments: payload.instruments,
|
||||
assignments: payload.assignments,
|
||||
}))
|
||||
.digest('hex');
|
||||
}
|
||||
|
||||
function preconditions(): { masterUrl: string; id: string; token: string } | null {
|
||||
const masterUrl = getSetting('master_dashboard_url');
|
||||
const token = getSetting('master_backup_token');
|
||||
// No uuid yet = the master has never said welcome; there is nothing stable
|
||||
// to key the backup by, so wait for the first hello to complete.
|
||||
const id = getSetting('instance_id');
|
||||
if (!masterUrl || !token || !id) return null;
|
||||
return { masterUrl, id, token };
|
||||
}
|
||||
|
||||
export async function pushBackup(force = false): Promise<{ ok: boolean; skipped?: boolean; error?: string }> {
|
||||
const pre = preconditions();
|
||||
if (!pre) return { ok: false, skipped: true, error: 'master url, backup token or instance uuid not set' };
|
||||
|
||||
const payload = collectBackup();
|
||||
const hash = contentHash(payload);
|
||||
if (!force && hash === g.__backupLastHash) return { ok: true, skipped: true };
|
||||
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(BACKUP_PATH, pre.masterUrl);
|
||||
} catch {
|
||||
return { ok: false, error: `master_dashboard_url is not a URL: ${pre.masterUrl}` };
|
||||
}
|
||||
|
||||
try {
|
||||
const res = await fetch(url, {
|
||||
method: 'PUT',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
authorization: `Bearer ${pre.token}`,
|
||||
},
|
||||
body: JSON.stringify({
|
||||
id: pre.id,
|
||||
name: getSetting('instance_name') ?? '',
|
||||
payload,
|
||||
}),
|
||||
signal: AbortSignal.timeout(FETCH_TIMEOUT_MS),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const detail = await res.json().then((b) => b?.error).catch(() => '');
|
||||
throw new Error(`master returned ${res.status}${detail ? ` — ${detail}` : ''}`);
|
||||
}
|
||||
g.__backupLastHash = hash;
|
||||
setSetting('backup_last_push_at', new Date().toISOString());
|
||||
setSetting('backup_last_error', '');
|
||||
console.log('[backup] pushed to master');
|
||||
return { ok: true };
|
||||
} catch (err) {
|
||||
const error = err instanceof Error ? err.message : String(err);
|
||||
console.error('[backup] push failed:', error);
|
||||
setSetting('backup_last_error', error);
|
||||
return { ok: false, error };
|
||||
}
|
||||
}
|
||||
|
||||
/** Debounced push for the mutation routes: a settings PATCH or credential PUT
|
||||
* lands on the master seconds later instead of at the next poll. */
|
||||
export function scheduleBackupPush(): void {
|
||||
if (g.__backupDebounceTimer) clearTimeout(g.__backupDebounceTimer);
|
||||
g.__backupDebounceTimer = setTimeout(() => { void pushBackup(); }, DEBOUNCE_MS);
|
||||
}
|
||||
|
||||
export function startBackupSync(): void {
|
||||
stopBackupSync();
|
||||
const tick = async () => {
|
||||
await pushBackup();
|
||||
// Jitter so a fleet restarting together doesn't push in lockstep.
|
||||
g.__backupSyncTimer = setTimeout(tick, POLL_INTERVAL_MS + Math.floor(Math.random() * 30_000));
|
||||
};
|
||||
void tick();
|
||||
}
|
||||
|
||||
export function stopBackupSync(): void {
|
||||
if (g.__backupSyncTimer) {
|
||||
clearTimeout(g.__backupSyncTimer);
|
||||
g.__backupSyncTimer = undefined;
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,8 @@ import { getTradableFirms, startCatalogSync } from './firm-catalog';
|
||||
import { resolveFirmAssignments } from './account-type-resolver';
|
||||
import { resolveContracts } from './contract-resolver';
|
||||
import { startReporter, sendReport } from './reporter';
|
||||
import { startBackupSync } from './backup-sync';
|
||||
import { applyBrokerProxy } from './outbound-proxy';
|
||||
import { startMasterLink } from './master-link';
|
||||
import { resumeSchedulerIfPersisted } from './auto-trade';
|
||||
|
||||
@@ -128,6 +130,10 @@ export function getClients(): Map<string, PlatformClient> {
|
||||
if (!g.__tradovateClientsInitialized) {
|
||||
g.__tradovateClientsInitialized = true;
|
||||
try {
|
||||
// Before anything dials out: put the operator's proxy (if any)
|
||||
// under the broker clients. Master traffic stays direct by design.
|
||||
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
|
||||
@@ -155,6 +161,10 @@ export function getClients(): Map<string, PlatformClient> {
|
||||
// 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(); });
|
||||
|
||||
@@ -1,9 +1,22 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import { mkdirSync } from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
const db = new Database(path.join(process.cwd(), 'autotrader.sqlite'));
|
||||
// AUTOFIRMER_DB lets several instances share one checkout, each with its own
|
||||
// database. It is read once, at module load — schema creation and seeding are
|
||||
// import side effects — so it must be in the environment at process start.
|
||||
const DB_PATH = process.env.AUTOFIRMER_DB
|
||||
? path.resolve(process.env.AUTOFIRMER_DB)
|
||||
: path.join(process.cwd(), 'autotrader.sqlite');
|
||||
|
||||
mkdirSync(path.dirname(DB_PATH), { recursive: true });
|
||||
const db = new Database(DB_PATH);
|
||||
|
||||
db.pragma('foreign_keys = ON');
|
||||
// If two processes are ever misconfigured onto the same file, WAL + a busy
|
||||
// timeout turns SQLITE_BUSY crashes into brief contention.
|
||||
db.pragma('journal_mode = WAL');
|
||||
db.pragma('busy_timeout = 5000');
|
||||
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS instruments (
|
||||
@@ -123,6 +136,11 @@ export function listAssignments(firmId: string): AssignmentRow[] {
|
||||
return db.prepare('SELECT * FROM account_assignments WHERE firm_id = ?').all(firmId) as AssignmentRow[];
|
||||
}
|
||||
|
||||
/** Every assignment, for the backup pushed to the master. */
|
||||
export function listAllAssignments(): AssignmentRow[] {
|
||||
return db.prepare('SELECT * FROM account_assignments').all() as AssignmentRow[];
|
||||
}
|
||||
|
||||
export function setAssignment(
|
||||
accountId: number,
|
||||
firmId: string,
|
||||
@@ -179,12 +197,31 @@ seedSetting.run('twocaptcha_api_key', '');
|
||||
// Blank falls back to the proxy hardcoded in app/api/autobuyer/captcha; `none`
|
||||
// solves without one.
|
||||
seedSetting.run('twocaptcha_proxy', '');
|
||||
// Bearer for pushing/pulling the encrypted settings+credentials backup on the
|
||||
// master (its INSTANCE_BACKUP_TOKEN). Blank = backups off.
|
||||
seedSetting.run('master_backup_token', '');
|
||||
// Routes broker traffic (TopstepX, Tradovate, Yahoo quotes) through a proxy:
|
||||
// http(s):// or socks5(h)://, user:pass ok. The master link stays direct on
|
||||
// purpose. Blank = direct. Takes effect on restart.
|
||||
seedSetting.run('proxy_url', '');
|
||||
// Mirrors the in-memory scheduler state so a restart can resume it.
|
||||
seedSetting.run('scheduler_running', '0');
|
||||
seedSetting.run('scheduler_action', 'Buy');
|
||||
seedSetting.run('scheduler_symbol', 'NQ');
|
||||
seedSetting.run('scheduler_stop_after_all', '0');
|
||||
|
||||
/** Wraps several writes in one transaction (used by the backup restore, so a
|
||||
* half-applied backup cannot survive a crash). */
|
||||
export function runInTransaction<T>(fn: () => T): T {
|
||||
return db.transaction(fn)();
|
||||
}
|
||||
|
||||
/** The whole table, for the backup pushed to the master. */
|
||||
export function listSettings(): Record<string, string> {
|
||||
const rows = db.prepare('SELECT key, value FROM settings').all() as { key: string; value: string }[];
|
||||
return Object.fromEntries(rows.map((r) => [r.key, r.value]));
|
||||
}
|
||||
|
||||
export function getSetting(key: string): string | null {
|
||||
const row = db.prepare('SELECT value FROM settings WHERE key = ?').get(key) as { value: string } | undefined;
|
||||
return row?.value ?? null;
|
||||
|
||||
@@ -20,6 +20,7 @@ import { runTrade, startScheduler, stopScheduler, getSchedulerStatus } from './a
|
||||
import { FIRMS, automationKey } from './automations';
|
||||
import { POINT_VALUES } from './trading-logic';
|
||||
import { queueRun } from './queue-run';
|
||||
import { pushBackup } from './backup-sync';
|
||||
import { getBuildVersion } from './version';
|
||||
import type {
|
||||
Command,
|
||||
@@ -119,6 +120,9 @@ async function execute(command: Command): Promise<unknown> {
|
||||
case 'report-now':
|
||||
return { reported: true };
|
||||
|
||||
case 'backup-now':
|
||||
return pushBackup(true);
|
||||
|
||||
default: {
|
||||
// A newer master may know commands this build does not. Say so
|
||||
// rather than failing silently.
|
||||
|
||||
@@ -198,7 +198,9 @@ export type Command =
|
||||
| { kind: 'trade'; action: 'Buy' | 'Sell' | 'Auto'; symbol: string; stopAfterAll?: boolean }
|
||||
| { kind: 'stop-trading' }
|
||||
| { kind: 'autobuyer-run'; automationId: string; inputs?: Record<string, number> }
|
||||
| { kind: 'report-now' };
|
||||
| { kind: 'report-now' }
|
||||
/** Push a fresh settings/credentials backup right now. */
|
||||
| { kind: 'backup-now' };
|
||||
|
||||
export type MasterMessage =
|
||||
/** personId is the person's stable uuid; the instance stores it and
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* Routes BROKER traffic through the operator's proxy — and nothing else.
|
||||
*
|
||||
* Scope is deliberate: TopstepX REST + its SignalR hub, Tradovate REST + its
|
||||
* websocket, and the Yahoo quote lookups go through the proxy so each instance
|
||||
* can present its own IP to the firms. The master link, reporter, catalog sync
|
||||
* and backup push stay direct — the control channel must keep working when the
|
||||
* proxy dies, and there is no setGlobalDispatcher here for exactly that
|
||||
* reason. (Corollary: never call the master through axios — axios defaults
|
||||
* carry the proxy agents. Master calls use fetch, which stays untouched.)
|
||||
*
|
||||
* Configured by the `proxy_url` setting: http://, https://, socks5:// or
|
||||
* socks5h:// with optional user:pass. Read once per process — changing it
|
||||
* requires a restart, which the settings UI says out loud. Blank = direct.
|
||||
*/
|
||||
|
||||
import axios from 'axios';
|
||||
import { ProxyAgent, type Dispatcher } from 'undici';
|
||||
import { socksDispatcher } from 'fetch-socks';
|
||||
import { HttpProxyAgent } from 'http-proxy-agent';
|
||||
import { HttpsProxyAgent } from 'https-proxy-agent';
|
||||
import { SocksProxyAgent } from 'socks-proxy-agent';
|
||||
import type { Agent } from 'node:http';
|
||||
import { getSetting } from './db';
|
||||
|
||||
const SCHEMES = ['http:', 'https:', 'socks5:', 'socks5h:'] as const;
|
||||
|
||||
interface ProxyState {
|
||||
url: URL | null;
|
||||
dispatcher: Dispatcher | null;
|
||||
httpAgent: Agent | null;
|
||||
httpsAgent: Agent | null;
|
||||
applied: boolean;
|
||||
}
|
||||
|
||||
// HMR-safe, and the "read once per process" contract lives here.
|
||||
const g = globalThis as typeof globalThis & { __outboundProxy?: ProxyState };
|
||||
|
||||
function state(): ProxyState {
|
||||
if (g.__outboundProxy) return g.__outboundProxy;
|
||||
|
||||
const s: ProxyState = { url: null, dispatcher: null, httpAgent: null, httpsAgent: null, applied: false };
|
||||
g.__outboundProxy = s;
|
||||
|
||||
const raw = getSetting('proxy_url')?.trim();
|
||||
if (!raw) return s;
|
||||
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(raw);
|
||||
} catch {
|
||||
console.error(`[proxy] proxy_url is not a URL — trading DIRECT: ${raw}`);
|
||||
return s;
|
||||
}
|
||||
if (!SCHEMES.includes(url.protocol as (typeof SCHEMES)[number])) {
|
||||
console.error(`[proxy] proxy_url scheme must be one of ${SCHEMES.join(' ')} — trading DIRECT: ${url.protocol}`);
|
||||
return s;
|
||||
}
|
||||
|
||||
const socks = url.protocol.startsWith('socks');
|
||||
if (socks) {
|
||||
s.dispatcher = socksDispatcher({
|
||||
type: 5,
|
||||
host: url.hostname,
|
||||
port: Number(url.port || 1080),
|
||||
userId: url.username ? decodeURIComponent(url.username) : undefined,
|
||||
password: url.password ? decodeURIComponent(url.password) : undefined,
|
||||
});
|
||||
const agent = new SocksProxyAgent(url);
|
||||
s.httpAgent = agent;
|
||||
s.httpsAgent = agent;
|
||||
} else {
|
||||
s.dispatcher = new ProxyAgent(url.href);
|
||||
s.httpAgent = new HttpProxyAgent(url);
|
||||
s.httpsAgent = new HttpsProxyAgent(url);
|
||||
}
|
||||
s.url = url;
|
||||
console.log(`[proxy] broker traffic will use ${url.protocol}//${url.host} (master link stays direct)`);
|
||||
return s;
|
||||
}
|
||||
|
||||
export function getProxyUrl(): string | null {
|
||||
const { url } = state();
|
||||
return url ? url.href : null;
|
||||
}
|
||||
|
||||
/** Puts the proxy under every axios call. All axios usage in this app is
|
||||
* broker/quote traffic (TopstepX, Tradovate REST, contract resolver), so the
|
||||
* process-wide default is the scoped choice, not a global one. Idempotent;
|
||||
* called first in the getClients() bootstrap. */
|
||||
export function applyBrokerProxy(): void {
|
||||
const s = state();
|
||||
if (s.applied || !s.url) return;
|
||||
s.applied = true;
|
||||
axios.defaults.proxy = false; // agents own the routing; ignore env HTTP(S)_PROXY handling
|
||||
axios.defaults.httpAgent = s.httpAgent!;
|
||||
axios.defaults.httpsAgent = s.httpsAgent!;
|
||||
}
|
||||
|
||||
/** A websocket to a BROKER endpoint: the built-in (undici) WebSocket, dialled
|
||||
* through the proxy when one is set. The master link keeps using plain
|
||||
* `new WebSocket(...)` — do not use this for it. */
|
||||
export function brokerWebSocket(wsUrl: string): WebSocket {
|
||||
const { dispatcher } = state();
|
||||
if (!dispatcher) return new WebSocket(wsUrl);
|
||||
// The runtime accepts { dispatcher } fine; only the DOM lib type doesn't.
|
||||
return new (WebSocket as unknown as new (u: string, o: object) => WebSocket)(wsUrl, { dispatcher });
|
||||
}
|
||||
|
||||
/** A `ws`-package constructor with the proxy agent baked in, for SignalR's
|
||||
* `withUrl({ WebSocket })` option. undefined = no proxy, let SignalR use its
|
||||
* own require('ws'). */
|
||||
export function signalrWebSocket(): unknown {
|
||||
const s = state();
|
||||
if (!s.httpsAgent) return undefined;
|
||||
const agent = s.httpsAgent;
|
||||
// eslint-disable-next-line @typescript-eslint/no-require-imports
|
||||
const WS = require('ws') as new (address: string, protocols?: unknown, options?: object) => unknown;
|
||||
|
||||
return class ProxiedWebSocket extends (WS as new (...args: never[]) => object) {
|
||||
constructor(address: string, protocols?: unknown, options?: Record<string, unknown>) {
|
||||
// ws also allows (address, options) — normalize before merging.
|
||||
if (protocols && typeof protocols === 'object' && !Array.isArray(protocols)) {
|
||||
options = protocols as Record<string, unknown>;
|
||||
protocols = undefined;
|
||||
}
|
||||
super(...[address, protocols, { ...(options ?? {}), agent }] as never[]);
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -24,6 +24,7 @@ import axios from 'axios';
|
||||
import * as signalR from '@microsoft/signalr';
|
||||
import { currentTradeDateCT } from './trading-logic';
|
||||
import { getCachedContract } from './contract-resolver';
|
||||
import { signalrWebSocket } from './outbound-proxy';
|
||||
import { saveDailyPnL, loadDailyPnL, loadAccountMeta, saveAccountMeta, loadFundTransactions, saveFundTransactions } from './db';
|
||||
import { getAccountConfig } from './account-type-resolver';
|
||||
import type { PlatformClient, PlatformAccount, PlatformContract } from './platform-client';
|
||||
@@ -258,11 +259,16 @@ export class TopstepXClient implements PlatformClient {
|
||||
// the package is listed in next.config serverExternalPackages and any
|
||||
// residual failure here is caught by the caller (REST polling covers
|
||||
// everything the hub does, just slower).
|
||||
// With skipNegotiation the hub is a single websocket to a broker
|
||||
// endpoint, so it follows the operator's proxy like the REST calls do.
|
||||
const proxiedWs = signalrWebSocket();
|
||||
const hub = new signalR.HubConnectionBuilder()
|
||||
.withUrl(USER_HUB, {
|
||||
skipNegotiation: true,
|
||||
transport: signalR.HttpTransportType.WebSockets,
|
||||
accessTokenFactory: () => this.token ?? '',
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
...(proxiedWs ? { WebSocket: proxiedWs as any } : {}),
|
||||
})
|
||||
.withAutomaticReconnect()
|
||||
.configureLogging(signalR.LogLevel.Warning)
|
||||
|
||||
@@ -6,6 +6,7 @@ import { computeSec, randomUUIDV4 } from './tradovate-helpers';
|
||||
import { POINT_VALUES } from './trading-logic';
|
||||
import { getCachedContract, resolveContracts } from './contract-resolver';
|
||||
import { saveDailyPnL, loadDailyPnL, saveAccountMeta, loadAccountMeta, saveFundTransactions, loadFundTransactions } from './db';
|
||||
import { brokerWebSocket } from './outbound-proxy';
|
||||
|
||||
/**
|
||||
* Entity types Tradovate sends over the user-sync socket. Keep this as the
|
||||
@@ -147,7 +148,9 @@ export class TradovateClient {
|
||||
this.syncComplete = false;
|
||||
|
||||
const randomnumber = Math.random().toString(36).substring(2, 15);
|
||||
this.ws = new WebSocket(`wss://demo.tradovateapi.com/v1/websocket?r=${randomnumber}`);
|
||||
// Broker endpoint — dialled through the operator's proxy when one
|
||||
// is configured (lib/outbound-proxy.ts).
|
||||
this.ws = brokerWebSocket(`wss://demo.tradovateapi.com/v1/websocket?r=${randomnumber}`);
|
||||
|
||||
this.ws.onopen = () => {
|
||||
console.log('Connected to websocket');
|
||||
|
||||
Reference in New Issue
Block a user