Assessor_2.0/server/gpo/gpo-inventory-worker-client.ts

import { Worker } from 'node:worker_threads';
import { defaultImportResourcePolicy } from '../../src/importers/resource-policy';
import { ImportError } from '../../src/shared/errors';
import type { MicrosoftSecurityBaselineAnalysis } from '../../src/domain/microsoft-security-baseline-analysis';
import type {
  GpoContainerInventoryResult,
  GpoContainerRecognitionResult
} from './gpo-container-inventory';
import type {
  GpoBackupWorkerExecutionRequest,
  GpoBackupWorkerResult,
  GpoBackupWorkerExecutor,
  GpoInventoryExecutionRequest,
  GpoInventoryExecutor,
  GpoRecognitionExecutionRequest,
  MicrosoftBaselineWorkerExecutionRequest,
  MicrosoftBaselineWorkerExecutor
} from './gpo-inventory-executor';

interface WorkerResult {
  ok: boolean;
  inventory?: GpoContainerInventoryResult;
  backup?: GpoBackupWorkerResult;
  recognition?: GpoContainerRecognitionResult;
  microsoftBaseline?: MicrosoftSecurityBaselineAnalysis;
  error?: { message: string; diagnostics: ConstructorParameters<typeof ImportError>[1] };
}

type GpoWorkerExecutionRequest =
  | GpoInventoryExecutionRequest
  | GpoRecognitionExecutionRequest
  | GpoBackupWorkerExecutionRequest
  | MicrosoftBaselineWorkerExecutionRequest;

function workerErrorCode(error: unknown): string | undefined {
  if (!error || typeof error !== 'object' || !('code' in error) || typeof error.code !== 'string') return undefined;
  return /^[A-Z0-9_]{1,80}$/.test(error.code) ? error.code : undefined;
}

function workerErrorName(error: unknown): string {
  return error instanceof Error && /^[A-Za-z][A-Za-z0-9]{0,79}$/.test(error.name) ? error.name : 'Error';
}

function isWorkerMemoryError(error: unknown): boolean {
  if (workerErrorCode(error) === 'ERR_WORKER_OUT_OF_MEMORY') return true;
  const message = error instanceof Error ? error.message : '';
  return /out of memory|heap.{0,40}memory|memory limit|allocation failed/i.test(message);
}

export interface NodeGpoInventoryWorkerExecutorOptions {
  workerUrl: URL;
  timeoutMs?: number;
  maximumConcurrent?: number;
  maximumSourceBytes?: number;
  maximumOldGenerationMb?: number;
  maximumYoungGenerationMb?: number;
}

export class NodeGpoInventoryWorkerExecutor implements GpoInventoryExecutor, GpoBackupWorkerExecutor, MicrosoftBaselineWorkerExecutor {
  private readonly timeoutMs: number;
  private readonly maximumConcurrent: number;
  private readonly maximumSourceBytes: number;
  private readonly maximumOldGenerationMb: number;
  private readonly maximumYoungGenerationMb: number;
  private readonly workers = new Set<Worker>();
  private active = 0;
  private closed = false;

  constructor(private readonly options: NodeGpoInventoryWorkerExecutorOptions) {
    this.timeoutMs = options.timeoutMs ?? 30_000;
    this.maximumConcurrent = options.maximumConcurrent ?? 1;
    this.maximumSourceBytes = options.maximumSourceBytes ?? defaultImportResourcePolicy.maximumFileBytes;
    this.maximumOldGenerationMb = options.maximumOldGenerationMb ?? 1_024;
    this.maximumYoungGenerationMb = options.maximumYoungGenerationMb ?? 128;
  }

  execute(request: GpoInventoryExecutionRequest): Promise<GpoContainerInventoryResult> {
    return this.executeBounded(request, result => result.inventory);
  }

  recognize(request: GpoInventoryExecutionRequest): Promise<GpoContainerRecognitionResult> {
    const workerRequest: GpoRecognitionExecutionRequest = { ...request, operation: 'recognition' };
    return this.executeBounded(workerRequest, result => result.recognition);
  }

