import type { AssetClass, CanonicalQuote, ComparabilityClass, Observation, ObservationType, RealtimeStatus, RightsStatus, SourceContribution } from "@market-atlas/market-model";
import { CANONICAL_VERSIONS, COMPARABILITY, PROXY_TYPES, PUBLICLY_REDISTRIBUTABLE } from "@market-atlas/market-model";
import { config } from "../config.js";

export interface SourceProfile {
  family: string | null;
  reliability: number; // 0..1 operational reliability (from health engine)
  realtimeStatus: RealtimeStatus;
  isOfficial: boolean;
}

interface Latest {
  obs: Observation;
}

/** Fresh window per realtime class: realtime ticks age out fast, EOD values live for days. */
const FRESH_WINDOW_MS: Record<RealtimeStatus, number> = {
  REALTIME: config.consensusFreshWindowMs,
  DELAYED: 30 * 60_000,
  INDICATIVE: 60 * 60_000,
  END_OF_DAY: 10 * 86_400_000, // daily/weekly-refreshed official series stay "last close" over long weekends and weekly lake reloads
  STALE: 0,
  UNKNOWN: 60_000,
};

const REALTIME_RANK: Record<RealtimeStatus, number> = { REALTIME: 4, DELAYED: 3, INDICATIVE: 2, END_OF_DAY: 1, UNKNOWN: 0, STALE: -1 };
/** Same-class observations must be close in time to be compared: live within 2 minutes, fixings/closes on the same UTC day. */
const LIVE_TOLERANCE_MS = 120_000;

/** Infer the observation type when a connector did not declare one. */
export function inferObservationType(o: Pick<Observation, "realtimeStatus" | "observationType" | "field">): ObservationType {
  if (o.observationType) return o.observationType;
  if (o.field === "PREVIOUS_CLOSE") return "EOD_CLOSE";
  switch (o.realtimeStatus) {
    case "END_OF_DAY":
      return "EOD_CLOSE";
    case "INDICATIVE":
      return "INDICATIVE";
    default:
      return "TRADE";
  }
}

interface Candidate {
  value: number;
  weight: number;
  family: string;
  obs: Observation;
  type: ObservationType;
  proxy: boolean;
  validator: boolean;
  contrib: SourceContribution;
}

/**
 * Per-instrument consensus. Keeps the latest observation of each (source, field), keeps only
 * *comparable* observations (same class — live / official fix / end-of-day — and close in time),
 * computes a weighted median (one vote per upstream family, liquidity-aware for crypto), rejects
 * outliers, uses restricted-rights sources and proxies as confirmations only, and exposes the full
 * contribution list ("Why this price?").
 */
export class ConsensusEngine {
  private latest = new Map<string, Map<string, Latest>>(); // instrumentId -> `${sourceId}|${field}` -> latest
  private session = new Map<string, { high: number; low: number; startedAt: number }>();

  /**
   * @param profiles     source reliability/family lookup
   * @param marketOpen   true/false when the instrument's venue has a session calendar, null for continuous or unknown venues
   * @param assetClassOf asset class of an instrument (liquidity weighting for crypto, official dominance for rates)
   */
  constructor(
    private profiles: (sourceId: string) => SourceProfile,
    private marketOpen: (instrumentId: string) => boolean | null = () => null,
    private assetClassOf: (instrumentId: string) => AssetClass | null = () => null,
  ) {}

  ingest(o: Observation): void {
    let m = this.latest.get(o.instrumentId);
    if (!m) {
      m = new Map();
      this.latest.set(o.instrumentId, m);
    }
    const key = `${o.sourceId}|${o.field}`;
    const prev = m.get(key);
    // Ordering: prefer source timestamp, fall back to receive time. Ignore out-of-order values.
    if (prev) {
      const pt = prev.obs.sourceTimestamp ?? prev.obs.receivedAt;
      const nt = o.sourceTimestamp ?? o.receivedAt;
      if (nt < pt) return;
    }
    m.set(key, { obs: o });
  }

  resetSession(instrumentId: string, at: number) {
    this.session.set(instrumentId, { high: -Infinity, low: Infinity, startedAt: at });
  }

  sessionOf(instrumentId: string) {
    return this.session.get(instrumentId);
  }

  instrumentsTracked(): number {
    return this.latest.size;
  }

  sourcesFor(instrumentId: string): string[] {
    const m = this.latest.get(instrumentId);
    if (!m) return [];
    return [...new Set([...m.keys()].map((k) => k.split("|")[0]!))];
  }

  /** Latest LAST_PRICE observation per source (for lineage inference and coverage). */
  latestPrices(instrumentId: string): Observation[] {
    const m = this.latest.get(instrumentId);
    if (!m) return [];
    return [...m.entries()].filter(([k]) => k.endsWith("|LAST_PRICE")).map(([, l]) => l.obs);
  }

