import type { Bar, Instrument, NormalizedObservation, Observation, RawObservation } from "@market-atlas/market-model";
import { CANONICAL_VERSIONS, NormalizedObservationSchema, PUBLICLY_REDISTRIBUTABLE } from "@market-atlas/market-model";
import { observationFingerprint, type ConnectorDefinition, type NormalizedBatch } from "@market-atlas/connector-sdk";
import { config } from "../config.js";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { barAggregator, upsertBars } from "./bars.js";
import { bus } from "./bus.js";
import { calendar } from "./calendar.js";
import { ConsensusEngine } from "./consensus.js";
import { eventEngine } from "./events.js";
import { health } from "./health.js";
import { instruments } from "./instruments.js";
import { observationWriter } from "./observations.js";
import { quoteStore } from "./quotes.js";
import { telemetry } from "./telemetry.js";
import { lineage } from "./lineage.js";

export interface SourceInfo {
  family: string | null;
  isOfficial: boolean;
  realtimeStatus: Observation["realtimeStatus"];
}

/**
 * L1→L4 pipeline for one normalized batch: validate, resolve instruments, persist observations,
 * update consensus + canonical quotes, feed the event engine and the bar aggregator, and handle
 * side outputs (proposed events, instrument master updates, filings, holidays).
 */
export class Pipeline {
  readonly consensus: ConsensusEngine;
  private sourceInfo = new Map<string, SourceInfo>();
  private recomputeQueue = new Map<string, Instrument>();
  private recomputeTimer: NodeJS.Timeout | null = null;

  constructor() {
    this.consensus = new ConsensusEngine(
      (sourceId) => {
        const info = this.sourceInfo.get(sourceId);
        return {
          family: lineage.familyOf(sourceId) ?? info?.family ?? null,
          reliability: health.sourceReliability(sourceId),
          realtimeStatus: info?.realtimeStatus ?? "UNKNOWN",
          isOfficial: info?.isOfficial ?? false,
        };
      },
      (instrumentId) => {
        const inst = instruments.get(instrumentId);
        const ex = inst?.exchangeId ? calendar.exchange(inst.exchangeId) : undefined;
        if (!ex || ex.sessions.continuous) return null;
        return calendar.state(ex.id) === "OPEN";
      },
      (instrumentId) => instruments.get(instrumentId)?.assetClass ?? null,
    );
    bus.subscribe("canonical.quote", (q) => barAggregator.onQuote(q));
  }

  registerSource(sourceId: string, info: SourceInfo) {
    this.sourceInfo.set(sourceId, info);
  }

  async process(def: ConnectorDefinition, raw: RawObservation, batch: NormalizedBatch, rawRef: string | null): Promise<number> {
    const connectorId = def.metadata.id;
    const sourceId = def.metadata.sourceId;
    let accepted = 0;
    const touched = new Map<string, Instrument>();
    const ids: string[] = [];

    if (batch.instruments?.length) await this.applyInstruments(batch, sourceId);

    for (const n of batch.observations) {
      const parsed = NormalizedObservationSchema.safeParse(n);
      if (!parsed.success) {
        telemetry.inc("observations_invalid_total", 1, { connector: connectorId });
        continue;
      }
      const inst = n.instrumentId ? instruments.get(n.instrumentId) : await instruments.resolveOrCreate(n.symbol, sourceId, n.instrumentHint);
      if (!inst) {
        telemetry.inc("observations_unresolved_total", 1, { connector: connectorId });
        continue;
      }
      if (n.instrumentId && !instruments.resolve(n.symbol, sourceId)) await instruments.addAlias(n.symbol, sourceId, inst.id);
      if (!plausible(n, inst)) {
        telemetry.inc("observations_rejected_total", 1, { connector: connectorId, reason: "implausible" });
        continue;
      }
      const obs = this.toObservation(n, inst, raw, rawRef);
      if (this.shouldPersist(obs)) observationWriter.enqueue(obs);
      else telemetry.inc("observations_sampled_out_total");
      this.consensus.ingest(obs);
      lineage.observe(obs);
      bus.publish("normalized.observation", obs);
      touched.set(inst.id, inst);
      ids.push(inst.id);
      accepted++;
    }
    telemetry.inc("observations_accepted_total", accepted);
    const latency = batch.observations.find((o) => o.sourceTimestamp)?.sourceTimestamp;
    health.message(connectorId, latency ? raw.receivedAt - latency : null, ids);
    for (const [id, inst] of touched) this.recomputeQueue.set(id, inst);
    this.scheduleRecompute();

    if (batch.events?.length) await this.applyEvents(batch, sourceId);
    if (batch.filings?.length) await this.applyFilings(batch, sourceId);
    if (batch.holidays?.length) await this.applyHolidays(batch, sourceId);
    if (batch.bars?.length) await this.applyBars(batch, def);
    return accepted;
  }

