Assessor_2.0/server/compliance/regulatory-auto-mapping.ts

import { createHash, randomUUID } from 'node:crypto';
import { mkdir, readFile, rename, rm, writeFile } from 'node:fs/promises';
import { dirname } from 'node:path';
import type { ComplianceReadPort } from '../../src/compliance/compliance-store';
import { AUTO_MAPPING_BATCH_KEY_PREFIXES, type RegulatoryMappingBatchCreateRequest } from '../../src/compliance/regulatory-mapping';
import type { RegulatoryMappingService } from './regulatory-mapping-service';

const [SCAN_KEY_PREFIX, GPO_KEY_PREFIX] = AUTO_MAPPING_BATCH_KEY_PREFIXES;
const DEFAULT_POLL_INTERVAL_MS = 10_000;
const PAGE_SIZE = 500;

interface AutoMappingSettings { version: 1; enabled: boolean }
interface PendingBatch { projectId: string; request: RegulatoryMappingBatchCreateRequest }

export interface RegulatoryAutoMappingOptions {
  /** Engine-owned settings file; no Core table or column stores the switch. */
  settingsPath: string;
  read: Pick<ComplianceReadPort, 'listProjects' | 'listAssessments' | 'listProjectScanRuns' | 'queryUnifiedResults'>;
  mapping: Pick<RegulatoryMappingService, 'generationEnabled' | 'startBatch'>;
  pollIntervalMs?: number;
}

/**
 * Maps newly uploaded scans and GPO evaluations automatically while its Data Hub switch is on (default off).
 * It reads the Core only through the read-only port: committed scan runs, and the GPO deviation rows of each
 * timeslot as filtered Result DB pages (GPO backups have no scan run). For every new scan run, and every
 * timeslot with new GPO deviation rows, it starts the existing ISO batch on those deviations; rules mapped
 * before are reused, so only new rules reach the model. What exists when the watcher starts or is switched
 * on forms the baseline and is not mapped retroactively.
 */
export class RegulatoryAutoMapping {
  private enabled = false;
  private knownScanRuns: Set<string> | null = null;
  private knownGpoRows: Set<string> | null = null;
  private readonly pending = new Map<string, PendingBatch>();
  private timer?: ReturnType<typeof setInterval>;
  private polling?: Promise<void>;

  constructor(private readonly options: RegulatoryAutoMappingOptions) {}

  get isEnabled(): boolean { return this.enabled; }

  async start(): Promise<void> {
    const settings = await this.readSettings();
    if (settings?.enabled) await this.activate();
  }

  async setEnabled(enabled: boolean): Promise<void> {
    if (typeof enabled !== 'boolean') throw new TypeError('enabled must be a boolean.');
    await this.writeSettings({ version: 1, enabled });
    if (enabled && !this.enabled) await this.activate();
    if (!enabled && this.enabled) await this.stop();
  }

  async stop(): Promise<void> {
    this.enabled = false;
    if (this.timer) clearInterval(this.timer);
    this.timer = undefined;
    await this.polling;
    this.knownScanRuns = null;
    this.knownGpoRows = null;
    this.pending.clear();
  }

  /** One watcher pass; serialized so a slow pass never overlaps the next. */
  poll(): Promise<void> {
    this.polling ??= this.pass().catch(() => undefined).finally(() => { this.polling = undefined; });
    return this.polling;
  }

  private async activate(): Promise<void> {
    this.enabled = true;
    await this.poll();
    this.timer = setInterval(() => { void this.poll(); }, this.options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS);
    this.timer.unref?.();
  }

