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

/**
 * Batched, idempotent observation writer. Bounded memory: when the queue exceeds the hard cap,
 * P2 tick observations are dropped (counted), never events or non-REALTIME values.
 */
export class ObservationWriter {
  private queue: Observation[] = [];
  private timer: NodeJS.Timeout | null = null;
  private flushing = false;
  private partitionsReady = new Set<string>();
  private seen = new Map<string, number>(); // fingerprint -> ts (dedupe replays for 10 min)
  readonly hardCap = 200_000;

  enqueue(o: Observation): void {
    const fp = o.observationId;
    const now = Date.now();
    const prev = this.seen.get(fp);
    if (prev && now - prev < 600_000) {
      telemetry.inc("observations_duplicate_total");
      return;
    }
    this.seen.set(fp, now);
    if (this.seen.size > 500_000) this.pruneSeen(now);
    if (this.queue.length >= this.hardCap) {
      if (o.realtimeStatus === "REALTIME") {
        telemetry.inc("observations_dropped_total", 1, { reason: "backpressure" });
        return;
      }
    }
    this.queue.push(o);
    if (this.queue.length >= config.writerMaxBatch) void this.flush();
    else if (!this.timer) this.timer = setTimeout(() => void this.flush(), config.writerFlushMs);
  }

  private pruneSeen(now: number) {
    for (const [k, t] of this.seen) if (now - t > 600_000) this.seen.delete(k);
  }

  depth(): number {
    return this.queue.length;
  }

  async flush(): Promise<void> {
    if (this.timer) {
      clearTimeout(this.timer);
      this.timer = null;
    }
    if (this.flushing || !this.queue.length) return;
    this.flushing = true;
    const batch = this.queue.splice(0, config.writerMaxBatch);
    const started = Date.now();
    try {
      const days = new Set(batch.map((o) => new Date(o.receivedAt).toISOString().slice(0, 10)));
      for (const d of days) {
        if (!this.partitionsReady.has(d)) {
          await ensureObservationPartition(Date.parse(d));
          this.partitionsReady.add(d);
        }
      }
      const cols = 18;
      const values: unknown[] = [];
      const tuples: string[] = [];
      batch.forEach((o, i) => {
        const b = i * cols;
        tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);
        values.push(
          new Date(o.receivedAt).toISOString(),
          o.sourceTimestamp == null ? null : new Date(o.sourceTimestamp).toISOString(),
          o.instrumentId,
          o.field,
          o.value,
          o.currency ?? null,
          o.sourceId,
          o.connectorId,
          o.rightsStatus,
          o.realtimeStatus,
          o.timestampTrust,
          o.confidence ?? null,
          o.latencyMs,
          o.sequence == null ? null : String(o.sequence),
          o.observationId,
          o.rawRef,
          o.normalizerVersion,
          o.observationType ?? null,
        );
      });
      await pool.query(
        `insert into observations (received_at, source_ts, instrument_id, field, value, currency, source_id, connector_id, rights_status, realtime_status, timestamp_trust, confidence, latency_ms, sequence, fingerprint, raw_ref, normalizer_version, observation_type)
         values ${tuples.join(",")}`,
        values,
      );
      await pool.query(
        `insert into daily_stats (day, observations) values (current_date, $1)
         on conflict (day) do update set observations = daily_stats.observations + excluded.observations, updated_at = now()`,
        [batch.length],
      );
      telemetry.inc("observations_written_total", batch.length);
      telemetry.observe("db_insert_latency_ms", Date.now() - started);
    } catch (err) {
      telemetry.inc("observations_write_errors_total");
      logger.error({ err: err instanceof Error ? err.message : String(err), size: batch.length }, "observation batch insert failed");
      // Re-queue once at the front (bounded), so a transient DB hiccup does not lose canonical history.
      if (this.queue.length + batch.length <= this.hardCap) this.queue.unshift(...batch);
    } finally {
      this.flushing = false;
      telemetry.gauge("observation_queue_depth", this.queue.length);
      if (this.queue.length) this.timer = setTimeout(() => void this.flush(), config.writerFlushMs);
    }
  }
}

export const observationWriter = new ObservationWriter();