  private async applyBars(batch: NormalizedBatch, def: ConnectorDefinition) {
    const out: Bar[] = [];
    for (const b of batch.bars ?? []) {
      const inst = await instruments.resolveOrCreate(b.symbol, def.metadata.sourceId, b.instrumentHint);
      if (!inst) continue;
      if (!(b.high >= b.low && b.open > 0 && b.close > 0)) continue;
      out.push({ instrumentId: inst.id, resolution: b.resolution, ts: b.ts, open: b.open, high: b.high, low: b.low, close: b.close, volume: b.volume, sourceCount: 1, producer: def.metadata.id, version: CANONICAL_VERSIONS.normalizer });
    }
    await upsertBars(out);
    if (out.length) telemetry.inc("bars_backfilled_total", out.length, { connector: def.metadata.id });
  }

  private lastPersisted = new Map<string, number>();

  /** REALTIME streaming ticks are stored at most once per (source, instrument, field) per interval; everything else always. */
  private shouldPersist(o: Observation): boolean {
    if (o.realtimeStatus !== "REALTIME" || config.tickPersistIntervalMs <= 0) return true;
    const key = `${o.sourceId}|${o.instrumentId}|${o.field}`;
    const last = this.lastPersisted.get(key) ?? 0;
    if (o.receivedAt - last < config.tickPersistIntervalMs) return false;
    this.lastPersisted.set(key, o.receivedAt);
    if (this.lastPersisted.size > 200_000) this.lastPersisted.clear();
    return true;
  }

  private toObservation(n: NormalizedObservation, inst: Instrument, raw: RawObservation, rawRef: string | null): Observation {
    const fp = observationFingerprint({
      sourceId: raw.sourceId,
      instrumentId: inst.id,
      field: n.field,
      value: n.value,
      sourceTimestamp: n.sourceTimestamp,
      sequence: n.sequence ?? null,
    });
    return {
      observationId: fp,
      instrumentId: inst.id,
      symbol: n.symbol,
      field: n.field,
      value: n.value,
      currency: n.currency ?? inst.currency ?? null,
      observationType: n.observationType,
      sourceTimestamp: n.sourceTimestamp,
      timestampTrust: n.timestampTrust,
      sequence: n.sequence ?? null,
      rightsStatus: n.rightsStatus,
      realtimeStatus: n.realtimeStatus,
      confidence: n.confidence,
      meta: n.meta,
      sourceId: raw.sourceId,
      connectorId: raw.connectorId,
      receivedAt: raw.receivedAt,
      latencyMs: n.sourceTimestamp == null ? null : Math.max(0, raw.receivedAt - n.sourceTimestamp),
      rawRef,
      normalizerVersion: CANONICAL_VERSIONS.normalizer,
    };
  }

  /** Recompute canonical quotes at most ~10×/s per instrument (UI batching happens downstream too). */
  private scheduleRecompute() {
    if (this.recomputeTimer) return;
    this.recomputeTimer = setTimeout(() => {
      this.recomputeTimer = null;
      const batch = [...this.recomputeQueue.values()];
      this.recomputeQueue.clear();
      const now = Date.now();
      for (const inst of batch) {
        const q = this.consensus.compute(inst.id, inst.symbol, now);
        if (!q) continue;
        // Public redistribution guard: values built only from restricted sources stay internal.
        quoteStore.update(q);
        if (PUBLICLY_REDISTRIBUTABLE.has(q.rightsStatus)) eventEngine.onQuote(q, inst.name);
      }
      telemetry.gauge("consensus_instruments", this.consensus.instrumentsTracked());
    }, 100);
  }

  /** Recompute every tracked instrument (used by the freshness sweeper so stale quotes decay). */
  sweep(): void {
    for (const q of quoteStore.all()) {
      const inst = instruments.get(q.instrumentId);
      if (inst) this.recomputeQueue.set(inst.id, inst);
    }
    this.scheduleRecompute();
  }

  private async applyInstruments(batch: NormalizedBatch, sourceId: string) {
    let created = 0;
    for (const p of batch.instruments ?? []) {
      const before = instruments.count();
      const existing = instruments.resolve(p.symbol, sourceId);
      if (!existing && p.createIfMissing === false) continue;
      const inst = existing ?? (await instruments.resolveOrCreate(p.symbol, sourceId, p.hint));
      if (!inst) continue;
      if (instruments.count() > before) created++;
      for (const a of p.aliases ?? []) await instruments.addAlias(a, "*", inst.id);
      // Enrichment of an existing instrument: company/CIK, name, exchange when previously unknown.
      if (existing) {
        let companyId = existing.companyId;
        if (!companyId && p.hint.companyName) companyId = await instruments.upsertCompany({ name: p.hint.companyName, cik: p.hint.cik ?? null, country: p.hint.country ?? null });
        else if (companyId && p.hint.cik) await instruments.upsertCompany({ name: p.hint.companyName ?? existing.name, cik: p.hint.cik, country: p.hint.country ?? null });
        const patch: Partial<Instrument> = {};
        if (companyId !== existing.companyId) patch.companyId = companyId;
        if (!existing.exchangeId && p.hint.exchangeId) patch.exchangeId = p.hint.exchangeId;
        if (!existing.mic && p.hint.mic) patch.mic = p.hint.mic;
        if (existing.name === existing.symbol && p.hint.name) patch.name = p.hint.name;
        if (p.isActive === false && existing.isActive) patch.isActive = false;
        if (Object.keys(patch).length) await instruments.upsert({ ...existing, ...patch });
      }
    }
    if (created) logger.info({ source: sourceId, created, proposed: batch.instruments?.length }, "instrument master updated");
  }

