import { appendRiskAcceptanceReview, assertRiskAcceptanceUnchanged, type RiskAcceptanceCommand } from '../../src/domain/risk-acceptance'; import { randomUUID } from 'node:crypto'; import { appendScopedRiskAcceptance, scopedRiskTargetFromOccurrences, type ScopedRiskAcceptance, type ScopedRiskAcceptanceTarget, type ScopedRiskAcceptanceWrite } from '../../src/domain/scoped-risk-acceptance'; import type { Pool, PoolClient, QueryResult, QueryResultRow } from 'pg'; import type { Assessment, AssessmentProject, CanonicalAssessmentModel, EvidenceRecord, FindingDefinition, FindingOccurrence, FindingReview, ImportDiagnostic, ImportFileRecord, ScanRun, SourceTableRecord } from '../../src/domain/canonical-model'; import type { LocalComparisonRule } from '../../src/domain/comparison'; import type { ImportFingerprint, ImportFingerprintInput } from '../../src/domain/import-workflow'; import type { ImportArtifact, ImportProcessingRun, IngestionJob, IngestionJobClaim } from '../../src/domain/ingestion-persistence'; import type { StoredGpoBackupInstance, StoredGpoPolicyAnalysis } from '../../src/domain/gpo-analysis-persistence'; import type { MicrosoftSecurityBaselineClassification, MicrosoftSecurityBaselineRecord } from '../../src/domain/microsoft-security-baseline'; import type { StoredMicrosoftSecurityBaselineAnalysis } from '../../src/domain/microsoft-security-baseline-analysis'; import type { EnvironmentProfile, StoredPolicySnapshotAggregation } from '../../src/domain/gpo-snapshot-aggregation'; import type { PolicyEvaluationResult, PolicyEvaluationRun, StoredPolicyEvaluation } from '../../src/domain/policy-evaluation'; import { POLICY_EVALUATION_ENGINE, storedPolicyEvaluationValidationErrors } from '../../src/domain/policy-evaluation'; import type { StoredPolicyFindingRun } from '../../src/domain/policy-finding'; import type { EnvironmentObservation, EnvironmentSnapshot, EnvironmentSnapshotQuery, ScannerReportSection } from '../../src/domain/environment-context'; import { environmentCoverage } from '../../src/domain/environment-context'; import type { CveEnrichmentRecord } from '../../src/domain/cve-enrichment'; import type { AiOverviewRun } from '../../src/domain/ai-overview'; import { aiOverviewRunValidationErrors } from '../../src/domain/ai-overview'; import type { RegulatoryMappingBatchRun, RegulatoryMappingDecision, RegulatoryMappingRun } from '../../src/domain/regulatory-mapping'; import { regulatoryMappingBatchRunValidationErrors, regulatoryMappingDecisionValidationErrors, regulatoryMappingRunValidationErrors } from '../../src/domain/regulatory-mapping'; import { cveEnrichmentValidationErrors, cveIdFromRuleId, normalizeCveId } from '../../src/domain/cve-enrichment'; import { storedPolicyFindingRunValidationErrors } from '../../src/domain/policy-finding'; import { environmentProfileValidationErrors, storedPolicySnapshotAggregationValidationErrors } from '../../src/domain/gpo-snapshot-aggregation'; import { storedMicrosoftSecurityBaselineAnalysisValidationErrors } from '../../src/domain/microsoft-security-baseline-analysis'; import type { DbStructureColumn, DbStructureForeignKey, DbStructureLayerId, DbStructureResponse, DbStructureTable } from '../../src/domain/db-structure'; import { storedGpoBackupInstanceValidationErrors, storedGpoPolicyAnalysisValidationErrors } from '../../src/domain/gpo-analysis-persistence'; import { createEnvironmentUnifiedResult, createPolicyEvaluationUnifiedResult, createCurrentPolicyObservationUnifiedResults, createPolicyUnifiedResult, createScannerUnifiedResult, deriveUnifiedResultFacets, filterUnifiedResults, listUnifiedResultCategories, queryUnifiedResultItems, outcomeFromLegacyStatus, scannerResultCategories, technicalStateFromLegacyStatus, type UnifiedResultCategory, type UnifiedResultFacetResult, type UnifiedResultProjection, type UnifiedResultQuery, type UnifiedResultQueryResult } from '../../src/domain/unified-result'; import { createPolicyConsistencyUnifiedResults } from '../../src/evaluation/policy-consistency-results'; import { deriveExposureState, deriveIsIoe, deriveResultType } from '../../src/domain/result-contract-mapping'; import { importArtifactValidationErrors, ingestionJobValidationErrors, processingRunValidationErrors } from '../../src/domain/ingestion-persistence'; import { assertCanonicalAssessmentModel } from '../../src/domain/canonical-validation'; import { processingScopeIssues } from '../../src/domain/processing-compatibility'; import { compareReimportSourceHashes } from '../../src/domain/reimport-source'; import { planReimportReviewTransfer, type ReimportReviewTransferPlan, type ReimportReviewTransferSummary } from '../../src/domain/reimport-review-transfer'; import { canonicalJson } from '../../src/shared/canonical-json'; import { validateLocalComparisonRules } from '../../src/comparison/local-mapping-rules'; import type { AssessorRepository, AssessmentDeleteConfirmation, ArtifactDeletionRequest, CanonicalImportCommit, ControlledDeleteConfirmation, FindingOccurrenceQuery, FindingOccurrenceQueryResult, ImportCommitResult, ImportSession, ImportSessionStart, ImportSessionStatus, ImportStagingData, ProfileRegistration, ProjectDataExport, ScanRunDeleteConfirmation, StoredComparison } from '../../src/persistence/assessor-repository'; import { ImportSessionStateError, RepositoryConflictError, RepositoryNotFoundError, RepositoryValidationError } from '../../src/persistence/repository-errors'; interface Queryable { query(text: string, values?: unknown[]): Promise>; } interface DataRow extends QueryResultRow { data: T; } interface RevisionRow extends QueryResultRow { revision: string | number; } const emptyStaging = (): Required => ({ importFiles: [], sourceTables: [], reportSections: [], environmentObservations: [], findingDefinitions: [], findingOccurrences: [], evidence: [], diagnostics: [] }); function clone(value: T): T { return JSON.parse(JSON.stringify(value)) as T; } function json(value: unknown): string { return JSON.stringify(value); } function dataRows(result: QueryResult>): T[] { return result.rows.map(row => clone(row.data)); } function dataRow(result: QueryResult>): T | null { return result.rows[0] ? clone(result.rows[0].data) : null; } async function persistTransferredReviews(client: PoolClient, reviews: readonly FindingReview[]): Promise { for (const review of reviews) { const definitionId = review.target.kind === 'definition' ? review.target.definitionId : null; const occurrenceId = review.target.kind === 'occurrence' ? review.target.occurrenceId : null; await client.query(` INSERT INTO assessor_reviews (id, target_kind, definition_id, occurrence_id, revision, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb) ON CONFLICT (id) DO NOTHING `, [review.id, review.target.kind, definitionId, occurrenceId, review.revision, review.updatedAt, json(review)]); } } async function persistScannerContext( queryable: Queryable, sections: ScannerReportSection[], observations: EnvironmentObservation[] ): Promise { for (const section of sections) await queryable.query(` INSERT INTO assessor_scanner_report_sections (id, scan_run_id, import_file_id, tool_id, section_key, section_state, ordinal, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) `, [section.id, section.scanRunId, section.importFileId, section.tool.id, section.sectionKey, section.state, section.ordinal, json(section)]); for (const observation of observations) await queryable.query(` INSERT INTO assessor_environment_observations (id, scan_run_id, import_file_id, tool_id, subject_type, property_key, observation_state, observed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) `, [observation.id, observation.scanRunId, observation.importFileId, observation.tool.id, observation.subject.type, observation.property, observation.state, observation.observedAt ?? null, json(observation)]); } function persistedOutcome(occurrence: FindingOccurrence) { return occurrence.outcome?.normalizedValue ?? outcomeFromLegacyStatus(occurrence.status.normalizedValue ?? 'unknown'); } function persistedTechnicalExecutionState(occurrence: FindingOccurrence) { return occurrence.technicalExecutionState?.normalizedValue ?? technicalStateFromLegacyStatus(occurrence.status.normalizedValue ?? 'unknown'); } /** * RESULT_CONTRACT-Facetten (docs/RESULT_CONTRACT.md). Wie bei * persistedOutcome/persistedTechnicalExecutionState gilt: ein am Vorkommen * explizit gesetzter Wert wird übernommen, sonst ausschließlich über * src/domain/result-contract-mapping.ts abgeleitet — nie aus einem rohen * Quellstring. */ function persistedResultType(occurrence: FindingOccurrence) { return occurrence.resultType ?? deriveResultType(occurrence.sourceIdentity.tool.id); } function persistedExposureState(occurrence: FindingOccurrence) { if (occurrence.exposureState !== undefined) return occurrence.exposureState; return deriveExposureState(persistedResultType(occurrence), persistedOutcome(occurrence)) ?? null; } function persistedIsIoe(occurrence: FindingOccurrence) { if (occurrence.isIoe !== undefined) return occurrence.isIoe; return deriveIsIoe(persistedResultType(occurrence), persistedExposureState(occurrence) ?? undefined).isIoe; } function persistedIoeRuleId(occurrence: FindingOccurrence) { if (occurrence.ioeRuleId !== undefined) return occurrence.ioeRuleId; return deriveIsIoe(persistedResultType(occurrence), persistedExposureState(occurrence) ?? undefined).ioeRuleId; } function persistedIoeRuleVersion(occurrence: FindingOccurrence) { if (occurrence.ioeRuleVersion !== undefined) return occurrence.ioeRuleVersion; return deriveIsIoe(persistedResultType(occurrence), persistedExposureState(occurrence) ?? undefined).ioeRuleVersion; } function persistedIoeDerivationMethod(occurrence: FindingOccurrence) { if (occurrence.ioeDerivationMethod !== undefined) return occurrence.ioeDerivationMethod; return deriveIsIoe(persistedResultType(occurrence), persistedExposureState(occurrence) ?? undefined).ioeDerivationMethod; } /** * Nachvollziehbarkeit des angewendeten Feld-Mappings je Vorkommen: ein am * Vorkommen explizit gesetzter Wert wird übernommen, sonst aus * processedWith.mappingVersion der zugehörigen ImportFileRecord. */ function persistedMappingRuleVersion(occurrence: FindingOccurrence, importFile?: ImportFileRecord): string { return occurrence.mappingRuleVersion ?? importFile?.processedWith.mappingVersion ?? 'unknown'; } async function persistOccurrenceCategories( client: Queryable, occurrence: FindingOccurrence, knownDefinition?: FindingDefinition, knownImportFile?: ImportFileRecord ): Promise { const definition = knownDefinition ?? dataRow(await client.query>( 'SELECT data FROM assessor_finding_definitions WHERE id = $1', [occurrence.definitionId] )); const importFile = knownImportFile ?? dataRow(await client.query>( 'SELECT data FROM assessor_import_files WHERE id = $1', [occurrence.importFileId] )); if (!definition || !importFile) return; const categories = scannerResultCategories(definition, occurrence, importFile); for (const [ordinal, category] of categories.entries()) { const profileId = category.sourceProfileId ?? importFile.processedWith.profileId; const profileVersion = category.sourceProfileVersion ?? importFile.processedWith.profileVersion; await client.query(` INSERT INTO assessor_result_categories ( namespace, category_code, category_label, category_type, source_tool, source_profile_id, source_profile_version ) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (namespace, category_code, source_profile_id, source_profile_version) DO UPDATE SET category_label = EXCLUDED.category_label `, [ category.namespace, category.categoryCode, category.categoryLabel, category.categoryType, category.sourceTool ?? null, profileId, profileVersion ]); await client.query(` INSERT INTO assessor_finding_occurrence_categories ( occurrence_id, namespace, category_code, source_profile_id, source_profile_version, is_primary, ordinal ) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING `, [ occurrence.id, category.namespace, category.categoryCode, profileId, profileVersion, category.isPrimary, ordinal ]); } } function isRepositoryError(error: unknown): error is Error { return error instanceof RepositoryConflictError || error instanceof RepositoryNotFoundError || error instanceof RepositoryValidationError || error instanceof ImportSessionStateError; } function translateDatabaseError(error: unknown): Error { if (isRepositoryError(error)) return error; const code = typeof error === 'object' && error && 'code' in error ? String(error.code) : ''; if (['23505', '40001', '40P01'].includes(code)) return new RepositoryConflictError('Die Daten wurden parallel geändert oder verletzen eine Eindeutigkeitsregel.'); if (code === '23503') return new RepositoryNotFoundError('Eine referenzierte Entität ist nicht vorhanden.'); if (['22007', '22008', '22P02', '23502', '23514'].includes(code)) return new RepositoryValidationError('Die Daten verletzen eine Datenbank- oder Formatregel.'); return error instanceof Error ? error : new Error('Unbekannter PostgreSQL-Fehler.'); } function validateStaging(start: ImportSessionStart, staged: Required): void { if (!staged.importFiles.length) throw new RepositoryValidationError('Ein abgeschlossener Import benötigt mindestens eine Importdatei.'); if (staged.importFiles.some(file => file.scanRunId !== start.scanRun.id)) throw new RepositoryValidationError('Alle Importdateien müssen zum gestarteten Scan-Lauf gehören.'); const fileIds = new Set(staged.importFiles.map(file => file.id)); const tableIds = new Set(staged.sourceTables.map(table => table.id)); const occurrenceIds = new Set(staged.findingOccurrences.map(occurrence => occurrence.id)); if (staged.sourceTables.some(table => !fileIds.has(table.importFileId))) throw new RepositoryValidationError('Alle Quelltabellen müssen zu einer Datei derselben Importsitzung gehören.'); if (staged.reportSections.some(section => section.scanRunId !== start.scanRun.id || !fileIds.has(section.importFileId))) { throw new RepositoryValidationError('Alle Scanner-Reportabschnitte müssen zum gestarteten Scan-Lauf und dessen Importdateien gehören.'); } if (staged.environmentObservations.some(observation => observation.scanRunId !== start.scanRun.id || !fileIds.has(observation.importFileId))) { throw new RepositoryValidationError('Alle Umgebungsbeobachtungen müssen zum gestarteten Scan-Lauf und dessen Importdateien gehören.'); } if (staged.findingOccurrences.some(occurrence => occurrence.scanRunId !== start.scanRun.id || !fileIds.has(occurrence.importFileId))) { throw new RepositoryValidationError('Alle Finding-Vorkommen müssen zum gestarteten Scan-Lauf und dessen Importdateien gehören.'); } if (staged.findingOccurrences.some(occurrence => occurrence.sourceTableId && !tableIds.has(occurrence.sourceTableId))) { throw new RepositoryValidationError('Finding-Vorkommen dürfen nur Quelltabellen derselben Importsitzung referenzieren.'); } if (staged.evidence.some(record => !occurrenceIds.has(record.occurrenceId))) throw new RepositoryValidationError('Evidenz muss zu einem Finding derselben Importsitzung gehören.'); if (staged.diagnostics.some(diagnostic => !fileIds.has(diagnostic.importFileId) || (diagnostic.sourceTableId && !tableIds.has(diagnostic.sourceTableId)))) { throw new RepositoryValidationError('Importdiagnosen müssen zu Dateien und Tabellen derselben Importsitzung gehören.'); } } class PostgresImportSession implements ImportSession { private currentStatus: ImportSessionStatus = 'pending'; private staged = emptyStaging(); constructor( readonly id: string, private readonly onCommit: (staged: Required) => Promise, private readonly onRollback: () => void ) {} get status(): ImportSessionStatus { return this.currentStatus; } private ensurePending(): void { if (this.currentStatus !== 'pending') throw new ImportSessionStateError(`Importsitzung ${this.id} ist bereits ${this.currentStatus}.`); } async stage(data: ImportStagingData): Promise { this.ensurePending(); this.staged.importFiles.push(...clone(data.importFiles ?? [])); this.staged.sourceTables.push(...clone(data.sourceTables ?? [])); this.staged.reportSections.push(...clone(data.reportSections ?? [])); this.staged.environmentObservations.push(...clone(data.environmentObservations ?? [])); this.staged.findingDefinitions.push(...clone(data.findingDefinitions ?? [])); this.staged.findingOccurrences.push(...clone(data.findingOccurrences ?? [])); this.staged.evidence.push(...clone(data.evidence ?? [])); this.staged.diagnostics.push(...clone(data.diagnostics ?? [])); } async commit(): Promise { this.ensurePending(); const result = await this.onCommit(clone(this.staged)); this.currentStatus = 'committed'; return result; } async rollback(): Promise { this.ensurePending(); this.onRollback(); this.currentStatus = 'rolled-back'; this.staged = emptyStaging(); } } const DB_STRUCTURE_LAYER_BY_TABLE: Record = { assessor_scan_runs: 'import', assessor_import_files: 'import', assessor_source_tables: 'import', assessor_scanner_report_sections: 'import', assessor_environment_observations: 'canonical', assessor_import_fingerprints: 'import', assessor_import_diagnostics: 'import', assessor_import_artifacts: 'import', assessor_import_processing_runs: 'import', assessor_ingestion_jobs: 'import', assessor_gpo_backup_instances: 'import', assessor_policy_identities: 'canonical', assessor_policy_identity_analysis_runs: 'canonical', assessor_policy_identity_resolutions: 'canonical', assessor_microsoft_security_baselines: 'governance', assessor_microsoft_security_baseline_classifications: 'governance', assessor_microsoft_security_baseline_analyses: 'governance', assessor_admx_catalog_versions: 'governance', assessor_microsoft_security_baseline_settings: 'governance', assessor_artifact_deletion_outbox: 'import', assessor_profile_registrations: 'mapping', assessor_local_comparison_rules: 'mapping', assessor_canonical_assets: 'canonical', assessor_asset_properties: 'canonical', assessor_asset_edges: 'canonical', assessor_edge_coverage: 'canonical', assessor_finding_definitions: 'result', assessor_cve_enrichments: 'result', assessor_regulatory_mappings: 'result', assessor_regulatory_mapping_decisions: 'governance', assessor_regulatory_mapping_batches: 'governance', assessor_ai_overview_runs: 'governance', assessor_finding_occurrences: 'result', assessor_result_categories: 'result', assessor_finding_occurrence_categories: 'result', assessor_evidence: 'result', assessor_reviews: 'governance', assessor_scoped_risk_acceptances: 'governance', assessor_comparisons: 'governance', assessor_projects: 'core', assessor_assessments: 'core', assessor_repository_state: 'core' }; const DB_STRUCTURE_EXCLUDED_TABLES = new Set(['assessor_schema_migrations']); function dbStructureLayer(tableName: string): DbStructureLayerId { return DB_STRUCTURE_LAYER_BY_TABLE[tableName] ?? 'other'; } export class PostgresAssessorRepository implements AssessorRepository { private cachedRevision: number; private readonly activeImportSessions = new Set(); private constructor(private readonly pool: Pool, initialRevision: number) { this.cachedRevision = initialRevision; } static async create(pool: Pool): Promise { const result = await pool.query('SELECT revision FROM assessor_repository_state WHERE singleton = true'); if (!result.rows[0]) throw new Error('assessor_repository_state fehlt; Datenbankmigrationen wurden nicht ausgeführt.'); return new PostgresAssessorRepository(pool, Number(result.rows[0].revision)); } get revision(): number { return this.cachedRevision; } private async mutate( work: (client: PoolClient) => Promise, expectedRevision?: number, changed: (result: T) => boolean = () => true ): Promise { const client = await this.pool.connect(); try { await client.query('BEGIN'); const revisionResult = await client.query( 'SELECT revision FROM assessor_repository_state WHERE singleton = true FOR UPDATE' ); const databaseRevision = Number(revisionResult.rows[0]?.revision ?? -1); if (expectedRevision !== undefined && databaseRevision !== expectedRevision) { throw new RepositoryConflictError('Repository wurde seit Beginn des Vorgangs geändert; der Vorgang muss erneut gestartet werden.'); } const result = await work(client); if (changed(result)) { const updated = await client.query( 'UPDATE assessor_repository_state SET revision = revision + 1 WHERE singleton = true RETURNING revision' ); this.cachedRevision = Number(updated.rows[0].revision); } else { this.cachedRevision = databaseRevision; } await client.query('COMMIT'); return result; } catch (error) { try { await client.query('ROLLBACK'); } catch { /* connection failure keeps original error */ } throw translateDatabaseError(error); } finally { client.release(); } } async listProjects(): Promise { return dataRows(await this.pool.query>('SELECT data FROM assessor_projects ORDER BY id')); } async getProject(projectId: string): Promise { return dataRow(await this.pool.query>('SELECT data FROM assessor_projects WHERE id = $1', [projectId])); } async saveProject(project: AssessmentProject): Promise { if (!project.id.trim() || !project.name.trim()) throw new RepositoryValidationError('Projekt-ID und Projektname sind erforderlich.'); await this.mutate(async client => { await client.query(` INSERT INTO assessor_projects (id, name, created_at, updated_at, attributes, data) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, updated_at = EXCLUDED.updated_at, attributes = EXCLUDED.attributes, data = EXCLUDED.data `, [project.id, project.name, project.createdAt, project.updatedAt, json(project.attributes), json(project)]); }); } async listAssessments(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_assessments WHERE project_id = $1 ORDER BY id', [projectId] )); } async listProjectScanRuns(projectId: string): Promise { return dataRows(await this.pool.query>(` SELECT run.data FROM assessor_scan_runs run JOIN assessor_assessments assessment ON assessment.id = run.assessment_id WHERE assessment.project_id = $1 AND run.data->>'supersededByScanRunId' IS NULL ORDER BY run.sequence, run.id `, [projectId])); } async listProjectImportFiles(projectId: string): Promise { return dataRows(await this.pool.query>(` SELECT file.data FROM assessor_import_files file JOIN assessor_scan_runs run ON run.id = file.scan_run_id JOIN assessor_assessments assessment ON assessment.id = run.assessment_id WHERE assessment.project_id = $1 AND run.data->>'supersededByScanRunId' IS NULL ORDER BY COALESCE(file.data->>'lastModifiedAt', file.data->>'importedAt') DESC, file.id `, [projectId])); } async saveAssessment(assessment: Assessment): Promise { if (!assessment.id.trim() || !assessment.name.trim()) throw new RepositoryValidationError('Assessment-ID und Assessmentname sind erforderlich.'); await this.mutate(async client => { const existing = await client.query<{ project_id: string } & QueryResultRow>( 'SELECT project_id FROM assessor_assessments WHERE id = $1 FOR UPDATE', [assessment.id] ); if (existing.rows[0] && existing.rows[0].project_id !== assessment.projectId) { throw new RepositoryConflictError('Ein Assessment kann nicht in ein anderes Projekt verschoben werden.'); } await client.query(` INSERT INTO assessor_assessments (id, project_id, name, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, updated_at = EXCLUDED.updated_at, data = EXCLUDED.data `, [assessment.id, assessment.projectId, assessment.name, assessment.createdAt, assessment.updatedAt, json(assessment)]); }); } private async assessmentModel(queryable: Queryable, assessmentId: string): Promise { const assessment = dataRow(await queryable.query>( 'SELECT data FROM assessor_assessments WHERE id = $1', [assessmentId] )); if (!assessment) return null; const project = dataRow(await queryable.query>( 'SELECT data FROM assessor_projects WHERE id = $1', [assessment.projectId] )); if (!project) return null; const runs = await queryable.query>( `SELECT data FROM assessor_scan_runs WHERE assessment_id = $1 AND data->>'supersededByScanRunId' IS NULL ORDER BY id`, [assessmentId] ); const files = await queryable.query>( `SELECT f.data FROM assessor_import_files f JOIN assessor_scan_runs r ON r.id = f.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY f.id`, [assessmentId] ); const tables = await queryable.query>( `SELECT t.data FROM assessor_source_tables t JOIN assessor_import_files f ON f.id = t.import_file_id JOIN assessor_scan_runs r ON r.id = f.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY t.id`, [assessmentId] ); const reportSections = await queryable.query>( `SELECT s.data FROM assessor_scanner_report_sections s JOIN assessor_scan_runs r ON r.id = s.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY s.ordinal, s.id`, [assessmentId] ); const environmentObservations = await queryable.query>( `SELECT o.data FROM assessor_environment_observations o JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY o.property_key, o.id`, [assessmentId] ); const definitions = await queryable.query>( `SELECT DISTINCT d.data, d.id FROM assessor_finding_definitions d JOIN assessor_finding_occurrences o ON o.definition_id = d.id JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY d.id`, [assessmentId] ); const occurrences = await queryable.query>( `SELECT o.data FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY o.id`, [assessmentId] ); const occurrenceRows = dataRows(occurrences); const cveIds = [...new Set(occurrenceRows.flatMap(occurrence => { const id = cveIdFromRuleId(occurrence.sourceIdentity.canonicalRuleId ?? occurrence.sourceIdentity.originalRuleId); return id ? [id] : []; }))]; const cveEnrichments = cveIds.length ? await queryable.query>( 'SELECT data FROM assessor_cve_enrichments WHERE cve_id = ANY($1::text[]) ORDER BY cve_id', [cveIds] ) : null; const evidence = await queryable.query>( `SELECT e.data FROM assessor_evidence e JOIN assessor_finding_occurrences o ON o.id = e.occurrence_id JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY e.id`, [assessmentId] ); const reviews = await queryable.query>(` SELECT DISTINCT rv.data, rv.id FROM assessor_reviews rv WHERE EXISTS ( SELECT 1 FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL AND (rv.occurrence_id = o.id OR rv.definition_id = o.definition_id) ) ORDER BY rv.id `, [assessmentId]); const diagnostics = await queryable.query>( `SELECT d.data FROM assessor_import_diagnostics d JOIN assessor_import_files f ON f.id = d.import_file_id JOIN assessor_scan_runs r ON r.id = f.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL ORDER BY d.id`, [assessmentId] ); return { schemaVersion: 1, project, assessment, scanRuns: dataRows(runs), importFiles: dataRows(files), sourceTables: dataRows(tables), reportSections: dataRows(reportSections), environmentObservations: dataRows(environmentObservations), findingDefinitions: dataRows(definitions), findingOccurrences: occurrenceRows, cveEnrichments: cveEnrichments ? dataRows(cveEnrichments) : [], evidence: dataRows(evidence), reviews: dataRows(reviews), diagnostics: dataRows(diagnostics) }; } loadAssessmentModel(assessmentId: string): Promise { return this.assessmentModel(this.pool, assessmentId); } async getCveEnrichment(value: string): Promise { const cveId = normalizeCveId(value); if (!cveId) return null; return dataRow(await this.pool.query>( 'SELECT data FROM assessor_cve_enrichments WHERE cve_id = $1', [cveId] )); } async listCveEnrichments(values: string[]): Promise { const cveIds = [...new Set(values.map(normalizeCveId).filter((value): value is string => Boolean(value)))]; if (!cveIds.length) return []; return dataRows(await this.pool.query>( 'SELECT data FROM assessor_cve_enrichments WHERE cve_id = ANY($1::text[]) ORDER BY cve_id', [cveIds] )); } async saveCveEnrichment(record: CveEnrichmentRecord): Promise { const errors = cveEnrichmentValidationErrors(record); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { await client.query(` INSERT INTO assessor_cve_enrichments (cve_id, source, last_modified_at, fetched_at, data) VALUES ($1, $2, $3, $4, $5::jsonb) ON CONFLICT (cve_id) DO UPDATE SET source = EXCLUDED.source, last_modified_at = EXCLUDED.last_modified_at, fetched_at = EXCLUDED.fetched_at, data = EXCLUDED.data `, [record.cveId, record.source, record.lastModifiedAt, record.fetchedAt, json(record)]); }); } async createAiOverviewRun(run: AiOverviewRun): Promise { const errors = aiOverviewRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const idempotent = dataRow(await client.query>(` SELECT data FROM assessor_ai_overview_runs WHERE project_id = $1 AND scope_key = $2 AND idempotency_key = $3 `, [run.projectId, run.scopeKey, run.idempotencyKey])); if (idempotent) return idempotent; await client.query(` INSERT INTO assessor_ai_overview_runs (id, project_id, scope_key, assessment_ids, scan_run_ids, idempotency_key, run_status, input_hash, prompt_version, output_schema_version, model, created_at, updated_at, data) VALUES ($1, $2, $3, $4::text[], $5::text[], $6, $7, $8, $9, $10, $11, $12, $13, $14::jsonb) `, [run.id, run.projectId, run.scopeKey, run.assessmentIds, run.scanRunIds, run.idempotencyKey, run.status, run.inputHash, run.promptVersion, run.outputSchemaVersion, run.model, run.createdAt, run.updatedAt, json(run)]); return clone(run); }, undefined, result => result.id === run.id); } async updateAiOverviewRun(run: AiOverviewRun): Promise { const errors = aiOverviewRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_ai_overview_runs WHERE id = $1 FOR UPDATE', [run.id] )); if (!existing) throw new RepositoryNotFoundError(`AI overview run ${run.id} does not exist.`); if (existing.projectId !== run.projectId || existing.scopeKey !== run.scopeKey || existing.inputHash !== run.inputHash || existing.idempotencyKey !== run.idempotencyKey || existing.model !== run.model) { throw new RepositoryConflictError('Immutable AI overview run identity cannot be changed.'); } if (['succeeded', 'failed'].includes(existing.status)) throw new RepositoryConflictError('A completed AI overview run is immutable.'); if (existing.status === 'queued' && !['running', 'failed'].includes(run.status) || existing.status === 'running' && !['running', 'succeeded', 'failed'].includes(run.status)) { throw new RepositoryConflictError(`Invalid AI overview transition from ${existing.status} to ${run.status}.`); } await client.query(` UPDATE assessor_ai_overview_runs SET run_status = $2, updated_at = $3, data = $4::jsonb WHERE id = $1 `, [run.id, run.status, run.updatedAt, json(run)]); return clone(run); }); } async listAiOverviewRuns(projectId: string, scopeKey?: string): Promise { return dataRows(await this.pool.query>(` SELECT data FROM assessor_ai_overview_runs WHERE project_id = $1 AND ($2::text IS NULL OR scope_key = $2) ORDER BY created_at DESC, id DESC `, [projectId, scopeKey ?? null])); } async getRegulatoryMappingRun(id: string): Promise { return dataRow(await this.pool.query>( 'SELECT data FROM assessor_regulatory_mappings WHERE id = $1', [id] )); } async claimRegulatoryMappingRun(run: RegulatoryMappingRun, forceRegeneration: boolean): Promise<{ run: RegulatoryMappingRun; claimed: boolean }> { const errors = regulatoryMappingRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const inserted = await client.query(` INSERT INTO assessor_regulatory_mappings (id, rule_key, source_tool, source_rule_id, run_status, query_hash, corpus_id, corpus_version, embedding_model, retrieval_version, prompt_version, output_schema_version, model, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb) ON CONFLICT DO NOTHING RETURNING id `, [run.id, run.ruleKey, run.sourceTool, run.sourceRuleId ?? null, run.status, run.queryHash, run.corpusId, run.corpusVersion, run.embeddingModel, run.retrievalVersion, run.promptVersion, run.outputSchemaVersion, run.model, run.createdAt, run.updatedAt, json(run)]); if (inserted.rowCount) return { run: clone(run), claimed: true }; const existing = dataRow(await client.query>( 'SELECT data FROM assessor_regulatory_mappings WHERE id = $1 FOR UPDATE', [run.id] )); if (!existing) throw new RepositoryConflictError('Regulatory mapping identity conflicted with another pipeline tuple.'); if (['queued', 'running'].includes(existing.status) || existing.status === 'succeeded' && !forceRegeneration) { return { run: existing, claimed: false }; } await client.query(` UPDATE assessor_regulatory_mappings SET run_status = $2, query_hash = $3, created_at = $4, updated_at = $5, data = $6::jsonb WHERE id = $1 `, [run.id, run.status, run.queryHash, run.createdAt, run.updatedAt, json(run)]); return { run: clone(run), claimed: true }; }); } async saveRegulatoryMappingRun(run: RegulatoryMappingRun): Promise { const errors = regulatoryMappingRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { await client.query(` INSERT INTO assessor_regulatory_mappings (id, rule_key, source_tool, source_rule_id, run_status, query_hash, corpus_id, corpus_version, embedding_model, retrieval_version, prompt_version, output_schema_version, model, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb) ON CONFLICT (id) DO UPDATE SET run_status = excluded.run_status, updated_at = excluded.updated_at, data = excluded.data `, [run.id, run.ruleKey, run.sourceTool, run.sourceRuleId ?? null, run.status, run.queryHash, run.corpusId, run.corpusVersion, run.embeddingModel, run.retrievalVersion, run.promptVersion, run.outputSchemaVersion, run.model, run.createdAt, run.updatedAt, json(run)]); return clone(run); }); } async listRegulatoryMappingRuns(): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_regulatory_mappings ORDER BY created_at DESC, id DESC' )); } async createRegulatoryMappingBatchRun(run: RegulatoryMappingBatchRun): Promise { const errors = regulatoryMappingBatchRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const inserted = await client.query(` INSERT INTO assessor_regulatory_mapping_batches (id, project_id, scope_key, filter_hash, idempotency_key, run_status, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) ON CONFLICT DO NOTHING RETURNING data `, [run.id, run.projectId, run.scopeKey, run.filterHash, run.idempotencyKey, run.status, run.createdAt, run.updatedAt, json(run)]); if (inserted.rowCount) return clone(run); const existing = dataRow(await client.query>(` SELECT data FROM assessor_regulatory_mapping_batches WHERE project_id = $1 AND (idempotency_key = $2 OR (scope_key = $3 AND run_status IN ('queued', 'running'))) ORDER BY (idempotency_key = $2) DESC, created_at DESC, id DESC LIMIT 1 `, [run.projectId, run.idempotencyKey, run.scopeKey])); if (!existing) throw new RepositoryConflictError('Regulatory mapping batch conflicted with another active request.'); return existing; }); } async updateRegulatoryMappingBatchRun(run: RegulatoryMappingBatchRun): Promise { const errors = regulatoryMappingBatchRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_regulatory_mapping_batches WHERE id = $1 FOR UPDATE', [run.id] )); if (!existing) throw new RepositoryNotFoundError('Regulatory mapping batch does not exist.'); if (existing.projectId !== run.projectId || existing.scopeKey !== run.scopeKey || existing.idempotencyKey !== run.idempotencyKey) { throw new RepositoryConflictError('Regulatory mapping batch identity is immutable.'); } if (['completed', 'failed'].includes(existing.status)) throw new RepositoryConflictError('A completed regulatory mapping batch is immutable.'); const allowed = existing.status === 'queued' ? ['running', 'failed'] : ['running', 'completed', 'failed']; if (!allowed.includes(run.status)) throw new RepositoryConflictError(`Invalid regulatory mapping batch transition from ${existing.status} to ${run.status}.`); await client.query(` UPDATE assessor_regulatory_mapping_batches SET run_status = $2, updated_at = $3, data = $4::jsonb WHERE id = $1 `, [run.id, run.status, run.updatedAt, json(run)]); return clone(run); }); } async listRegulatoryMappingBatchRuns(projectId: string, scopeKey?: string): Promise { return dataRows(await this.pool.query>(` SELECT data FROM assessor_regulatory_mapping_batches WHERE project_id = $1 AND ($2::text IS NULL OR scope_key = $2) ORDER BY created_at DESC, id DESC `, [projectId, scopeKey ?? null])); } async createRegulatoryMappingDecision( decision: RegulatoryMappingDecision, expectedLatestDecisionId?: string ): Promise { const errors = regulatoryMappingDecisionValidationErrors(decision); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const run = dataRow(await client.query>( 'SELECT data FROM assessor_regulatory_mappings WHERE id = $1 FOR UPDATE', [decision.mappingId] )); if (!run || run.ruleKey !== decision.ruleKey || run.status !== 'succeeded') { throw new RepositoryNotFoundError('Completed regulatory mapping does not exist for this rule.'); } if (!run.candidates.some(candidate => candidate.controlId === decision.controlId)) { throw new RepositoryValidationError('Decision control was not a grounded candidate.'); } const latest = dataRow(await client.query>(` SELECT data FROM assessor_regulatory_mapping_decisions WHERE rule_key = $1 AND control_id = $2 ORDER BY created_at DESC, id DESC LIMIT 1 `, [decision.ruleKey, decision.controlId])); if ((expectedLatestDecisionId ?? undefined) !== (latest?.id ?? undefined)) { throw new RepositoryConflictError('Regulatory mapping decision changed before this update.'); } const stored = { ...decision, ...(latest ? { previousDecisionId: latest.id } : {}) }; await client.query(` INSERT INTO assessor_regulatory_mapping_decisions (id, rule_key, source_tool, source_rule_id, mapping_id, control_id, decision, previous_decision_id, created_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb) `, [stored.id, stored.ruleKey, stored.sourceTool, stored.sourceRuleId ?? null, stored.mappingId, stored.controlId, stored.decision, stored.previousDecisionId ?? null, stored.createdAt, json(stored)]); return clone(stored); }); } async listRegulatoryMappingDecisions(ruleKey: string): Promise { return dataRows(await this.pool.query>(` SELECT data FROM assessor_regulatory_mapping_decisions WHERE rule_key = $1 ORDER BY created_at DESC, id DESC `, [ruleKey])); } async beginImport(start: ImportSessionStart): Promise { if (!start.sessionId.trim()) throw new RepositoryValidationError('Eine Importsitzung benötigt eine ID.'); if (this.activeImportSessions.has(start.sessionId)) throw new RepositoryConflictError(`Importsitzung ${start.sessionId} ist bereits aktiv.`); const context = await this.pool.query<{ revision: string; project_exists: boolean; assessment_exists: boolean; run_exists: boolean } & QueryResultRow>(` SELECT s.revision, EXISTS (SELECT 1 FROM assessor_projects WHERE id = $1) AS project_exists, EXISTS (SELECT 1 FROM assessor_assessments WHERE id = $2 AND project_id = $1) AS assessment_exists, EXISTS (SELECT 1 FROM assessor_scan_runs WHERE id = $3 OR (assessment_id = $2 AND sequence = $4)) AS run_exists FROM assessor_repository_state s WHERE s.singleton = true `, [start.projectId, start.assessmentId, start.scanRun.id, start.scanRun.sequence]); const row = context.rows[0]; if (!row?.project_exists || !row.assessment_exists) throw new RepositoryNotFoundError('Projekt oder Assessment für den Import fehlt.'); if (start.scanRun.assessmentId !== start.assessmentId) throw new RepositoryValidationError('Der Scan-Lauf gehört nicht zum gewählten Assessment.'); if (row.run_exists) throw new RepositoryConflictError('Scan-Lauf-ID oder Scan-Reihenfolge ist bereits belegt.'); const expectedRevision = Number(row.revision); this.activeImportSessions.add(start.sessionId); return new PostgresImportSession( start.sessionId, staged => this.commitImport(start, staged, expectedRevision), () => { this.activeImportSessions.delete(start.sessionId); } ); } private async commitImport(start: ImportSessionStart, staged: Required, expectedRevision: number): Promise { validateStaging(start, staged); try { return await this.mutate(async client => { await client.query( 'INSERT INTO assessor_scan_runs (id, assessment_id, sequence, imported_at, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [start.scanRun.id, start.assessmentId, start.scanRun.sequence, start.scanRun.importedAt, json(start.scanRun)] ); for (const file of staged.importFiles) await client.query( 'INSERT INTO assessor_import_files (id, scan_run_id, file_name, format, imported_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb)', [file.id, file.scanRunId, file.fileName, file.format, file.importedAt, json(file)] ); for (const table of staged.sourceTables) await client.query( 'INSERT INTO assessor_source_tables (id, import_file_id, table_index, name, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [table.id, table.importFileId, table.index, table.name, json(table)] ); await persistScannerContext(client, staged.reportSections, staged.environmentObservations); for (const definition of staged.findingDefinitions) { const result = await client.query(` INSERT INTO assessor_finding_definitions (id, canonical_key, data) VALUES ($1, $2, $3::jsonb) ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data WHERE assessor_finding_definitions.canonical_key = EXCLUDED.canonical_key RETURNING id `, [definition.id, definition.canonicalKey, json(definition)]); if (!result.rowCount) throw new RepositoryConflictError(`Finding-Definition ${definition.id} besitzt einen widersprüchlichen kanonischen Schlüssel.`); } const stagedImportFileById = new Map(staged.importFiles.map(file => [file.id, file])); const stagedDefinitionById = new Map(staged.findingDefinitions.map(definition => [definition.id, definition])); for (const occurrence of staged.findingOccurrences) { const occurrenceImportFile = stagedImportFileById.get(occurrence.importFileId); await client.query(` INSERT INTO assessor_finding_occurrences ( id, definition_id, scan_run_id, import_file_id, source_table_id, tool_id, severity, finding_status, outcome_normalized, technical_execution_state, result_type, is_ioe, ioe_rule_id, ioe_rule_version, ioe_derivation_method, exposure_state, mapping_rule_version, data ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb) `, [ occurrence.id, occurrence.definitionId, occurrence.scanRunId, occurrence.importFileId, occurrence.sourceTableId ?? null, occurrence.sourceIdentity.tool.id, occurrence.severity.normalizedValue, occurrence.status.normalizedValue, persistedOutcome(occurrence), persistedTechnicalExecutionState(occurrence), persistedResultType(occurrence), persistedIsIoe(occurrence), persistedIoeRuleId(occurrence), persistedIoeRuleVersion(occurrence), persistedIoeDerivationMethod(occurrence), persistedExposureState(occurrence), persistedMappingRuleVersion(occurrence, occurrenceImportFile), json(occurrence) ]); await persistOccurrenceCategories( client, occurrence, stagedDefinitionById.get(occurrence.definitionId), occurrenceImportFile ); } for (const record of staged.evidence) await client.query( 'INSERT INTO assessor_evidence (id, occurrence_id, data) VALUES ($1, $2, $3::jsonb)', [record.id, record.occurrenceId, json(record)] ); for (const diagnostic of staged.diagnostics) await client.query(` INSERT INTO assessor_import_diagnostics (id, import_file_id, source_table_id, severity, code, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) `, [diagnostic.id, diagnostic.importFileId, diagnostic.sourceTableId ?? null, diagnostic.severity, diagnostic.code, json(diagnostic)]); const model = await this.assessmentModel(client, start.assessmentId); if (!model) throw new RepositoryValidationError('Das importierte Assessment konnte nicht aufgebaut werden.'); try { assertCanonicalAssessmentModel(model); } catch (error) { throw new RepositoryValidationError(error instanceof Error ? error.message : 'Importmodell ist ungültig.'); } return { sessionId: start.sessionId, scanRunId: start.scanRun.id, repositoryRevision: expectedRevision + 1, importedFiles: staged.importFiles.length, importedOccurrences: staged.findingOccurrences.length }; }, expectedRevision); } finally { this.activeImportSessions.delete(start.sessionId); } } async commitCanonicalImport(input: CanonicalImportCommit): Promise { const { model } = input; const run = input.replacesScanRunId ? { ...model.scanRuns[0], supersedesScanRunId: input.replacesScanRunId } : model.scanRuns[0]; model.scanRuns = [run]; const fingerprintHashes = new Set(input.fingerprints.map(entry => entry.sha256)); if (model.importFiles.some(file => !file.sha256 || !fingerprintHashes.has(file.sha256)) || input.fingerprints.some(entry => !model.importFiles.some(file => file.sha256 === entry.sha256))) { throw new RepositoryValidationError('Kanonische Importdateien und Fingerprint-Auswahl stimmen nicht überein.'); } if (model.scanRuns.length !== 1) throw new RepositoryValidationError('Ein kanonischer Batchimport benötigt genau einen Scan-Lauf.'); try { assertCanonicalAssessmentModel(model); } catch (error) { throw new RepositoryValidationError(error instanceof Error ? error.message : 'Importmodell ist ungültig.'); } input.fingerprints.forEach(entry => { if (!/^[a-f0-9]{64}$/.test(entry.sha256)) throw new RepositoryValidationError('SHA-256-Fingerabdruck ist ungültig.'); }); const expectedRevision = this.cachedRevision; return this.mutate(async client => { let reviewTransfer: ReimportReviewTransferPlan | undefined; if (input.replacesScanRunId) { const replacedResult = await client.query<{ assessment_id: string; data: ScanRun } & QueryResultRow>(` SELECT assessment_id, data FROM assessor_scan_runs WHERE id = $1 FOR UPDATE `, [input.replacesScanRunId]); const replaced = replacedResult.rows[0]; if (!replaced || replaced.assessment_id !== model.assessment.id || replaced.data.supersededByScanRunId) { throw new RepositoryValidationError('Der zu ersetzende Scanbericht ist nicht mehr als aktiver Lauf vorhanden.'); } const replacedFiles = await client.query<{ data: ImportFileRecord } & QueryResultRow>( 'SELECT data FROM assessor_import_files WHERE scan_run_id = $1 ORDER BY id', [input.replacesScanRunId] ); if (compareReimportSourceHashes( replacedFiles.rows.map(row => row.data.sha256), model.importFiles.map(file => file.sha256) ) !== 'match') { throw new RepositoryValidationError('Der Reimport muss kryptografisch exakt denselben Scanbericht ersetzen.'); } const previousOccurrences = dataRows(await client.query>( 'SELECT data FROM assessor_finding_occurrences WHERE scan_run_id = $1 ORDER BY id', [input.replacesScanRunId] )); const oldOccurrenceIds = previousOccurrences.map(item => item.id); const oldDefinitionIds = [...new Set(previousOccurrences.map(item => item.definitionId))]; const reviews = dataRows(await client.query>(` SELECT data FROM assessor_reviews WHERE occurrence_id = ANY($1::text[]) OR definition_id = ANY($2::text[]) ORDER BY id `, [oldOccurrenceIds, oldDefinitionIds])); reviewTransfer = planReimportReviewTransfer(previousOccurrences, model.findingOccurrences, reviews, run.importedAt); if (model.importFiles[0]) model.importFiles[0].reimportReviewTransfer = { replacesScanRunId: input.replacesScanRunId, ...reviewTransfer.summary }; const superseded = { ...replaced.data, supersededByScanRunId: run.id, supersededAt: run.importedAt }; await client.query('UPDATE assessor_scan_runs SET data = $2::jsonb WHERE id = $1', [input.replacesScanRunId, json(superseded)]); } const existingFiles = await client.query<{ data: ImportFileRecord }>(` SELECT f.data FROM assessor_import_files f JOIN assessor_scan_runs r ON r.id = f.scan_run_id WHERE r.assessment_id = $1 AND r.data->>'supersededByScanRunId' IS NULL `, [model.assessment.id]); if (!input.replacesScanRunId && processingScopeIssues([...existingFiles.rows.map(row => row.data), ...model.importFiles]).length) { throw new RepositoryValidationError('Der Timeslot enthält unbekannte oder semantisch inkompatible Verarbeitungsstände. Bitte vollständig mit dem aktuellen Profil reimportieren oder einen getrennten Timeslot verwenden.'); } await client.query(` INSERT INTO assessor_projects (id, name, created_at, updated_at, attributes, data) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, updated_at = EXCLUDED.updated_at, attributes = EXCLUDED.attributes, data = EXCLUDED.data `, [model.project.id, model.project.name, model.project.createdAt, model.project.updatedAt, json(model.project.attributes), json(model.project)]); const assessmentResult = await client.query(` INSERT INTO assessor_assessments (id, project_id, name, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, updated_at = EXCLUDED.updated_at, data = EXCLUDED.data WHERE assessor_assessments.project_id = EXCLUDED.project_id RETURNING id `, [model.assessment.id, model.assessment.projectId, model.assessment.name, model.assessment.createdAt, model.assessment.updatedAt, json(model.assessment)]); if (!assessmentResult.rowCount) throw new RepositoryConflictError('Ein Assessment kann nicht in ein anderes Projekt verschoben werden.'); await client.query( 'INSERT INTO assessor_scan_runs (id, assessment_id, sequence, imported_at, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [run.id, run.assessmentId, run.sequence, run.importedAt, json(run)] ); for (const file of model.importFiles) await client.query( 'INSERT INTO assessor_import_files (id, scan_run_id, file_name, format, imported_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb)', [file.id, file.scanRunId, file.fileName, file.format, file.importedAt, json(file)] ); for (const table of model.sourceTables) await client.query( 'INSERT INTO assessor_source_tables (id, import_file_id, table_index, name, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [table.id, table.importFileId, table.index, table.name, json(table)] ); await persistScannerContext(client, model.reportSections ?? [], model.environmentObservations ?? []); for (const definition of model.findingDefinitions) { const result = await client.query(` INSERT INTO assessor_finding_definitions (id, canonical_key, data) VALUES ($1, $2, $3::jsonb) ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data WHERE assessor_finding_definitions.canonical_key = EXCLUDED.canonical_key RETURNING id `, [definition.id, definition.canonicalKey, json(definition)]); if (!result.rowCount) throw new RepositoryConflictError(`Finding-Definition ${definition.id} besitzt einen widersprüchlichen kanonischen Schlüssel.`); } const importFileById = new Map(model.importFiles.map(file => [file.id, file])); const definitionById = new Map(model.findingDefinitions.map(definition => [definition.id, definition])); for (const occurrence of model.findingOccurrences) { const occurrenceImportFile = importFileById.get(occurrence.importFileId); await client.query(` INSERT INTO assessor_finding_occurrences ( id, definition_id, scan_run_id, import_file_id, source_table_id, tool_id, severity, finding_status, outcome_normalized, technical_execution_state, result_type, is_ioe, ioe_rule_id, ioe_rule_version, ioe_derivation_method, exposure_state, mapping_rule_version, data ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb) `, [ occurrence.id, occurrence.definitionId, occurrence.scanRunId, occurrence.importFileId, occurrence.sourceTableId ?? null, occurrence.sourceIdentity.tool.id, occurrence.severity.normalizedValue, occurrence.status.normalizedValue, persistedOutcome(occurrence), persistedTechnicalExecutionState(occurrence), persistedResultType(occurrence), persistedIsIoe(occurrence), persistedIoeRuleId(occurrence), persistedIoeRuleVersion(occurrence), persistedIoeDerivationMethod(occurrence), persistedExposureState(occurrence), persistedMappingRuleVersion(occurrence, occurrenceImportFile), json(occurrence) ]); await persistOccurrenceCategories( client, occurrence, definitionById.get(occurrence.definitionId), occurrenceImportFile ); } if (reviewTransfer) await persistTransferredReviews(client, reviewTransfer.reviews); for (const record of model.evidence) await client.query( 'INSERT INTO assessor_evidence (id, occurrence_id, data) VALUES ($1, $2, $3::jsonb)', [record.id, record.occurrenceId, json(record)] ); for (const diagnostic of model.diagnostics) await client.query(` INSERT INTO assessor_import_diagnostics (id, import_file_id, source_table_id, severity, code, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) `, [diagnostic.id, diagnostic.importFileId, diagnostic.sourceTableId ?? null, diagnostic.severity, diagnostic.code, json(diagnostic)]); for (const entry of input.fingerprints) { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_import_fingerprints WHERE sha256 = $1 FOR UPDATE', [entry.sha256] )); const fingerprint: ImportFingerprint = existing ? { ...existing, ...entry, lastImportedAt: run.importedAt, importCount: existing.importCount + 1 } : { ...entry, firstImportedAt: run.importedAt, lastImportedAt: run.importedAt, importCount: 1 }; await client.query(` INSERT INTO assessor_import_fingerprints (sha256, file_name, size_bytes, last_modified, first_imported_at, last_imported_at, import_count, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) ON CONFLICT (sha256) DO UPDATE SET file_name = EXCLUDED.file_name, size_bytes = EXCLUDED.size_bytes, last_modified = EXCLUDED.last_modified, last_imported_at = EXCLUDED.last_imported_at, import_count = EXCLUDED.import_count, data = EXCLUDED.data `, [fingerprint.sha256, fingerprint.fileName, fingerprint.sizeBytes, fingerprint.lastModified, fingerprint.firstImportedAt, fingerprint.lastImportedAt, fingerprint.importCount, json(fingerprint)]); } const stored = await this.assessmentModel(client, model.assessment.id); if (!stored) throw new RepositoryValidationError('Das importierte Assessment konnte nicht aufgebaut werden.'); try { assertCanonicalAssessmentModel(stored); } catch (error) { throw new RepositoryValidationError(error instanceof Error ? error.message : 'Importmodell ist ungültig.'); } return { sessionId: input.sessionId, scanRunId: run.id, repositoryRevision: expectedRevision + 1, importedFiles: model.importFiles.length, importedOccurrences: model.findingOccurrences.length, reimportReviewTransfer: reviewTransfer?.summary }; }, expectedRevision); } async findImportFingerprints(hashes: string[]): Promise { if (!hashes.length) return []; return dataRows(await this.pool.query>( 'SELECT data FROM assessor_import_fingerprints WHERE sha256 = ANY($1::text[]) ORDER BY sha256', [hashes] )); } async recordImportFingerprints(entries: ImportFingerprintInput[], importedAt = new Date().toISOString()): Promise { entries.forEach(entry => { if (!/^[a-f0-9]{64}$/.test(entry.sha256)) throw new RepositoryValidationError('SHA-256-Fingerabdruck ist ungültig.'); }); await this.mutate(async client => { for (const entry of entries) { const existing = await client.query>( 'SELECT data FROM assessor_import_fingerprints WHERE sha256 = $1 FOR UPDATE', [entry.sha256] ); const previous = dataRow(existing); const fingerprint: ImportFingerprint = previous ? { ...previous, ...entry, lastImportedAt: importedAt, importCount: previous.importCount + 1 } : { ...entry, firstImportedAt: importedAt, lastImportedAt: importedAt, importCount: 1 }; await client.query(` INSERT INTO assessor_import_fingerprints (sha256, file_name, size_bytes, last_modified, first_imported_at, last_imported_at, import_count, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) ON CONFLICT (sha256) DO UPDATE SET file_name = EXCLUDED.file_name, size_bytes = EXCLUDED.size_bytes, last_modified = EXCLUDED.last_modified, last_imported_at = EXCLUDED.last_imported_at, import_count = EXCLUDED.import_count, data = EXCLUDED.data `, [fingerprint.sha256, fingerprint.fileName, fingerprint.sizeBytes, fingerprint.lastModified, fingerprint.firstImportedAt, fingerprint.lastImportedAt, fingerprint.importCount, json(fingerprint)]); } }); return this.findImportFingerprints(entries.map(entry => entry.sha256)); } async findOccurrences(query: FindingOccurrenceQuery = {}): Promise { const offset = query.offset ?? 0; const limit = query.limit ?? 1000; if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(limit) || limit < 0) { throw new RepositoryValidationError('Offset und Limit müssen nichtnegative Ganzzahlen sein.'); } const values: unknown[] = []; const conditions: string[] = ["r.data->>'supersededByScanRunId' IS NULL"]; const add = (expression: string, value: unknown) => { values.push(value); conditions.push(expression.replace('?', `$${values.length}`)); }; if (query.projectId) add('a.project_id = ?', query.projectId); if (query.assessmentId) add('r.assessment_id = ?', query.assessmentId); if (query.scanRunId) add('o.scan_run_id = ?', query.scanRunId); if (query.definitionId) add('o.definition_id = ?', query.definitionId); if (query.toolId) add('lower(o.tool_id) = lower(?)', query.toolId); if (query.severity) add('o.severity = ?', query.severity); if (query.status) add('o.finding_status = ?', query.status); if (query.search?.trim()) { values.push(query.search.trim()); conditions.push(`(o.search_document @@ websearch_to_tsquery('simple'::regconfig, $${values.length}) OR d.search_document @@ websearch_to_tsquery('simple'::regconfig, $${values.length}))`); } values.push(Math.min(10000, limit), offset); const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : ''; const result = await this.pool.query & { total: string }>(` SELECT o.data, count(*) OVER() AS total FROM assessor_finding_occurrences o JOIN assessor_finding_definitions d ON d.id = o.definition_id JOIN assessor_scan_runs r ON r.id = o.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id ${where} ORDER BY o.scan_run_id, o.id LIMIT $${values.length - 1} OFFSET $${values.length} `, values); return { total: Number(result.rows[0]?.total ?? 0), items: result.rows.map(row => clone(row.data)) }; } private async unifiedScannerResults( query: UnifiedResultQuery = {}, occurrenceId?: string ): Promise { if (query.sourceKind?.length && !query.sourceKind.includes('scanner-finding')) return []; const values: unknown[] = []; const conditions: string[] = ["r.data->>'supersededByScanRunId' IS NULL"]; const add = (expression: string, value: unknown) => { values.push(value); conditions.push(expression.replace('?', `$${values.length}`)); }; if (query.projectId) add('p.id = ?', query.projectId); if (query.assessmentId) add('a.id = ?', query.assessmentId); if (query.scanRunId) add('r.id = ?', query.scanRunId); if (query.scanRunIds?.length) add('r.id = ANY(?::text[])', query.scanRunIds); if (query.sourceTool?.length) add('lower(o.tool_id) = ANY(?::text[])', query.sourceTool.map(tool => tool.toLocaleLowerCase('en-US'))); if (query.resultType?.length) add('o.result_type = ANY(?::text[])', query.resultType); if (query.isIoe !== undefined) add('o.is_ioe = ?', query.isIoe); // Abweichung ist nicht persistiert, sondern ein versioniertes Praedikat // (src/domain/deviation-predicate.ts). Die SQL-Vorfilterung bildet es // deterministisch aus denselben beiden Spalten ab; die endgueltige // Entscheidung faellt weiterhin in filterUnifiedResults. if (query.deviation !== undefined) { add(`(o.is_ioe OR o.outcome_normalized IN ('fail', 'warning')) = ?`, query.deviation); } if (occurrenceId) add('o.id = ?', occurrenceId); const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : ''; const result = await this.pool.query<{ project: AssessmentProject; assessment: Assessment; scan_run: ScanRun; import_file: ImportFileRecord; definition: FindingDefinition; occurrence: FindingOccurrence; reviews: FindingReview[]; cve_enrichment: CveEnrichmentRecord | null; } & QueryResultRow>(` SELECT p.data AS project, a.data AS assessment, r.data AS scan_run, f.data AS import_file, d.data AS definition, o.data AS occurrence, cve.data AS cve_enrichment, COALESCE(( SELECT jsonb_agg(rv.data ORDER BY rv.id) FROM assessor_reviews rv WHERE rv.occurrence_id = o.id OR rv.definition_id = d.id ), '[]'::jsonb) AS reviews FROM assessor_finding_occurrences o JOIN assessor_finding_definitions d ON d.id = o.definition_id JOIN assessor_scan_runs r ON r.id = o.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id JOIN assessor_projects p ON p.id = a.project_id JOIN assessor_import_files f ON f.id = o.import_file_id LEFT JOIN assessor_cve_enrichments cve ON cve.cve_id = upper(COALESCE(o.data #>> '{sourceIdentity,canonicalRuleId}', o.data #>> '{sourceIdentity,originalRuleId}', '')) ${where} ORDER BY o.scan_run_id, o.id `, values); return result.rows.map(row => createScannerUnifiedResult({ project: clone(row.project), assessment: clone(row.assessment), scanRun: clone(row.scan_run), importFile: clone(row.import_file), definition: clone(row.definition), occurrence: clone(row.occurrence), reviews: clone(row.reviews), ...(row.cve_enrichment ? { cveEnrichment: clone(row.cve_enrichment) } : {}) })); } private async unifiedEnvironmentResults( query: UnifiedResultQuery = {}, observationId?: string ): Promise { if (query.sourceKind?.length && !query.sourceKind.includes('environment-observation')) return []; const values: unknown[] = []; const conditions: string[] = ["r.data->>'supersededByScanRunId' IS NULL"]; const add = (expression: string, value: unknown) => { values.push(value); conditions.push(expression.replace('?', `$${values.length}`)); }; if (query.projectId) add('p.id = ?', query.projectId); if (query.assessmentId) add('a.id = ?', query.assessmentId); if (query.scanRunId) add('r.id = ?', query.scanRunId); if (query.scanRunIds?.length) add('r.id = ANY(?::text[])', query.scanRunIds); if (query.sourceTool?.length) add('lower(o.tool_id) = ANY(?::text[])', query.sourceTool.map(tool => tool.toLocaleLowerCase('en-US'))); if (query.environmentState?.length) add('o.observation_state = ANY(?::text[])', query.environmentState); if (query.environmentProperty?.length) add('o.property_key = ANY(?::text[])', query.environmentProperty); if (query.environmentSubjectType?.length) add('o.subject_type = ANY(?::text[])', query.environmentSubjectType); if (observationId) add('o.id = ?', observationId); const result = await this.pool.query<{ project: AssessmentProject; assessment: Assessment; scan_run: ScanRun; import_file: ImportFileRecord; observation: EnvironmentObservation; } & QueryResultRow>(` SELECT p.data AS project, a.data AS assessment, r.data AS scan_run, f.data AS import_file, o.data AS observation FROM assessor_environment_observations o JOIN assessor_scan_runs r ON r.id = o.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id JOIN assessor_projects p ON p.id = a.project_id JOIN assessor_import_files f ON f.id = o.import_file_id WHERE ${conditions.join(' AND ')} ORDER BY o.scan_run_id, o.property_key, o.id `, values); return result.rows.flatMap(row => { const projection = createEnvironmentUnifiedResult({ project: clone(row.project), assessment: clone(row.assessment), scanRun: clone(row.scan_run), importFile: clone(row.import_file), observation: clone(row.observation) }); return projection ? [projection] : []; }); } private async unifiedPolicyResults( query: UnifiedResultQuery = {}, occurrenceId?: string ): Promise { if (query.sourceKind?.length && !query.sourceKind.includes('policy-evaluation')) return []; const values: unknown[] = []; const conditions: string[] = []; const add = (expression: string, value: unknown) => { values.push(value); conditions.push(expression.replace('?', `$${values.length}`)); }; if (query.projectId) add('r.project_id = ?', query.projectId); if (query.evaluationRunId) add('r.evaluation_run_id = ?', query.evaluationRunId); if (occurrenceId) add('o.id = ?', occurrenceId); const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : ''; const result = await this.pool.query<{ definition: FindingDefinition; occurrence: StoredPolicyFindingRun['occurrences'][number]; comparison: PolicyEvaluationResult | null; } & QueryResultRow>(` SELECT d.data AS definition, o.data AS occurrence, comparison.data AS comparison FROM assessor_policy_finding_occurrences o JOIN assessor_policy_finding_runs r ON r.id = o.factory_run_id JOIN assessor_finding_definitions d ON d.id = o.definition_id LEFT JOIN assessor_policy_evaluation_results comparison ON comparison.id = o.evaluation_result_id AND comparison.evaluation_run_id = r.evaluation_run_id ${where} ORDER BY r.completed_at, o.id `, values); return result.rows.map(row => createPolicyUnifiedResult({ definition: clone(row.definition), occurrence: clone(row.occurrence), comparison: row.comparison ? clone(row.comparison) : undefined })); } private async unifiedPolicyEvaluationComparisonResults( query: UnifiedResultQuery = {}, evaluationResultId?: string ): Promise { if (query.sourceKind?.length && !query.sourceKind.includes('policy-evaluation')) return []; const values: unknown[] = [POLICY_EVALUATION_ENGINE.version]; const conditions = ['finding.id IS NULL']; const add = (expression: string, value: unknown) => { values.push(value); conditions.push(expression.replace('?', `$${values.length}`)); }; if (query.projectId) add('selected.project_id = ?', query.projectId); if (query.evaluationRunId) add('selected.id = ?', query.evaluationRunId); if (evaluationResultId) add('result.id = ?', evaluationResultId); const rows = await this.pool.query<{ run: PolicyEvaluationRun; result: PolicyEvaluationResult; snapshot_id: string; assessment_ids: string[]; artifact_ids: string[]; } & QueryResultRow>(` WITH ranked AS ( SELECT evaluation.*, row_number() OVER ( PARTITION BY evaluation.aggregation_run_id -- Match selectCurrentPolicyEvaluations: multiple analysis generations -- of the same compared baseline do not increase benchmark coverage. -- Only fall back to declared references when no result names a baseline. ORDER BY COALESCE(NULLIF(( SELECT count(DISTINCT NULLIF(compared.data->>'baselineId', '')) FROM assessor_policy_evaluation_results compared WHERE compared.evaluation_run_id = evaluation.id ), 0), jsonb_array_length(evaluation.data->'run'->'baselineAnalysisIds')) DESC, evaluation.completed_at DESC, evaluation.id DESC ) AS run_rank FROM assessor_policy_evaluation_runs evaluation WHERE evaluation.evaluator_version = $1 AND NOT EXISTS ( SELECT 1 FROM assessor_policy_snapshot_aggregation_sources source JOIN assessor_policy_identity_analysis_runs newer ON newer.supersedes_analysis_run_id = source.analysis_run_id WHERE source.aggregation_run_id = evaluation.aggregation_run_id ) ), selected AS MATERIALIZED ( SELECT id, project_id, aggregation_run_id, completed_at, data->'run' AS run FROM ranked WHERE run_rank = 1 ), aggregation_context AS MATERIALIZED ( SELECT source.aggregation_run_id, array_agg(DISTINCT artifact.assessment_id) FILTER (WHERE artifact.assessment_id IS NOT NULL) AS assessment_ids, array_agg(DISTINCT artifact.id) AS artifact_ids FROM assessor_policy_snapshot_aggregation_sources source JOIN assessor_policy_identity_analysis_runs analysis ON analysis.id = source.analysis_run_id JOIN assessor_import_artifacts artifact ON artifact.id = analysis.artifact_id GROUP BY source.aggregation_run_id ) SELECT selected.run, result.data AS result, aggregation.snapshot_id, coalesce(context.assessment_ids, ARRAY[]::text[]) AS assessment_ids, coalesce(context.artifact_ids, ARRAY[]::text[]) AS artifact_ids FROM selected JOIN assessor_policy_evaluation_results result ON result.evaluation_run_id = selected.id JOIN assessor_policy_snapshot_aggregation_runs aggregation ON aggregation.id = selected.aggregation_run_id LEFT JOIN aggregation_context context ON context.aggregation_run_id = selected.aggregation_run_id LEFT JOIN assessor_policy_finding_occurrences finding ON finding.evaluation_result_id = result.id WHERE ${conditions.join(' AND ')} ORDER BY selected.completed_at, result.id `, values); return rows.rows.map(row => createPolicyEvaluationUnifiedResult(row.run, row.result, { assessmentId: row.assessment_ids.length === 1 ? row.assessment_ids[0] : undefined, importArtifactId: row.artifact_ids.length === 1 ? row.artifact_ids[0] : undefined, snapshotId: row.snapshot_id })); } private async currentPolicyObservationResults(projectId?: string): Promise { const where = projectId ? 'WHERE project_id = $1' : ''; const values = projectId ? [projectId] : []; const [artifactRows, processingRows, backupRows, analysisRows] = await Promise.all([ this.pool.query>(`SELECT data FROM assessor_import_artifacts ${where} ORDER BY created_at, id`, values), this.pool.query>(`SELECT data FROM assessor_import_processing_runs ${where} ORDER BY completed_at, id`, values), this.pool.query>(`SELECT data FROM assessor_gpo_backup_instances ${where} ORDER BY created_at, id`, values), this.pool.query>(`SELECT data FROM assessor_policy_identity_analysis_runs ${where} ORDER BY completed_at, id`, values) ]); return createCurrentPolicyObservationUnifiedResults( dataRows(artifactRows), dataRows(processingRows), dataRows(backupRows), dataRows(analysisRows) ); } private async unifiedPolicyObservationResults( query: UnifiedResultQuery = {}, observationId?: string ): Promise { if (query.audience !== 'internal-analysis') return []; if (query.sourceKind?.length && !query.sourceKind.includes('policy-observation')) return []; const results = await this.currentPolicyObservationResults(query.projectId); return observationId ? results.filter(item => item.sourceResultId === observationId) : results; } private async unifiedPolicyConsistencyResults(query: UnifiedResultQuery = {}): Promise { if (query.audience === 'internal-analysis') return []; if (query.sourceKind?.length && !query.sourceKind.includes('policy-evaluation')) return []; return createPolicyConsistencyUnifiedResults(await this.currentPolicyObservationResults(query.projectId)); } private async allUnifiedResults(query: UnifiedResultQuery = {}): Promise { const [scanner, environment, policyObservations, policyConsistency, policyEvaluations, policy] = await Promise.all([ this.unifiedScannerResults(query), this.unifiedEnvironmentResults(query), this.unifiedPolicyObservationResults(query), this.unifiedPolicyConsistencyResults(query), this.unifiedPolicyEvaluationComparisonResults(query), this.unifiedPolicyResults(query) ]); return [...scanner, ...environment, ...policyObservations, ...policyConsistency, ...policyEvaluations, ...policy]; } async queryUnifiedResults(query: UnifiedResultQuery = {}): Promise { return queryUnifiedResultItems(await this.allUnifiedResults(query), query); } async queryUnifiedResultFacets(query: UnifiedResultQuery = {}): Promise { return deriveUnifiedResultFacets(await this.allUnifiedResults(query), query); } async getUnifiedResultById(resultProjectionId: string): Promise { const scannerPrefix = 'scanner-finding:'; const policyPrefix = 'policy-evaluation:'; const policyComparisonPrefix = 'policy-evaluation-comparison:'; const policyObservationPrefix = 'policy-observation:'; const policyConsistencyPrefix = 'policy-consistency:'; const environmentPrefix = 'environment-observation:'; const items = resultProjectionId.startsWith(scannerPrefix) ? await this.unifiedScannerResults({}, resultProjectionId.slice(scannerPrefix.length)) : resultProjectionId.startsWith(policyComparisonPrefix) ? await this.unifiedPolicyEvaluationComparisonResults({}, resultProjectionId.slice(policyComparisonPrefix.length)) : resultProjectionId.startsWith(policyPrefix) ? await this.unifiedPolicyResults({}, resultProjectionId.slice(policyPrefix.length)) : resultProjectionId.startsWith(policyObservationPrefix) ? await this.unifiedPolicyObservationResults({}, resultProjectionId.slice(policyObservationPrefix.length)) : resultProjectionId.startsWith(policyConsistencyPrefix) ? await this.unifiedPolicyConsistencyResults({}) : resultProjectionId.startsWith(environmentPrefix) ? await this.unifiedEnvironmentResults({}, resultProjectionId.slice(environmentPrefix.length)) : []; return items.find(item => item.resultProjectionId === resultProjectionId) ?? null; } async listResultCategories(query: UnifiedResultQuery = {}): Promise { const results = await this.allUnifiedResults(query); return listUnifiedResultCategories(filterUnifiedResults(results, query)); } async adoptScanRunReplacement(projectId: string, replacedScanRunId: string, replacementScanRunId: string, supersededAt: string): Promise { if (replacedScanRunId === replacementScanRunId || !Number.isFinite(Date.parse(supersededAt))) { throw new RepositoryValidationError('Die Scan-Ersatzverknüpfung ist ungültig.'); } return this.mutate(async client => { const selected = await client.query<{ assessment_id: string; project_id: string; data: ScanRun } & QueryResultRow>(` SELECT run.assessment_id, assessment.project_id, run.data FROM assessor_scan_runs run JOIN assessor_assessments assessment ON assessment.id = run.assessment_id WHERE run.id = ANY($1::text[]) FOR UPDATE OF run, assessment `, [[replacedScanRunId, replacementScanRunId]]); const replaced = selected.rows.find(row => row.data.id === replacedScanRunId); const replacement = selected.rows.find(row => row.data.id === replacementScanRunId); if (!replaced || !replacement || replaced.project_id !== projectId || replacement.project_id !== projectId) { throw new RepositoryNotFoundError('Die beiden Scanberichte sind im bestätigten Mandanten nicht vorhanden.'); } if (replaced.assessment_id !== replacement.assessment_id || replaced.data.tool.id !== replacement.data.tool.id) { throw new RepositoryValidationError('Als Ersatz ist nur ein Scan desselben Werkzeugs im selben Timeslot zulässig.'); } const sourceFiles = await client.query<{ scan_run_id: string; data: ImportFileRecord } & QueryResultRow>(` SELECT scan_run_id, data FROM assessor_import_files WHERE scan_run_id = ANY($1::text[]) ORDER BY scan_run_id, id `, [[replacedScanRunId, replacementScanRunId]]); if (compareReimportSourceHashes( sourceFiles.rows.filter(row => row.scan_run_id === replacedScanRunId).map(row => row.data.sha256), sourceFiles.rows.filter(row => row.scan_run_id === replacementScanRunId).map(row => row.data.sha256) ) !== 'match') { throw new RepositoryValidationError('Als Ersatz ist nur kryptografisch exakt derselbe Scanbericht zulässig.'); } if (replaced.data.supersededByScanRunId || replacement.data.supersededByScanRunId || replacement.data.supersedesScanRunId || Date.parse(replacement.data.importedAt) <= Date.parse(replaced.data.importedAt)) { throw new RepositoryConflictError('Die Scanberichte sind nicht mehr als eindeutige aktive Vorgänger-/Nachfolgerkombination verfügbar.'); } const updatedReplaced = { ...replaced.data, supersededByScanRunId: replacementScanRunId, supersededAt }; const updatedReplacement = { ...replacement.data, supersedesScanRunId: replacedScanRunId }; const occurrenceRows = await client.query<{ scan_run_id: string; data: FindingOccurrence } & QueryResultRow>(` SELECT scan_run_id, data FROM assessor_finding_occurrences WHERE scan_run_id = ANY($1::text[]) ORDER BY scan_run_id, id `, [[replacedScanRunId, replacementScanRunId]]); const previousOccurrences = occurrenceRows.rows.filter(row => row.scan_run_id === replacedScanRunId).map(row => row.data); const replacementOccurrences = occurrenceRows.rows.filter(row => row.scan_run_id === replacementScanRunId).map(row => row.data); const occurrenceIds = occurrenceRows.rows.map(row => row.data.id); const definitionIds = [...new Set(occurrenceRows.rows.map(row => row.data.definitionId))]; const reviews = dataRows(await client.query>(` SELECT data FROM assessor_reviews WHERE occurrence_id = ANY($1::text[]) OR definition_id = ANY($2::text[]) ORDER BY id `, [occurrenceIds, definitionIds])); const reviewTransfer = planReimportReviewTransfer(previousOccurrences, replacementOccurrences, reviews, supersededAt); await persistTransferredReviews(client, reviewTransfer.reviews); const replacementFile = sourceFiles.rows.find(row => row.scan_run_id === replacementScanRunId)?.data; if (replacementFile) { const updatedFile = { ...replacementFile, reimportReviewTransfer: { replacesScanRunId: replacedScanRunId, ...reviewTransfer.summary } }; await client.query('UPDATE assessor_import_files SET data = $2::jsonb WHERE id = $1', [updatedFile.id, json(updatedFile)]); } await client.query('UPDATE assessor_scan_runs SET data = $2::jsonb WHERE id = $1', [replacedScanRunId, json(updatedReplaced)]); await client.query('UPDATE assessor_scan_runs SET data = $2::jsonb WHERE id = $1', [replacementScanRunId, json(updatedReplacement)]); return reviewTransfer.summary; }); } async getEnvironmentSnapshot(query: EnvironmentSnapshotQuery): Promise { const context = await this.pool.query<{ project_exists: boolean; assessment_exists: boolean } & QueryResultRow>(` SELECT EXISTS (SELECT 1 FROM assessor_projects WHERE id = $1) AS project_exists, ($2::text IS NULL OR EXISTS (SELECT 1 FROM assessor_assessments WHERE id = $2 AND project_id = $1)) AS assessment_exists `, [query.projectId, query.assessmentId ?? null]); if (!context.rows[0]?.project_exists) throw new RepositoryNotFoundError(`Projekt ${query.projectId} ist nicht vorhanden.`); if (!context.rows[0]?.assessment_exists) throw new RepositoryValidationError('Environment Snapshot verweist auf ein Assessment außerhalb des Projekts.'); const values: unknown[] = [query.projectId]; const conditions = ['a.project_id = $1']; if (query.assessmentId) { values.push(query.assessmentId); conditions.push(`a.id = $${values.length}`); } if (query.scanRunIds?.length) { values.push(query.scanRunIds); conditions.push(`r.id = ANY($${values.length}::text[])`); const valid = await this.pool.query<{ count: string } & QueryResultRow>(` SELECT count(*)::text AS count FROM assessor_scan_runs r JOIN assessor_assessments a ON a.id = r.assessment_id WHERE ${conditions.join(' AND ')} `, values); if (Number(valid.rows[0]?.count ?? 0) !== new Set(query.scanRunIds).size) { throw new RepositoryValidationError('Environment Snapshot verweist auf einen Scan-Lauf außerhalb des Projekts oder Assessments.'); } } const where = conditions.join(' AND '); const [runs, sectionsResult, observationsResult] = await Promise.all([ this.pool.query<{ id: string } & QueryResultRow>(`SELECT r.id FROM assessor_scan_runs r JOIN assessor_assessments a ON a.id = r.assessment_id WHERE ${where} ORDER BY r.id`, values), this.pool.query>(`SELECT s.data FROM assessor_scanner_report_sections s JOIN assessor_scan_runs r ON r.id = s.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id WHERE ${where} ORDER BY s.ordinal, s.id`, values), this.pool.query>(`SELECT o.data FROM assessor_environment_observations o JOIN assessor_scan_runs r ON r.id = o.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id WHERE ${where} ORDER BY o.property_key, o.id`, values) ]); const sections = dataRows(sectionsResult); const observations = dataRows(observationsResult); return { version: 1, projectId: query.projectId, assessmentId: query.assessmentId, scanRunIds: runs.rows.map(row => row.id), generatedAt: new Date().toISOString(), sections, observations, coverage: environmentCoverage(sections, observations) }; } async saveImportArtifact(artifact: ImportArtifact): Promise { const errors = importArtifactValidationErrors(artifact); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const context = await client.query<{ project_exists: boolean; assessment_valid: boolean; run_valid: boolean; file_valid: boolean; parent_valid: boolean; } & QueryResultRow>(` SELECT EXISTS (SELECT 1 FROM assessor_projects WHERE id = $1) AS project_exists, ($2::text IS NULL OR EXISTS (SELECT 1 FROM assessor_assessments WHERE id = $2 AND project_id = $1)) AS assessment_valid, ($3::text IS NULL OR EXISTS ( SELECT 1 FROM assessor_scan_runs r JOIN assessor_assessments a ON a.id = r.assessment_id WHERE r.id = $3 AND a.project_id = $1 )) AS run_valid, ($4::text IS NULL OR EXISTS ( SELECT 1 FROM assessor_import_files f JOIN assessor_scan_runs r ON r.id = f.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id WHERE f.id = $4 AND a.project_id = $1 )) AS file_valid, ($5::text IS NULL OR EXISTS (SELECT 1 FROM assessor_import_artifacts WHERE id = $5 AND project_id = $1)) AS parent_valid `, [artifact.projectId, artifact.assessmentId ?? null, artifact.scanRunId ?? null, artifact.importFileId ?? null, artifact.parentArtifactId ?? null]); const valid = context.rows[0]; if (!valid?.project_exists || !valid.assessment_valid || !valid.run_valid || !valid.file_valid || !valid.parent_valid) { throw new RepositoryNotFoundError('Artefaktreferenzen gehören nicht vollständig zum angegebenen Projekt.'); } await client.query(` INSERT INTO assessor_import_artifacts (id, project_id, assessment_id, scan_run_id, import_file_id, parent_artifact_id, artifact_kind, relative_path, sha256, size_bytes, sensitivity, parse_status, storage_kind, storage_key, created_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb) `, [ artifact.id, artifact.projectId, artifact.assessmentId ?? null, artifact.scanRunId ?? null, artifact.importFileId ?? null, artifact.parentArtifactId ?? null, artifact.kind, artifact.relativePath, artifact.sha256, artifact.sizeBytes, artifact.sensitivity, artifact.parseStatus, artifact.storage.kind, artifact.storage.kind === 'backend-controlled' ? artifact.storage.storageKey : null, artifact.createdAt, json(artifact) ]); }); } async saveImportArtifactAndEnqueueJob(artifact: ImportArtifact, job: IngestionJob) { const errors = [...importArtifactValidationErrors(artifact), ...ingestionJobValidationErrors(job)]; if (job.status !== 'queued' || job.attempts !== 0) errors.push('Ein neuer Job muss ohne Versuche in queued beginnen.'); if (job.projectId !== artifact.projectId || job.artifactId !== artifact.id) errors.push('Artefakt und Job müssen dieselbe Projekt- und Artefaktidentität besitzen.'); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); return this.mutate(async client => { const context = await client.query<{ project_exists: boolean; assessment_valid: boolean; run_valid: boolean; file_valid: boolean; parent_valid: boolean; } & QueryResultRow>(` SELECT EXISTS (SELECT 1 FROM assessor_projects WHERE id = $1) AS project_exists, ($2::text IS NULL OR EXISTS (SELECT 1 FROM assessor_assessments WHERE id = $2 AND project_id = $1)) AS assessment_valid, ($3::text IS NULL OR EXISTS ( SELECT 1 FROM assessor_scan_runs r JOIN assessor_assessments a ON a.id = r.assessment_id WHERE r.id = $3 AND a.project_id = $1 )) AS run_valid, ($4::text IS NULL OR EXISTS ( SELECT 1 FROM assessor_import_files f JOIN assessor_scan_runs r ON r.id = f.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id WHERE f.id = $4 AND a.project_id = $1 )) AS file_valid, ($5::text IS NULL OR EXISTS (SELECT 1 FROM assessor_import_artifacts WHERE id = $5 AND project_id = $1)) AS parent_valid `, [artifact.projectId, artifact.assessmentId ?? null, artifact.scanRunId ?? null, artifact.importFileId ?? null, artifact.parentArtifactId ?? null]); const valid = context.rows[0]; if (!valid?.project_exists || !valid.assessment_valid || !valid.run_valid || !valid.file_valid || !valid.parent_valid) { throw new RepositoryNotFoundError('Artefaktreferenzen gehören nicht vollständig zum angegebenen Projekt.'); } const artifactInsert = await client.query>(` INSERT INTO assessor_import_artifacts (id, project_id, assessment_id, scan_run_id, import_file_id, parent_artifact_id, artifact_kind, relative_path, sha256, size_bytes, sensitivity, parse_status, storage_kind, storage_key, created_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb) ON CONFLICT (id) DO NOTHING RETURNING data `, [ artifact.id, artifact.projectId, artifact.assessmentId ?? null, artifact.scanRunId ?? null, artifact.importFileId ?? null, artifact.parentArtifactId ?? null, artifact.kind, artifact.relativePath, artifact.sha256, artifact.sizeBytes, artifact.sensitivity, artifact.parseStatus, artifact.storage.kind, artifact.storage.kind === 'backend-controlled' ? artifact.storage.storageKey : null, artifact.createdAt, json(artifact) ]); let storedArtifact = artifactInsert.rows[0]?.data; if (!storedArtifact) { storedArtifact = (await client.query>( 'SELECT data FROM assessor_import_artifacts WHERE id = $1 FOR UPDATE', [artifact.id] )).rows[0]?.data; if (!storedArtifact || storedArtifact.projectId !== artifact.projectId || storedArtifact.relativePath !== artifact.relativePath || storedArtifact.sha256 !== artifact.sha256 || storedArtifact.sizeBytes !== artifact.sizeBytes || storedArtifact.kind !== artifact.kind || storedArtifact.storage.kind !== 'backend-controlled') { throw new RepositoryConflictError(`Importartefakt ${artifact.id} ist bereits mit abweichender Identität vorhanden.`); } } const jobInsert = await client.query>(` INSERT INTO assessor_ingestion_jobs (id, project_id, artifact_id, job_kind, idempotency_key, job_status, attempts, max_attempts, available_at, lease_owner, lease_expires_at, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL, NULL, $10, $11, $12::jsonb) ON CONFLICT (project_id, idempotency_key) DO NOTHING RETURNING data `, [job.id, job.projectId, job.artifactId ?? null, job.kind, job.idempotencyKey, job.status, job.attempts, job.maxAttempts, job.availableAt, job.createdAt, job.updatedAt, json(job)]); let storedJob = jobInsert.rows[0]?.data; if (!storedJob) { storedJob = (await client.query>( 'SELECT data FROM assessor_ingestion_jobs WHERE project_id = $1 AND idempotency_key = $2 FOR UPDATE', [job.projectId, job.idempotencyKey] )).rows[0]?.data; if (!storedJob || storedJob.kind !== job.kind || storedJob.artifactId !== storedArtifact.id) { throw new RepositoryConflictError('Idempotenzschlüssel bezeichnet einen anderen Job.'); } } return { artifact: clone(storedArtifact), job: clone(storedJob), artifactCreated: Boolean(artifactInsert.rows[0]), jobCreated: Boolean(jobInsert.rows[0]) }; }, undefined, result => result.artifactCreated || result.jobCreated); } async listImportArtifacts(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_import_artifacts WHERE project_id = $1 ORDER BY created_at, id', [projectId] )); } async saveProcessingRun(run: ImportProcessingRun): Promise { const errors = processingRunValidationErrors(run); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const artifact = await client.query('SELECT 1 FROM assessor_import_artifacts WHERE id = $1 AND project_id = $2', [run.artifactId, run.projectId]); if (!artifact.rowCount) throw new RepositoryNotFoundError('Processing-Artefakt ist nicht im Projekt vorhanden.'); if (run.supersedesRunId) { const predecessor = await client.query( 'SELECT 1 FROM assessor_import_processing_runs WHERE id = $1 AND artifact_id = $2', [run.supersedesRunId, run.artifactId] ); if (!predecessor.rowCount) throw new RepositoryNotFoundError('Der ersetzte Processing Run gehört nicht zum Artefakt.'); } await client.query(` INSERT INTO assessor_import_processing_runs (id, project_id, artifact_id, supersedes_run_id, run_status, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) `, [run.id, run.projectId, run.artifactId, run.supersedesRunId ?? null, run.status, run.startedAt, run.completedAt, json(run)]); }); } async listProcessingRuns(artifactId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_import_processing_runs WHERE artifact_id = $1 ORDER BY started_at, id', [artifactId] )); } async listProjectProcessingRuns(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_import_processing_runs WHERE project_id = $1 ORDER BY started_at, id', [projectId] )); } async saveGpoBackupAnalysis(run: ImportProcessingRun, instances: StoredGpoBackupInstance[], policyAnalysis?: StoredGpoPolicyAnalysis): Promise { const errors = [ ...processingRunValidationErrors(run), ...instances.flatMap(storedGpoBackupInstanceValidationErrors), ...(policyAnalysis ? storedGpoPolicyAnalysisValidationErrors(policyAnalysis) : []) ]; if (run.processors.every(processor => processor.kind !== 'artifact-parser')) errors.push('Ein GPO-Processing Run benötigt einen Artefaktparser.'); if (policyAnalysis && (policyAnalysis.run.projectId !== run.projectId || policyAnalysis.run.artifactId !== run.artifactId || policyAnalysis.run.sourceProcessingRunId !== run.id)) errors.push('Die Policy-Identitätsanalyse gehört nicht zum GPO-Processing Run.'); if (instances.some(record => record.projectId !== run.projectId || record.artifactId !== run.artifactId || record.processingRunId !== run.id)) { errors.push('GPO-Backupinstanzen gehören nicht vollständig zum Processing Run.'); } if (new Set(instances.map(record => record.id)).size !== instances.length || new Set(instances.map(record => record.instance.id)).size !== instances.length) errors.push('GPO-Backupinstanzen eines Processing Runs müssen eindeutig sein.'); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const existing = await client.query>( 'SELECT data FROM assessor_import_processing_runs WHERE id = $1 FOR UPDATE', [run.id] ); if (existing.rows[0]) { const storedInstances = dataRows(await client.query>( 'SELECT data FROM assessor_gpo_backup_instances WHERE processing_run_id = $1 ORDER BY id', [run.id] )); const storedPolicy = dataRow(await client.query>( 'SELECT data FROM assessor_policy_identity_analysis_runs WHERE source_processing_run_id = $1', [run.id] )); const expected = instances.slice().sort((left, right) => left.id.localeCompare(right.id)); if (canonicalJson(existing.rows[0].data) === canonicalJson(run) && canonicalJson(storedInstances) === canonicalJson(expected)) { if (!policyAnalysis && !storedPolicy) return { inserted: false }; if (policyAnalysis && storedPolicy && canonicalJson(storedPolicy) === canonicalJson(policyAnalysis)) return { inserted: false }; } throw new RepositoryConflictError(`GPO-Processing Run ${run.id} ist bereits mit abweichenden Daten vorhanden.`); } const artifact = await client.query( 'SELECT 1 FROM assessor_import_artifacts WHERE id = $1 AND project_id = $2 FOR UPDATE', [run.artifactId, run.projectId] ); if (!artifact.rowCount) throw new RepositoryNotFoundError('GPO-Processing-Artefakt ist nicht im Projekt vorhanden.'); if (run.supersedesRunId) { const predecessor = await client.query('SELECT 1 FROM assessor_import_processing_runs WHERE id = $1 AND project_id = $2 AND artifact_id = $3', [run.supersedesRunId, run.projectId, run.artifactId]); if (!predecessor.rowCount) throw new RepositoryNotFoundError('Der zu ersetzende GPO-Processing Run ist nicht vorhanden.'); } await client.query(` INSERT INTO assessor_import_processing_runs (id, project_id, artifact_id, supersedes_run_id, run_status, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) `, [run.id, run.projectId, run.artifactId, run.supersedesRunId ?? null, run.status, run.startedAt, run.completedAt, json(run)]); for (const record of instances) { await client.query(` INSERT INTO assessor_gpo_backup_instances (id, project_id, artifact_id, processing_run_id, instance_id, created_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb) `, [record.id, record.projectId, record.artifactId, record.processingRunId, record.instance.id, record.createdAt, json(record)]); } if (policyAnalysis) { const analysisRun = policyAnalysis.run; await client.query(` INSERT INTO assessor_policy_identity_analysis_runs (id, project_id, artifact_id, source_processing_run_id, supersedes_analysis_run_id, analyzer_version, run_status, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb) `, [analysisRun.id, analysisRun.projectId, analysisRun.artifactId, analysisRun.sourceProcessingRunId, analysisRun.supersedesAnalysisRunId ?? null, analysisRun.analyzer.version, analysisRun.status, analysisRun.startedAt, analysisRun.completedAt, json(policyAnalysis)]); for (const resolution of policyAnalysis.resolutions) { const identity = resolution.resolution.status === 'resolved' ? resolution.resolution.identity : undefined; if (identity) { await client.query(` INSERT INTO assessor_policy_identities (id, resolver_id, resolver_version, scope, target_kind, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) ON CONFLICT (id) DO NOTHING `, [identity.id, identity.resolver.id, identity.resolver.version, identity.scope, identity.target.kind, json({ version: identity.version, id: identity.id, resolver: identity.resolver, scope: identity.scope, target: identity.target })]); } await client.query(` INSERT INTO assessor_policy_identity_resolutions (id, analysis_run_id, project_id, source_processing_run_id, backup_instance_id, observation_id, identity_id, resolution_status, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) `, [resolution.id, resolution.analysisRunId, resolution.projectId, resolution.sourceProcessingRunId, resolution.backupInstanceId, resolution.observationId, identity?.id ?? null, resolution.resolution.status, json(resolution)]); } } return { inserted: true }; }, undefined, result => result.inserted); } async listGpoBackupInstances(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_gpo_backup_instances WHERE project_id = $1 ORDER BY created_at, id', [projectId] )); } async listGpoPolicyAnalyses(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_policy_identity_analysis_runs WHERE project_id = $1 ORDER BY completed_at, id', [projectId] )); } async saveEnvironmentProfile(profile: EnvironmentProfile): Promise { const errors = environmentProfileValidationErrors(profile); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const project = await client.query('SELECT 1 FROM assessor_projects WHERE id = $1 FOR SHARE', [profile.projectId]); if (!project.rowCount) throw new RepositoryNotFoundError('Das Projekt des Environment Profiles ist nicht vorhanden.'); const existing = dataRow(await client.query>( 'SELECT data FROM assessor_environment_profiles WHERE id = $1 FOR UPDATE', [profile.id] )); if (existing) { if (canonicalJson(existing) === canonicalJson(profile)) return false; throw new RepositoryConflictError('Das Environment Profile ist bereits mit abweichenden Daten vorhanden.'); } let predecessor: EnvironmentProfile | undefined; if (profile.supersedesProfileId) { predecessor = dataRow(await client.query>( 'SELECT data FROM assessor_environment_profiles WHERE id = $1 AND project_id = $2 FOR UPDATE', [profile.supersedesProfileId, profile.projectId] )) ?? undefined; if (!predecessor) throw new RepositoryNotFoundError('Das zu ersetzende Environment Profile ist nicht vorhanden.'); } if (profile.revision !== (predecessor?.revision ?? 0) + 1) { throw new RepositoryValidationError('Die Revision des Environment Profiles ist nicht fortlaufend.'); } await client.query( 'INSERT INTO assessor_environment_profiles (id, project_id, revision, supersedes_profile_id, created_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb)', [profile.id, profile.projectId, profile.revision, profile.supersedesProfileId ?? null, profile.createdAt, json(profile)] ); return true; }, undefined, changed => changed); } async listEnvironmentProfiles(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_environment_profiles WHERE project_id = $1 ORDER BY revision, created_at, id', [projectId] )); } async savePolicySnapshotAggregation(record: StoredPolicySnapshotAggregation): Promise { const errors = storedPolicySnapshotAggregationValidationErrors(record); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_policy_snapshot_aggregation_runs WHERE id = $1 FOR UPDATE', [record.run.id] )); if (existing) { if (canonicalJson(existing) === canonicalJson(record)) return false; throw new RepositoryConflictError('Der Policy-Snapshot-Aggregationslauf ist bereits mit abweichenden Daten vorhanden.'); } const profile = await client.query( 'SELECT 1 FROM assessor_environment_profiles WHERE id = $1 AND project_id = $2 FOR SHARE', [record.run.environmentProfileId, record.run.projectId] ); if (!profile.rowCount) throw new RepositoryNotFoundError('Das Environment Profile des Aggregationslaufs ist nicht vorhanden.'); const sources = await client.query<{ id: string; source_processing_run_id: string } & QueryResultRow>( 'SELECT id, source_processing_run_id FROM assessor_policy_identity_analysis_runs WHERE project_id = $1 AND id = ANY($2::text[]) FOR SHARE', [record.run.projectId, record.run.sourceAnalysisRunIds] ); if (sources.rowCount !== record.run.sourceAnalysisRunIds.length) { throw new RepositoryNotFoundError('Mindestens ein Quellanalyselauf der Policyaggregation ist nicht vorhanden.'); } const processingRunIds = sources.rows.map(row => row.source_processing_run_id); const snapshotInstances = await client.query<{ instance_id: string } & QueryResultRow>( 'SELECT instance_id FROM assessor_gpo_backup_instances WHERE project_id = $1 AND processing_run_id = ANY($2::text[])', [record.run.projectId, processingRunIds] ); const validInstanceIds = new Set(snapshotInstances.rows.map(row => row.instance_id)); if (record.snapshot.instanceIds.some(id => !validInstanceIds.has(id))) { throw new RepositoryValidationError('Der Aggregationssnapshot verweist auf eine fremde Backupinstanz.'); } const identityIds = [...new Set(record.states.map(state => state.identity.id))].sort(); if (identityIds.length) { const identities = await client.query<{ identity_id: string } & QueryResultRow>( 'SELECT DISTINCT identity_id FROM assessor_policy_identity_resolutions WHERE analysis_run_id = ANY($1::text[]) AND identity_id = ANY($2::text[])', [record.run.sourceAnalysisRunIds, identityIds] ); if (identities.rowCount !== identityIds.length) { throw new RepositoryValidationError('Ein aggregierter Policyzustand verweist auf eine fremde technische Identitaet.'); } } await client.query( 'INSERT INTO assessor_policy_snapshot_aggregation_runs (id, project_id, environment_profile_id, snapshot_id, aggregator_version, run_status, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb)', [record.run.id, record.run.projectId, record.run.environmentProfileId, record.run.snapshotId, record.run.aggregator.version, record.run.status, record.run.startedAt, record.run.completedAt, json(record)] ); for (const analysisRunId of record.run.sourceAnalysisRunIds) { await client.query( 'INSERT INTO assessor_policy_snapshot_aggregation_sources (aggregation_run_id, analysis_run_id, project_id) VALUES ($1, $2, $3)', [record.run.id, analysisRunId, record.run.projectId] ); } for (const state of record.states) { await client.query( 'INSERT INTO assessor_policy_snapshot_states (id, aggregation_run_id, project_id, identity_id, delivery_mechanism, state_status, data) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb)', [state.id, record.run.id, record.run.projectId, state.identity.id, state.deliveryMechanism, state.status, json(state)] ); } return true; }, undefined, changed => changed); } async listPolicySnapshotAggregations(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_policy_snapshot_aggregation_runs WHERE project_id = $1 ORDER BY completed_at, id', [projectId] )); } async savePolicyEvaluation(record: StoredPolicyEvaluation): Promise { const errors = storedPolicyEvaluationValidationErrors(record); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_policy_evaluation_runs WHERE id = $1 FOR UPDATE', [record.run.id] )); if (existing) { if (canonicalJson(existing) === canonicalJson(record)) return false; throw new RepositoryConflictError('Die Policy-Evaluation ist bereits mit abweichenden Daten vorhanden.'); } const aggregation = dataRow(await client.query>( 'SELECT data FROM assessor_policy_snapshot_aggregation_runs WHERE id = $1 AND project_id = $2 FOR SHARE', [record.run.aggregationRunId, record.run.projectId] )); if (!aggregation) throw new RepositoryNotFoundError('Der Aggregationslauf der Policy-Evaluation ist nicht vorhanden.'); if (aggregation.run.environmentProfileId !== record.run.environmentProfileId) { throw new RepositoryValidationError('Die Policy-Evaluation verweist auf ein anderes Environment Profile.'); } const sources = await client.query<{ id: string; baseline_id: string } & QueryResultRow>( 'SELECT id, baseline_id FROM assessor_microsoft_security_baseline_analyses WHERE id = ANY($1::text[]) FOR SHARE', [record.run.baselineAnalysisIds] ); if (sources.rowCount !== record.run.baselineAnalysisIds.length) { throw new RepositoryNotFoundError('Mindestens eine Baselineanalyse der Policy-Evaluation ist nicht vorhanden.'); } const states = await client.query<{ id: string; identity_id: string } & QueryResultRow>( 'SELECT id, identity_id FROM assessor_policy_snapshot_states WHERE aggregation_run_id = $1 FOR SHARE', [record.run.aggregationRunId] ); const stateIdentity = new Map(states.rows.map(row => [row.id, row.identity_id])); if (record.results.some(result => result.aggregationStateId ? stateIdentity.get(result.aggregationStateId) !== result.identity.id : result.customerState !== 'missing')) { throw new RepositoryValidationError('Ein Evaluationsergebnis verweist auf einen fremden Aggregationszustand.'); } await client.query( 'INSERT INTO assessor_policy_evaluation_runs (id, project_id, aggregation_run_id, environment_profile_id, evaluator_version, comparator_catalog_version, dependency_rules_version, risk_rules_version, rule_catalog_id, rule_catalog_version, run_status, projected_finding_count, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15::jsonb)', [record.run.id, record.run.projectId, record.run.aggregationRunId, record.run.environmentProfileId, record.run.evaluator.version, record.run.comparatorCatalog.version, record.run.dependencyRules.version, record.run.riskRules.version, record.run.ruleCatalog.id, record.run.ruleCatalog.catalog.version, record.run.status, record.run.projectedFindingCount, record.run.startedAt, record.run.completedAt, json(record)] ); const sourceBaselineIds = new Map(sources.rows.map(row => [row.id, row.baseline_id])); for (const analysisId of record.run.baselineAnalysisIds) { await client.query( 'INSERT INTO assessor_policy_evaluation_baseline_sources (evaluation_run_id, baseline_analysis_id, baseline_id) VALUES ($1, $2, $3)', [record.run.id, analysisId, sourceBaselineIds.get(analysisId)] ); } for (const result of record.results) { await client.query( 'INSERT INTO assessor_policy_evaluation_results (id, evaluation_run_id, aggregation_state_id, baseline_analysis_id, customer_state, identity_id, classification, baseline_state, comparator_kind, rule_ids, dependency_rule_ids, risk_rule_ids, finding_eligible, effective_state, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::text[], $11::text[], $12::text[], $13, $14, $15::jsonb)', [result.id, record.run.id, result.aggregationStateId ?? null, result.baselineAnalysisId, result.customerState, result.identity.id, result.classification, result.baselineState, result.comparator.comparator.kind, result.ruleIds, result.dependencyRuleIds, result.riskRuleIds, result.findingEligibility === 'eligible', result.effectiveState, json(result)] ); } for (const item of record.evidence) { await client.query( 'INSERT INTO assessor_policy_evaluation_evidence (id, evaluation_result_id, evidence_kind, source_id, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [item.id, item.evaluationResultId, item.kind, item.sourceId, json(item)] ); } return true; }, undefined, changed => changed); } async listPolicyEvaluations(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_policy_evaluation_runs WHERE project_id = $1 ORDER BY completed_at, id', [projectId] )); } async savePolicyFindingRun(record: StoredPolicyFindingRun): Promise { const errors = storedPolicyFindingRunValidationErrors(record); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const existing = dataRow(await client.query>( 'SELECT data FROM assessor_policy_finding_runs WHERE id = $1 FOR UPDATE', [record.run.id] )); if (existing) { if (canonicalJson(existing) === canonicalJson(record)) return false; throw new RepositoryConflictError('Der Policy-Finding-Lauf ist bereits mit abweichenden Daten vorhanden.'); } const evaluation = await client.query<{ project_id: string } & QueryResultRow>( 'SELECT project_id FROM assessor_policy_evaluation_runs WHERE id = $1 FOR SHARE', [record.run.evaluationRunId] ); if (!evaluation.rowCount || evaluation.rows[0].project_id !== record.run.projectId) { throw new RepositoryNotFoundError('Die Evaluation des Policy-Finding-Laufs ist nicht vorhanden.'); } const resultIds = record.occurrences.map(occurrence => occurrence.evaluationResultId); if (resultIds.length) { const results = await client.query<{ id: string } & QueryResultRow>( 'SELECT id FROM assessor_policy_evaluation_results WHERE evaluation_run_id = $1 AND id = ANY($2::text[]) FOR SHARE', [record.run.evaluationRunId, resultIds] ); if (results.rowCount !== resultIds.length) { throw new RepositoryValidationError('Ein Policy-Finding verweist auf ein fremdes Evaluationsergebnis.'); } } for (const definition of record.definitions) { const existingDefinition = await client.query<{ id: string; canonical_key: string; data: FindingDefinition; } & QueryResultRow>( 'SELECT id, canonical_key, data FROM assessor_finding_definitions WHERE id = $1 OR canonical_key = $2 FOR UPDATE', [definition.id, definition.canonicalKey] ); if (existingDefinition.rowCount) { const row = existingDefinition.rows[0]; if (row.id !== definition.id || row.canonical_key !== definition.canonicalKey || canonicalJson(row.data) !== canonicalJson(definition)) { throw new RepositoryConflictError('Eine Policy-Finding-Definition kollidiert mit einer vorhandenen Definition.'); } } else { await client.query( 'INSERT INTO assessor_finding_definitions (id, canonical_key, data) VALUES ($1, $2, $3::jsonb)', [definition.id, definition.canonicalKey, json(definition)] ); } } await client.query( 'INSERT INTO assessor_policy_finding_runs (id, project_id, evaluation_run_id, factory_id, factory_version, run_status, eligible_result_count, finding_count, rejected_result_count, started_at, completed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb)', [record.run.id, record.run.projectId, record.run.evaluationRunId, record.run.factory.id, record.run.factory.version, record.run.status, record.run.eligibleResultCount, record.run.findingCount, record.run.rejectedResultCount, record.run.startedAt, record.run.completedAt, json(record)] ); for (const occurrence of record.occurrences) { await client.query( 'INSERT INTO assessor_policy_finding_occurrences (id, factory_run_id, evaluation_result_id, definition_id, severity, classification, scope_state, effective_state, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb)', [occurrence.id, record.run.id, occurrence.evaluationResultId, occurrence.definitionId, occurrence.severity, occurrence.classification, occurrence.scopeState, occurrence.effectiveState, json(occurrence)] ); } return true; }, undefined, changed => changed); } async listPolicyFindingRuns(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_policy_finding_runs WHERE project_id = $1 ORDER BY completed_at, id', [projectId] )); } async saveMicrosoftSecurityBaseline(record: MicrosoftSecurityBaselineRecord): Promise { if (!record.id.trim() || !/^[a-f0-9]{64}$/.test(record.sha256) || !record.storageKey.trim()) throw new RepositoryValidationError('Microsoft baseline record is invalid.'); await this.mutate(async client => { const existing = dataRow(await client.query>('SELECT data FROM assessor_microsoft_security_baselines WHERE id = $1 OR sha256 = $2', [record.id, record.sha256])); if (existing) { if (canonicalJson(existing) === canonicalJson(record)) return false; throw new RepositoryConflictError('Microsoft baseline already exists with different metadata.'); } await client.query('INSERT INTO assessor_microsoft_security_baselines (id, sha256, storage_key, imported_at, data) VALUES ($1, $2, $3, $4, $5::jsonb)', [record.id, record.sha256, record.storageKey, record.importedAt, json(record)]); return true; }, undefined, changed => changed); } async listMicrosoftSecurityBaselines(): Promise { return dataRows(await this.pool.query>('SELECT data FROM assessor_microsoft_security_baselines ORDER BY imported_at DESC, id')); } async deleteMicrosoftSecurityBaseline(baselineId: string): Promise { return this.mutate(async client => { const record = dataRow(await client.query>( 'SELECT data FROM assessor_microsoft_security_baselines WHERE id = $1 FOR UPDATE', [baselineId] )); if (!record) return null; // Evaluations and findings are reproducible derivatives. Explicit baseline deletion // removes only chains that consume this baseline; customer snapshots and imports remain. await client.query(` DELETE FROM assessor_policy_evaluation_runs WHERE id IN ( SELECT evaluation_run_id FROM assessor_policy_evaluation_baseline_sources WHERE baseline_id = $1 ) `, [baselineId]); await client.query('UPDATE assessor_microsoft_security_baseline_classifications SET supersedes_classification_id = NULL WHERE baseline_id = $1', [baselineId]); await client.query('UPDATE assessor_microsoft_security_baseline_analyses SET supersedes_analysis_id = NULL WHERE baseline_id = $1', [baselineId]); await client.query('DELETE FROM assessor_microsoft_security_baselines WHERE id = $1', [baselineId]); return record; }, undefined, result => result !== null); } async saveMicrosoftSecurityBaselineClassification(classification: MicrosoftSecurityBaselineClassification): Promise { await this.mutate(async client => { const baseline = await client.query('SELECT 1 FROM assessor_microsoft_security_baselines WHERE id = $1', [classification.baselineId]); if (!baseline.rowCount) throw new RepositoryNotFoundError('Microsoft baseline package is not present.'); const existing = dataRow(await client.query>('SELECT data FROM assessor_microsoft_security_baseline_classifications WHERE id = $1 OR (baseline_id = $2 AND detector_version = $3)', [classification.id, classification.baselineId, classification.detectorVersion])); if (existing) { if (canonicalJson(existing) === canonicalJson(classification)) return false; throw new RepositoryConflictError('Microsoft baseline classification already exists with different data.'); } if (classification.supersedesClassificationId) { const superseded = await client.query('SELECT 1 FROM assessor_microsoft_security_baseline_classifications WHERE id = $1 AND baseline_id = $2', [classification.supersedesClassificationId, classification.baselineId]); if (!superseded.rowCount) throw new RepositoryNotFoundError('Superseded baseline classification is not present.'); } await client.query('INSERT INTO assessor_microsoft_security_baseline_classifications (id, baseline_id, detector_version, supersedes_classification_id, created_at, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb)', [classification.id, classification.baselineId, classification.detectorVersion, classification.supersedesClassificationId ?? null, classification.createdAt, json(classification)]); return true; }, undefined, changed => changed); } async listMicrosoftSecurityBaselineClassifications(): Promise { return dataRows(await this.pool.query>('SELECT data FROM assessor_microsoft_security_baseline_classifications ORDER BY created_at, id')); } async saveMicrosoftSecurityBaselineAnalysis(record: StoredMicrosoftSecurityBaselineAnalysis): Promise { const errors = storedMicrosoftSecurityBaselineAnalysisValidationErrors(record); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); await this.mutate(async client => { const baseline = await client.query<{ sha256: string } & QueryResultRow>( 'SELECT sha256 FROM assessor_microsoft_security_baselines WHERE id = $1 FOR UPDATE', [record.analysis.baselineId] ); if (baseline.rows[0]?.sha256 !== record.analysis.packageSha256) throw new RepositoryNotFoundError('Das Microsoft-Baselinepaket der Analyse ist nicht vorhanden.'); const existing = dataRow(await client.query>( 'SELECT data FROM assessor_microsoft_security_baseline_analyses WHERE id = $1', [record.analysis.id] )); if (existing) { if (canonicalJson(existing) === canonicalJson(record)) return false; throw new RepositoryConflictError('Die Microsoft-Baselineanalyse ist bereits mit abweichenden Daten vorhanden.'); } if (record.supersedesAnalysisId) { const predecessor = await client.query('SELECT 1 FROM assessor_microsoft_security_baseline_analyses WHERE id = $1 AND baseline_id = $2', [record.supersedesAnalysisId, record.analysis.baselineId]); if (!predecessor.rowCount) throw new RepositoryNotFoundError('Die zu ersetzende Microsoft-Baselineanalyse ist nicht vorhanden.'); } const analysis = record.analysis; await client.query(` INSERT INTO assessor_microsoft_security_baseline_analyses (id, baseline_id, package_sha256, analyzer_version, supersedes_analysis_id, analysis_status, analyzed_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) `, [analysis.id, analysis.baselineId, analysis.packageSha256, analysis.analyzer.version, record.supersedesAnalysisId ?? null, analysis.status, analysis.analyzedAt, json(record)]); await client.query(` INSERT INTO assessor_admx_catalog_versions (id, analysis_id, baseline_id, package_sha256, parser_version, catalog_status, data) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb) `, [analysis.admxCatalog.id, analysis.id, analysis.baselineId, analysis.packageSha256, analysis.admxCatalog.parser.version, analysis.admxCatalog.status, json(analysis.admxCatalog)]); for (const setting of analysis.settings) { const identity = setting.identity.status === 'resolved' ? setting.identity.identity : undefined; if (identity) { await client.query(` INSERT INTO assessor_policy_identities (id, resolver_id, resolver_version, scope, target_kind, data) VALUES ($1, $2, $3, $4, $5, $6::jsonb) ON CONFLICT (id) DO NOTHING `, [identity.id, identity.resolver.id, identity.resolver.version, identity.scope, identity.target.kind, json({ version: identity.version, id: identity.id, resolver: identity.resolver, scope: identity.scope, target: identity.target })]); } await client.query(` INSERT INTO assessor_microsoft_security_baseline_settings (id, analysis_id, catalog_id, baseline_id, identity_id, backup_id, scope, source_path, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) `, [setting.id, analysis.id, analysis.admxCatalog.id, analysis.baselineId, identity?.id ?? null, setting.backupId, setting.observation.scope, setting.observation.sourcePath, json(setting)]); } return true; }, undefined, changed => changed); } async listMicrosoftSecurityBaselineAnalyses(baselineId?: string): Promise { const result = baselineId ? await this.pool.query>( 'SELECT data FROM assessor_microsoft_security_baseline_analyses WHERE baseline_id = $1 ORDER BY analyzed_at, id', [baselineId]) : await this.pool.query>( 'SELECT data FROM assessor_microsoft_security_baseline_analyses ORDER BY analyzed_at, id'); return dataRows(result); } async enqueueIngestionJob(job: IngestionJob): Promise { const errors = ingestionJobValidationErrors(job); if (errors.length) throw new RepositoryValidationError(errors.join(' ')); if (job.status !== 'queued' || job.attempts !== 0) throw new RepositoryValidationError('Ein neuer Job muss ohne Versuche in queued beginnen.'); const outcome = await this.mutate(async client => { const context = await client.query<{ project_exists: boolean; artifact_valid: boolean; } & QueryResultRow>(`SELECT EXISTS (SELECT 1 FROM assessor_projects WHERE id = $1) AS project_exists, ($2::text IS NULL OR EXISTS (SELECT 1 FROM assessor_import_artifacts WHERE id = $2 AND project_id = $1)) AS artifact_valid `, [job.projectId, job.artifactId ?? null]); if (!context.rows[0]?.project_exists || !context.rows[0].artifact_valid) { throw new RepositoryNotFoundError('Jobreferenzen gehören nicht zum angegebenen Projekt.'); } const inserted = await client.query>(` INSERT INTO assessor_ingestion_jobs (id, project_id, artifact_id, job_kind, idempotency_key, job_status, attempts, max_attempts, available_at, lease_owner, lease_expires_at, created_at, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL, NULL, $10, $11, $12::jsonb) ON CONFLICT (project_id, idempotency_key) DO NOTHING RETURNING data `, [job.id, job.projectId, job.artifactId ?? null, job.kind, job.idempotencyKey, job.status, job.attempts, job.maxAttempts, job.availableAt, job.createdAt, job.updatedAt, json(job)]); if (inserted.rows[0]) return { job: clone(inserted.rows[0].data), inserted: true }; const existing = await client.query>( 'SELECT data FROM assessor_ingestion_jobs WHERE project_id = $1 AND idempotency_key = $2 FOR UPDATE', [job.projectId, job.idempotencyKey] ); const stored = existing.rows[0]?.data; if (!stored || stored.kind !== job.kind || stored.artifactId !== job.artifactId) { throw new RepositoryConflictError('Idempotenzschlüssel bezeichnet einen anderen Job.'); } return { job: clone(stored), inserted: false }; }, undefined, result => result.inserted); return outcome.job; } async claimNextIngestionJob(claim: IngestionJobClaim): Promise { if (!claim.workerId.trim() || !Number.isInteger(claim.leaseDurationMs) || claim.leaseDurationMs < 1000 || claim.leaseDurationMs > 3600000 || !Number.isFinite(Date.parse(claim.now))) { throw new RepositoryValidationError('Worker, Zeitpunkt oder Lease-Dauer sind ungültig.'); } const outcome = await this.mutate(async client => { const exhausted = await client.query(` UPDATE assessor_ingestion_jobs SET job_status = 'dead-letter', lease_owner = NULL, lease_expires_at = NULL, updated_at = $1, data = (data - 'leaseOwner' - 'leaseExpiresAt') || jsonb_build_object( 'status', 'dead-letter', 'lastErrorCode', coalesce(data->>'lastErrorCode', 'lease-expired'), 'updatedAt', $1::text ) WHERE job_status = 'running' AND attempts >= max_attempts AND lease_expires_at <= $1 AND ($2::text IS NULL OR project_id = $2) `, [claim.now, claim.projectId ?? null]); const selected = await client.query>(` SELECT data FROM assessor_ingestion_jobs WHERE attempts < max_attempts AND ($2::text[] IS NULL OR job_kind = ANY($2::text[])) AND ($3::text IS NULL OR project_id = $3) AND ((job_status = 'queued' AND available_at <= $1) OR (job_status = 'running' AND lease_expires_at <= $1)) ORDER BY available_at, created_at, id FOR UPDATE SKIP LOCKED LIMIT 1 `, [claim.now, claim.kinds?.length ? claim.kinds : null, claim.projectId ?? null]); const current = selected.rows[0]?.data; if (!current) return { job: null, cleaned: Boolean(exhausted.rowCount) }; const updated: IngestionJob = { ...current, status: 'running', attempts: current.attempts + 1, leaseOwner: claim.workerId, leaseExpiresAt: new Date(Date.parse(claim.now) + claim.leaseDurationMs).toISOString(), updatedAt: claim.now }; await client.query(` UPDATE assessor_ingestion_jobs SET job_status = 'running', attempts = $2, lease_owner = $3, lease_expires_at = $4, updated_at = $5, data = $6::jsonb WHERE id = $1 `, [updated.id, updated.attempts, updated.leaseOwner, updated.leaseExpiresAt, updated.updatedAt, json(updated)]); return { job: updated, cleaned: Boolean(exhausted.rowCount) }; }, undefined, result => result.cleaned || result.job !== null); return outcome.job; } async completeIngestionJob(jobId: string, workerId: string, completedAt: string): Promise { return this.mutate(async client => { const selected = await client.query>('SELECT data FROM assessor_ingestion_jobs WHERE id = $1 FOR UPDATE', [jobId]); const current = selected.rows[0]?.data; if (!current) throw new RepositoryNotFoundError(`Ingestion Job ${jobId} ist nicht vorhanden.`); if (current.status !== 'running' || current.leaseOwner !== workerId || !Number.isFinite(Date.parse(completedAt)) || Date.parse(completedAt) > Date.parse(current.leaseExpiresAt ?? '')) throw new RepositoryConflictError('Job-Lease ist nicht mehr gültig oder gehört einem anderen Worker.'); const updated: IngestionJob = { ...current, status: 'succeeded', updatedAt: completedAt }; delete updated.leaseOwner; delete updated.leaseExpiresAt; await client.query(`UPDATE assessor_ingestion_jobs SET job_status = 'succeeded', lease_owner = NULL, lease_expires_at = NULL, updated_at = $2, data = $3::jsonb WHERE id = $1`, [jobId, completedAt, json(updated)]); return updated; }); } async failIngestionJob(jobId: string, workerId: string, failedAt: string, errorCode: string, nextAvailableAt: string, retryable = true): Promise { return this.mutate(async client => { const selected = await client.query>('SELECT data FROM assessor_ingestion_jobs WHERE id = $1 FOR UPDATE', [jobId]); const current = selected.rows[0]?.data; if (!current) throw new RepositoryNotFoundError(`Ingestion Job ${jobId} ist nicht vorhanden.`); const failed = Date.parse(failedAt); const nextAvailable = Date.parse(nextAvailableAt); if (current.status !== 'running' || current.leaseOwner !== workerId || !errorCode.trim() || !Number.isFinite(failed) || !Number.isFinite(nextAvailable) || nextAvailable < failed || failed > Date.parse(current.leaseExpiresAt ?? '')) throw new RepositoryConflictError('Job-Fehler kann ohne gültigen Lease und Retry-Zeitpunkt nicht gespeichert werden.'); const exhausted = !retryable || current.attempts >= current.maxAttempts; const updated: IngestionJob = { ...current, status: exhausted ? 'dead-letter' : 'queued', availableAt: exhausted ? current.availableAt : nextAvailableAt, lastErrorCode: errorCode.trim(), updatedAt: failedAt }; delete updated.leaseOwner; delete updated.leaseExpiresAt; await client.query(`UPDATE assessor_ingestion_jobs SET job_status = $2, available_at = $3, lease_owner = NULL, lease_expires_at = NULL, updated_at = $4, data = $5::jsonb WHERE id = $1`, [jobId, updated.status, updated.availableAt, updated.updatedAt, json(updated)]); return updated; }); } async listIngestionJobs(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_ingestion_jobs WHERE project_id = $1 ORDER BY created_at, id', [projectId] )); } async retryIngestionJob(projectId: string, jobId: string, retriedAt: string): Promise { if (!Number.isFinite(Date.parse(retriedAt))) throw new RepositoryValidationError('Der Retry-Zeitpunkt ist ungültig.'); return this.mutate(async client => { const selected = await client.query>( `SELECT data FROM assessor_ingestion_jobs WHERE id = $1 AND project_id = $2 AND job_kind = 'analyze-gpo-backup' FOR UPDATE`, [jobId, projectId] ); const current = selected.rows[0]?.data; if (!current) throw new RepositoryNotFoundError('Der GPO-Analysejob ist im Mandanten nicht vorhanden.'); if (current.status !== 'dead-letter') throw new RepositoryConflictError('Nur ein endgültig fehlgeschlagener GPO-Analysejob kann erneut gestartet werden.'); const updated: IngestionJob = { ...current, status: 'queued', attempts: 0, availableAt: retriedAt, updatedAt: retriedAt }; delete updated.lastErrorCode; await client.query(`UPDATE assessor_ingestion_jobs SET job_status = 'queued', attempts = 0, available_at = $2, lease_owner = NULL, lease_expires_at = NULL, updated_at = $2, data = $3::jsonb WHERE id = $1`, [jobId, retriedAt, json(updated)]); return updated; }); } async deleteGpoArtifact(projectId: string, artifactId: string, request: ArtifactDeletionRequest) { return this.mutate(async client => { const selected = await client.query>( 'SELECT data FROM assessor_import_artifacts WHERE id = $1 AND project_id = $2 FOR UPDATE', [artifactId, projectId] ); const artifact = selected.rows[0]?.data; if (!artifact || artifact.storage.kind !== 'backend-controlled' || artifact.kind !== 'source-file' || !artifact.relativePath.startsWith('gpo-backups/')) { throw new RepositoryNotFoundError('Das GPO-Artefakt ist im Mandanten nicht vorhanden.'); } const jobs = dataRows(await client.query>( `SELECT data FROM assessor_ingestion_jobs WHERE artifact_id = $1 AND project_id = $2 AND job_kind = 'analyze-gpo-backup' FOR UPDATE`, [artifactId, projectId] )); if (jobs.some(item => item.status === 'queued' || item.status === 'running')) throw new RepositoryConflictError('Ein laufender GPO-Import kann nicht gelöscht werden.'); if (request.version !== 1 || request.projectId !== projectId || request.artifactId !== artifactId || request.storageKey !== artifact.storage.storageKey || request.attempts !== 0 || !request.id.trim() || !Number.isFinite(Date.parse(request.createdAt)) || !Number.isFinite(Date.parse(request.nextAttemptAt))) { throw new RepositoryValidationError('Der Artefakt-Löschauftrag ist ungültig.'); } const inserted = await client.query>(` INSERT INTO assessor_artifact_deletion_outbox (id, project_id, artifact_id, storage_key, attempts, next_attempt_at, created_at, last_error_code, data) VALUES ($1, $2, $3, $4, 0, $5, $6, NULL, $7::jsonb) ON CONFLICT (storage_key) DO NOTHING RETURNING data `, [request.id, request.projectId, request.artifactId, request.storageKey, request.nextAttemptAt, request.createdAt, json(request)]); let storedRequest = inserted.rows[0]?.data; if (!storedRequest) storedRequest = (await client.query>( 'SELECT data FROM assessor_artifact_deletion_outbox WHERE storage_key = $1 FOR UPDATE', [request.storageKey] )).rows[0]?.data; if (!storedRequest || storedRequest.artifactId !== artifactId || storedRequest.projectId !== projectId) throw new RepositoryConflictError('Speicherobjekt besitzt einen abweichenden Löschauftrag.'); await client.query(` DELETE FROM assessor_policy_snapshot_aggregation_runs aggregation WHERE EXISTS ( SELECT 1 FROM assessor_policy_snapshot_aggregation_sources source JOIN assessor_policy_identity_analysis_runs analysis ON analysis.id = source.analysis_run_id WHERE source.aggregation_run_id = aggregation.id AND analysis.project_id = $1 AND analysis.artifact_id = $2 ) `, [projectId, artifactId]); await client.query( 'UPDATE assessor_policy_identity_analysis_runs SET supersedes_analysis_run_id = NULL WHERE project_id = $1 AND artifact_id = $2', [projectId, artifactId] ); await client.query( 'UPDATE assessor_import_processing_runs SET supersedes_run_id = NULL WHERE project_id = $1 AND artifact_id = $2', [projectId, artifactId] ); await client.query('DELETE FROM assessor_import_artifacts WHERE id = $1 AND project_id = $2', [artifactId, projectId]); await client.query(` DELETE FROM assessor_finding_definitions definition WHERE NOT EXISTS (SELECT 1 FROM assessor_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) AND NOT EXISTS (SELECT 1 FROM assessor_policy_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) `); return { artifact: clone(artifact), request: clone(storedRequest) }; }); } async listArtifactDeletionRequests(dueAt: string, limit: number): Promise { if (!Number.isFinite(Date.parse(dueAt)) || !Number.isInteger(limit) || limit < 1 || limit > 1_000) throw new RepositoryValidationError('Löschabfrage ist ungültig.'); return dataRows(await this.pool.query>( 'SELECT data FROM assessor_artifact_deletion_outbox WHERE next_attempt_at <= $1 ORDER BY next_attempt_at, created_at, id LIMIT $2', [dueAt, limit] )); } async completeArtifactDeletionRequest(requestId: string): Promise { const result = await this.mutate(client => client.query('DELETE FROM assessor_artifact_deletion_outbox WHERE id = $1', [requestId]), undefined, outcome => Boolean(outcome.rowCount)); if (!result.rowCount) throw new RepositoryNotFoundError('Artefakt-Löschauftrag ist nicht vorhanden.'); } async failArtifactDeletionRequest(requestId: string, failedAt: string, errorCode: string, nextAttemptAt: string): Promise { if (!errorCode.trim() || !Number.isFinite(Date.parse(failedAt)) || !Number.isFinite(Date.parse(nextAttemptAt)) || Date.parse(nextAttemptAt) < Date.parse(failedAt)) { throw new RepositoryValidationError('Fehlgeschlagener Löschauftrag ist ungültig.'); } await this.mutate(async client => { const selected = await client.query>('SELECT data FROM assessor_artifact_deletion_outbox WHERE id = $1 FOR UPDATE', [requestId]); const current = selected.rows[0]?.data; if (!current) throw new RepositoryNotFoundError('Artefakt-Löschauftrag ist nicht vorhanden.'); const updated: ArtifactDeletionRequest = { ...current, attempts: current.attempts + 1, lastErrorCode: errorCode.trim(), nextAttemptAt }; await client.query(`UPDATE assessor_artifact_deletion_outbox SET attempts = $2, next_attempt_at = $3, last_error_code = $4, data = $5::jsonb WHERE id = $1`, [requestId, updated.attempts, nextAttemptAt, updated.lastErrorCode, json(updated)]); }); } async saveReview(review: FindingReview, expectedRevision?: number): Promise { await this.mutate(async client => { const existing = await client.query<{ revision: number; data: FindingReview } & QueryResultRow>( 'SELECT revision, data FROM assessor_reviews WHERE id = $1 FOR UPDATE', [review.id] ); assertRiskAcceptanceUnchanged(existing.rows[0]?.data, review); const currentRevision = existing.rows[0]?.revision ?? 0; if (expectedRevision !== undefined && expectedRevision !== currentRevision) throw new RepositoryConflictError(`Review ${review.id} wurde zwischenzeitlich geändert.`); if (review.revision !== currentRevision + 1) throw new RepositoryValidationError(`Reviewrevision muss ${currentRevision + 1} sein.`); const definitionId = review.target.kind === 'definition' ? review.target.definitionId : null; const occurrenceId = review.target.kind === 'occurrence' ? review.target.occurrenceId : null; await client.query(` INSERT INTO assessor_reviews (id, target_kind, definition_id, occurrence_id, revision, updated_at, data) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb) ON CONFLICT (id) DO UPDATE SET target_kind = EXCLUDED.target_kind, definition_id = EXCLUDED.definition_id, occurrence_id = EXCLUDED.occurrence_id, revision = EXCLUDED.revision, updated_at = EXCLUDED.updated_at, data = EXCLUDED.data `, [review.id, review.target.kind, definitionId, occurrenceId, review.revision, review.updatedAt, json(review)]); }); } async recordRiskAcceptance(projectId: string, occurrenceId: string, command: RiskAcceptanceCommand, expectedRevision: number): Promise { return this.mutate(async client => { // Lock the occurrence as well: two first decisions must not create parallel review histories. const target = await client.query(`SELECT o.id FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id=o.scan_run_id JOIN assessor_assessments a ON a.id=r.assessment_id WHERE o.id=$1 AND a.project_id=$2 FOR UPDATE OF o`, [occurrenceId, projectId]); if (!target.rowCount) throw new RepositoryNotFoundError('Finding is not in this project.'); const existing = await client.query>(`SELECT data FROM assessor_reviews WHERE occurrence_id=$1 ORDER BY revision DESC, updated_at DESC, id LIMIT 1 FOR UPDATE`, [occurrenceId]); const current = existing.rows[0]?.data; if (!Number.isSafeInteger(expectedRevision) || expectedRevision !== (current?.revision ?? 0)) throw new RepositoryConflictError('The review changed. Reload it before recording a decision.'); const review = appendRiskAcceptanceReview(current, occurrenceId, command, new Date().toISOString()); await client.query(`INSERT INTO assessor_reviews (id, target_kind, definition_id, occurrence_id, revision, updated_at, data) VALUES ($1, 'occurrence', NULL, $2, $3, $4, $5::jsonb) ON CONFLICT (id) DO UPDATE SET revision=EXCLUDED.revision, updated_at=EXCLUDED.updated_at, data=EXCLUDED.data`, [review.id, occurrenceId, review.revision, review.updatedAt, json(review)]); return review; }); } async recordScopedRiskAcceptance(projectId: string, write: ScopedRiskAcceptanceWrite): Promise { return this.mutate(async client => { const existing = write.acceptanceId ? await client.query>(`SELECT data FROM assessor_scoped_risk_acceptances WHERE id=$1 AND project_id=$2 FOR UPDATE`, [write.acceptanceId, projectId]) : undefined; const current = existing?.rows[0]?.data; if (write.acceptanceId && !current) throw new RepositoryNotFoundError('Scoped risk acceptance is not available.'); if (!Number.isSafeInteger(write.expectedRevision) || write.expectedRevision !== (current?.revision ?? 0)) { throw new RepositoryConflictError('The scoped risk acceptance changed. Reload it before recording a decision.'); } let target: ScopedRiskAcceptanceTarget; if (current) { if (write.target) throw new RepositoryValidationError('An existing scoped risk acceptance cannot be retargeted.'); target = current.target; } else { if (!write.target) throw new RepositoryValidationError('A scoped risk target is required.'); let occurrences: FindingOccurrence[]; let memberOccurrenceIds: Set | undefined; if (write.target.kind === 'device-finding') { const result = await client.query>(`SELECT o.data FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id=o.scan_run_id JOIN assessor_assessments a ON a.id=r.assessment_id WHERE o.id=$1 AND a.project_id=$2 FOR UPDATE OF o`, [write.target.occurrenceId, projectId]); const selected = dataRows(result)[0]; occurrences = selected ? dataRows(await client.query>(`SELECT o.data FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id=o.scan_run_id JOIN assessor_assessments a ON a.id=r.assessment_id WHERE o.scan_run_id=$1 AND a.project_id=$2 ORDER BY o.id FOR UPDATE OF o`, [selected.scanRunId, projectId])) : []; memberOccurrenceIds = selected ? new Set([selected.id]) : undefined; } else { if (!write.target.scanRunIds.length) throw new RepositoryValidationError('A non-empty scan-run scope is required.'); const runCheck = await client.query<{ id: string } & QueryResultRow>(`SELECT r.id FROM assessor_scan_runs r JOIN assessor_assessments a ON a.id=r.assessment_id WHERE a.project_id=$1 AND r.id=ANY($2::text[]) FOR UPDATE OF r`, [projectId, write.target.scanRunIds]); if (new Set(runCheck.rows.map(row => row.id)).size !== new Set(write.target.scanRunIds).size) { throw new RepositoryValidationError('The selected scan-run scope is not valid for this project.'); } occurrences = dataRows(await client.query>(`SELECT o.data FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id=o.scan_run_id JOIN assessor_assessments a ON a.id=r.assessment_id WHERE a.project_id=$1 AND o.scan_run_id=ANY($2::text[]) ORDER BY o.id FOR UPDATE OF o`, [projectId, write.target.scanRunIds])); } const now = new Date().toISOString(); try { target = scopedRiskTargetFromOccurrences(write.target.kind, write.target.deviceKey, occurrences, now, memberOccurrenceIds); } catch (error) { throw new RepositoryValidationError(error instanceof Error ? error.message : 'Invalid scoped risk target.'); } } const now = new Date().toISOString(); let saved: ScopedRiskAcceptance; try { saved = appendScopedRiskAcceptance(current, projectId, current?.id ?? `scoped-risk-${randomUUID()}`, target, write.command, now); } catch (error) { throw new RepositoryValidationError(error instanceof Error ? error.message : 'Invalid scoped risk acceptance.'); } await client.query(`INSERT INTO assessor_scoped_risk_acceptances (id, project_id, scope_kind, device_key, revision, created_at, updated_at, data) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb) ON CONFLICT (id) DO UPDATE SET revision=EXCLUDED.revision, updated_at=EXCLUDED.updated_at, data=EXCLUDED.data`, [saved.id, projectId, saved.target.kind, saved.target.device.key, saved.revision, saved.createdAt, saved.updatedAt, json(saved)]); return saved; }); } async listScopedRiskAcceptances(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_scoped_risk_acceptances WHERE project_id=$1 ORDER BY updated_at DESC, id', [projectId] )); } async listReviews(projectId: string): Promise { return dataRows(await this.pool.query>(` SELECT DISTINCT rv.data, rv.id FROM assessor_reviews rv WHERE EXISTS ( SELECT 1 FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id = o.scan_run_id JOIN assessor_assessments a ON a.id = r.assessment_id WHERE a.project_id = $1 AND (rv.occurrence_id = o.id OR rv.definition_id = o.definition_id) ) ORDER BY rv.id `, [projectId])); } async saveProfileRegistration(profile: ProfileRegistration): Promise { if (!profile.profileId.trim() || !profile.profileVersion.trim() || profile.schemaVersion < 1) throw new RepositoryValidationError('Profil-ID, Version und Schemaversion sind erforderlich.'); await this.mutate(async client => { await client.query(` INSERT INTO assessor_profile_registrations (profile_id, profile_version, schema_version, data) VALUES ($1, $2, $3, $4::jsonb) ON CONFLICT (profile_id, profile_version) DO UPDATE SET schema_version = EXCLUDED.schema_version, data = EXCLUDED.data `, [profile.profileId, profile.profileVersion, profile.schemaVersion, json(profile)]); }); } async listProfileRegistrations(): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_profile_registrations ORDER BY profile_id, profile_version' )); } async saveLocalComparisonRule(projectId: string, rule: LocalComparisonRule): Promise { const issues = validateLocalComparisonRules([rule]); if (issues.length) throw new RepositoryValidationError(issues.join(' ')); await this.mutate(async client => { await client.query(` INSERT INTO assessor_local_comparison_rules (project_id, id, data) VALUES ($1, $2, $3::jsonb) ON CONFLICT (project_id, id) DO UPDATE SET data = EXCLUDED.data `, [projectId, rule.id, json(rule)]); }); } async listLocalComparisonRules(projectId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_local_comparison_rules WHERE project_id = $1 ORDER BY id', [projectId] )); } async saveComparison(stored: StoredComparison): Promise { await this.mutate(async client => { const context = await client.query<{ valid_context: boolean; valid_runs: boolean; valid_occurrences: boolean } & QueryResultRow>(` SELECT EXISTS (SELECT 1 FROM assessor_assessments WHERE id = $1 AND project_id = $2) AS valid_context, (SELECT count(*) = 2 FROM assessor_scan_runs WHERE assessment_id = $1 AND id = ANY($3::text[])) AS valid_runs, NOT EXISTS ( SELECT 1 FROM unnest($4::text[]) occurrence_id WHERE occurrence_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM assessor_finding_occurrences o JOIN assessor_scan_runs r ON r.id = o.scan_run_id WHERE o.id = occurrence_id AND r.assessment_id = $1 ) ) AS valid_occurrences `, [ stored.assessmentId, stored.projectId, [stored.comparison.baselineRunId, stored.comparison.targetRunId], stored.comparison.results.flatMap(result => [result.baselineOccurrenceId, result.targetOccurrenceId, result.historicalOccurrenceId]).filter(Boolean) ]); const row = context.rows[0]; if (!row?.valid_context) throw new RepositoryNotFoundError('Projekt oder Assessment für den Vergleich fehlt.'); if (!row.valid_runs || !row.valid_occurrences) throw new RepositoryValidationError('Vergleich verweist auf Scan-Läufe oder Findings eines anderen Assessments.'); const existing = await client.query<{ project_id: string; assessment_id: string } & QueryResultRow>( 'SELECT project_id, assessment_id FROM assessor_comparisons WHERE id = $1 FOR UPDATE', [stored.id] ); if (existing.rows[0] && (existing.rows[0].project_id !== stored.projectId || existing.rows[0].assessment_id !== stored.assessmentId)) { throw new RepositoryConflictError('Ein Vergleich kann nicht in ein anderes Projekt oder Assessment verschoben werden.'); } await client.query(` INSERT INTO assessor_comparisons (id, project_id, assessment_id, data) VALUES ($1, $2, $3, $4::jsonb) ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data `, [stored.id, stored.projectId, stored.assessmentId, json(stored)]); }); } async loadComparison(comparisonId: string): Promise { return dataRow(await this.pool.query>( 'SELECT data FROM assessor_comparisons WHERE id = $1', [comparisonId] )); } async listComparisons(assessmentId: string): Promise { return dataRows(await this.pool.query>( 'SELECT data FROM assessor_comparisons WHERE assessment_id = $1 ORDER BY id', [assessmentId] )); } async exportProject(projectId: string): Promise { const project = await this.getProject(projectId); if (!project) throw new RepositoryNotFoundError(`Projekt ${projectId} ist nicht vorhanden.`); const assessments = await this.listAssessments(projectId); const models = (await Promise.all(assessments.map(assessment => this.loadAssessmentModel(assessment.id)))) .filter((model): model is CanonicalAssessmentModel => model !== null); const policyFindingRuns = await this.listPolicyFindingRuns(projectId); const uniqueDefinitions = new Map([ ...models.flatMap(model => model.findingDefinitions), ...policyFindingRuns.flatMap(record => record.definitions) ].map(item => [item.id, item])); return { version: 1, project, assessments, scanRuns: models.flatMap(model => model.scanRuns), importFiles: models.flatMap(model => model.importFiles), sourceTables: models.flatMap(model => model.sourceTables), reportSections: models.flatMap(model => model.reportSections ?? []), environmentObservations: models.flatMap(model => model.environmentObservations ?? []), findingDefinitions: [...uniqueDefinitions.values()].sort((left, right) => left.id.localeCompare(right.id)), findingOccurrences: models.flatMap(model => model.findingOccurrences), cveEnrichments: [...new Map(models.flatMap(model => model.cveEnrichments ?? []).map(item => [item.cveId, item])).values()] .sort((left, right) => left.cveId.localeCompare(right.cveId)), evidence: models.flatMap(model => model.evidence), reviews: await this.listReviews(projectId), scopedRiskAcceptances: await this.listScopedRiskAcceptances(projectId), diagnostics: models.flatMap(model => model.diagnostics), comparisons: dataRows(await this.pool.query>( 'SELECT data FROM assessor_comparisons WHERE project_id = $1 ORDER BY id', [projectId] )), localComparisonRules: await this.listLocalComparisonRules(projectId), importArtifacts: await this.listImportArtifacts(projectId), processingRuns: dataRows(await this.pool.query>( 'SELECT data FROM assessor_import_processing_runs WHERE project_id = $1 ORDER BY started_at, id', [projectId] )), ingestionJobs: await this.listIngestionJobs(projectId), gpoBackupInstances: await this.listGpoBackupInstances(projectId), gpoPolicyAnalyses: await this.listGpoPolicyAnalyses(projectId), environmentProfiles: await this.listEnvironmentProfiles(projectId), policySnapshotAggregations: await this.listPolicySnapshotAggregations(projectId), policyEvaluations: await this.listPolicyEvaluations(projectId), policyFindingRuns, aiOverviewRuns: (await this.listAiOverviewRuns(projectId)).filter(item => item.status === 'succeeded'), regulatoryMappingBatchRuns: await this.listRegulatoryMappingBatchRuns(projectId), // Regulatory mappings and decisions are global reference data, not project-scoped. regulatoryMappingRuns: (await this.listRegulatoryMappingRuns()).filter(item => item.status === 'succeeded'), regulatoryMappingDecisions: dataRows(await this.pool.query>( 'SELECT data FROM assessor_regulatory_mapping_decisions ORDER BY created_at, id' )) }; } async deleteScanRun(projectId: string, scanRunId: string, confirmation: ScanRunDeleteConfirmation): Promise { if (confirmation.projectId !== projectId || confirmation.scanRunId !== scanRunId || confirmation.cascade !== true) { throw new RepositoryValidationError('Scanberichtslöschung wurde nicht für das exakte Ziel bestätigt.'); } return this.mutate(async client => { const context = await client.query<{ assessment_id: string; project_id: string } & QueryResultRow>(` SELECT run.assessment_id, assessment.project_id FROM assessor_scan_runs run JOIN assessor_assessments assessment ON assessment.id = run.assessment_id WHERE run.id = $1 FOR UPDATE OF run, assessment `, [scanRunId]); const row = context.rows[0]; if (!row) return false; if (row.project_id !== projectId) throw new RepositoryValidationError('Der Scanbericht gehört nicht zum bestätigten Projekt.'); const hashes = await client.query<{ sha256: string | null } & QueryResultRow>( `SELECT data->>'sha256' AS sha256 FROM assessor_import_files WHERE scan_run_id = $1`, [scanRunId] ); await client.query('DELETE FROM assessor_comparisons WHERE assessment_id = $1', [row.assessment_id]); await client.query('DELETE FROM assessor_ai_overview_runs WHERE project_id = $1 AND $2 = ANY(scan_run_ids)', [projectId, scanRunId]); const deleted = await client.query('DELETE FROM assessor_scan_runs WHERE id = $1 RETURNING id', [scanRunId]); if (!deleted.rowCount) throw new RepositoryConflictError('Scanbericht wurde zwischen Prüfung und Löschung bereits entfernt.'); await client.query(` DELETE FROM assessor_assessments assessment WHERE assessment.id = $1 AND NOT EXISTS (SELECT 1 FROM assessor_scan_runs run WHERE run.assessment_id = assessment.id) `, [row.assessment_id]); await client.query(` DELETE FROM assessor_finding_definitions definition WHERE NOT EXISTS (SELECT 1 FROM assessor_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) AND NOT EXISTS (SELECT 1 FROM assessor_policy_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) `); const removedHashes = hashes.rows.map(item => item.sha256).filter((value): value is string => Boolean(value)); if (removedHashes.length) await client.query(` DELETE FROM assessor_import_fingerprints fingerprint WHERE fingerprint.sha256 = ANY($1::text[]) AND NOT EXISTS ( SELECT 1 FROM assessor_import_files file WHERE file.data->>'sha256' = fingerprint.sha256 ) `, [removedHashes]); return true; }, undefined, Boolean); } async deleteAssessment(projectId: string, assessmentId: string, confirmation: AssessmentDeleteConfirmation): Promise { if (confirmation.projectId !== projectId || confirmation.assessmentId !== assessmentId || confirmation.cascade !== true) { throw new RepositoryValidationError('Timeslot deletion was not confirmed for the exact target.'); } return this.mutate(async client => { const context = await client.query<{ id: string; project_id: string } & QueryResultRow>( 'SELECT id, project_id FROM assessor_assessments WHERE id = $1 FOR UPDATE', [assessmentId] ); const assessment = context.rows[0]; if (!assessment) return false; if (assessment.project_id !== projectId) throw new RepositoryValidationError('The timeslot does not belong to the confirmed client.'); const hashes = await client.query<{ sha256: string | null } & QueryResultRow>(` SELECT file.data->>'sha256' AS sha256 FROM assessor_import_files file JOIN assessor_scan_runs run ON run.id = file.scan_run_id WHERE run.assessment_id = $1 `, [assessmentId]); await client.query('DELETE FROM assessor_asset_edges WHERE assessment_id = $1', [assessmentId]); await client.query('DELETE FROM assessor_edge_coverage WHERE assessment_id = $1', [assessmentId]); await client.query('DELETE FROM assessor_canonical_assets WHERE assessment_id = $1', [assessmentId]); await client.query('DELETE FROM assessor_ai_overview_runs WHERE project_id = $1 AND $2 = ANY(assessment_ids)', [projectId, assessmentId]); const deleted = await client.query('DELETE FROM assessor_assessments WHERE id = $1 RETURNING id', [assessmentId]); if (!deleted.rowCount) throw new RepositoryConflictError('The timeslot was removed between validation and deletion.'); await client.query(` DELETE FROM assessor_finding_definitions definition WHERE NOT EXISTS (SELECT 1 FROM assessor_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) AND NOT EXISTS (SELECT 1 FROM assessor_policy_finding_occurrences occurrence WHERE occurrence.definition_id = definition.id) `); const removedHashes = hashes.rows.map(item => item.sha256).filter((value): value is string => Boolean(value)); if (removedHashes.length) await client.query(` DELETE FROM assessor_import_fingerprints fingerprint WHERE fingerprint.sha256 = ANY($1::text[]) AND NOT EXISTS (SELECT 1 FROM assessor_import_files file WHERE file.data->>'sha256' = fingerprint.sha256) `, [removedHashes]); return true; }, undefined, Boolean); } async deleteProject(projectId: string, confirmation: ControlledDeleteConfirmation): Promise { if (confirmation.projectId !== projectId || confirmation.cascade !== true) throw new RepositoryValidationError('Projektlöschung wurde nicht für das exakte Ziel bestätigt.'); const exists = await this.getProject(projectId); if (!exists) return false; await this.mutate(async client => { const deleted = await client.query('DELETE FROM assessor_projects WHERE id = $1 RETURNING id', [projectId]); if (!deleted.rowCount) throw new RepositoryConflictError('Projekt wurde zwischen Prüfung und Löschung bereits entfernt.'); await client.query(` DELETE FROM assessor_finding_definitions d WHERE NOT EXISTS (SELECT 1 FROM assessor_finding_occurrences o WHERE o.definition_id = d.id) AND NOT EXISTS (SELECT 1 FROM assessor_policy_finding_occurrences o WHERE o.definition_id = d.id) `); }); return true; } async getDatabaseStructure(): Promise { const tablesResult = await this.pool.query<{ table_name: string }>(` SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_type = 'BASE TABLE' ORDER BY table_name `); const tableNames = tablesResult.rows .map(row => row.table_name) .filter(name => !DB_STRUCTURE_EXCLUDED_TABLES.has(name)); if (tableNames.length === 0) { return { generatedAt: new Date().toISOString(), persistence: 'postgresql', tables: [] }; } const columnsResult = await this.pool.query<{ table_name: string; column_name: string; data_type: string; is_nullable: string }>(` SELECT table_name, column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = 'public' AND table_name = ANY($1::text[]) ORDER BY table_name, ordinal_position `, [tableNames]); const foreignKeysResult = await this.pool.query<{ table_name: string; column_name: string; referenced_table_name: string; referenced_column_name: string; }>(` SELECT source_table.relname AS table_name, source_column.attname AS column_name, target_table.relname AS referenced_table_name, target_column.attname AS referenced_column_name FROM pg_constraint constraint_info JOIN pg_class source_table ON source_table.oid = constraint_info.conrelid JOIN pg_class target_table ON target_table.oid = constraint_info.confrelid JOIN LATERAL unnest(constraint_info.conkey, constraint_info.confkey) WITH ORDINALITY AS cols(source_attnum, target_attnum, ordinal) ON true JOIN pg_attribute source_column ON source_column.attrelid = source_table.oid AND source_column.attnum = cols.source_attnum JOIN pg_attribute target_column ON target_column.attrelid = target_table.oid AND target_column.attnum = cols.target_attnum WHERE constraint_info.contype = 'f' AND source_table.relnamespace = 'public'::regnamespace AND source_table.relname = ANY($1::text[]) ORDER BY source_table.relname, cols.ordinal `, [tableNames]); const rowCountResult = await this.pool.query<{ relname: string; reltuples: string }>(` SELECT relname, reltuples FROM pg_class WHERE relkind = 'r' AND relname = ANY($1::text[]) `, [tableNames]); const columnsByTable = new Map(); for (const row of columnsResult.rows) { const columns = columnsByTable.get(row.table_name) ?? []; columns.push({ name: row.column_name, dataType: row.data_type, nullable: row.is_nullable === 'YES' }); columnsByTable.set(row.table_name, columns); } const foreignKeysByTable = new Map(); for (const row of foreignKeysResult.rows) { const foreignKeys = foreignKeysByTable.get(row.table_name) ?? []; foreignKeys.push({ column: row.column_name, referencedTable: row.referenced_table_name, referencedColumn: row.referenced_column_name }); foreignKeysByTable.set(row.table_name, foreignKeys); } const rowCountByTable = new Map(rowCountResult.rows.map(row => [row.relname, Math.max(0, Math.round(Number(row.reltuples)))])); const tables: DbStructureTable[] = tableNames.map(name => ({ name, layer: dbStructureLayer(name), columns: columnsByTable.get(name) ?? [], foreignKeys: foreignKeysByTable.get(name) ?? [], approximateRowCount: rowCountByTable.get(name) ?? null })); return { generatedAt: new Date().toISOString(), persistence: 'postgresql', tables }; } }