tyx_AI_xhs/apps/worker/src/structured-log.ts

import {
  appendFileSync,
  existsSync,
  mkdirSync,
  readdirSync,
  renameSync,
  rmSync,
  statSync,
} from "node:fs";
import { join } from "node:path";

export const LOG_SIZE_LIMIT_BYTES = 10 * 1024 * 1024;
export const LOG_FILE_LIMIT = 10;
export const LOG_RETENTION_DAYS = 30;

const statusCategories = new Set(["starting", "ready", "completed", "failed", "unavailable", "degraded", "blocked", "stopping", "stopped"]);
const errorCategories = new Set(["none", "invalid_input", "storage_unavailable", "log_write_failed", "service_unavailable", "timeout", "internal_error"]);
const identifierPattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/;

export interface StructuredLogInput {
  correlation_id?: string;
  duration_ms?: number;
  error_category?: string;
  object_id?: string;
  status_category: string;
}

export class LogWriteError extends Error {
  constructor(options?: ErrorOptions) {
    super("log_write_failed", options);
  }
}

export class StructuredJsonlLogger {
  private readonly directory: string;
  private readonly now: () => Date;
  private readonly onWriteFailure: () => void;
  private initialized = false;
  private currentSize = 0;

  constructor(options: { component: "worker"; directory: string; now?: () => Date; onWriteFailure?: () => void }) {
    this.directory = options.directory;
    this.now = options.now ?? (() => new Date());
    this.onWriteFailure = options.onWriteFailure ?? (() => undefined);
  }

  write(input: StructuredLogInput | Record<string, unknown>) {
    try {
      this.initialize();
      const line = `${JSON.stringify(sanitizeWorkerLogEntry(input, this.now()))}\n`;
      const bytes = Buffer.byteLength(line);
      if (bytes > LOG_SIZE_LIMIT_BYTES) throw new Error("log_entry_too_large");
      if (this.currentSize > 0 && this.currentSize + bytes > LOG_SIZE_LIMIT_BYTES) this.rotate();
      appendFileSync(this.activePath(), line, { encoding: "utf8" });
      this.currentSize += bytes;
    } catch (error) {
      if (error instanceof LogWriteError) throw error;
      this.fail(error);
    }
  }

  maintain() {
    try {
      mkdirSync(this.directory, { recursive: true });
      const cutoff = this.now().getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000;
      for (const file of this.files()) {
        const path = join(this.directory, file.name);
        if (file.index > LOG_FILE_LIMIT - 1 || statSync(path).mtimeMs < cutoff) rmSync(path, { force: true });
      }
      for (const file of this.files().slice(LOG_FILE_LIMIT)) rmSync(join(this.directory, file.name), { force: true });
      this.currentSize = existsSync(this.activePath()) ? statSync(this.activePath()).size : 0;
      this.initialized = true;
    } catch (error) {
      if (error instanceof LogWriteError) throw error;
      this.fail(error);
    }
  }

  private initialize() {
    if (!this.initialized) this.maintain();
  }

  private rotate() {
    rmSync(join(this.directory, `worker.${LOG_FILE_LIMIT - 1}.jsonl`), { force: true });
    for (let index = LOG_FILE_LIMIT - 2; index >= 1; index -= 1) {
      const source = join(this.directory, `worker.${index}.jsonl`);
      if (existsSync(source)) renameSync(source, join(this.directory, `worker.${index + 1}.jsonl`));
    }
    if (existsSync(this.activePath())) renameSync(this.activePath(), join(this.directory, "worker.1.jsonl"));
    this.currentSize = 0;
  }

  private files() {
    if (!existsSync(this.directory)) return [];
    return readdirSync(this.directory)
      .map((name) => ({ match: /^worker(?:\.(\d+))?\.jsonl$/.exec(name), name }))
      .filter((entry): entry is { match: RegExpExecArray; name: string } => entry.match !== null)
      .map(({ match, name }) => ({ index: match[1] === undefined ? 0 : Number(match[1]), name }))
      .sort((left, right) => left.index - right.index);
  }

  private activePath() {
    return join(this.directory, "worker.jsonl");
  }

  private fail(error: unknown): never {
    try {
      this.onWriteFailure();
    } catch {
      // State persistence failure must not replace the stable redacted error.
    }
    throw new LogWriteError({ cause: error });
  }
}

function sanitizeWorkerLogEntry(input: StructuredLogInput | Record<string, unknown>, now: Date) {
  const value = input as Record<string, unknown>;
  const entry: Record<string, unknown> = {
    component: "worker",
    schema_version: "1.0",
    status_category: statusCategories.has(String(value.status_category)) ? value.status_category : "failed",
    timestamp: now.toISOString(),
  };
  if (typeof value.correlation_id === "string" && identifierPattern.test(value.correlation_id)) entry.correlation_id = value.correlation_id;
  if (typeof value.object_id === "string" && identifierPattern.test(value.object_id)) entry.object_id = value.object_id;
  if (typeof value.duration_ms === "number" && Number.isSafeInteger(value.duration_ms) && value.duration_ms >= 0) entry.duration_ms = value.duration_ms;
  if (typeof value.error_category === "string") entry.error_category = errorCategories.has(value.error_category) ? value.error_category : "internal_error";
  return entry;
}