import type { Company, Exchange, Instrument, InstrumentHint, SymbolAlias } from "@market-atlas/market-model";
import { canonicalSymbol, makeCompanyId, makeInstrumentId } from "@market-atlas/market-model";
import { pool, query, type Queryable } from "../db/pool.js";
import { logger } from "../logger.js";
import { telemetry } from "./telemetry.js";
import { masterMutex } from "./mutex.js";

interface Row {
  id: string;
  symbol: string;
  name: string;
  asset_class: Instrument["assetClass"];
  exchange_id: string | null;
  mic: string | null;
  currency: string | null;
  country: string | null;
  company_id: string | null;
  isin: string | null;
  security_type: string | null;
  base: string | null;
  quote: string | null;
  is_active: boolean;
  metadata: Record<string, unknown>;
}

export const rowToInstrument = (r: Row): Instrument => ({
  id: r.id,
  symbol: r.symbol,
  name: r.name,
  assetClass: r.asset_class,
  exchangeId: r.exchange_id,
  mic: r.mic,
  currency: r.currency,
  country: r.country,
  companyId: r.company_id,
  isin: r.isin,
  securityType: r.security_type,
  base: r.base,
  quote: r.quote,
  isActive: r.is_active,
  metadata: r.metadata ?? {},
});

/**
 * Instrument master + resolver. Keeps an in-memory index (id, alias per source, canonical symbol)
 * so that the hot path never queries Postgres per tick.
 */
export class InstrumentStore {
  private byId = new Map<string, Instrument>();
  private byAlias = new Map<string, string>(); // `${sourceId}|${canonicalAlias}` or `*|alias` -> id
  private bySymbol = new Map<string, string[]>(); // canonical symbol -> ids
  private exchangesById = new Map<string, Exchange>();
  private creating = new Map<string, Promise<Instrument>>();

  async load(): Promise<void> {
    const [inst, aliases, ex] = await Promise.all([
      query<Row>("select * from instruments"),
      query<{ alias: string; source_id: string; instrument_id: string }>("select alias, source_id, instrument_id from symbol_aliases"),
      query<any>("select * from exchanges"),
    ]);
    this.byId.clear();
    this.byAlias.clear();
    this.bySymbol.clear();
    for (const r of inst.rows) this.index(rowToInstrument(r));
    for (const a of aliases.rows) this.byAlias.set(`${a.source_id}|${canonicalSymbol(a.alias)}`, a.instrument_id);
    this.exchangesById.clear();
    for (const e of ex.rows) this.exchangesById.set(e.id, rowToExchange(e));
    telemetry.gauge("instruments_total", this.byId.size);
    logger.info({ instruments: this.byId.size, aliases: this.byAlias.size, exchanges: this.exchangesById.size }, "instrument master loaded");
  }

  private index(i: Instrument) {
    this.byId.set(i.id, i);
    const cs = canonicalSymbol(i.symbol);
    const arr = this.bySymbol.get(cs) ?? [];
    if (!arr.includes(i.id)) arr.push(i.id);
    this.bySymbol.set(cs, arr);
    this.byAlias.set(`*|${cs}`, i.id);
  }

  get(id: string): Instrument | undefined {
    return this.byId.get(id);
  }

  all(): Instrument[] {
    return [...this.byId.values()];
  }

  count(): number {
    return this.byId.size;
  }

  exchange(id: string | null | undefined): Exchange | undefined {
    return id ? this.exchangesById.get(id) : undefined;
  }

  exchanges(): Exchange[] {
    return [...this.exchangesById.values()];
  }

  /** Resolve a source-native symbol. Source-specific aliases win over global ones. */
  resolve(symbol: string, sourceId: string): Instrument | undefined {
    const cs = canonicalSymbol(symbol);
    const id = this.byAlias.get(`${sourceId}|${cs}`) ?? this.byAlias.get(`*|${cs}`);
    return id ? this.byId.get(id) : undefined;
  }

  /** Ids sharing a canonical symbol (e.g. AAPL on several venues). */
  bySymbolCandidates(symbol: string): Instrument[] {
    return (this.bySymbol.get(canonicalSymbol(symbol)) ?? []).map((id) => this.byId.get(id)!).filter(Boolean);
  }

  /** Resolve, or create from a hint (deterministic id). Concurrent creations of the same id are coalesced. */
  async resolveOrCreate(symbol: string, sourceId: string, hint: InstrumentHint | undefined): Promise<Instrument | undefined> {
    const found = this.resolve(symbol, sourceId);
    if (found) return found;
    if (!hint) return undefined;
    const id = makeInstrumentId(hint.assetClass, { symbol, ...hint });
    const existing = this.byId.get(id);
    if (existing) {
      await this.addAlias(symbol, sourceId, id);
      return existing;
    }
    let p = this.creating.get(id);
    if (!p) {
      p = this.create(id, symbol, sourceId, hint).finally(() => this.creating.delete(id));
      this.creating.set(id, p);
    }
    return p;
  }

