import { createWriteStream, existsSync, mkdirSync, createReadStream } from 'node:fs';
import { mkdir, stat } from 'node:fs/promises';
import path from 'node:path';
import { createGzip, createGunzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';
import { createInterface } from 'node:readline';
import { dataDir } from '@cancerindex/shared';

interface Writer {
  path: string;
  stream: ReturnType<typeof createWriteStream>;
  gz: ReturnType<typeof createGzip>;
  lines: number;
  part: number;
  bytes: number;
  /** Single pending drain promise: every writer waiting for back-pressure shares it (no listener pile-up). */
  drain: Promise<void> | null;
}

const PART_BYTES = 64 * 1024 * 1024;

/**
 * Raw data lake (CLAUDE.md §26): every source payload retained when licensing allows.
 * Layout: {CI_DATA_DIR}/raw/{source}/{YYYY-MM-DD}/{entity}/{runId}-{part}.jsonl.gz
 * A record's `rawPath` is "<file>#<line>" so any canonical value can be traced to its raw payload.
 *
 * Writes are serialised through one promise chain: concurrent `put()` calls (e.g. `Promise.all`
 * over a batch) are appended in call order and back-pressure is awaited once, through a single
 * shared drain promise — never one `drain` listener per caller (that produced
 * `MaxListenersExceededWarning: 11 drain listeners added to [Gzip]`).
 */
export class RawLake {
  private readonly root: string;
  private writers = new Map<string, Writer>();
  private chain: Promise<unknown> = Promise.resolve();
  constructor(
    private readonly source: string,
    private readonly runId: string,
    private readonly date = new Date(),
    root?: string,
  ) {
    this.root = root ?? path.join(dataDir(), 'raw');
  }

  dir(entity: string): string {
    const day = this.date.toISOString().slice(0, 10);
    return path.join(this.root, this.source, day, entity);
  }

  /** Run `fn` after every previously queued lake operation (FIFO). */
  private serialize<T>(fn: () => Promise<T>): Promise<T> {
    const next = this.chain.then(fn, fn);
    this.chain = next.catch(() => undefined);
    return next;
  }

  /** Append one raw JSON payload; returns its rawPath reference. */
  put(entity: string, payload: unknown): Promise<string> {
    return this.serialize(() => this.write(entity, payload));
  }

  private async write(entity: string, payload: unknown): Promise<string> {
    let w = this.writers.get(entity);
    if (!w || w.bytes > PART_BYTES) {
      if (w) await this.closeWriterNow(entity);
      const part = (w?.part ?? 0) + 1;
      const dir = this.dir(entity);
      if (!existsSync(dir)) mkdirSync(dir, { recursive: true });
      const file = path.join(dir, `${this.runId}-${String(part).padStart(3, '0')}.jsonl.gz`);
      const stream = createWriteStream(file);
      const gz = createGzip({ level: 6 });
      gz.pipe(stream);
      w = { path: file, stream, gz, lines: 0, part, bytes: 0, drain: null };
      this.writers.set(entity, w);
    }
    const line = JSON.stringify(payload) + '\n';
    w.bytes += line.length;
    const ref = `${w.path}#${w.lines}`;
    w.lines++;
    if (!w.gz.write(line)) await this.awaitDrain(w);
    return ref;
  }

  private awaitDrain(w: Writer): Promise<void> {
    if (!w.drain) {
      w.drain = new Promise<void>((resolve) => {
        w.gz.once('drain', () => {
          w.drain = null;
          resolve();
        });
      });
    }
    return w.drain;
  }

  private async closeWriterNow(entity: string): Promise<void> {
    const w = this.writers.get(entity);
    if (!w) return;
    await new Promise<void>((resolve, reject) => {
      w.stream.once('finish', () => resolve());
      w.stream.once('error', reject);
      w.gz.end();
    });
    this.writers.delete(entity);
  }

  /**
   * Flush compressed data written so far to disk (Z_SYNC_FLUSH) without closing the files — used on
   * abort so the lines already referenced by `source_records.raw_path` are readable even if the
   * process is killed before `close()`.
   */
  flush(): Promise<void> {
    return this.serialize(async () => {
      for (const w of this.writers.values()) await new Promise<void>((resolve) => w.gz.flush(() => resolve()));
    });
  }

  close(): Promise<void> {
    return this.serialize(async () => {
      for (const entity of [...this.writers.keys()]) await this.closeWriterNow(entity);
    });
  }

  /**
   * Read a raw payload back by reference (admin TRACE, reprocessing). Tolerates a file whose gzip
   * trailer is missing (run killed before `close()`): lines flushed before the cut are still
   * returned, later ones yield null instead of an unhandled zlib error.
   */
  static async read(ref: string): Promise<unknown | null> {
    const [file, lineStr] = ref.split('#');
    if (!file || lineStr === undefined) return null;
    const target = Number(lineStr);
    // A missing or unreadable file must resolve to null: a stream error that only destroys the
    // gunzip leaves the readline iterator pending forever, the event loop drains and the process
    // exits 0 mid-run without finishing the ingest run (observed 2026-09-11 on a prod copy whose
    // lake lives on another host).
    if (!existsSync(file)) return null;
    const src = createReadStream(file);
    const gunzip = createGunzip();
    const rl = createInterface({ input: gunzip, crlfDelay: Infinity });
    const swallow = () => {
      rl.close();
      gunzip.destroy();
    };
    src.on('error', swallow);
    gunzip.on('error', swallow);
    src.pipe(gunzip);
    try {
      let i = 0;
      for await (const line of rl) {
        if (i === target) return JSON.parse(line);
        i++;
      }
      return null;
    } finally {
      rl.close();
      gunzip.destroy();
      src.destroy();
    }
  }
}

/** Download a bulk file to the lake (CLAUDE.md §228-230) with checksum + size recorded. */
export async function downloadBulk(url: string, dest: string, fetchImpl: typeof fetch = fetch, headers: Record<string, string> = {}): Promise<{ path: string; bytes: number; sha256: string }> {
  await mkdir(path.dirname(dest), { recursive: true });
  const res = await fetchImpl(url, { headers: { 'user-agent': 'CancerIndex/0.1 (+https://www.cancerindex.io)', ...headers } });
  if (!res.ok || !res.body) throw new Error(`bulk download failed ${res.status} ${url}`);
  const { createHash } = await import('node:crypto');
  const hash = createHash('sha256');
  const { Transform } = await import('node:stream');
  const tap = new Transform({
    transform(chunk, _enc, cb) {
      hash.update(chunk);
      cb(null, chunk);
    },
  });
  const { Readable } = await import('node:stream');
  await pipeline(Readable.fromWeb(res.body as never), tap, createWriteStream(dest));
  const s = await stat(dest);
  return { path: dest, bytes: s.size, sha256: hash.digest('hex') };
}