LogPipeline

export class LogPipeline {
  readonly #parsers: readonly LogParser[];
  readonly #enrichers: readonly LogEnricher[];
  readonly #redactors: readonly LogRedactor[];
  readonly buffer: LogRingBuffer;
  readonly #history: string[] = [];
  readonly #saved = new Map<string, string>();
  readonly #deduplicate: boolean;
  readonly #sampleEvery: number;
  readonly #maxPerSecond?: number;
  readonly #queryHistoryLimit: number;
  readonly #now: () => number;
  #ingested = 0;
  #rateWindow = { second: -1, count: 0 };

  constructor(
    options: {
      readonly parsers?: readonly LogParser[];
      readonly enrichers?: readonly LogEnricher[];
      readonly redactors?: readonly LogRedactor[];
      readonly capacity?: number;
      readonly deduplicate?: boolean;
      readonly sampleEvery?: number;
      readonly maxPerSecond?: number;
      readonly queryHistoryLimit?: number;
      readonly now?: () => number;
    } = {},
  ) {
    this.#parsers = Object.freeze([...(options.parsers ?? builtInLogParsers)]);
    this.#enrichers = Object.freeze([...(options.enrichers ?? [])]);
    this.#redactors = Object.freeze([...(options.redactors ?? [])]);
    this.buffer = new LogRingBuffer(options.capacity);
    this.#deduplicate = options.deduplicate ?? false;
    this.#sampleEvery = positiveLogCount(
      "sampleEvery",
      options.sampleEvery ?? 1,
    );
    this.#maxPerSecond =
      options.maxPerSecond === undefined
        ? undefined
        : positiveLogCount("maxPerSecond", options.maxPerSecond);
    this.#queryHistoryLimit = positiveLogCount(
      "queryHistoryLimit",
      options.queryHistoryLimit ?? 100,
    );
    this.#now = options.now ?? Date.now;
  }

  ingest(input: string, parserId?: string): readonly LogRecord[] {
    const parser = parserId
      ? this.#parsers.find((item) => item.id === parserId)
      : this.#parsers.toSorted(
          (left, right) => right.detect(input) - left.detect(input),
        )[0];
    if (!parser) {
      throw new Error(`Log parser "${parserId ?? "detected"}" is unavailable`);
    }
    const output = parser.parse(input).map((item) => this.#ingestRecord(item));
    return Object.freeze(output);
  }

  #ingestRecord(item: LogRecord): LogRecord {
    const transformed = this.#applyPipeline(item);
    this.#ingested += 1;
    if (this.#ingested % this.#sampleEvery !== 0) {
      this.buffer.recordSampled();
      return record({ ...transformed, sampled: true });
    }
    const duplicate = this.#deduplicatedRecord(transformed);
    if (duplicate) return duplicate;
    const rateLimited = this.#applyRateLimit(transformed);
    if (rateLimited.rateLimited) {
      this.buffer.recordRateLimited();
      return rateLimited;
    }
    this.buffer.push(rateLimited);
    return rateLimited;
  }

  #applyPipeline(item: LogRecord): LogRecord {
    let transformed = item;
    for (const enricher of this.#enrichers)
      transformed = enricher.enrich(transformed);
    for (const redactor of this.#redactors)
      transformed = redactor.redact(transformed);
    return record({ ...transformed });
  }

  #applyRateLimit(item: LogRecord): LogRecord {
    const second = Math.floor(this.#now() / 1_000);
    if (this.#rateWindow.second !== second) {
      this.#rateWindow = { second, count: 0 };
    }
    this.#rateWindow.count += 1;
    return this.#maxPerSecond !== undefined &&
      this.#rateWindow.count > this.#maxPerSecond
      ? record({ ...item, rateLimited: true })
      : item;
  }

  #deduplicatedRecord(item: LogRecord): LogRecord | undefined {
    const previous = this.buffer.last();
    if (!this.#deduplicate || !previous || !sameLogRecord(previous, item))
      return undefined;
    const duplicate = record({
      ...previous,
      duplicateCount: (previous.duplicateCount ?? 1) + 1,
    });
    this.buffer.replaceLast(duplicate);
    return duplicate;
  }

  query(source: string): readonly LogRecord[] {
    this.#history.push(source);
    if (this.#history.length > this.#queryHistoryLimit) this.#history.shift();
    return this.filter(source);
  }

  filter(source: string): readonly LogRecord[] {
    const query = compileLogQuery(source);
    if (query.diagnostics.length > 0) return [];
    return Object.freeze(this.buffer.records().filter(query.predicate));
  }

  saveSearch(name: string, source: string): void {
    this.#saved.set(name, source);
  }

  savedSearches(): Readonly<Record<string, string>> {
    return Object.freeze(Object.fromEntries(this.#saved));
  }

  history(): readonly string[] {
    return Object.freeze([...this.#history]);
  }

  clear(): void {
    this.#ingested = 0;
    this.#rateWindow = { second: -1, count: 0 };
    this.buffer.clear();
  }

  replay(records: readonly LogRecord[]): void {
    for (const item of records) this.buffer.push(this.#applyPipeline(item));
  }

  export(
    records: readonly LogRecord[] = this.buffer.records(),
    format: "jsonl" | "text" = "jsonl",
  ): string {
    if (format === "text") {
      return records.map((item) => String(item.body)).join("\n");
    }
    return records
      .map((item) =>
        JSON.stringify(item, (_key, value) =>
          typeof value === "bigint" ? value.toString() : value,
        ),
      )
      .join("\n");
  }
}