226 lines
12 KiB
TypeScript
226 lines
12 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'node:fs/promises';
|
|
import { join, resolve, sep } from 'node:path';
|
|
import type { RegulatoryCorpusV2 } from '../../src/compliance/regulatory-rag';
|
|
import type { RegulatoryCorpusJob, RegulatoryCorpusStatus, RegulatoryCorpusSummary } from '../../src/compliance/regulatory-corpus-management';
|
|
import type { IsoPdfCorpusOptions } from './iso-pdf-corpus';
|
|
import { prepareIsoCorpusInWorker } from './iso-pdf-worker-client';
|
|
import type { RegulatoryIngestionService } from './regulatory-rag-service';
|
|
import type { RegulatoryMappingService } from './regulatory-mapping-service';
|
|
import type { RegulatoryAutoMapping } from './regulatory-auto-mapping';
|
|
|
|
const MAXIMUM_PDF_BYTES = 32 * 1024 * 1024;
|
|
const MAXIMUM_CORPUS_BYTES = 64 * 1024 * 1024;
|
|
|
|
export class RegulatoryCorpusUploadError extends Error {
|
|
constructor(readonly code: string, message: string, readonly status = 400) {
|
|
super(message);
|
|
}
|
|
}
|
|
|
|
export interface RegulatoryCorpusUploadOptions {
|
|
root: string;
|
|
ingestion?: RegulatoryIngestionService;
|
|
mapping: RegulatoryMappingService;
|
|
autoMapping?: Pick<RegulatoryAutoMapping, 'isEnabled' | 'setEnabled'>;
|
|
embeddingModel?: string;
|
|
disabledReason?: string;
|
|
countIndexed?: (corpusId: string, corpusVersion: string) => Promise<number>;
|
|
pruneIndexed?: (corpusId: string, keepCorpusVersion: string) => Promise<void>;
|
|
prepare?: (options: IsoPdfCorpusOptions, signal?: AbortSignal) => Promise<RegulatoryCorpusV2>;
|
|
now?: () => Date;
|
|
}
|
|
|
|
export class RegulatoryCorpusUploadService {
|
|
private active: RegulatoryCorpusSummary | null = null;
|
|
private job: RegulatoryCorpusJob | null = null;
|
|
private task?: Promise<void>;
|
|
private abort?: AbortController;
|
|
private readonly now: () => Date;
|
|
private readonly prepare: NonNullable<RegulatoryCorpusUploadOptions['prepare']>;
|
|
|
|
constructor(private readonly options: RegulatoryCorpusUploadOptions) {
|
|
this.now = options.now ?? (() => new Date());
|
|
this.prepare = options.prepare ?? prepareIsoCorpusInWorker;
|
|
}
|
|
|
|
async initialize(): Promise<void> {
|
|
await mkdir(this.options.root, { recursive: true, mode: 0o700 });
|
|
const root = resolve(this.options.root);
|
|
for (const entry of await readdir(root, { withFileTypes: true })) {
|
|
if (!entry.isDirectory() || !/^\.upload-[a-f0-9-]{36}$/.test(entry.name)) continue;
|
|
const temporary = resolve(root, entry.name);
|
|
if (!temporary.startsWith(root + sep)) throw new Error('ISO temporary path escaped its private storage root.');
|
|
await rm(temporary, { recursive: true, force: true });
|
|
}
|
|
this.active = await this.readState<RegulatoryCorpusSummary>('active.json');
|
|
this.job = await this.readState<RegulatoryCorpusJob>('job.json');
|
|
if (this.job?.status === 'running') {
|
|
this.job = { ...this.job, status: 'failed', completedAt: this.now().toISOString(), error: 'Indexing was interrupted. Upload both PDFs again.' };
|
|
await this.writeState('job.json', this.job);
|
|
}
|
|
if (this.active && await this.corpusExists(this.active.corpusVersion)) {
|
|
if (this.active.embeddingModel === this.options.embeddingModel) {
|
|
await this.options.mapping.activateCorpus(this.active.corpusId, this.active.corpusVersion);
|
|
} else if (this.options.ingestion) {
|
|
this.job = { id: randomUUID(), status: 'failed', phase: 'complete', startedAt: this.now().toISOString(),
|
|
completedAt: this.now().toISOString(), error: 'The embedding model changed. Upload both PDFs to rebuild the index.' };
|
|
}
|
|
} else if (this.active) {
|
|
this.active = null;
|
|
this.job = { id: randomUUID(), status: 'failed', phase: 'complete', startedAt: this.now().toISOString(),
|
|
completedAt: this.now().toISOString(), error: 'The active corpus source is missing. Upload both PDFs again.' };
|
|
}
|
|
}
|
|
|
|
status(): RegulatoryCorpusStatus {
|
|
return {
|
|
version: 1,
|
|
configured: Boolean(this.options.ingestion),
|
|
...(!this.options.ingestion ? { disabledReason: this.options.disabledReason ?? 'Embedding or Qdrant is not configured.' } : {}),
|
|
active: this.active,
|
|
job: this.job,
|
|
...(this.options.autoMapping ? { autoMapping: { enabled: this.options.autoMapping.isEnabled } } : {})
|
|
};
|
|
}
|
|
|
|
async setAutoMapping(enabled: boolean): Promise<RegulatoryCorpusStatus> {
|
|
if (!this.options.autoMapping) throw new RegulatoryCorpusUploadError('regulatory_auto_mapping_unavailable', 'Automatic ISO mapping is not configured.', 503);
|
|
if (typeof enabled !== 'boolean') throw new RegulatoryCorpusUploadError('invalid_auto_mapping', 'enabled must be true or false.');
|
|
await this.options.autoMapping.setEnabled(enabled);
|
|
return this.liveStatus();
|
|
}
|
|
|
|
async liveStatus(): Promise<RegulatoryCorpusStatus> {
|
|
const status = this.status();
|
|
if (!this.active || !this.options.countIndexed) return status;
|
|
try {
|
|
const count = await this.options.countIndexed(this.active.corpusId, this.active.corpusVersion);
|
|
return { ...status, indexStatus: count === this.active.chunkCount ? 'ready' : 'incomplete' };
|
|
} catch { return { ...status, indexStatus: 'unavailable' }; }
|
|
}
|
|
|
|
async submit(iso27001: Buffer, iso27002: Buffer): Promise<RegulatoryCorpusStatus> {
|
|
if (!this.options.ingestion) throw new RegulatoryCorpusUploadError('regulatory_ingestion_unavailable', this.options.disabledReason ?? 'Embedding or Qdrant is not configured.', 503);
|
|
if (this.task) throw new RegulatoryCorpusUploadError('regulatory_ingestion_busy', 'A corpus upload is already running.', 409);
|
|
for (const [label, bytes] of [['ISO/IEC 27001', iso27001], ['ISO/IEC 27002', iso27002]] as const) {
|
|
if (!Buffer.isBuffer(bytes) || bytes.length < 1 || bytes.length > MAXIMUM_PDF_BYTES || !bytes.subarray(0, 1024).includes('%PDF-')) {
|
|
throw new RegulatoryCorpusUploadError('invalid_iso_pdf', `${label} must be a PDF no larger than 32 MiB.`);
|
|
}
|
|
}
|
|
this.job = { id: randomUUID(), status: 'running', phase: 'validating', startedAt: this.now().toISOString() };
|
|
const abort = new AbortController();
|
|
this.abort = abort;
|
|
// Claim the task slot before the first await so overlapping submissions receive 409.
|
|
const persisted = this.writeState('job.json', this.job);
|
|
this.task = persisted.then(() => this.run(iso27001, iso27002, abort.signal), () => undefined)
|
|
.finally(() => { this.task = undefined; });
|
|
await persisted;
|
|
return this.status();
|
|
}
|
|
|
|
async close(): Promise<void> { await this.task; }
|
|
|
|
async stop(): Promise<void> {
|
|
this.abort?.abort(new Error('Indexing was interrupted by a server shutdown. Upload both PDFs again.'));
|
|
await this.task;
|
|
}
|
|
|
|
private async run(iso27001: Buffer, iso27002: Buffer, signal: AbortSignal): Promise<void> {
|
|
const temporaryRoot = join(this.options.root, `.upload-${this.job!.id}`);
|
|
try {
|
|
await mkdir(temporaryRoot, { mode: 0o700 });
|
|
const iso27001Path = join(temporaryRoot, '27001.pdf');
|
|
const iso27002Path = join(temporaryRoot, '27002.pdf');
|
|
await Promise.all([writeFile(iso27001Path, iso27001, { mode: 0o600 }), writeFile(iso27002Path, iso27002, { mode: 0o600 })]);
|
|
const corpus = await this.prepare({ iso27001Path, iso27002Path }, signal);
|
|
signal.throwIfAborted();
|
|
try { await rm(temporaryRoot, { recursive: true, force: true }); }
|
|
catch { throw new Error('Temporary ISO PDFs could not be removed after parsing.'); }
|
|
if (!/^[a-z0-9][a-z0-9._-]{0,100}$/.test(corpus.corpusVersion)) throw new Error('The generated corpus version is invalid.');
|
|
const existing = this.active;
|
|
if (existing?.corpusId === corpus.corpusId && existing.corpusVersion === corpus.corpusVersion
|
|
&& existing.embeddingModel === this.options.embeddingModel && existing.parserVersion === corpus.parserVersion
|
|
&& this.options.countIndexed) {
|
|
const count = await this.options.countIndexed(corpus.corpusId, corpus.corpusVersion).catch(() => 0);
|
|
if (count === existing.chunkCount) {
|
|
await this.finish();
|
|
return;
|
|
}
|
|
}
|
|
const serialized = JSON.stringify(corpus);
|
|
if (Buffer.byteLength(serialized) > MAXIMUM_CORPUS_BYTES) throw new Error('The generated corpus exceeds 64 MiB.');
|
|
await this.writeState(`${corpus.corpusVersion}.json`, corpus);
|
|
await this.updatePhase('embedding');
|
|
const result = await this.options.ingestion!.ingest(corpus.documents, async (completedChunks, totalChunks) => {
|
|
signal.throwIfAborted();
|
|
this.job = { ...this.job!, completedChunks, totalChunks };
|
|
await this.writeState('job.json', this.job);
|
|
});
|
|
await this.updatePhase('activating');
|
|
const active: RegulatoryCorpusSummary = {
|
|
corpusId: corpus.corpusId, corpusVersion: corpus.corpusVersion,
|
|
embeddingModel: this.options.embeddingModel!, parserVersion: corpus.parserVersion,
|
|
documentCount: result.documentCount, chunkCount: result.chunkCount,
|
|
activatedAt: this.now().toISOString(),
|
|
sources: corpus.sources.map(source => ({ standard: source.standard, sha256: source.sha256, pageCount: source.pageCount }))
|
|
};
|
|
await this.writeState('active.json', active);
|
|
await this.options.mapping.activateCorpus(corpus.corpusId, corpus.corpusVersion);
|
|
this.active = active;
|
|
// Superseded versions are no longer retrievable; a failed prune is retried by the next activation.
|
|
await this.options.pruneIndexed?.(corpus.corpusId, corpus.corpusVersion).catch(() => undefined);
|
|
await this.finish();
|
|
} catch (error) {
|
|
this.job = { ...this.job!, status: 'failed', completedAt: this.now().toISOString(),
|
|
error: error instanceof Error ? error.message.slice(0, 300) : 'Corpus ingestion failed.' };
|
|
await this.writeState('job.json', this.job).catch(() => undefined);
|
|
} finally {
|
|
iso27001.fill(0);
|
|
iso27002.fill(0);
|
|
const root = resolve(this.options.root);
|
|
const temporary = resolve(temporaryRoot);
|
|
if (!temporary.startsWith(root + sep)) throw new Error('ISO temporary path escaped its private storage root.');
|
|
try { await rm(temporary, { recursive: true, force: true }); }
|
|
catch {
|
|
this.job = { ...this.job!, status: 'failed', completedAt: this.now().toISOString(),
|
|
error: 'Temporary ISO PDFs could not be removed from private storage. Restart the server after fixing storage access.' };
|
|
await this.writeState('job.json', this.job).catch(() => undefined);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async updatePhase(phase: RegulatoryCorpusJob['phase']): Promise<void> {
|
|
this.job = { ...this.job!, phase };
|
|
await this.writeState('job.json', this.job);
|
|
}
|
|
|
|
private async finish(): Promise<void> {
|
|
this.job = { ...this.job!, status: 'succeeded', phase: 'complete', completedAt: this.now().toISOString() };
|
|
await this.writeState('job.json', this.job);
|
|
}
|
|
|
|
private async corpusExists(version: string): Promise<boolean> {
|
|
if (!/^[a-z0-9][a-z0-9._-]{0,100}$/.test(version)) return false;
|
|
try { return (await stat(join(this.options.root, `${version}.json`))).isFile(); }
|
|
catch { return false; }
|
|
}
|
|
|
|
private async readState<T>(name: string): Promise<T | null> {
|
|
try { return JSON.parse(await readFile(join(this.options.root, name), 'utf8')) as T; }
|
|
catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private async writeState(name: string, value: RegulatoryCorpusV2 | RegulatoryCorpusSummary | RegulatoryCorpusJob): Promise<void> {
|
|
const path = join(this.options.root, name);
|
|
const temporary = `${path}.${randomUUID()}.tmp`;
|
|
try {
|
|
await writeFile(temporary, JSON.stringify(value), { mode: 0o600, flag: 'wx' });
|
|
await rename(temporary, path);
|
|
} finally { await rm(temporary, { force: true }); }
|
|
}
|
|
}
|