Assessor_2.0/server/gpo/gpo-backup-job-processor.ts

import { createHash } from 'node:crypto';
import type { StoredGpoBackupInstance, StoredGpoPolicyAnalysis } from '../../src/domain/gpo-analysis-persistence';
import { GPO_POLICY_IDENTITY_RESOLVER } from '../../src/domain/gpo-policy-identity';
import type { ImportProcessingRun, IngestionJob } from '../../src/domain/ingestion-persistence';
import type { AssessorRepository } from '../../src/persistence/assessor-repository';
import { RepositoryNotFoundError, RepositoryValidationError } from '../../src/persistence/repository-errors';
import { ImportError } from '../../src/shared/errors';
import type { BackendArtifactStore } from '../artifacts/backend-artifact-store';
import { GPO_BACKUP_ANALYZER_VERSION } from './gpo-backup-submission-service';
import type { GpoBackupWorkerExecutor } from './gpo-inventory-executor';

export const GPO_BACKUP_JOB_PROCESSOR_VERSION = '2.2.4';

export interface GpoBackupJobProcessorOptions {
  leaseDurationMs?: number;
  retryDelayMs?: number;
  maximumArtifactBytes: number;
  now?: () => Date;
  afterAnalysis?: (projectId: string, processingRunId: string) => Promise<unknown>;
}

export interface GpoBackupJobProcessResult {
  jobId: string;
  status: IngestionJob['status'];
  processingRunId?: string;
  backupInstanceCount?: number;
  errorCode?: string;
}

function stableId(prefix: string, value: string): string {
  return `${prefix}-${createHash('sha256').update(value).digest('hex').slice(0, 32)}`;
}

function errorCode(error: unknown): string {
  if (error instanceof ImportError) return error.diagnostics[0]?.code ?? 'gpo-analysis-rejected';
  if (error instanceof RepositoryNotFoundError) return 'gpo-analysis-reference-missing';
  if (error instanceof RepositoryValidationError) return 'gpo-analysis-persistence-invalid';
  return 'gpo-analysis-failed';
}

export class GpoBackupJobProcessor {
  private readonly leaseDurationMs: number;
  private readonly retryDelayMs: number;
  private readonly maximumArtifactBytes: number;
  private readonly now: () => Date;

  constructor(
    private readonly repository: AssessorRepository,
    private readonly artifactStore: BackendArtifactStore,
    private readonly executor: GpoBackupWorkerExecutor,
    private readonly options: GpoBackupJobProcessorOptions
  ) {
    this.leaseDurationMs = options.leaseDurationMs ?? 60_000;
    this.retryDelayMs = options.retryDelayMs ?? 5_000;
    this.maximumArtifactBytes = options.maximumArtifactBytes;
    this.now = options.now ?? (() => new Date());
    if (!Number.isInteger(this.leaseDurationMs) || this.leaseDurationMs < 1_000 || this.leaseDurationMs > 3_600_000) {
      throw new Error('Die GPO-Job-Leasedauer ist ungültig.');
    }
    if (!Number.isInteger(this.retryDelayMs) || this.retryDelayMs < 0 || !Number.isSafeInteger(this.maximumArtifactBytes) || this.maximumArtifactBytes < 1) {
      throw new Error('GPO-Job-Retry- oder Artefaktlimit ist ungültig.');
    }
  }

