spb/market-atlas
Public
TypeScript 96.7%
SQL 1.6%
CSS 0.8%
JavaScript 0.5%
1import type { RawObservation } from "@market-atlas/market-model";2import {3 HttpClient,4 ManagedWebSocket,5 RateLimiter,6 backoffMs,7 schemaFingerprint,8 type ConnectorContext,9 type ConnectorDefinition,10 type ManagedWebSocketOptions,11} from "@market-atlas/connector-sdk";12import { config } from "../config.js";13import { pool } from "../db/pool.js";14import { logger } from "../logger.js";15import { calendar } from "./calendar.js";16import { eventEngine } from "./events.js";17import { health } from "./health.js";18import { pipeline } from "./pipeline.js";19import { rawArchive } from "./raw-archive.js";20import { telemetry } from "./telemetry.js";2122interface Runtime {23 def: ConnectorDefinition;24 ctx: ConnectorContext;25 sockets: ManagedWebSocket[];26 pollTimer: NodeJS.Timeout | null;27 running: boolean;28 paused: boolean;29 consecutiveFailures: number;30 driftStrikes: number;31 symbols: string[];32 stateCache: Record<string, unknown>;33 stateDirty: boolean;34}3536/**37 * Runs connectors: lifecycle, adaptive polling, per-host rate limits, circuit breaker with38 * jittered backoff, schema-drift detection, L0 archiving and hand-off to the pipeline.39 */40export class ConnectorManager {41 private runtimes = new Map<string, Runtime>();42 readonly limiter = new RateLimiter({ ratePerSec: 2, burst: 4 });43 readonly http = new HttpClient({44 userAgent: config.userAgent,45 limiter: this.limiter,46 onRequest: (i) => {47 telemetry.inc("http_requests_total", 1, { host: i.host, status: String(i.status) });48 telemetry.observe("http_request_ms", i.durationMs, { host: i.host });49 },50 });5152 list(): Runtime[] {53 return [...this.runtimes.values()];54 }5556 get(id: string): Runtime | undefined {57 return this.runtimes.get(id);58 }5960 async register(def: ConnectorDefinition, opts: { paused: boolean; state: Record<string, unknown>; fingerprints: Record<string, string[]>; symbols?: string[] }): Promise<void> {61 const id = def.metadata.id;62 for (const [host, rate] of Object.entries(def.rateLimits ?? {})) this.limiter.configure(host, rate);63 health.register(id, def.metadata.sourceId);64 health.loadFingerprints(id, opts.fingerprints);65 pipeline.registerSource(def.metadata.sourceId, {66 family: def.metadata.sourceFamily,67 isOfficial: def.metadata.rightsStatus === "OFFICIAL_OPEN_DATA",68 realtimeStatus: def.metadata.realtimeStatus,69 });70 const rt: Runtime = {71 def,72 ctx: null as unknown as ConnectorContext,73 sockets: [],74 pollTimer: null,75 running: false,76 paused: opts.paused,77 consecutiveFailures: 0,78 driftStrikes: 0,79 symbols: opts.symbols ?? def.defaultSymbols ?? [],80 stateCache: opts.state ?? {},81 stateDirty: false,82 };83 rt.ctx = this.makeContext(rt);84 this.runtimes.set(id, rt);85 }8687 private makeContext(rt: Runtime): ConnectorContext {88 const id = rt.def.metadata.id;89 const log = logger.child({ connector: id });90 return {91 connectorId: id,92 logger: {93 debug: (m, d) => log.debug(d ?? {}, m),94 info: (m, d) => log.info(d ?? {}, m),95 warn: (m, d) => log.warn(d ?? {}, m),96 error: (m, d) => log.error(d ?? {}, m),97 },98 http: this.http,99 state: {100 get: async (k) => rt.stateCache[k] as never,101 set: async (k, v) => {102 rt.stateCache[k] = v;103 rt.stateDirty = true;104 },105 },106 emit: (raw) => {107 for (const r of Array.isArray(raw) ? raw : [raw]) void this.handleRaw(rt, r, false);108 },109 openWebSocket: (url: string, o: ManagedWebSocketOptions) => {110 const ws = new ManagedWebSocket(url, {111 ...o,112 onError: (err) => {113 health.error(id, err);114 o.onError?.(err);115 },116 });117 ws.on((ev, detail) => {118 if (ev === "connected") {119 rt.consecutiveFailures = 0;120 health.setState(id, "HEALTHY");121 } else if (ev === "reconnecting") {122 health.reconnect(id);123 health.setState(id, "RECONNECTING");124 if ((detail?.attempt as number) >= 5) this.markFailed(rt, `reconnect attempt ${detail?.attempt}`);125 } else if (ev === "stale") {126 health.setState(id, "STALE");127 }128 });129 rt.sockets.push(ws);130 return ws;131 },132 isMarketOpen: (exchangeId) => calendar.isOpen(exchangeId),133 watchedSymbols: () => rt.symbols,134 reportError: (err, context) => {135 health.error(id, err);136 log.warn({ err: err instanceof Error ? err.message : String(err), ...context }, "connector reported error");137 },138 secret: (name) => process.env[name],139 now: () => Date.now(),140 };141 }142143 async startAll(): Promise<void> {144 for (const rt of this.runtimes.values()) {145 if (rt.paused) {146 health.setState(rt.def.metadata.id, "PAUSED");147 continue;148 }149 await this.start(rt.def.metadata.id);150 }151 }152153 async start(id: string): Promise<void> {154 const rt = this.runtimes.get(id);155 if (!rt || rt.running) return;156 rt.running = true;157 rt.paused = false;158 health.setState(id, "STARTING");159 try {160 if (rt.def.start) {161 await rt.def.start(rt.ctx);162 if (!rt.sockets.length) health.setState(id, "HEALTHY");163 }164 if (rt.def.poll) this.schedulePoll(rt, 500 + Math.random() * 3000);165 } catch (err) {166 health.error(id, err);167 this.markFailed(rt, err instanceof Error ? err.message : String(err));168 this.scheduleRestart(rt);169 }170 }171172 async stop(id: string, reason = "stop"): Promise<void> {173 const rt = this.runtimes.get(id);174 if (!rt) return;175 rt.running = false;176 if (rt.pollTimer) clearTimeout(rt.pollTimer);177 rt.pollTimer = null;178 for (const s of rt.sockets) s.close();179 rt.sockets = [];180 try {181 await rt.def.stop?.(rt.ctx);182 } catch (err) {183 logger.warn({ connector: id, err }, "connector stop failed");184 }185 await this.persistState(rt);186 logger.info({ connector: id, reason }, "connector stopped");187 }188189 async pause(id: string): Promise<void> {190 const rt = this.runtimes.get(id);191 if (!rt) return;192 await this.stop(id, "pause");193 rt.paused = true;194 health.setState(id, "PAUSED");195 await pool.query("update connectors set paused = true, updated_at = now() where id = $1", [id]);196 }197198 async resume(id: string): Promise<void> {199 const rt = this.runtimes.get(id);200 if (!rt) return;201 await pool.query("update connectors set paused = false, updated_at = now() where id = $1", [id]);202 rt.consecutiveFailures = 0;203 rt.driftStrikes = 0;204 await this.start(id);205 }206207 async restart(id: string): Promise<void> {208 await this.stop(id, "restart");209 const rt = this.runtimes.get(id);210 if (rt) {211 rt.consecutiveFailures = 0;212 await this.start(id);213 }214 }215216 /** Run one poll immediately (admin "run test"); returns accepted observation count. */217 async runOnce(id: string): Promise<{ accepted: number; raws: number; durationMs: number }> {218 const rt = this.runtimes.get(id);219 if (!rt) throw new Error("unknown connector");220 const started = Date.now();221 let accepted = 0;222 let raws = 0;223 if (rt.def.poll) {224 const out = await rt.def.poll(rt.ctx);225 raws = out.length;226 for (const r of out) accepted += await this.handleRaw(rt, r, true);227 } else if (rt.def.healthCheck) {228 const res = await rt.def.healthCheck(rt.ctx);229 if (!res.ok) throw new Error(res.detail ?? "health check failed");230 }231 return { accepted, raws, durationMs: Date.now() - started };232 }233234 private schedulePoll(rt: Runtime, delayMs: number) {235 if (!rt.running) return;236 if (rt.pollTimer) clearTimeout(rt.pollTimer);237 const at = Date.now() + delayMs;238 health.poll(rt.def.metadata.id, at);239 rt.pollTimer = setTimeout(() => void this.poll(rt), delayMs);240 }241242 private nextInterval(rt: Runtime): number {243 const s = rt.def.schedule!;244 if ("intervalMs" in s) return s.intervalMs;245 if (!s.exchangeId) return s.openMs;246 const st = calendar.status(s.exchangeId);247 if (st.state === "OPEN" || st.state === "PRE" || st.state === "POST") return s.openMs;248 const day = new Date().getUTCDay();249 if (day === 0 || day === 6 || st.isHoliday) return s.weekendMs;250 return s.closedMs;251 }252253 private async poll(rt: Runtime) {254 const id = rt.def.metadata.id;255 if (!rt.running || !rt.def.poll) return;256 const started = Date.now();257 try {258 const raws = await rt.def.poll(rt.ctx);259 let accepted = 0;260 for (const r of raws) accepted += await this.handleRaw(rt, r, true);261 rt.consecutiveFailures = 0;262 if (health.state(id) !== "HEALTHY") health.setState(id, "HEALTHY");263 telemetry.observe("connector_poll_ms", Date.now() - started, { connector: id });264 logger.debug({ connector: id, raws: raws.length, accepted, ms: Date.now() - started }, "poll done");265 await this.persistState(rt);266 this.schedulePoll(rt, this.nextInterval(rt));267 } catch (err) {268 rt.consecutiveFailures++;269 health.error(id, err);270 // Retry sooner than the regular cadence (a daily connector must not wait 24 h after a transient network error).271 const wait = Math.max(5000, Math.min(this.nextInterval(rt), backoffMs(rt.consecutiveFailures, 15_000, 30 * 60_000)));272 logger.warn({ connector: id, err: err instanceof Error ? err.message : String(err), failures: rt.consecutiveFailures, retryInMs: wait }, "poll failed");273 if (rt.consecutiveFailures >= 3) this.markFailed(rt, err instanceof Error ? err.message : String(err));274 else health.setState(id, "DEGRADED");275 this.schedulePoll(rt, wait);276 }277 }278279 private markFailed(rt: Runtime, reason: string) {280 const id = rt.def.metadata.id;281 if (health.state(id) === "FAILED") return;282 health.setState(id, "FAILED");283 eventEngine.system("SOURCE_FAILURE", `${id}:${Math.floor(Date.now() / 3_600_000)}`, Date.now(), "WARNING", `Source failure: ${rt.def.metadata.name}`, { connectorId: id, reason: reason.slice(0, 200) }, [rt.def.metadata.sourceId]);284 }285286 private scheduleRestart(rt: Runtime) {287 const wait = backoffMs(Math.min(8, rt.consecutiveFailures + 1), 10_000, 15 * 60_000);288 rt.consecutiveFailures++;289 setTimeout(() => {290 if (!rt.paused) {291 rt.running = false;292 void this.start(rt.def.metadata.id);293 }294 }, wait);295 }296297 /** Every raw payload: fingerprint, archive, normalize, hand to the pipeline. Returns accepted count. */298 private async handleRaw(rt: Runtime, raw: RawObservation, forceArchive: boolean): Promise<number> {299 const id = rt.def.metadata.id;300 telemetry.inc("raw_total", 1, { connector: id });301 const fp = schemaFingerprint(raw.payload);302 const status = health.recordFingerprint(id, raw.kind, fp);303 let batch;304 try {305 batch = rt.def.normalize(raw);306 health.parse(id, true);307 } catch (err) {308 health.parse(id, false);309 health.error(id, err);310 rt.driftStrikes++;311 this.recordSchemaChange(rt, raw, fp, status).catch(() => {});312 if (rt.driftStrikes >= 20) this.driftPause(rt, raw.kind);313 return 0;314 }315 const produced = batch.observations.length + (batch.events?.length ?? 0) + (batch.instruments?.length ?? 0) + (batch.filings?.length ?? 0) + (batch.holidays?.length ?? 0);316 if (status === "new") {317 await this.recordSchemaChange(rt, raw, fp, status);318 if (produced === 0) {319 rt.driftStrikes++;320 if (rt.driftStrikes >= 5) this.driftPause(rt, raw.kind);321 }322 }323 if (produced > 0) rt.driftStrikes = Math.max(0, rt.driftStrikes - 1);324 const rawRef = rawArchive.store(raw, forceArchive || rt.def.metadata.sourceType !== "WEBSOCKET");325 const st = health.state(id);326 if (st === "STALE" || st === "DEGRADED" || st === "STARTING" || st === "RECONNECTING") health.setState(id, "HEALTHY");327 return pipeline.process(rt.def, raw, batch, rawRef);328 }329330 private async recordSchemaChange(rt: Runtime, raw: RawObservation, fp: string, status: "known" | "new" | "first") {331 if (status === "known") return;332 const id = rt.def.metadata.id;333 const known = health.fingerprints(id)[raw.kind] ?? [];334 await pool.query(335 `insert into connector_schema_changes (connector_id, kind, old_fingerprint, new_fingerprint, sample) values ($1,$2,$3,$4,$5)`,336 [id, raw.kind, known.filter((k) => k !== fp).at(-1) ?? null, fp, JSON.stringify(truncate(raw.payload))],337 );338 if (status === "new") {339 telemetry.inc("schema_changes_total", 1, { connector: id });340 logger.warn({ connector: id, kind: raw.kind, fingerprint: fp }, "schema change detected");341 }342 }343344 private driftPause(rt: Runtime, kind: string) {345 const id = rt.def.metadata.id;346 logger.error({ connector: id, kind }, "schema drift: pausing connector to avoid producing wrong values");347 eventEngine.system("SCHEMA_DRIFT", `${id}:${kind}:${Math.floor(Date.now() / 3_600_000)}`, Date.now(), "CRITICAL", `Schema drift: ${rt.def.metadata.name}`, { connectorId: id, kind }, [rt.def.metadata.sourceId]);348 void this.pause(id);349 }350351 private async persistState(rt: Runtime) {352 if (!rt.stateDirty) return;353 rt.stateDirty = false;354 try {355 await pool.query("update connectors set state = $2, updated_at = now() where id = $1", [rt.def.metadata.id, JSON.stringify(rt.stateCache)]);356 } catch (err) {357 rt.stateDirty = true;358 logger.warn({ connector: rt.def.metadata.id, err }, "state persist failed");359 }360 }361362 async stopAll(): Promise<void> {363 await Promise.all([...this.runtimes.keys()].map((id) => this.stop(id, "shutdown")));364 }365}366367function truncate(v: unknown): unknown {368 const s = JSON.stringify(v);369 if (!s) return null;370 return s.length > 4000 ? { truncated: s.slice(0, 4000) } : v;371}372373export const connectorManager = new ConnectorManager();374