  private async applyEvents(batch: NormalizedBatch, sourceId: string) {
    for (const e of batch.events ?? []) {
      const ids = new Set<string>(e.instrumentIds ?? []);
      for (const s of e.symbols ?? []) {
        const inst = instruments.resolve(s, sourceId) ?? instruments.bySymbolCandidates(s).find((i) => i.assetClass === "EQUITY" || i.assetClass === "ETF");
        if (inst) ids.add(inst.id);
      }
      eventEngine.proposed(e, sourceId, [...ids]);
    }
  }

  private async applyFilings(batch: NormalizedBatch, sourceId: string) {
    let inserted = 0;
    for (const f of batch.filings ?? []) {
      const ids: string[] = [];
      if (f.cik) {
        const r = await pool.query<{ id: string }>("select i.id from instruments i join companies c on c.id = i.company_id where c.cik = $1 and i.is_active limit 5", [f.cik.replace(/^0+/, "")]);
        ids.push(...r.rows.map((x) => x.id));
      }
      for (const s of f.symbols ?? []) {
        const inst = instruments.bySymbolCandidates(s)[0];
        if (inst) ids.push(inst.id);
      }
      const res = await pool.query(
        `insert into filings (id, source_id, cik, company_name, form_type, filed_at, url, instrument_ids, metadata) values ($1,$2,$3,$4,$5,$6,$7,$8,$9) on conflict (id) do nothing`,
        [f.id, sourceId, f.cik, f.companyName, f.formType, new Date(f.filedAt).toISOString(), f.url, ids, JSON.stringify(f.metadata ?? {})],
      );
      if (res.rowCount) {
        inserted++;
        if (f.metadata?.material === false) continue;
        eventEngine.proposed(
          {
            type: "FILING_PUBLISHED",
            timestamp: f.filedAt,
            title: `${f.companyName}: ${f.formType} filing`,
            summary: null,
            severity: "INFO",
            confidence: 1,
            dedupeKey: `filing:${f.id}`,
            data: { formType: f.formType, cik: f.cik, url: f.url, companyName: f.companyName, filingId: f.id },
          },
          sourceId,
          ids,
        );
      }
    }
    if (inserted) {
      await pool.query(`insert into daily_stats (day, filings) values (current_date, $1) on conflict (day) do update set filings = daily_stats.filings + excluded.filings, updated_at = now()`, [inserted]);
      telemetry.inc("filings_total", inserted);
    }
  }

  private async applyHolidays(batch: NormalizedBatch, sourceId: string) {
    for (const h of batch.holidays ?? []) {
      await pool.query(
        `insert into exchange_holidays (exchange_id, date, name, kind, close_time, source_id) values ($1,$2,$3,$4,$5,$6)
         on conflict (exchange_id, date) do update set name = excluded.name, kind = excluded.kind, close_time = excluded.close_time, source_id = excluded.source_id`,
        [h.exchangeId, h.date, h.name, h.kind, h.closeTime ?? null, sourceId],
      );
      calendar.addHoliday({ ...h, sourceId });
    }
  }
}

/** Reject values that cannot be right for the field/instrument (guards against schema drift producing garbage). */
function plausible(n: NormalizedObservation, inst: Instrument): boolean {
  if (!Number.isFinite(n.value)) return false;
  switch (n.field) {
    case "LAST_PRICE":
    case "OPEN":
    case "HIGH":
    case "LOW":
    case "PREVIOUS_CLOSE":
    case "CLOSE":
    case "BID":
    case "ASK":
    case "VWAP":
    case "NAV":
      if (n.value <= 0 && inst.assetClass !== "INTEREST_RATE" && inst.assetClass !== "TREASURY") return false;
      if (n.value > 1e9) return false;
      return true;
    case "VOLUME":
    case "BID_SIZE":
    case "ASK_SIZE":
    case "OPEN_INTEREST":
    case "MARKET_CAP":
      return n.value >= 0;
    case "CHANGE_PERCENT":
      return Math.abs(n.value) < 1000;
    case "YIELD":
    case "RATE":
      return n.value > -20 && n.value < 200;
    default:
      return true;
  }
}

export const pipeline = new Pipeline();