import { createWriteStream, mkdirSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import { gunzipSync, gzipSync } from "node:zlib";
import type { RawObservation } from "@market-atlas/market-model";
import { redactObject, sha256 } from "@market-atlas/connector-sdk";
import { config } from "../config.js";
import { telemetry } from "./telemetry.js";

/**
 * L0 archive: raw payloads stored gzip-compressed, content-addressed, one directory per day
 * and connector. Streaming frames are sampled (MA_RAW_SAMPLE_RATE); polled/bulk payloads are
 * always kept. Secrets are redacted before writing.
 */
export class RawArchive {
  private pending = new Map<string, Buffer>();
  private timer: NodeJS.Timeout | null = null;

  constructor(private root = join(config.dataDir, "raw")) {}

  /** Returns a raw_ref (relative path) or null when the frame was not sampled. */
  store(raw: RawObservation, force = false): string | null {
    if (!force && Math.random() > config.rawSampleRate) return null;
    const day = new Date(raw.receivedAt).toISOString().slice(0, 10);
    const body = JSON.stringify({
      connector_id: raw.connectorId,
      source_id: raw.sourceId,
      kind: raw.kind,
      received_at: new Date(raw.receivedAt).toISOString(),
      meta: raw.meta ? redactObject(raw.meta) : undefined,
      payload: redactObject(raw.payload),
    });
    const hash = sha256(body).slice(0, 32);
    const rel = join(day, raw.connectorId, `${hash}.json.gz`);
    if (!this.pending.has(rel)) {
      this.pending.set(rel, gzipSync(body, { level: 6 }));
      this.schedule();
    }
    return rel;
  }

  private schedule() {
    if (this.timer) return;
    this.timer = setTimeout(() => {
      this.timer = null;
      this.flush();
    }, 500);
  }

  flush(): void {
    const batch = [...this.pending.entries()];
    this.pending.clear();
    for (const [rel, buf] of batch) {
      const abs = join(this.root, rel);
      mkdirSync(join(abs, ".."), { recursive: true });
      const ws = createWriteStream(abs, { flags: "wx" });
      ws.on("error", () => {}); // already exists → content-addressed, fine
      ws.end(buf);
      telemetry.inc("raw_archived_total");
      telemetry.inc("raw_archived_bytes", buf.length);
    }
  }

  async read(ref: string): Promise<unknown> {
    if (ref.includes("..")) throw new Error("invalid ref");
    const buf = await readFile(join(this.root, ref));
    return JSON.parse(gunzipSync(buf).toString("utf8"));
  }
}

export const rawArchive = new RawArchive();