import type { Bar, BarResolution, CanonicalQuote } from "@market-atlas/market-model";
import { CANONICAL_VERSIONS, RESOLUTION_MS, floorTo } from "@market-atlas/market-model";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { telemetry } from "./telemetry.js";
import { barsMutex } from "./mutex.js";

/**
 * Live 1-minute bar aggregation from canonical prices; higher resolutions are rolled up in SQL
 * by the scheduler (reproducible from the 1m series).
 */
export class BarAggregator {
  private open = new Map<string, Bar>(); // instrumentId -> current 1m bar
  private pending: Bar[] = [];
  private timer: NodeJS.Timeout | null = null;

  onQuote(q: CanonicalQuote): void {
    if (q.price == null || q.realtimeStatus === "STALE" || q.realtimeStatus === "END_OF_DAY") return;
    const ts = floorTo(q.sourceTimestamp ?? q.updatedAt, RESOLUTION_MS["1m"]);
    const cur = this.open.get(q.instrumentId);
    if (cur && cur.ts !== ts) {
      this.pending.push(cur);
      this.open.delete(q.instrumentId);
    }
    if (!this.open.has(q.instrumentId)) {
      this.open.set(q.instrumentId, {
        instrumentId: q.instrumentId,
        resolution: "1m",
        ts,
        open: q.price,
        high: q.price,
        low: q.price,
        close: q.price,
        volume: null,
        sourceCount: q.sourceCount,
        producer: "consensus",
        version: CANONICAL_VERSIONS.consensus,
      });
    } else {
      const b = this.open.get(q.instrumentId)!;
      b.high = Math.max(b.high, q.price);
      b.low = Math.min(b.low, q.price);
      b.close = q.price;
      b.sourceCount = Math.max(b.sourceCount, q.sourceCount);
    }
    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 5000);
  }

  /** Flush completed bars plus a snapshot of open bars (upserted again when they close). */
  async flush(): Promise<void> {
    this.timer = null;
    const batch = [...this.pending, ...this.open.values()];
    this.pending = [];
    if (!batch.length) return;
    await upsertBars(batch);
    if (this.open.size) this.timer = setTimeout(() => void this.flush(), 5000);
  }
}

export async function upsertBars(input: Bar[]): Promise<void> {
  if (!input.length) return;
  // One row per key (a late tick can re-open an already flushed minute) and a deterministic lock order.
  const byKey = new Map<string, Bar>();
  for (const b of input) byKey.set(`${b.instrumentId}|${b.resolution}|${b.ts}`, b);
  const bars = [...byKey.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId) || a.resolution.localeCompare(b.resolution) || a.ts - b.ts);
  await barsMutex.run(() => upsertSorted(bars));
}

async function upsertSorted(bars: Bar[]): Promise<void> {
  const cols = 11;
  for (let i = 0; i < bars.length; i += 1000) {
    const chunk = bars.slice(i, i + 1000);
    const tuples: string[] = [];
    const values: unknown[] = [];
    chunk.forEach((b, j) => {
      const base = j * cols;
      tuples.push(`(${Array.from({ length: cols }, (_, k) => `$${base + k + 1}`).join(",")})`);
      values.push(b.instrumentId, b.resolution, new Date(b.ts).toISOString(), b.open, b.high, b.low, b.close, b.volume, b.sourceCount, b.producer, b.version);
    });
    try {
      await pool.query(
        `insert into bars (instrument_id, resolution, ts, open, high, low, close, volume, source_count, producer, version) values ${tuples.join(",")}
         on conflict (instrument_id, resolution, ts) do update set open = excluded.open,
           high = excluded.high, low = excluded.low, close = excluded.close, volume = coalesce(excluded.volume, bars.volume), source_count = excluded.source_count, producer = excluded.producer, version = excluded.version
         where bars.producer = 'consensus' or excluded.producer <> 'consensus'`,
        values,
      );
      telemetry.inc("bars_written_total", chunk.length);
    } catch (err) {
      logger.error({ err }, "bar upsert failed");
    }
  }
}

/** Roll 1m bars into 5m/15m/1h and 1m→1d (UTC days) for the last `hours` hours. */
export async function rollupBars(hours = 3): Promise<void> {
  await barsMutex.run(() => rollupInner(hours));
}

async function rollupInner(hours: number): Promise<void> {
  const targets: Array<[BarResolution, string]> = [
    ["5m", "5 minutes"],
    ["15m", "15 minutes"],
    ["1h", "1 hour"],
    ["1d", "1 day"],
  ];
  for (const [res, interval] of targets) {
    await pool.query(
      `insert into bars (instrument_id, resolution, ts, open, high, low, close, volume, source_count, producer, version)
       select instrument_id, $1, bucket, (array_agg(open order by ts))[1], max(high), min(low), (array_agg(close order by ts desc))[1], sum(volume), max(source_count), 'consensus', $3
       from (select *, date_bin($2::interval, ts, timestamptz '2000-01-01') as bucket from bars where resolution = '1m' and producer = 'consensus' and ts > now() - ($4 || ' hours')::interval) b
       group by instrument_id, bucket
       order by instrument_id, bucket
       on conflict (instrument_id, resolution, ts) do update set open = excluded.open, high = excluded.high, low = excluded.low, close = excluded.close, volume = excluded.volume, source_count = excluded.source_count
       where bars.producer = 'consensus'`,
      [res, interval, CANONICAL_VERSIONS.consensus, String(hours)],
    );
  }
}

export const barAggregator = new BarAggregator();