gongxue-base/apps/worker/src/jobs/platform-audit-alerts.ts

import { pool } from '../db.js';
import { config } from '../config.js';

interface PlatformAuditAlertWorkerResult {
  processed: number;
  created: number;
  skipped: number;
}

interface AlertCandidate {
  auditLogId: string;
  tenantId: string | null;
  action: string;
  targetType: string | null;
  targetId: string | null;
  auditDetails: Record<string, unknown> | null;
  createdAt: string;
  ruleId: string;
  ruleCode: string;
  ruleName: string;
  severity: string;
  conditions: Record<string, unknown> | null;
}

function positiveInteger(value: number, fallback: number, max: number) {
  if (!Number.isFinite(value) || value <= 0) return fallback;
  return Math.min(Math.trunc(value), max);
}

function nonNegativeInteger(value: number, fallback: number, max: number) {
  if (!Number.isFinite(value) || value < 0) return fallback;
  return Math.min(Math.trunc(value), max);
}

function truncate(value: unknown, max = 900) {
  return String(value ?? '').slice(0, max);
}

function redactAuditAlertValue(value: unknown, parentKey = '', depth = 0): unknown {
  if (value === null || value === undefined) return value;
  if (depth > 8) return '[REDACTED_DEPTH_LIMIT]';
  if (
    /(?:password|passwd|secret|token|credential|private[_-]?key|api[_-]?key|app[_-]?secret|authorization|cookie|session|cert|signature|nonce)$/i
      .test(parentKey)
  ) {
    return '[REDACTED]';
  }
  if (Array.isArray(value)) return value.map(item => redactAuditAlertValue(item, parentKey, depth + 1));
  if (typeof value === 'object') {
    const output: Record<string, unknown> = {};
    for (const [key, item] of Object.entries(value as Record<string, unknown>)) {
      output[key] = redactAuditAlertValue(item, key, depth + 1);
    }
    return output;
  }
  return value;
}

function alertTitle(candidate: AlertCandidate) {
  return `${candidate.ruleName}: ${candidate.action}`;
}

function alertSummary(candidate: AlertCandidate) {
  const target = [candidate.targetType, candidate.targetId].filter(Boolean).join(':');
  return target ? `${candidate.action} -> ${target}` : candidate.action;
}

async function loadAlertCandidates(params: {
  limit: number;
  lookbackDays: number;
}) {
  const result = await pool.query<AlertCandidate>(
    `
    with matched as (
      select al.id as audit_log_id,
             al.tenant_id,
             al.action,
             al.target_type,
             al.target_id,
             al.details as audit_details,
             al.created_at,
             r.id as rule_id,
             r.code as rule_code,
             r.name as rule_name,
             r.severity,
             r.conditions,
             row_number() over (
               partition by al.id
               order by
                 case r.severity when 'critical' then 1 when 'high' then 2 when 'medium' then 3 else 4 end,
                 r.created_at asc
             ) as rn
      from public.audit_logs al
      join public.platform_audit_alert_rules r
        on r.enabled = true
       and (r.tenant_id is null or r.tenant_id = al.tenant_id)
       and (
         cardinality(r.action_patterns) = 0
         or exists (
           select 1
           from unnest(r.action_patterns) as pattern
           where al.action = pattern
              or (right(pattern, 1) = '*' and al.action like left(pattern, length(pattern) - 1) || '%')
         )
       )
       and (
         cardinality(r.target_types) = 0
         or al.target_type = any(r.target_types)
       )
      left join public.platform_audit_alerts existing
        on existing.rule_id = r.id
       and existing.audit_log_id = al.id
      where existing.id is null
        and al.action like 'platform.%'
        and al.created_at >= now() - ($2::integer * interval '1 day')
    )
    select audit_log_id as "auditLogId",
           tenant_id as "tenantId",
           action,
           target_type as "targetType",
           target_id as "targetId",
           audit_details as "auditDetails",
           created_at as "createdAt",
           rule_id as "ruleId",
           rule_code as "ruleCode",
           rule_name as "ruleName",
           severity,
           conditions
    from matched
    where rn = 1
    order by created_at asc
    limit $1
    `,
    [params.limit, params.lookbackDays],
  );
  return result.rows;
}

async function createAlert(candidate: AlertCandidate) {
  const details = {
    auditDetails: redactAuditAlertValue(candidate.auditDetails || {}),
    ruleCode: candidate.ruleCode,
    conditions: redactAuditAlertValue(candidate.conditions || {}),
    workerId: config.platformAuditAlertWorkerId,
  };

  const client = await pool.connect();
  try {
    await client.query('begin');
    const inserted = await client.query<{ id: string }>(
      `
      insert into public.platform_audit_alerts (
        rule_id, audit_log_id, tenant_id, severity, status,
        action, target_type, target_id, title, summary,
        details, first_seen_at, last_seen_at
      )
      values (
        $1, $2, $3::uuid, $4, 'open',
        $5, $6, $7, $8, $9,
        $10::jsonb, $11::timestamptz, $11::timestamptz
      )
      on conflict (rule_id, audit_log_id) do nothing
      returning id
      `,
      [
        candidate.ruleId,
        candidate.auditLogId,
        candidate.tenantId,
        candidate.severity,
        candidate.action,
        candidate.targetType,
        candidate.targetId,
        alertTitle(candidate),
        alertSummary(candidate),
        JSON.stringify(details),
        candidate.createdAt,
      ],
    );

    const alertId = inserted.rows[0]?.id;
    if (!alertId) {
      await client.query('rollback');
      return 'skipped' as const;
    }

    await client.query(
      `
      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
      values ($1::uuid, null, 'platform.audit.alert_created', 'platform_audit_alert', $2, $3::jsonb)
      `,
      [
        candidate.tenantId,
        alertId,
        JSON.stringify({
          ruleId: candidate.ruleId,
          ruleCode: candidate.ruleCode,
          auditLogId: candidate.auditLogId,
          severity: candidate.severity,
          sourceAction: candidate.action,
          workerId: config.platformAuditAlertWorkerId,
        }),
      ],
    );
    await client.query('commit');
    return 'created' as const;
  } catch (error) {
    await client.query('rollback').catch(() => {});
    await pool.query(
      `
      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
      values ($1::uuid, null, 'platform.audit.alert_failed', 'audit_logs', $2, $3::jsonb)
      `,
      [
        candidate.tenantId,
        candidate.auditLogId,
        JSON.stringify({
          ruleId: candidate.ruleId,
          ruleCode: candidate.ruleCode,
          code: typeof error === 'object' && error !== null && 'code' in error ? String((error as { code?: unknown }).code) : 'PLATFORM_AUDIT_ALERT_FAILED',
          message: truncate(error instanceof Error ? error.message : String(error)),
          workerId: config.platformAuditAlertWorkerId,
        }),
      ],
    ).catch(() => {});
    return 'skipped' as const;
  } finally {
    client.release();
  }
}

export async function processPlatformAuditAlertBatch(options: {
  limit?: number;
  lookbackDays?: number;
} = {}): Promise<PlatformAuditAlertWorkerResult> {
  const limit = positiveInteger(options.limit ?? config.platformAuditAlertBatchSize, 200, 1000);
  const lookbackDays = nonNegativeInteger(options.lookbackDays ?? config.platformAuditAlertLookbackDays, 14, 365);
  const candidates = await loadAlertCandidates({ limit, lookbackDays });
  const result: PlatformAuditAlertWorkerResult = {
    processed: candidates.length,
    created: 0,
    skipped: 0,
  };

  for (const candidate of candidates) {
    const status = await createAlert(candidate);
    result[status] += 1;
  }

  return result;
}