wulf-pulse/lib/services/rmm/worker.ts

/**
 * RMM Overshell poller. Runs in-process alongside the analyzer worker.
 *
 * For each row in `running` status whose timeout_at hasn't passed:
 *   1. Call dattoClient.getJobResults(job_uid, target_device_uid).
 *   2. If Datto says still-running → leave alone, retry next tick.
 *   3. If Datto returns stdout/stderr → redact, run script.parseOutput,
 *      persist via markExecutionComplete (status complete | failed
 *      depending on exit_code).
 *   4. Catch all errors so a single bad job doesn't kill the loop.
 *
 * Stale-row sweep: any row past timeout_at gets marked timeout regardless
 * of whether Datto is responding. Stops the table from accumulating
 * orphaned 'running' rows.
 */

import { getDattoRMMClient } from '@/lib/services/datto-rmm-factory';
import { redact } from '@/lib/services/analyzer/itglue-redact';
import { getScript } from './scripts';
import {
  listRunningExecutions,
  listTimedOutExecutions,
  markExecutionComplete,
  markExecutionTimeout,
} from './persistence';
import { audit } from '@/lib/services/audit';

const POLL_INTERVAL_MS = 5_000;

type DattoJobResults = {
  jobStatus?: string;
  status?: string;
  // Datto returns per-target stdout/stderr in different shapes depending on
  // tenant config; we accept both top-level and nested under
  // results/results[0]/stdOut etc.
  stdOut?: string | null;
  stdErr?: string | null;
  errorCode?: number | null;
  results?: Array<{
    deviceUid?: string;
    stdOut?: string | null;
    stdErr?: string | null;
    errorCode?: number | null;
    jobStatus?: string;
  }>;
};

interface ExtractedResult {
  done: boolean;
  exitCode: number | null;
  stdout: string | null;
  stderr: string | null;
}

function extractResult(payload: DattoJobResults, deviceUid: string): ExtractedResult {
  // Per-device result if available.
  const perDevice = payload.results?.find((r) => r.deviceUid === deviceUid)
    ?? payload.results?.[0];

  const candidate = perDevice ?? payload;
  const status = (candidate.jobStatus ?? payload.jobStatus ?? payload.status ?? '')
    .toString()
    .toLowerCase();

  // Datto's terminal states: succeeded, failed, cancelled. Anything else =
  // not done yet.
  const done =
    status === 'succeeded' ||
    status === 'success' ||
    status === 'failed' ||
    status === 'cancelled' ||
    status === 'canceled';

  return {
    done,
    exitCode:
      typeof candidate.errorCode === 'number'
        ? candidate.errorCode
        : status === 'failed' || status === 'cancelled' || status === 'canceled'
          ? 1
          : status === 'succeeded' || status === 'success'
            ? 0
            : null,
    stdout: candidate.stdOut ?? null,
    stderr: candidate.stdErr ?? null,
  };
}

class RmmOvershellWorker {
  private timer: NodeJS.Timeout | null = null;
  private running = false;
  private inFlight = false;

  async start(): Promise<void> {
    if (this.running) return;
    this.running = true;
    console.log('[RMM-WORKER] starting; polling every 5s');
    this.scheduleNextPoll(0);
  }

  async stop(): Promise<void> {
    this.running = false;
    if (this.timer) {
      clearTimeout(this.timer);
      this.timer = null;
    }
  }

  private scheduleNextPoll(delay: number): void {
    if (!this.running) return;
    this.timer = setTimeout(() => {
      void this.poll();
    }, delay);
  }

  private async poll(): Promise<void> {
    if (!this.running) return;
    if (this.inFlight) {
      this.scheduleNextPoll(POLL_INTERVAL_MS);
      return;
    }
    this.inFlight = true;
    try {
      // First — sweep any rows that timed out.
      const timed = await listTimedOutExecutions();
      for (const row of timed) {
        try {
          await markExecutionTimeout(row.id);
          await audit.log({
            userId: row.performedByUserId ?? undefined,
            action: 'rmm.execute.timeout',
            resource: 'datto_device',
            resourceId: row.targetDeviceUid,
            details: { execution_id: row.id, script_id: row.scriptId },
          });
        } catch (err) {
          console.warn('[RMM-WORKER] timeout-mark failed:', err);
        }
      }

      // Then poll the running ones. Skip b2_upload rows — those are flipped
      // to complete by the LogLift webhook handler when the upload lands;
      // there's no stdout to poll for. The 5-minute timeout sweep above
      // still catches stuck rows.
      const running = (await listRunningExecutions()).filter(
        (r) => r.transport !== 'b2_upload'
      );
      if (running.length === 0) {
        return;
      }
      const client = getDattoRMMClient();
      for (const row of running) {
        if (!row.jobUid) continue;
        try {
          const payload = (await client.getJobResults(
            row.jobUid,
            row.targetDeviceUid
          )) as DattoJobResults;
          const r = extractResult(payload, row.targetDeviceUid);
          if (!r.done) continue; // still running, try next tick.

          // Redact outputs before persistence — defensive against scripts
          // that might dump credentials. The script library is curated, but
          // belt + braces.
          const redactedStdout = r.stdout
            ? (redact({ s: r.stdout }) as { s: string }).s
            : null;
          const redactedStderr = r.stderr
            ? (redact({ s: r.stderr }) as { s: string }).s
            : null;

          // Parse via the script's own parser. Failure is non-fatal — we
          // still persist the raw output.
          let parsed: unknown = null;
          let parseError: string | null = null;
          if (redactedStdout && r.exitCode === 0) {
            const script = getScript(row.scriptId);
            if (script) {
              try {
                parsed = script.parseOutput(redactedStdout);
              } catch (err) {
                parseError = err instanceof Error ? err.message : String(err);
              }
            }
          }

          await markExecutionComplete({
            id: row.id,
            exitCode: r.exitCode,
            rawStdout: redactedStdout,
            rawStderr: redactedStderr,
            parsedEvidence: parsed,
            parseError,
          });

          await audit.log({
            userId: row.performedByUserId ?? undefined,
            action: r.exitCode === 0 ? 'rmm.execute.complete' : 'rmm.execute.failed',
            resource: 'datto_device',
            resourceId: row.targetDeviceUid,
            details: {
              execution_id: row.id,
              script_id: row.scriptId,
              exit_code: r.exitCode,
              parse_error: parseError,
            },
          });
        } catch (err) {
          console.warn(
            `[RMM-WORKER] poll error for execution ${row.id}:`,
            err instanceof Error ? err.message : err
          );
          // Don't mark failed yet — could be transient. timeout_at will
          // catch it eventually.
        }
      }
    } catch (err) {
      console.error('[RMM-WORKER] poll loop error:', err);
    } finally {
      this.inFlight = false;
      this.scheduleNextPoll(POLL_INTERVAL_MS);
    }
  }
}

export const rmmOvershellWorker = new RmmOvershellWorker();

function shouldAutoStart(): boolean {
  if (typeof window !== 'undefined') return false;
  if (process.env.VITEST === 'true') return false;
  if (process.env.NODE_ENV === 'production') return true;
  return process.env.RMM_WORKER_AUTOSTART === '1';
}

if (shouldAutoStart()) {
  rmmOvershellWorker.start().catch((err) => {
    console.error('[RMM-WORKER] failed to auto-start:', err);
  });
}

export const _RMM_WORKER_INTERNALS = { POLL_INTERVAL_MS, extractResult };