spb/cancerindex

1import { createWriteStream, existsSync, mkdirSync, createReadStream } from 'node:fs';2import { mkdir, stat } from 'node:fs/promises';3import path from 'node:path';4import { createGzip, createGunzip } from 'node:zlib';5import { pipeline } from 'node:stream/promises';6import { createInterface } from 'node:readline';7import { dataDir } from '@cancerindex/shared';89interface Writer {10  path: string;11  stream: ReturnType<typeof createWriteStream>;12  gz: ReturnType<typeof createGzip>;13  lines: number;14  part: number;15  bytes: number;16  /** Single pending drain promise: every writer waiting for back-pressure shares it (no listener pile-up). */17  drain: Promise<void> | null;18}1920const PART_BYTES = 64 * 1024 * 1024;2122/**23 * Raw data lake (CLAUDE.md §26): every source payload retained when licensing allows.24 * Layout: {CI_DATA_DIR}/raw/{source}/{YYYY-MM-DD}/{entity}/{runId}-{part}.jsonl.gz25 * A record's `rawPath` is "<file>#<line>" so any canonical value can be traced to its raw payload.26 *27 * Writes are serialised through one promise chain: concurrent `put()` calls (e.g. `Promise.all`28 * over a batch) are appended in call order and back-pressure is awaited once, through a single29 * shared drain promise — never one `drain` listener per caller (that produced30 * `MaxListenersExceededWarning: 11 drain listeners added to [Gzip]`).31 */32export class RawLake {33  private readonly root: string;34  private writers = new Map<string, Writer>();35  private chain: Promise<unknown> = Promise.resolve();36  constructor(37    private readonly source: string,38    private readonly runId: string,39    private readonly date = new Date(),40    root?: string,41  ) {42    this.root = root ?? path.join(dataDir(), 'raw');43  }4445  dir(entity: string): string {46    const day = this.date.toISOString().slice(0, 10);47    return path.join(this.root, this.source, day, entity);48  }4950  /** Run `fn` after every previously queued lake operation (FIFO). */51  private serialize<T>(fn: () => Promise<T>): Promise<T> {52    const next = this.chain.then(fn, fn);53    this.chain = next.catch(() => undefined);54    return next;55  }5657  /** Append one raw JSON payload; returns its rawPath reference. */58  put(entity: string, payload: unknown): Promise<string> {59    return this.serialize(() => this.write(entity, payload));60  }6162  private async write(entity: string, payload: unknown): Promise<string> {63    let w = this.writers.get(entity);64    if (!w || w.bytes > PART_BYTES) {65      if (w) await this.closeWriterNow(entity);66      const part = (w?.part ?? 0) + 1;67      const dir = this.dir(entity);68      if (!existsSync(dir)) mkdirSync(dir, { recursive: true });69      const file = path.join(dir, `${this.runId}-${String(part).padStart(3, '0')}.jsonl.gz`);70      const stream = createWriteStream(file);71      const gz = createGzip({ level: 6 });72      gz.pipe(stream);73      w = { path: file, stream, gz, lines: 0, part, bytes: 0, drain: null };74      this.writers.set(entity, w);75    }76    const line = JSON.stringify(payload) + '\n';77    w.bytes += line.length;78    const ref = `${w.path}#${w.lines}`;79    w.lines++;80    if (!w.gz.write(line)) await this.awaitDrain(w);81    return ref;82  }8384  private awaitDrain(w: Writer): Promise<void> {85    if (!w.drain) {86      w.drain = new Promise<void>((resolve) => {87        w.gz.once('drain', () => {88          w.drain = null;89          resolve();90        });91      });92    }93    return w.drain;94  }9596  private async closeWriterNow(entity: string): Promise<void> {97    const w = this.writers.get(entity);98    if (!w) return;99    await new Promise<void>((resolve, reject) => {100      w.stream.once('finish', () => resolve());101      w.stream.once('error', reject);102      w.gz.end();103    });104    this.writers.delete(entity);105  }106107  /**108   * Flush compressed data written so far to disk (Z_SYNC_FLUSH) without closing the files — used on109   * abort so the lines already referenced by `source_records.raw_path` are readable even if the110   * process is killed before `close()`.111   */112  flush(): Promise<void> {113    return this.serialize(async () => {114      for (const w of this.writers.values()) await new Promise<void>((resolve) => w.gz.flush(() => resolve()));115    });116  }117118  close(): Promise<void> {119    return this.serialize(async () => {120      for (const entity of [...this.writers.keys()]) await this.closeWriterNow(entity);121    });122  }123124  /**125   * Read a raw payload back by reference (admin TRACE, reprocessing). Tolerates a file whose gzip126   * trailer is missing (run killed before `close()`): lines flushed before the cut are still127   * returned, later ones yield null instead of an unhandled zlib error.128   */129  static async read(ref: string): Promise<unknown | null> {130    const [file, lineStr] = ref.split('#');131    if (!file || lineStr === undefined) return null;132    const target = Number(lineStr);133    // A missing or unreadable file must resolve to null: a stream error that only destroys the134    // gunzip leaves the readline iterator pending forever, the event loop drains and the process135    // exits 0 mid-run without finishing the ingest run (observed 2026-09-11 on a prod copy whose136    // lake lives on another host).137    if (!existsSync(file)) return null;138    const src = createReadStream(file);139    const gunzip = createGunzip();140    const rl = createInterface({ input: gunzip, crlfDelay: Infinity });141    const swallow = () => {142      rl.close();143      gunzip.destroy();144    };145    src.on('error', swallow);146    gunzip.on('error', swallow);147    src.pipe(gunzip);148    try {149      let i = 0;150      for await (const line of rl) {151        if (i === target) return JSON.parse(line);152        i++;153      }154      return null;155    } finally {156      rl.close();157      gunzip.destroy();158      src.destroy();159    }160  }161}162163/** Download a bulk file to the lake (CLAUDE.md §228-230) with checksum + size recorded. */164export async function downloadBulk(url: string, dest: string, fetchImpl: typeof fetch = fetch, headers: Record<string, string> = {}): Promise<{ path: string; bytes: number; sha256: string }> {165  await mkdir(path.dirname(dest), { recursive: true });166  const res = await fetchImpl(url, { headers: { 'user-agent': 'CancerIndex/0.1 (+https://www.cancerindex.io)', ...headers } });167  if (!res.ok || !res.body) throw new Error(`bulk download failed ${res.status} ${url}`);168  const { createHash } = await import('node:crypto');169  const hash = createHash('sha256');170  const { Transform } = await import('node:stream');171  const tap = new Transform({172    transform(chunk, _enc, cb) {173      hash.update(chunk);174      cb(null, chunk);175    },176  });177  const { Readable } = await import('node:stream');178  await pipeline(Readable.fromWeb(res.body as never), tap, createWriteStream(dest));179  const s = await stat(dest);180  return { path: dest, bytes: s.size, sha256: hash.digest('hex') };181}182