  private async create(id: string, symbol: string, sourceId: string, hint: InstrumentHint): Promise<Instrument> {
    let companyId: string | null = null;
    if (hint.companyName) companyId = await this.upsertCompany({ name: hint.companyName, cik: hint.cik ?? null, country: hint.country ?? null });
    const inst: Instrument = {
      id,
      symbol: symbol.replace(/^[_^]/, ""),
      name: hint.name ?? symbol,
      assetClass: hint.assetClass,
      exchangeId: hint.exchangeId ?? null,
      mic: hint.mic ?? null,
      currency: hint.currency ?? null,
      country: hint.country ?? null,
      companyId,
      isin: null,
      securityType: hint.securityType ?? null,
      base: hint.base ?? null,
      quote: hint.quote ?? null,
      isActive: true,
      metadata: hint.metadata ?? {},
    };
    await this.upsert(inst);
    await this.addAlias(symbol, sourceId, id);
    telemetry.inc("instruments_created_total");
    return inst;
  }

  async upsert(i: Instrument, client: Queryable = pool): Promise<void> {
    await masterMutex.run(() => query(
      `insert into instruments (id, symbol, name, asset_class, exchange_id, mic, currency, country, company_id, isin, security_type, base, quote, is_active, metadata)
       values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
       on conflict (id) do update set
         name = case when excluded.name <> excluded.symbol then excluded.name else instruments.name end,
         exchange_id = coalesce(excluded.exchange_id, instruments.exchange_id),
         mic = coalesce(excluded.mic, instruments.mic),
         currency = coalesce(excluded.currency, instruments.currency),
         country = coalesce(excluded.country, instruments.country),
         company_id = coalesce(excluded.company_id, instruments.company_id),
         isin = coalesce(excluded.isin, instruments.isin),
         security_type = coalesce(excluded.security_type, instruments.security_type),
         is_active = excluded.is_active,
         metadata = instruments.metadata || excluded.metadata,
         updated_at = now()`,
      [i.id, i.symbol, i.name, i.assetClass, i.exchangeId, i.mic, i.currency, i.country, i.companyId, i.isin, i.securityType, i.base, i.quote, i.isActive, JSON.stringify(i.metadata ?? {})],
      client,
    ));
    const merged = { ...(this.byId.get(i.id) ?? i), ...stripNulls(i) } as Instrument;
    this.index(merged);
    telemetry.gauge("instruments_total", this.byId.size);
  }

  async addAlias(alias: string, sourceId: string, instrumentId: string): Promise<void> {
    const key = `${sourceId}|${canonicalSymbol(alias)}`;
    if (this.byAlias.get(key) === instrumentId) return;
    this.byAlias.set(key, instrumentId);
    await masterMutex.run(() =>
      query(
        `insert into symbol_aliases (alias, source_id, instrument_id) values ($1,$2,$3)
         on conflict (alias, source_id) do update set instrument_id = excluded.instrument_id`,
        [alias, sourceId, instrumentId],
      ),
    );
  }

  aliasesOf(instrumentId: string): SymbolAlias[] {
    const out: SymbolAlias[] = [];
    for (const [k, v] of this.byAlias) {
      if (v !== instrumentId) continue;
      const [sourceId, alias] = k.split("|");
      out.push({ alias: alias!, sourceId: sourceId === "*" ? null : sourceId!, instrumentId });
    }
    return out;
  }

  async upsertCompany(c: { name: string; cik: string | null; country: string | null; sector?: string | null; website?: string | null }): Promise<string> {
    const id = makeCompanyId(c.name);
    await masterMutex.run(() => query(
      `insert into companies (id, name, cik, country, sector, website) values ($1,$2,$3,$4,$5,$6)
       on conflict (id) do update set cik = coalesce(excluded.cik, companies.cik), country = coalesce(excluded.country, companies.country),
         sector = coalesce(excluded.sector, companies.sector), website = coalesce(excluded.website, companies.website), updated_at = now()`,
      [id, c.name, c.cik, c.country, c.sector ?? null, c.website ?? null],
    ));
    return id;
  }

  async getCompany(id: string): Promise<Company | null> {
    const r = await query<any>("select * from companies where id = $1", [id]);
    const c = r.rows[0];
    return c ? { id: c.id, name: c.name, cik: c.cik, country: c.country, sector: c.sector, industry: c.industry, website: c.website } : null;
  }
}

function stripNulls<T extends object>(o: T): Partial<T> {
  return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== null && v !== undefined)) as Partial<T>;
}

export function rowToExchange(e: any): Exchange {
  return {
    id: e.id,
    mic: e.mic,
    name: e.name,
    operator: e.operator,
    country: e.country,
    city: e.city,
    timezone: e.timezone,
    currency: e.currency,
    website: e.website,
    sessions: e.sessions,
    lat: e.lat,
    lon: e.lon,
    assetClasses: e.asset_classes ?? [],
  };
}

export const instruments = new InstrumentStore();