  async processNext(workerId: string, projectId?: string): Promise<GpoBackupJobProcessResult | null> {
    const claimedAt = this.now().toISOString();
    const job = await this.repository.claimNextIngestionJob({
      workerId,
      projectId,
      now: claimedAt,
      leaseDurationMs: this.leaseDurationMs,
      kinds: ['analyze-gpo-backup']
    });
    if (!job) return null;
    const processingRunId = stableId('gpo-processing', `${job.id}\0${GPO_BACKUP_JOB_PROCESSOR_VERSION}`);
    try {
      if (!job.artifactId) throw new RepositoryValidationError('Der GPO-Analysejob besitzt kein Artefakt.');
      const existingRun = (await this.repository.listProcessingRuns(job.artifactId)).find(run => run.id === processingRunId);
      if (existingRun) {
        await this.options.afterAnalysis?.(job.projectId, processingRunId);
        const completed = await this.repository.completeIngestionJob(job.id, workerId, this.now().toISOString());
        const existingInstances = (await this.repository.listGpoBackupInstances(job.projectId))
          .filter(record => record.processingRunId === processingRunId);
        return { jobId: job.id, status: completed.status, processingRunId, backupInstanceCount: existingInstances.length };
      }
      const artifact = (await this.repository.listImportArtifacts(job.projectId)).find(item => item.id === job.artifactId);
      if (!artifact) throw new RepositoryNotFoundError('Das GPO-Jobartefakt ist nicht im Projekt vorhanden.');
      if (artifact.storage.kind !== 'backend-controlled') throw new RepositoryValidationError('Das GPO-Jobartefakt wurde nicht backendkontrolliert aufbewahrt.');
      const bytes = await this.artifactStore.read({
        storageKey: artifact.storage.storageKey,
        expectedSha256: artifact.sha256,
        maximumBytes: this.maximumArtifactBytes
      });
      const analysis = await this.executor.executeBackup({
        bytes,
        options: {
          projectId: job.projectId,
          sourceArtifactId: artifact.id,
          discoveredAt: artifact.createdAt
        }
      });
      const completedAt = this.now().toISOString();
      const diagnosticCodes = analysis.diagnosticCodes;
      const partial = analysis.backupInstances.length === 0 || analysis.backupInstances.some(instance =>
        instance.inventory.status !== 'complete'
        || instance.inventory.entries.some(entry => entry.disposition !== 'supported')
      );
      const run: ImportProcessingRun = {
        version: 1,
        id: processingRunId,
        projectId: job.projectId,
        artifactId: artifact.id,
        supersedesRunId: job.supersedesProcessingRunId,
        status: partial ? 'partial' : 'complete',
        processors: [
          { kind: 'artifact-parser', id: 'gpo-backup-ingestion', version: GPO_BACKUP_JOB_PROCESSOR_VERSION },
          { kind: 'normalizer', id: GPO_POLICY_IDENTITY_RESOLVER.id, version: GPO_POLICY_IDENTITY_RESOLVER.version }
        ],
        diagnosticCodes,
        startedAt: claimedAt,
        completedAt
      };
      const instances: StoredGpoBackupInstance[] = analysis.backupInstances.map(instance => ({
        version: 1,
        id: stableId('stored-gpo-backup', `${processingRunId}\0${instance.id}`),
        projectId: job.projectId,
        artifactId: artifact.id,
        processingRunId,
        instance,
        createdAt: completedAt
      }));
      const previousPolicyAnalysis = job.supersedesProcessingRunId
        ? (await this.repository.listGpoPolicyAnalyses(job.projectId)).find(item => item.run.sourceProcessingRunId === job.supersedesProcessingRunId)
        : undefined;
      const policyAnalysisRunId = stableId('gpo-policy-analysis', `${processingRunId}\0${GPO_POLICY_IDENTITY_RESOLVER.version}`);
      const storedResolutions = analysis.identityResolutions.map(item => ({
        version: 1 as const,
        id: stableId('gpo-policy-resolution', `${policyAnalysisRunId}\0${item.observationId}`),
        projectId: job.projectId,
        analysisRunId: policyAnalysisRunId,
        sourceProcessingRunId: processingRunId,
        backupInstanceId: item.backupInstanceId,
        observationId: item.observationId,
        resolution: item.resolution
      }));
      const policyAnalysis: StoredGpoPolicyAnalysis = {
        run: {
          version: 1,
          id: policyAnalysisRunId,
          projectId: job.projectId,
          artifactId: artifact.id,
          sourceProcessingRunId: processingRunId,
          supersedesAnalysisRunId: previousPolicyAnalysis?.run.id,
          analyzer: { id: 'gpo-policy-identity-analysis', version: GPO_BACKUP_ANALYZER_VERSION },
          identityResolver: GPO_POLICY_IDENTITY_RESOLVER,
          status: storedResolutions.some(item => item.resolution.status === 'unresolved') ? 'partial' : 'complete',
          observationCount: storedResolutions.length,
          resolvedCount: storedResolutions.filter(item => item.resolution.status === 'resolved').length,
          diagnosticCodes: [...new Set(storedResolutions.flatMap(item => item.resolution.status === 'unresolved' ? [item.resolution.reasonCode] : []))].sort(),
          startedAt: claimedAt,
          completedAt
        },
        resolutions: storedResolutions
      };
      await this.repository.saveGpoBackupAnalysis(run, instances, policyAnalysis);
      await this.options.afterAnalysis?.(job.projectId, processingRunId);
      const completed = await this.repository.completeIngestionJob(job.id, workerId, this.now().toISOString());
      return { jobId: job.id, status: completed.status, processingRunId, backupInstanceCount: instances.length };
    } catch (error) {
      const failedAt = this.now();
      const code = errorCode(error);
      const retryAt = new Date(failedAt.getTime() + this.retryDelayMs * Math.max(1, 2 ** (job.attempts - 1))).toISOString();
      const failed = await this.repository.failIngestionJob(job.id, workerId, failedAt.toISOString(), code, retryAt,
        code !== 'artifact-store-object-missing');
      return { jobId: job.id, status: failed.status, processingRunId, errorCode: code };
    }
  }
}

export class GpoBackupJobRunner {
  private timer?: ReturnType<typeof setInterval>;
  private active?: Promise<void>;

  constructor(
    private readonly processor: GpoBackupJobProcessor,
    private readonly workerId: string,
    private readonly pollIntervalMs = 1_000
  ) {
    if (!workerId.trim() || !Number.isInteger(pollIntervalMs) || pollIntervalMs < 100 || pollIntervalMs > 60_000) {
      throw new Error('GPO-Job-Runner-Konfiguration ist ungültig.');
    }
  }

  start(): void {
    if (this.timer) return;
    const tick = () => {
      if (this.active) return;
      this.active = this.processor.processNext(this.workerId).then(() => undefined, () => undefined)
        .finally(() => { this.active = undefined; });
    };
    tick();
    this.timer = setInterval(tick, this.pollIntervalMs);
    this.timer.unref?.();
  }

  async stop(): Promise<void> {
    if (this.timer) clearInterval(this.timer);
    this.timer = undefined;
    await this.active;
  }
}