akurai-log / js/monitor-server.js

Latest change 2577f04e8d3b30f5b729c10273a73b0d54351113 - feat: enforce tenant-safe log contract by Ólafur Búi Ólafsson

// monitor-server.js — server-side log shipper for AkurAI-Monitor (Node/Bun).
//
// Zero dependencies: uses global `fetch` (Node >=18, Bun). Provides:
//   - createLogger(): a structured logger that also ships to Monitor
//   - clientProxy(): forwards browser logs (from monitor-client.js) to Monitor
//     with the ingest token attached server-side and origin=client.
//
// Config (env or explicit): MONITOR_INGEST_URL, MONITOR_INGEST_TOKEN,
// MONITOR_HOST_ID, MONITOR_APP_ID, MONITOR_LOG_LEVEL.

import {
  BATCH_MAX,
  FLUSH_MS,
  LEVELS,
  normalizeCorrelation,
  normalizeLevel,
  sanitizeLog,
} from './monitor-contract.js';

function resolve(opts = {}) {
  return {
    url: opts.url || process.env.MONITOR_INGEST_URL || '',
    token: opts.token || process.env.MONITOR_INGEST_TOKEN || '',
    hostId: numOrNull(opts.hostId ?? process.env.MONITOR_HOST_ID),
    appId: numOrNull(opts.appId ?? process.env.MONITOR_APP_ID),
    level: opts.level || process.env.MONITOR_LOG_LEVEL || 'debug',
    correlation: normalizeCorrelation(opts.correlation ?? {
      organization_id: opts.organizationId ?? process.env.MONITOR_ORGANIZATION_ID,
      workspace_id: opts.workspaceId ?? process.env.MONITOR_WORKSPACE_ID,
      installation_id: opts.installationId ?? process.env.MONITOR_INSTALLATION_ID,
      deployment_id: opts.deploymentId ?? process.env.MONITOR_DEPLOYMENT_ID,
      request_id: opts.requestId ?? process.env.MONITOR_REQUEST_ID,
    }),
  };
}

function numOrNull(v) {
  if (v === undefined || v === null || v === '') return null;
  const n = Number(v);
  return Number.isFinite(n) ? n : null;
}

async function ship(cfg, logs) {
  if (logs.length === 0) return { ok: true, inserted: 0 };
  if (!cfg.url || !cfg.token) {
    return { ok: false, error: 'Monitor ingest is not configured' };
  }
  try {
    const response = await fetch(`${cfg.url.replace(/\/$/, '')}/api/ingest`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${cfg.token}`,
      },
      body: JSON.stringify({ logs }),
    });
    if (!response.ok) {
      return { ok: false, status: response.status, error: `Monitor rejected logs (${response.status})` };
    }
    return { ok: true, inserted: logs.length, status: response.status };
  } catch {
    return { ok: false, error: 'Monitor ingest request failed' };
  }
}

// A batching, non-blocking structured logger.
export function createLogger(app, options = {}) {
  const cfg = resolve(options);
  const minRank = LEVELS[cfg.level] ?? 1;
  const source = app;
  let queue = [];
  let timer = null;

  function flush() {
    if (timer) {
      clearTimeout(timer);
      timer = null;
    }
    if (queue.length === 0) return;
    const batch = queue;
    queue = [];
    void ship(cfg, batch);
  }

  function emit(level, message, fields) {
    const line = typeof message === 'string' ? message : JSON.stringify(message);
    const log = sanitizeLog({
      source,
      line,
      level,
      target: (fields && fields.target) || app,
      origin: 'server',
      fields,
      ...normalizeCorrelation(fields),
      ts: Math.floor(Date.now() / 1000),
    }, cfg.correlation);
    const safeLine = sanitizeLog({ source, line, level, origin: 'server' })?.line || '[REDACTED]';
    // Mirror the privacy-safe line to stdout/stderr so journald still captures it.
    (level === 'error' || level === 'warn' ? console.error : console.log)(
      `[${level}] ${source}: ${safeLine}`
    );
    if ((LEVELS[level] ?? 1) < minRank || !log) return;
    queue.push({
      ...log,
      pid: process.pid,
      host_id: cfg.hostId,
      application_id: cfg.appId,
    });
    if (queue.length >= BATCH_MAX) flush();
    else if (!timer) timer = setTimeout(flush, FLUSH_MS);
  }

  return {
    trace: (m, f) => emit('trace', m, f),
    debug: (m, f) => emit('debug', m, f),
    info: (m, f) => emit('info', m, f),
    warn: (m, f) => emit('warn', m, f),
    error: (m, f) => emit('error', m, f),
    flush,
  };
}

// Build a handler that forwards browser-submitted logs to Monitor.
// `body` is the parsed JSON from monitor-client.js: { logs: [...] }.
// Returns { inserted } or throws on malformed input. Framework-agnostic:
// mount it on your own POST /api/client-logs route.
export function clientProxy(app, options = {}) {
  const cfg = resolve(options);
  const source = `${app}-client`;
  return async function handleClientLogs(body) {
    if (!body || !Array.isArray(body.logs)) {
      throw new Error('expected { logs: [...] }');
    }
    const logs = body.logs.slice(0, 500).flatMap((input) => {
      const l = input && typeof input === 'object' ? input : {};
      const log = sanitizeLog({
        source,
        line: l.line ?? '',
        level: normalizeLevel(l.level),
        target: typeof l.target === 'string' ? l.target : null,
        origin: 'client',
        fields: l.fields,
        ...normalizeCorrelation(l),
        ts: Number.isFinite(l.ts) ? l.ts : Math.floor(Date.now() / 1000),
      }, cfg.correlation);
      return log ? [{
        ...log,
        host_id: cfg.hostId,
        application_id: cfg.appId,
      }] : [];
    });
    const result = await ship(cfg, logs);
    if (!result.ok) {
      throw new Error(`Monitor log forwarding failed: ${result.error}`);
    }
    return { inserted: result.inserted };
  };
}