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) { 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, now: Date) { const value = input as Record; const entry: Record = { 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; }