  compute(instrumentId: string, symbol: string, now = Date.now()): CanonicalQuote | null {
    const m = this.latest.get(instrumentId);
    if (!m) return null;
    const closed = this.marketOpen(instrumentId) === false;
    const assetClass = this.assetClassOf(instrumentId);
    const price = this.aggregate(m, "LAST_PRICE", now, closed, assetClass);
    const pick = (field: string) => this.aggregate(m, field, now, closed, assetClass);
    const open = pick("OPEN");
    const high = pick("HIGH");
    const low = pick("LOW");
    const prevClose = pick("PREVIOUS_CLOSE");
    const volume = pick("VOLUME");
    const bid = pick("BID");
    const ask = pick("ASK");
    const changeObs = pick("CHANGE");
    const changePctObs = pick("CHANGE_PERCENT");
    const p = price.value;
    let change = changeObs.value;
    let changePercent = changePctObs.value;
    if (p != null && prevClose.value != null && prevClose.value !== 0) {
      change = p - prevClose.value;
      changePercent = (change / prevClose.value) * 100;
    } else if (p != null && change == null && open.value != null && open.value !== 0) {
      // Continuous markets (crypto) publish a rolling 24 h open instead of a previous close.
      change = p - open.value;
      changePercent = (change / open.value) * 100;
    }
    let sess = this.session.get(instrumentId);
    if (p != null) {
      if (!sess) {
        sess = { high: p, low: p, startedAt: now };
        this.session.set(instrumentId, sess);
      } else {
        if (p > sess.high) sess.high = p;
        if (p < sess.low) sess.low = p;
      }
    }
    const all = [...m.values()].map((l) => l.obs);
    const currency = all.find((o) => o.currency)?.currency ?? null;
    const anyNonPublic = price.contributions.some((c) => c.included && !PUBLICLY_REDISTRIBUTABLE.has(c.rightsStatus));
    const rightsBasis = price.contributions.find((c) => c.included) ?? [...price.contributions].sort((a, b) => a.ageMs - b.ageMs)[0];
    const rights: RightsStatus = anyNonPublic ? "INTERNAL_ONLY" : (rightsBasis?.rightsStatus ?? "UNKNOWN");
    return {
      instrumentId,
      symbol,
      price: p,
      open: open.value,
      high: high.value ?? (sess && Number.isFinite(sess.high) ? sess.high : null),
      low: low.value ?? (sess && Number.isFinite(sess.low) ? sess.low : null),
      previousClose: prevClose.value,
      change: change ?? null,
      changePercent: changePercent ?? null,
      volume: volume.value,
      bid: bid.value,
      ask: ask.value,
      currency,
      sourceCount: price.independentFamilies,
      observationCount: price.observationCount,
      proxyCount: price.proxyCount,
      validatorCount: price.validatorCount,
      comparability: price.comparability,
      dispersionBps: price.dispersionBps,
      confidence: price.confidence,
      freshnessMs: price.freshnessMs,
      realtimeStatus: price.realtimeStatus,
      rightsStatus: rights,
      updatedAt: now,
      sourceTimestamp: price.sourceTimestamp,
      contributions: price.contributions,
      consensusVersion: CANONICAL_VERSIONS.consensus,
      sessionHigh: sess && Number.isFinite(sess.high) ? sess.high : null,
      sessionLow: sess && Number.isFinite(sess.low) ? sess.low : null,
    };
  }

