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.0.0'; export interface GpoBackupJobProcessorOptions { leaseDurationMs?: number; retryDelayMs?: number; maximumArtifactBytes: number; now?: () => Date; } 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, 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): Promise { const claimedAt = this.now().toISOString(); const job = await this.repository.claimNextIngestionJob({ workerId, 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) { 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); 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); return { jobId: job.id, status: failed.status, processingRunId, errorCode: code }; } } } export class GpoBackupJobRunner { private timer?: ReturnType; private active?: Promise; 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 { if (this.timer) clearInterval(this.timer); this.timer = undefined; await this.active; } }