spb/market-atlas

1import type { ConnectorDefinition } from "@market-atlas/connector-sdk";2import { CONNECTORS, SOURCES } from "@market-atlas/connectors";3import { config } from "../config.js";4import { pool } from "../db/pool.js";5import { logger } from "../logger.js";6import { connectorManager } from "./connector-manager.js";7import { instruments } from "./instruments.js";89/** Syncs code-defined sources/connectors into the DB (metadata is code; runtime flags are DB) and registers them. */10export async function syncRegistry(): Promise<ConnectorDefinition[]> {11  for (const s of SOURCES) {12    await pool.query(13      `insert into sources (id, name, organization, source_type, homepage, jurisdiction, rights_status, realtime_status, family, category, enabled)14       values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)15       on conflict (id) do update set name = excluded.name, organization = excluded.organization, source_type = excluded.source_type, homepage = excluded.homepage,16         jurisdiction = excluded.jurisdiction, rights_status = excluded.rights_status, realtime_status = excluded.realtime_status, family = excluded.family, category = excluded.category, updated_at = now()`,17      [s.id, s.name, s.organization, s.sourceType, s.homepage, s.jurisdiction, s.rightsStatus, s.realtimeStatus, s.family, s.category, s.enabled],18    );19  }20  const registered: ConnectorDefinition[] = [];21  for (const def of CONNECTORS) {22    const m = def.metadata;23    const enabled = m.enabled && !config.disabledConnectors.has(m.id);24    await pool.query(25      `insert into connectors (id, source_id, name, version, source_type, rights_status, realtime_status, enabled, metadata)26       values ($1,$2,$3,$4,$5,$6,$7,$8,$9)27       on conflict (id) do update set source_id = excluded.source_id, name = excluded.name, version = excluded.version, source_type = excluded.source_type,28         rights_status = excluded.rights_status, realtime_status = excluded.realtime_status, enabled = excluded.enabled, metadata = excluded.metadata, updated_at = now()`,29      [m.id, m.sourceId, m.name, m.version, m.sourceType, m.rightsStatus, m.realtimeStatus, enabled, JSON.stringify(m)],30    );31    await pool.query(`insert into connector_versions (connector_id, version, metadata) values ($1,$2,$3) on conflict do nothing`, [m.id, m.version, JSON.stringify({ description: m.description })]);32    const row = (await pool.query<{ paused: boolean; state: Record<string, unknown>; schema_fingerprints: Record<string, string[]> }>("select paused, state, schema_fingerprints from connectors where id = $1", [m.id])).rows[0]!;33    if (!enabled) {34      logger.info({ connector: m.id }, "connector disabled");35      continue;36    }37    // Seed static instruments declared by the connector.38    for (const seed of def.seeds ?? []) {39      const inst = await instruments.resolveOrCreate(seed.symbol, m.sourceId, seed.hint);40      if (inst) for (const a of seed.aliases ?? []) await instruments.addAlias(a, "*", inst.id);41    }42    await connectorManager.register(def, { paused: row.paused, state: row.state ?? {}, fingerprints: row.schema_fingerprints ?? {} });43    registered.push(def);44  }45  logger.info({ connectors: registered.length, sources: SOURCES.length }, "registry synced");46  return registered;47}48