  private aggregate(m: Map<string, Latest>, field: string, now: number, marketClosed: boolean, assetClass: AssetClass | null) {
    const contributions: SourceContribution[] = [];
    const fresh: Candidate[] = [];
    const empty = {
      value: null as number | null,
      confidence: 0,
      dispersionBps: null as number | null,
      independentFamilies: 0,
      observationCount: 0,
      proxyCount: 0,
      validatorCount: 0,
      comparability: null as ComparabilityClass | null,
      freshnessMs: null as number | null,
      realtimeStatus: "STALE" as RealtimeStatus,
      sourceTimestamp: null as number | null,
      contributions,
    };

    for (const [key, l] of m) {
      if (!key.endsWith(`|${field}`)) continue;
      const o = l.obs;
      const prof = this.profiles(o.sourceId);
      const type = inferObservationType(o);
      const eventTs = o.sourceTimestamp ?? o.receivedAt;
      const ageMs = Math.max(0, now - eventTs);
      // While the venue is closed, the last session value is the truth for up to 4 days (long weekends).
      const window = marketClosed && (o.realtimeStatus === "REALTIME" || o.realtimeStatus === "DELAYED") ? 4 * 86_400_000 : (FRESH_WINDOW_MS[o.realtimeStatus] ?? 60_000);
      const isFresh = ageMs <= window;
      const base = 0.35 + 0.65 * prof.reliability;
      const tsQuality = o.timestampTrust === "EXCHANGE" ? 1 : o.timestampTrust === "SOURCE" ? 0.9 : 0.75;
      const rtQuality = o.realtimeStatus === "REALTIME" ? 1 : o.realtimeStatus === "DELAYED" ? 0.6 : o.realtimeStatus === "INDICATIVE" ? 0.5 : 0.4;
      const official = prof.isOfficial ? (assetClass === "TREASURY" || assetClass === "INTEREST_RATE" || assetClass === "BOND" ? 1.6 : 1.15) : 1;
      const decay = isFresh ? Math.exp(-ageMs / Math.max(1000, window)) : 0;
      const weight = base * tsQuality * rtQuality * official * (0.4 + 0.6 * decay) * (o.confidence ?? 1);
      const contrib: SourceContribution = {
        sourceId: o.sourceId,
        connectorId: o.connectorId,
        value: o.value,
        sourceTimestamp: o.sourceTimestamp,
        receivedAt: o.receivedAt,
        ageMs,
        weight: 0,
        included: false,
        observationType: type,
        deltaBps: null,
        realtimeStatus: isFresh ? o.realtimeStatus : "STALE",
        rightsStatus: o.rightsStatus,
        ...(isFresh ? {} : { reason: "stale" }),
      };
      contributions.push(contrib);
      if (isFresh) fresh.push({ value: o.value, weight, family: prof.family ?? o.sourceId, obs: o, type, proxy: PROXY_TYPES.has(type), validator: !PUBLICLY_REDISTRIBUTABLE.has(o.rightsStatus), contrib });
    }
    if (!fresh.length) {
      // Fall back to the most recent stale value so the UI can show "last known", clearly labelled STALE.
      const newest = [...contributions].sort((a, b) => a.ageMs - b.ageMs)[0];
      return { ...empty, value: newest?.value ?? null, freshnessMs: newest?.ageMs ?? null, sourceTimestamp: newest?.sourceTimestamp ?? null, contributions: contributions.sort((a, b) => a.ageMs - b.ageMs) };
    }

    // 1. Comparability class: live values from real markets win, then official fixings, then end-of-day closes,
    //    then live proxies alone (stablecoin markets over a weekend). Other classes are reported, never mixed.
    const cls = (c: Candidate) => COMPARABILITY[c.type];
    const hasRealLive = fresh.some((c) => cls(c) === "LIVE" && !c.proxy);
    const chosen: ComparabilityClass = hasRealLive ? "LIVE" : fresh.some((c) => cls(c) === "FIX") ? "FIX" : fresh.some((c) => cls(c) === "EOD") ? "EOD" : "LIVE";
    let pool = fresh.filter((c) => cls(c) === chosen);
    for (const c of fresh) if (cls(c) !== chosen) exclude(c, "not_comparable");

    // 2. Temporal tolerance inside the class.
    const tsOf = (c: Candidate) => c.obs.sourceTimestamp ?? c.obs.receivedAt;
    const freshestTs = Math.max(...pool.map(tsOf));
    pool = pool.filter((c) => {
      const ok = chosen === "LIVE" ? freshestTs - tsOf(c) <= LIVE_TOLERANCE_MS : sameUtcDay(tsOf(c), freshestTs);
      if (!ok) exclude(c, "temporal_mismatch");
      return ok;
    });

    // 3. Roles: restricted-rights sources validate only; proxies confirm with reduced weight when real sources exist.
    const validators = pool.filter((c) => c.validator);
    for (const c of validators) exclude(c, "validation_only");
    let voters = pool.filter((c) => !c.validator);
    const realVoters = voters.filter((c) => !c.proxy);
    const proxies = voters.filter((c) => c.proxy);
    if (realVoters.length) for (const c of proxies) c.weight *= 0.35;

    // 4. One vote per upstream family (the freshest keeps its weight, others are discounted).
    const seenFamily = new Set<string>();
    for (const c of [...voters].sort((a, b) => tsOf(b) - tsOf(a))) {
      if (seenFamily.has(c.family)) c.weight *= 0.25;
      else seenFamily.add(c.family);
    }
    // 5. Crypto: liquidity-aware — a venue's weight scales with its share of the 24 h volume among voters.
    if (assetClass === "CRYPTO" && voters.length > 1) {
      const vols = new Map<string, number>();
      for (const c of voters) {
        const v = m.get(`${c.obs.sourceId}|VOLUME`)?.obs.value;
        if (v != null && v > 0) vols.set(c.obs.sourceId, v * c.value); // notional
      }
      const max = Math.max(0, ...vols.values());
      if (max > 0) for (const c of voters) c.weight *= 0.6 + 0.8 * ((vols.get(c.obs.sourceId) ?? 0) / max);
    }

    // 6. Outlier rejection (> 2 % from the weighted median of real voters, when at least 3 voters).
    const reference = realVoters.length ? realVoters : voters;
    let median = weightedMedian(reference.map((c) => ({ v: c.value, w: c.weight })));
    if (voters.length >= 3 && median !== 0) {
      const kept = voters.filter((c) => Math.abs(c.value - median) / Math.abs(median) <= 0.02);
      for (const c of voters) if (!kept.includes(c)) exclude(c, "outlier");
      if (kept.length) voters = kept;
      median = weightedMedian((kept.filter((c) => !c.proxy).length ? kept.filter((c) => !c.proxy) : kept).map((c) => ({ v: c.value, w: c.weight })));
    }
    const finalVoters = voters;
    for (const c of finalVoters) {
      c.contrib.included = true;
      c.contrib.weight = round(c.weight, 4);
    }
    // Deltas vs canonical for every fresh contribution (included or not).
    for (const c of fresh) c.contrib.deltaBps = median ? round(((c.value - median) / Math.abs(median)) * 10_000, 2) : null;

    const realIncluded = finalVoters.filter((c) => !c.proxy);
    const spreadBasis = realIncluded.length ? realIncluded : finalVoters;
    const values = spreadBasis.map((c) => c.value);
    const spread = values.length > 1 ? (Math.max(...values) - Math.min(...values)) / Math.abs(median || 1) : 0;
    const families = new Set(realIncluded.map((c) => c.family)).size;
    const proxyCount = finalVoters.filter((c) => c.proxy && Math.abs(c.contrib.deltaBps ?? 1e9) <= 50).length;
    const validatorCount = validators.filter((c) => Math.abs(c.contrib.deltaBps ?? 1e9) <= 25).length;
    for (const c of validators) if (Math.abs(c.contrib.deltaBps ?? 0) > 50) c.contrib.reason = "validator_disagrees";
    const newest = Math.min(...finalVoters.map((c) => Math.max(0, now - tsOf(c))));
    let bestRt = finalVoters.map((c) => c.obs.realtimeStatus).sort((a, b) => REALTIME_RANK[b] - REALTIME_RANK[a])[0] ?? "UNKNOWN";
    if (chosen === "LIVE" && !realIncluded.length) bestRt = "INDICATIVE"; // proxies only
    // Confidence: agreement (dispersion), redundancy (independent families + confirmations), freshness, reliability. Never claims > 0.995.
    const agreement = Math.max(0, 1 - Math.min(1, spread / 0.01)); // 1% spread → 0
    const redundancy = 1 - Math.exp(-(families + 0.5 * proxyCount + 0.5 * validatorCount) / 2);
    const freshness = marketClosed ? 1 : Math.exp(-newest / 60_000);
    const avgReliability = finalVoters.reduce((a, c) => a + this.profiles(c.obs.sourceId).reliability, 0) / finalVoters.length;
    const confidence = round(Math.min(0.995, 0.15 + 0.35 * agreement + 0.25 * redundancy + 0.1 * freshness + 0.15 * avgReliability), 3);
    return {
      value: median,
      confidence,
      dispersionBps: round(spread * 10_000, 2),
      independentFamilies: families,
      observationCount: fresh.length,
      proxyCount,
      validatorCount,
      comparability: chosen,
      freshnessMs: Math.round(newest),
      realtimeStatus: bestRt,
      sourceTimestamp: finalVoters.map((c) => c.obs.sourceTimestamp).filter((t): t is number => t != null).sort((a, b) => b - a)[0] ?? null,
      contributions: contributions.sort((a, b) => Number(b.included) - Number(a.included) || a.ageMs - b.ageMs),
    };
  }
}

function exclude(c: Candidate, reason: string) {
  c.contrib.included = false;
  c.contrib.weight = 0;
  if (!c.contrib.reason) c.contrib.reason = reason;
}

function sameUtcDay(a: number, b: number): boolean {
  return Math.floor(a / 86_400_000) === Math.floor(b / 86_400_000);
}

export function weightedMedian(items: Array<{ v: number; w: number }>): number {
  if (!items.length) return NaN;
  const sorted = [...items].sort((a, b) => a.v - b.v);
  const total = sorted.reduce((a, b) => a + Math.max(b.w, 1e-9), 0);
  let acc = 0;
  for (const it of sorted) {
    acc += Math.max(it.w, 1e-9);
    if (acc >= total / 2) return it.v;
  }
  return sorted[sorted.length - 1]!.v;
}

export const round = (n: number, d: number) => Math.round(n * 10 ** d) / 10 ** d;