  executeBackup(request: GpoBackupWorkerExecutionRequest): Promise<GpoBackupWorkerResult> {
    return this.executeBounded(request, result => result.backup);
  }
  executeMicrosoftBaseline(request: Omit<MicrosoftBaselineWorkerExecutionRequest, 'operation'>): Promise<MicrosoftSecurityBaselineAnalysis> {
    return this.executeBounded({ ...request, operation: 'microsoft-baseline' }, result => result.microsoftBaseline);
  }

  async close(): Promise<void> {
    this.closed = true;
    await Promise.all([...this.workers].map(worker => worker.terminate().catch(() => undefined)));
  }

  private executeBounded<T>(
    request: GpoWorkerExecutionRequest,
    select: (result: WorkerResult) => T | undefined
  ): Promise<T> {
    if (this.closed) return Promise.reject(this.error('gpo-inventory-executor-closed', 'Der GPO-Inventardienst wird beendet.'));
    if (request.bytes.byteLength > this.maximumSourceBytes) {
      return Promise.reject(this.error('gpo-inventory-source-size-limit', 'Der GPO-Container überschreitet das Worker-Eingabelimit.'));
    }
    if (this.active >= this.maximumConcurrent) {
      return Promise.reject(this.error('gpo-inventory-capacity-limit', 'Die GPO-Inventarisierung ist ausgelastet. Bitte später erneut versuchen.'));
    }
    this.active += 1;
    return this.run(request, select).finally(() => { this.active -= 1; });
  }

  private error(code: string, message: string, details?: Record<string, string>): ImportError {
    return new ImportError(message, [{
      severity: 'error',
      code,
      message,
      ...(details ? { details } : {})
    }]);
  }

  private workerFailure(error: unknown): ImportError {
    const outOfMemory = isWorkerMemoryError(error);
    return this.error(
      outOfMemory ? 'gpo-inventory-memory-limit' : 'gpo-inventory-worker-error',
      outOfMemory ? 'Die GPO-Inventarisierung hat ihr Speicherlimit überschritten.' : 'Der GPO-Inventar-Worker ist fehlgeschlagen.',
      { workerErrorCode: workerErrorCode(error) ?? 'UNAVAILABLE', workerErrorName: workerErrorName(error) }
    );
  }

  private run<T>(
    request: GpoWorkerExecutionRequest,
    select: (result: WorkerResult) => T | undefined
  ): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      // Keep the caller-owned bytes intact, but transfer the worker copy instead of
      // structured-cloning it. Large uncompressed GPO transports would otherwise
      // temporarily occupy the full payload in both heaps during worker startup.
      const transferableBytes = new Uint8Array(request.bytes.byteLength);
      transferableBytes.set(request.bytes);
      const workerRequest = { ...request, bytes: transferableBytes };
      let worker: Worker;
      try {
        worker = new Worker(this.options.workerUrl, {
          workerData: workerRequest,
          transferList: [transferableBytes.buffer],
          resourceLimits: {
            maxOldGenerationSizeMb: this.maximumOldGenerationMb,
            maxYoungGenerationSizeMb: this.maximumYoungGenerationMb,
            stackSizeMb: 8
          }
        });
      } catch (error) {
        reject(this.workerFailure(error));
        return;
      }
      this.workers.add(worker);
      let settled = false;
      const settle = (action: () => void) => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        this.workers.delete(worker);
        action();
        void worker.terminate();
      };
      const timer = setTimeout(() => settle(() => reject(this.error(
        'gpo-inventory-time-limit', `Die GPO-Inventarisierung wurde nach ${this.timeoutMs} Millisekunden beendet.`
      ))), this.timeoutMs);
      worker.once('message', (result: WorkerResult) => settle(() => {
        const value = result.ok ? select(result) : undefined;
        if (value !== undefined) resolve(value);
        else reject(new ImportError(result.error?.message ?? 'GPO-Inventar-Worker fehlgeschlagen.', result.error?.diagnostics ?? []));
      }));
      worker.once('error', error => settle(() => reject(this.workerFailure(error))));
      worker.once('exit', code => {
        if (!settled) settle(() => reject(this.error('gpo-inventory-worker-exit', `Der GPO-Inventar-Worker wurde mit Status ${code} beendet.`)));
      });
    });
  }
}