  private async pass(): Promise<void> {
    if (!this.enabled) return;
    const scanRuns: Array<{ id: string; projectId: string; assessmentId: string }> = [];
    const gpoRows = new Map<string, { projectId: string; assessmentId: string; ids: string[] }>();
    for (const project of await this.options.read.listProjects()) {
      for (const run of await this.options.read.listProjectScanRuns(project.id)) {
        if (!run.supersededByScanRunId) scanRuns.push({ id: run.id, projectId: project.id, assessmentId: run.assessmentId });
      }
      for (const assessment of await this.options.read.listAssessments(project.id)) {
        const ids = await this.gpoDeviationRowIds(project.id, assessment.id);
        if (ids.length) gpoRows.set(assessment.id, { projectId: project.id, assessmentId: assessment.id, ids });
      }
    }
    if (!this.knownScanRuns || !this.knownGpoRows) {
      this.knownScanRuns = new Set(scanRuns.map(run => run.id));
      this.knownGpoRows = new Set([...gpoRows.values()].flatMap(entry => entry.ids));
      return;
    }
    for (const run of scanRuns) {
      if (this.knownScanRuns.has(run.id)) continue;
      this.knownScanRuns.add(run.id);
      this.pending.set(`scan:${run.id}`, { projectId: run.projectId, request: {
        // Stable per scan run, so a restart or a second pass never creates a duplicate batch.
        idempotencyKey: safeKey(`${SCAN_KEY_PREFIX}${run.id}`),
        assessmentIds: [run.assessmentId], scanRunIds: [run.id], filters: { deviation: true }
      } });
    }
    for (const { projectId, assessmentId, ids } of gpoRows.values()) {
      const added = ids.filter(id => !this.knownGpoRows!.has(id));
      if (!added.length) continue;
      for (const id of added) this.knownGpoRows.add(id);
      // Stable per set of new GPO rows; one batch maps every GPO deviation of the timeslot.
      const fingerprint = createHash('sha256').update(added.sort().join('\n'), 'utf8').digest('hex').slice(0, 16);
      this.pending.set(`gpo:${assessmentId}`, { projectId, request: {
        idempotencyKey: safeKey(`${GPO_KEY_PREFIX}${fingerprint}:${assessmentId}`),
        assessmentIds: [assessmentId], scanRunIds: [], filters: { sourceKind: ['policy-evaluation'], deviation: true }
      } });
    }
    // New imports wait until the ISO corpus and model are available.
    if (!this.options.mapping.generationEnabled) return;
    for (const [key, { projectId, request }] of [...this.pending]) {
      if (!this.enabled) return;
      this.pending.delete(key);
      try { await this.options.mapping.startBatch(projectId, request); }
      catch { /* A removed or superseded timeslot is skipped; the manual batch stays available. */ }
    }
  }

  private async gpoDeviationRowIds(projectId: string, assessmentId: string): Promise<string[]> {
    const ids: string[] = [];
    for (let offset = 0; ; offset += PAGE_SIZE) {
      const page = await this.options.read.queryUnifiedResults({ audience: 'result-db', projectId, assessmentIds: [assessmentId],
        sourceKind: ['policy-evaluation'], deviation: true, offset, limit: PAGE_SIZE });
      ids.push(...page.items.map(item => item.resultProjectionId));
      if (offset + page.items.length >= page.total || page.items.length === 0) return ids;
    }
  }

  private async readSettings(): Promise<AutoMappingSettings | null> {
    try {
      const parsed = JSON.parse(await readFile(this.options.settingsPath, 'utf8')) as Partial<AutoMappingSettings>;
      return { version: 1, enabled: parsed.enabled === true };
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null;
      throw error;
    }
  }

  private async writeSettings(settings: AutoMappingSettings): Promise<void> {
    await mkdir(dirname(this.options.settingsPath), { recursive: true, mode: 0o700 });
    const temporary = `${this.options.settingsPath}.${randomUUID()}.tmp`;
    try {
      await writeFile(temporary, JSON.stringify(settings), { mode: 0o600, flag: 'wx' });
      await rename(temporary, this.options.settingsPath);
    } finally { await rm(temporary, { force: true }); }
  }
}

function safeKey(value: string): string {
  return value.replace(/[^A-Za-z0-9._:-]/g, '-').slice(0, 128);
}