wulf-pulse/lib/services/rmm/loglift-receiver.ts

/**
 * LogLift webhook receiver business logic.
 *
 *   1. Validate the webhook payload + object key shape.
 *   2. Resolve linkages: Datto site → Autotask company; hostname → Datto
 *      device + IT Glue Configuration.
 *   3. Correlate to a Pulse-dispatched execution by run_id, or insert a
 *      fresh out-of-band row.
 *   4. Download the gzipped JSON from B2 (25MB cap).
 *   5. Decompress with a zip-bomb guard (refuse if the inflated payload
 *      claims > 100MB).
 *   6. Slim to a storage-friendly shape: system_context + summary + top 100
 *      events by severity. The full gzip stays in B2 forever.
 *   7. Persist the row complete + redact for safety.
 *   8. Auto-fire an asset-first audit when the Configuration matched
 *      uniquely.
 *
 * The full audit + correlation chain is intentionally synchronous so the
 * webhook response can include the executionId + matched_configuration_id.
 * The runAssetAudit() call is fire-and-forget so the agent doesn't wait
 * for an LLM call.
 */

import { gunzipSync } from 'node:zlib';

import { downloadToBuffer, OBJECT_KEY_REGEX } from '@/lib/services/b2/client';
import { redact } from '@/lib/services/analyzer/itglue-redact';
import { runAssetAudit } from '@/lib/services/analyzer/asset-audit/runner';
import { audit } from '@/lib/services/audit';
import postgresClient from '@/lib/services/postgres-client';
import {
  createOutOfBandUploadExecution,
  findExecutionByRunId,
  markExecutionFromB2Upload,
} from './persistence';
import {
  resolveConfigurationByHostname,
  resolveDattoDeviceByHostname,
  resolveDattoSiteByUid,
} from './loglift-matcher';

/** Hard cap on the *inflated* payload size — defense against zip bombs. */
const MAX_INFLATED_BYTES = 100 * 1024 * 1024; // 100 MB

/** Cap on how many events we copy into Postgres. The full set stays in B2. */
const TOP_EVENTS_LIMIT = 100;

export interface LogliftWebhookPayload {
  runId: string;
  clientId: string; // Datto site uuid
  computerName: string;
  deviceUid?: string | null;
  summary: {
    totalEvents?: number | null;
    criticalEvents?: number | null;
    errorCount?: number | null;
    warningCount?: number | null;
    timeRange?: string | null;
  };
  objectKey: string;
  collectedAt: string;
  rmmContext?: {
    siteName?: string | null;
    siteUid?: string | null;
    accountUid?: string | null;
  } | null;
  issueDescription?: string | null;
  ticketNumber?: string | null;
}

export interface LogliftReceiveResult {
  executionId: string;
  matched: {
    datto_site_id: number | null;
    datto_device_uid: string | null;
    autotask_company_id: string | null;
    configuration_id: string | null;
    configuration_single_match: boolean;
  };
  parsed: {
    total_events: number;
    critical_events: number;
    error_count: number | null;
    warning_count: number | null;
    time_range: string | null;
  };
  audit_id: string | null;
}

interface LogliftRawPayload {
  metadata?: unknown;
  systemContext?: unknown;
  summary?: {
    TotalEvents?: number;
    CriticalEvents?: number;
    ByLevel?: Record<string, number>;
    TimeRange?: unknown;
    TopEventIds?: unknown;
  };
  events?: Array<Record<string, unknown>>;
}

/**
 * Severity ranking. Critical > Error > Warning > Information > Verbose. The
 * Windows event log uses both the LevelDisplayName string and a numeric
 * Level (1=Critical, 2=Error, 3=Warning, 4=Information, 5=Verbose).
 */
function severityRank(ev: Record<string, unknown>): number {
  const lvlName = String(
    ev.LevelDisplayName ?? ev.levelDisplayName ?? ev.level ?? ''
  ).toLowerCase();
  if (lvlName.includes('critical')) return 5;
  if (lvlName.includes('error')) return 4;
  if (lvlName.includes('warn')) return 3;
  if (lvlName.includes('info')) return 2;
  if (lvlName.includes('verbose')) return 1;
  const numLevel = Number(ev.Level ?? ev.level);
  if (Number.isFinite(numLevel)) {
    if (numLevel === 1) return 5;
    if (numLevel === 2) return 4;
    if (numLevel === 3) return 3;
    if (numLevel === 4) return 2;
    if (numLevel === 5) return 1;
  }
  return 0;
}

