import type { Observation } from "@market-atlas/market-model";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { telemetry } from "./telemetry.js";

interface Sample {
  v: number;
  t: number; // source ts or receive ts
}

/**
 * Statistical shared-upstream detection. For every instrument, recent LAST_PRICE observations of each
 * source are kept; two sources whose values coincide (identical price within a rounding epsilon and
 * within 2 s) on the vast majority of aligned samples very likely share an upstream vendor. The result is
 * a `similarity` in [0, 1] per source pair; pairs above the threshold are treated as one family by the
 * consensus engine (independence weight 0.25) even without contractual knowledge of the vendor.
 */
export class LineageEngine {
  private samples = new Map<string, Map<string, Sample[]>>(); // instrumentId -> sourceId -> ring
  private similarity = new Map<string, { similarity: number; samples: number; instruments: number }>(); // "a|b" sorted
  private inferredFamily = new Map<string, string>(); // sourceId -> inferred family id
  readonly maxPerSource = 400;
  readonly threshold = 0.92;
  readonly minSamples = 120;

  observe(o: Observation): void {
    if (o.field !== "LAST_PRICE") return;
    let m = this.samples.get(o.instrumentId);
    if (!m) {
      m = new Map();
      this.samples.set(o.instrumentId, m);
    }
    let ring = m.get(o.sourceId);
    if (!ring) {
      ring = [];
      m.set(o.sourceId, ring);
    }
    ring.push({ v: o.value, t: o.sourceTimestamp ?? o.receivedAt });
    if (ring.length > this.maxPerSource) ring.splice(0, ring.length - this.maxPerSource);
  }

  /** Family override for a source when statistical lineage says it mirrors another source. */
  familyOf(sourceId: string): string | null {
    return this.inferredFamily.get(sourceId) ?? null;
  }

  pairs(): Array<{ sourceA: string; sourceB: string; similarity: number; samples: number; instruments: number; likelySharedUpstream: boolean }> {
    return [...this.similarity.entries()]
      .map(([k, v]) => {
        const [sourceA, sourceB] = k.split("|") as [string, string];
        return { sourceA, sourceB, ...v, likelySharedUpstream: v.similarity >= this.threshold && v.samples >= this.minSamples };
      })
      .sort((a, b) => b.similarity - a.similarity);
  }

  /** Recompute similarities (every 5 min) and persist them. */
  async recompute(): Promise<void> {
    const acc = new Map<string, { matched: number; total: number; instruments: Set<string> }>();
    for (const [instrumentId, bySource] of this.samples) {
      const sources = [...bySource.keys()];
      for (let i = 0; i < sources.length; i++) {
        for (let j = i + 1; j < sources.length; j++) {
          const a = bySource.get(sources[i]!)!;
          const b = bySource.get(sources[j]!)!;
          if (a.length < 30 || b.length < 30) continue;
          const { matched, total } = align(a, b);
          if (!total) continue;
          const key = [sources[i]!, sources[j]!].sort().join("|");
          const e = acc.get(key) ?? { matched: 0, total: 0, instruments: new Set<string>() };
          e.matched += matched;
          e.total += total;
          e.instruments.add(instrumentId);
          acc.set(key, e);
        }
      }
    }
    this.similarity.clear();
    this.inferredFamily.clear();
    const rows: unknown[][] = [];
    for (const [key, e] of acc) {
      const sim = e.matched / e.total;
      this.similarity.set(key, { similarity: Math.round(sim * 1000) / 1000, samples: e.total, instruments: e.instruments.size });
      if (sim >= this.threshold && e.total >= this.minSamples) {
        const [a, b] = key.split("|") as [string, string];
        const fam = this.inferredFamily.get(a) ?? this.inferredFamily.get(b) ?? `inferred:${a}+${b}`;
        this.inferredFamily.set(a, fam);
        this.inferredFamily.set(b, fam);
        telemetry.inc("lineage_shared_upstream_pairs_total");
      }
      const [a, b] = key.split("|");
      rows.push([a, b, Math.round(sim * 1000) / 1000, e.total, e.instruments.size]);
    }
    if (!rows.length) return;
    try {
      const tuples = rows.map((_, i) => `($${i * 5 + 1},$${i * 5 + 2},$${i * 5 + 3},$${i * 5 + 4},$${i * 5 + 5})`);
      await pool.query(
        `insert into source_lineage (source_a, source_b, similarity, samples, instruments) values ${tuples.join(",")}
         on conflict (source_a, source_b) do update set similarity = excluded.similarity, samples = excluded.samples, instruments = excluded.instruments, computed_at = now()`,
        rows.flat(),
      );
    } catch (err) {
      logger.warn({ err: err instanceof Error ? err.message : String(err) }, "lineage persist failed");
    }
  }
}

/** Count how many samples of `a` have a sample of `b` within 2 s with an identical value (relative 1e-6). */
function align(a: Sample[], b: Sample[]): { matched: number; total: number } {
  let j = 0;
  let matched = 0;
  let total = 0;
  const sb = [...b].sort((x, y) => x.t - y.t);
  for (const s of [...a].sort((x, y) => x.t - y.t)) {
    while (j < sb.length && sb[j]!.t < s.t - 2000) j++;
    let k = j;
    let hit = false;
    let found = false;
    while (k < sb.length && sb[k]!.t <= s.t + 2000) {
      found = true;
      if (Math.abs(sb[k]!.v - s.v) <= Math.abs(s.v) * 1e-6) {
        hit = true;
        break;
      }
      k++;
    }
    if (found) {
      total++;
      if (hit) matched++;
    }
  }
  return { matched, total };
}

export const lineage = new LineageEngine();