spb/market-atlas

1import { createWriteStream, mkdirSync } from "node:fs";2import { readFile } from "node:fs/promises";3import { join } from "node:path";4import { gunzipSync, gzipSync } from "node:zlib";5import type { RawObservation } from "@market-atlas/market-model";6import { redactObject, sha256 } from "@market-atlas/connector-sdk";7import { config } from "../config.js";8import { telemetry } from "./telemetry.js";910/**11 * L0 archive: raw payloads stored gzip-compressed, content-addressed, one directory per day12 * and connector. Streaming frames are sampled (MA_RAW_SAMPLE_RATE); polled/bulk payloads are13 * always kept. Secrets are redacted before writing.14 */15export class RawArchive {16  private pending = new Map<string, Buffer>();17  private timer: NodeJS.Timeout | null = null;1819  constructor(private root = join(config.dataDir, "raw")) {}2021  /** Returns a raw_ref (relative path) or null when the frame was not sampled. */22  store(raw: RawObservation, force = false): string | null {23    if (!force && Math.random() > config.rawSampleRate) return null;24    const day = new Date(raw.receivedAt).toISOString().slice(0, 10);25    const body = JSON.stringify({26      connector_id: raw.connectorId,27      source_id: raw.sourceId,28      kind: raw.kind,29      received_at: new Date(raw.receivedAt).toISOString(),30      meta: raw.meta ? redactObject(raw.meta) : undefined,31      payload: redactObject(raw.payload),32    });33    const hash = sha256(body).slice(0, 32);34    const rel = join(day, raw.connectorId, `${hash}.json.gz`);35    if (!this.pending.has(rel)) {36      this.pending.set(rel, gzipSync(body, { level: 6 }));37      this.schedule();38    }39    return rel;40  }4142  private schedule() {43    if (this.timer) return;44    this.timer = setTimeout(() => {45      this.timer = null;46      this.flush();47    }, 500);48  }4950  flush(): void {51    const batch = [...this.pending.entries()];52    this.pending.clear();53    for (const [rel, buf] of batch) {54      const abs = join(this.root, rel);55      mkdirSync(join(abs, ".."), { recursive: true });56      const ws = createWriteStream(abs, { flags: "wx" });57      ws.on("error", () => {}); // already exists → content-addressed, fine58      ws.end(buf);59      telemetry.inc("raw_archived_total");60      telemetry.inc("raw_archived_bytes", buf.length);61    }62  }6364  async read(ref: string): Promise<unknown> {65    if (ref.includes("..")) throw new Error("invalid ref");66    const buf = await readFile(join(this.root, ref));67    return JSON.parse(gunzipSync(buf).toString("utf8"));68  }69}7071export const rawArchive = new RawArchive();72