function eventTimestamp(ev: Record<string, unknown>): number {
  const t = ev.TimeCreated ?? ev.timeCreated ?? ev.timestamp ?? null;
  if (typeof t !== 'string') return 0;
  const n = Date.parse(t);
  return Number.isFinite(n) ? n : 0;
}

/**
 * Slim the LogLift payload to a storage-friendly shape. Drops the full
 * `events` array (often thousands), keeps the top N by severity then
 * recency. The full gzip lives in B2 for forensic replay.
 */
export function slimLogliftPayload(raw: LogliftRawPayload): {
  metadata: unknown;
  system_context: unknown;
  summary: unknown;
  top_events: Array<Record<string, unknown>>;
  event_count_total: number;
  top_events_truncated: boolean;
} {
  const events = Array.isArray(raw.events) ? raw.events : [];
  const total = events.length;

  const ranked = events
    .map((e, i) => ({ e, rank: severityRank(e), ts: eventTimestamp(e), i }))
    .sort((a, b) => {
      if (a.rank !== b.rank) return b.rank - a.rank;
      if (a.ts !== b.ts) return b.ts - a.ts;
      return a.i - b.i;
    })
    .slice(0, TOP_EVENTS_LIMIT)
    .map((x) => x.e);

  return {
    metadata: raw.metadata ?? null,
    system_context: raw.systemContext ?? null,
    summary: raw.summary ?? null,
    top_events: ranked,
    event_count_total: total,
    top_events_truncated: total > TOP_EVENTS_LIMIT,
  };
}

/**
 * Decompress the gzipped LogLift payload with a zip-bomb guard. The Node
 * zlib API doesn't expose a streaming size cap directly, so we check the
 * gzip's ISIZE trailer (last 4 bytes of the gzip stream — the modulo-2^32
 * inflated size). Refuse before decompressing if it claims > MAX_INFLATED_BYTES.
 */
export function decompressLogliftPayload(gz: Buffer): Buffer {
  if (gz.length < 18) {
    throw new Error('LogLift payload too small to be a valid gzip stream');
  }
  // gzip ISIZE: last 4 bytes, little-endian, = uncompressed size mod 2^32.
  // It's a rough hint — for streams > 4 GiB it wraps, but at our scale
  // (collector targets are < 25 MB compressed) it's a tight enough guard.
  const isize = gz.readUInt32LE(gz.length - 4);
  if (isize > MAX_INFLATED_BYTES) {
    throw new Error(
      `LogLift payload claims ${isize} bytes inflated; cap is ${MAX_INFLATED_BYTES}`
    );
  }
  const out = gunzipSync(gz);
  if (out.length > MAX_INFLATED_BYTES) {
    throw new Error(
      `LogLift payload inflated to ${out.length} bytes; cap is ${MAX_INFLATED_BYTES}`
    );
  }
  return out;
}

export async function processLogliftWebhook(
  payload: LogliftWebhookPayload
): Promise<LogliftReceiveResult> {
  if (!OBJECT_KEY_REGEX.test(payload.objectKey)) {
    throw new Error(`Invalid objectKey shape: ${payload.objectKey.slice(0, 200)}`);
  }

  // 1. Resolve the Datto site → Autotask company.
  const site = await resolveDattoSiteByUid(payload.clientId);
  // 2. Resolve the device by hostname within that site (best-effort — we
  //    fall back to the deviceUid the agent claimed).
  let deviceUid: string | null = null;
  let deviceHostname: string | null = payload.computerName;
  if (site) {
    const dev = await resolveDattoDeviceByHostname(site.datto_site_id, payload.computerName);
    if (dev) {
      deviceUid = dev.uid;
      deviceHostname = dev.hostname ?? deviceHostname;
    }
  }
  if (!deviceUid) {
    // Use the agent's claimed deviceUid as a last resort. The collector's
    // current PowerShell stuffs the hostname here — that's still useful for
    // joining downstream even if it's not the canonical Datto uid.
    deviceUid = payload.deviceUid ?? payload.computerName;
  }

  // 3. Resolve the IT Glue Configuration when we have a company anchor.
  let configMatch: Awaited<ReturnType<typeof resolveConfigurationByHostname>> = null;
  if (site?.autotask_company_id) {
    configMatch = await resolveConfigurationByHostname(
      site.autotask_company_id,
      payload.computerName
    );
  }

  // 4. Correlate to a Pulse-dispatched execution by run_id.
  const correlated = await findExecutionByRunId(payload.runId);
  let executionId: string;
  if (correlated) {
    executionId = correlated.id;
  } else {
    const created = await createOutOfBandUploadExecution({
      scriptId: 'loglift-eventlogs',
      scriptVersion: 1,
      runId: payload.runId,
      jobName: 'LogLift event-log collection (out-of-band)',
      targetDeviceUid: deviceUid,
      targetHostname: deviceHostname,
      targetCompanyId: site?.autotask_company_id ?? null,
      assetType: configMatch ? 'configuration' : null,
      assetId: configMatch?.id ?? null,
      variables: {
        runId: payload.runId,
        clientId: payload.clientId,
        computerName: payload.computerName,
        objectKey: payload.objectKey,
      },
    });
    executionId = created.id;
  }

  // 5. Download the gzipped payload from B2.
  const gz = await downloadToBuffer(payload.objectKey);
  // 6. Decompress with zip-bomb guard.
  const json = decompressLogliftPayload(gz);

  // 7. Parse + slim.
  let raw: LogliftRawPayload;
  try {
    raw = JSON.parse(json.toString('utf8')) as LogliftRawPayload;
  } catch (err) {
    throw new Error(
      `LogLift payload at ${payload.objectKey} is not valid JSON: ${
        err instanceof Error ? err.message : String(err)
      }`
    );
  }
  const slim = slimLogliftPayload(raw);

  // 8. Redact and persist. We keep the raw webhook summary alongside the
  //    slimmed-from-payload one so the audit pipeline can rely on either.
  const persistedEvidence = redact({
    schema_version: 1,
    transport: 'b2_upload',
    object_key: payload.objectKey,
    collected_at: payload.collectedAt,
    rmm_context: payload.rmmContext ?? null,
    issue_description: payload.issueDescription ?? null,
    ticket_number: payload.ticketNumber ?? null,
    webhook_summary: payload.summary,
    metadata: slim.metadata,
    system_context: slim.system_context,
    summary: slim.summary,
    top_events: slim.top_events,
    event_count_total: slim.event_count_total,
    top_events_truncated: slim.top_events_truncated,
  });

  await markExecutionFromB2Upload({
    id: executionId,
    evidenceObjectKey: payload.objectKey,
    parsedEvidence: persistedEvidence,
    targetCompanyId: site?.autotask_company_id ?? null,
    assetType: configMatch ? 'configuration' : null,
    assetId: configMatch?.id ?? null,
  });

  // 8a. Dual-write into the new endpoint data model. Resolve to a configuration_item
  //     via the device_external_ids xref — preferred by IT Glue config (when matched),
  //     falling back to the Datto device UID.
  let configurationItemId: string | null = null;
  if (configMatch?.id) {
    const r = await postgresClient.query<{ configuration_item_id: string | null }>(
      `SELECT configuration_item_id::text
         FROM device_external_ids
        WHERE source = 'itglue' AND source_id = $1
        LIMIT 1`,
      [String(configMatch.id)]
    );
    configurationItemId = r.rows[0]?.configuration_item_id ?? null;
  }
  if (!configurationItemId && deviceUid) {
    const r = await postgresClient.query<{ configuration_item_id: string | null }>(
      `SELECT configuration_item_id::text
         FROM device_external_ids
        WHERE source = 'datto_rmm' AND source_id = $1
        LIMIT 1`,
      [deviceUid]
    );
    configurationItemId = r.rows[0]?.configuration_item_id ?? null;
  }

  const obsRes = await postgresClient.query<{ id: string }>(
    `INSERT INTO device_observations
       (configuration_item_id, source, kind, collected_at, payload, evidence_object_key, run_id)
     VALUES ($1, 'loglift', 'loglift_eventlogs', $2::timestamptz, $3::jsonb, $4, $5)
     RETURNING id::text`,
    [
      configurationItemId,
      payload.collectedAt,
      JSON.stringify(persistedEvidence),
      payload.objectKey,
      payload.runId,
    ]
  );
  const observationId = obsRes.rows[0]?.id ?? null;

  await audit.log({
    action: 'rmm.loglift.received',
    resource: 'datto_device',
    resourceId: deviceUid,
    details: {
      execution_id: executionId,
      run_id: payload.runId,
      object_key: payload.objectKey,
      computer_name: payload.computerName,
      total_events: slim.event_count_total,
      critical_events: payload.summary.criticalEvents ?? null,
      datto_site_id: site?.datto_site_id ?? null,
      autotask_company_id: site?.autotask_company_id ?? null,
      configuration_id: configMatch?.id ?? null,
      configuration_single_match: configMatch?.single_match ?? false,
    },
  });

  if (configMatch && configMatch.single_match) {
    await audit.log({
      action: 'rmm.loglift.matched',
      resource: 'itg_configuration',
      resourceId: configMatch.id,
      details: {
        execution_id: executionId,
        run_id: payload.runId,
        autotask_company_id: site?.autotask_company_id ?? null,
        hostname: payload.computerName,
      },
    });
  }

  // 9. Auto-audit on a single-match Configuration. Fire-and-forget — the
  //    webhook returns immediately. The audit row will appear on the
  //    Configuration's audit page when the LLM call finishes.
  let auditId: string | null = null;
  if (configMatch && configMatch.single_match) {
    try {
      const result = await runAssetAudit({
        assetType: 'configuration',
        assetId: configMatch.id,
        generatedByUserId: null,
        provider: 'anthropic',
      });
      auditId = result.auditId;
      await audit.log({
        action: 'rmm.loglift.audit_triggered',
        resource: 'itg_configuration',
        resourceId: configMatch.id,
        details: {
          execution_id: executionId,
          audit_id: auditId,
          status: result.status,
        },
      });

      // 9a. Mirror the resulting itglue_asset_audits row into endpoint_audits
      //     so the new device-anchored model gets the same audit record. We
      //     keep both during the cutover — endpoint_audits.legacy_itglue_audit_id
      //     points back at the original.
      try {
        await postgresClient.query(
          `INSERT INTO endpoint_audits (
             configuration_item_id, itglue_configuration_id, organization_id,
             generated_by_user_id, generated_at, provider, model_used,
             asset_snapshot, observations_consumed, ticket_count,
             field_gaps, notes_promotions, contradictions,
             overall_score, estimated_cost_usd, total_input_tokens, total_output_tokens,
             status, error_message, triggered_by_ticket_number, triggered_by_analysis_id,
             triggered_by_observation_id, legacy_itglue_audit_id
           )
           SELECT
             $2::bigint, ia.asset_id, ia.organization_id,
             ia.generated_by_user_id, ia.generated_at, ia.provider, ia.model_used,
             ia.asset_snapshot, $3::uuid[], ia.ticket_count,
             ia.field_gaps, ia.notes_promotions, ia.contradictions,
             ia.overall_score, ia.estimated_cost_usd, ia.total_input_tokens, ia.total_output_tokens,
             ia.status, ia.error_message, ia.triggered_by_ticket_number, ia.triggered_by_analysis_id,
             $4::uuid, ia.id
             FROM itglue_asset_audits ia
            WHERE ia.id = $1::uuid`,
          [
            auditId,
            configurationItemId,
            observationId ? [observationId] : [],
            observationId,
          ]
        );
      } catch (err) {
        console.warn(
          `[LOGLIFT] endpoint_audits mirror failed for audit ${auditId}:`,
          err instanceof Error ? err.message : err
        );
      }
    } catch (err) {
      // Don't fail the webhook if the auto-audit blows up — the evidence
      // is already persisted; the user can re-run from the UI.
      console.warn(
        `[LOGLIFT] auto-audit failed for configuration ${configMatch.id}:`,
        err instanceof Error ? err.message : err
      );
    }
  }

  return {
    executionId,
    matched: {
      datto_site_id: site?.datto_site_id ?? null,
      datto_device_uid: deviceUid,
      autotask_company_id: site?.autotask_company_id ?? null,
      configuration_id: configMatch?.id ?? null,
      configuration_single_match: configMatch?.single_match ?? false,
    },
    parsed: {
      total_events: slim.event_count_total,
      critical_events: payload.summary.criticalEvents ?? 0,
      error_count: payload.summary.errorCount ?? null,
      warning_count: payload.summary.warningCount ?? null,
      time_range: payload.summary.timeRange ?? null,
    },
    audit_id: auditId,
  };
}

export const _LOGLIFT_INTERNALS = {
  severityRank,
  eventTimestamp,
  MAX_INFLATED_BYTES,
  TOP_EVENTS_LIMIT,
};