Assessor_2.0/server/regmap/ingest-catalog.mjs

// One-time regulation-catalog ingestion into Qdrant.
//
// Reads a LOCAL, git-ignored catalog JSONL (one embedding chunk per line),
// embeds each chunk's `embed_text` via an OpenAI-compatible endpoint, and
// upserts the vectors + metadata into a Qdrant collection. Idempotent: re-runs
// upsert the same deterministic point ids.
//
// The catalog file contains licensed authoritative text and must stay OUTSIDE
// the repository. This script is committed; the catalog is not.
//
// Usage:
//   node server/regmap/ingest-catalog.mjs [catalog.jsonl]
//
// Required env: ASSESSOR_QDRANT_URL, ASSESSOR_EMBEDDING_BASE_URL,
//   ASSESSOR_EMBEDDING_MODEL, ASSESSOR_EMBEDDING_DIMENSIONS
// Optional env: ASSESSOR_QDRANT_API_KEY, ASSESSOR_QDRANT_COLLECTION,
//   ASSESSOR_EMBEDDING_API_KEY, ASSESSOR_EMBEDDING_PASSAGE_PREFIX,
//   ASSESSOR_REGMAP_CATALOG_VERSION, ASSESSOR_REGMAP_CATALOG_FILE,
//   ASSESSOR_REGMAP_INGEST_BATCH (default 32)
//
// Prints counts only — never catalog text, vectors or payloads.

import { readFile } from 'node:fs/promises';
import { createHash } from 'node:crypto';

const env = process.env;

function required(name) {
  const value = env[name]?.trim();
  if (!value) throw new Error(`${name} ist erforderlich.`);
  return value;
}
function stripTrailingSlash(url) {
  return url.replace(/\/+$/, '');
}

const catalogFile = (process.argv[2] ?? env.ASSESSOR_REGMAP_CATALOG_FILE ?? '').trim();
if (!catalogFile) throw new Error('Katalogdatei fehlt: als Argument oder über ASSESSOR_REGMAP_CATALOG_FILE angeben.');

const qdrantUrl = stripTrailingSlash(required('ASSESSOR_QDRANT_URL'));
const qdrantApiKey = env.ASSESSOR_QDRANT_API_KEY?.trim() || undefined;
const collection = env.ASSESSOR_QDRANT_COLLECTION?.trim() || 'regulation_catalog_v1';
const embeddingBaseUrl = stripTrailingSlash(required('ASSESSOR_EMBEDDING_BASE_URL'));
const embeddingModel = required('ASSESSOR_EMBEDDING_MODEL');
const embeddingApiKey = env.ASSESSOR_EMBEDDING_API_KEY?.trim() || undefined;
const dimensions = Number(required('ASSESSOR_EMBEDDING_DIMENSIONS'));
if (!Number.isInteger(dimensions) || dimensions < 8) throw new Error('ASSESSOR_EMBEDDING_DIMENSIONS muss eine positive Ganzzahl sein.');
const passagePrefix = env.ASSESSOR_EMBEDDING_PASSAGE_PREFIX ?? '';
const expectedCatalogVersion = env.ASSESSOR_REGMAP_CATALOG_VERSION?.trim() || undefined;
const batchSize = Math.max(1, Math.min(256, Number(env.ASSESSOR_REGMAP_INGEST_BATCH ?? 32) || 32));

// Deterministic UUID (v5-like) from a stable chunk id, so re-runs upsert in place.
function pointId(chunkId) {
  const hex = createHash('sha256').update(chunkId).digest('hex').slice(0, 32).split('');
  hex[12] = '5';
  hex[16] = ((parseInt(hex[16], 16) & 0x3) | 0x8).toString(16);
  const h = hex.join('');
  return `${h.slice(0, 8)}-${h.slice(8, 12)}-${h.slice(12, 16)}-${h.slice(16, 20)}-${h.slice(20, 32)}`;
}

const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));

// Retry only transient connection-level failures (a thrown `fetch failed`),
// not HTTP status codes. Local Qdrant/embedding endpoints on WSL occasionally
// reset a connection mid-run; a fresh connection then succeeds.
async function fetchRetry(url, options, attempts = 12) {
  for (let attempt = 1; ; attempt += 1) {
    try {
      return await fetch(url, options);
    } catch (error) {
      if (attempt >= attempts) throw error;
      if (attempt === 1) process.stdout.write(`\n  (endpoint hiccup, retrying up to ${attempts - 1}x)`);
      await sleep(Math.min(3000, 750 * attempt));
    }
  }
}

async function embed(texts) {
  const response = await fetch(`${embeddingBaseUrl}/embeddings`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', ...(embeddingApiKey ? { Authorization: `Bearer ${embeddingApiKey}` } : {}) },
    body: JSON.stringify({ model: embeddingModel, input: texts.map(t => `${passagePrefix}${t}`) })
  });
  if (!response.ok) throw new Error(`Embedding-Dienst antwortete mit HTTP ${response.status}.`);
  const payload = await response.json();
  const rows = Array.isArray(payload.data) ? payload.data : [];
  if (rows.length !== texts.length) throw new Error('Embedding-Dienst lieferte eine unerwartete Anzahl an Vektoren.');
  return rows.map(row => {
    const vector = Array.isArray(row.embedding) ? row.embedding : [];
    if (vector.length !== dimensions) throw new Error(`Embedding lieferte ${vector.length} Dimensionen, erwartet ${dimensions}.`);
    return vector;
  });
}

async function qdrant(path, method, body) {
  return fetchRetry(`${qdrantUrl}${path}`, {
    method,
    headers: { 'Content-Type': 'application/json', ...(qdrantApiKey ? { 'api-key': qdrantApiKey } : {}) },
    body: body === undefined ? undefined : JSON.stringify(body)
  });
}

async function ensureCollection() {
  const info = await qdrant(`/collections/${encodeURIComponent(collection)}`, 'GET');
  if (info.ok) {
    const data = await info.json().catch(() => null);
    const size = data?.result?.config?.params?.vectors?.size;
    if (size !== undefined && size !== dimensions) {
      throw new Error(`Bestehende Collection hat Vektorgröße ${size}, erwartet ${dimensions}.`);
    }
    return;
  }
  if (info.status !== 404) throw new Error(`Qdrant antwortete mit HTTP ${info.status}.`);
  const created = await qdrant(`/collections/${encodeURIComponent(collection)}`, 'PUT', { vectors: { size: dimensions, distance: 'Cosine' } });
  if (!created.ok) throw new Error(`Collection konnte nicht angelegt werden (HTTP ${created.status}).`);
}

const PAYLOAD_FIELDS = [
  'unit_id', 'unit_version', 'regulation_id', 'regulation_title',
  'control_id', 'control_title', 'language', 'catalog_version',
  'chunk_id', 'chunk_text', 'authoritative_text', 'source_uri'
];

function toPayload(row) {
  const payload = {};
  for (const field of PAYLOAD_FIELDS) {
    if (row[field] !== undefined) payload[field] = row[field];
  }
  return payload;
}

async function main() {
  const raw = await readFile(catalogFile, 'utf8');
  const lines = raw.split(/\r?\n/).filter(line => line.trim());
  const rows = [];
  const units = new Set();
  lines.forEach((line, index) => {
    let row;
    try {
      row = JSON.parse(line);
    } catch {
      throw new Error(`Zeile ${index + 1} ist kein gültiges JSON.`);
    }
    if (!row.unit_id || !row.chunk_id || !row.embed_text?.trim()) {
      throw new Error(`Zeile ${index + 1} fehlen Pflichtfelder (unit_id, chunk_id, embed_text).`);
    }
    if (expectedCatalogVersion && row.catalog_version !== expectedCatalogVersion) {
      throw new Error(`Zeile ${index + 1} hat catalog_version, die nicht ${expectedCatalogVersion} entspricht.`);
    }
    units.add(row.unit_id);
    rows.push(row);
  });
  if (rows.length === 0) throw new Error('Die Katalogdatei enthält keine Zeilen.');

  await ensureCollection();

  let upserted = 0;
  for (let start = 0; start < rows.length; start += batchSize) {
    const batch = rows.slice(start, start + batchSize);
    const vectors = await embed(batch.map(row => row.embed_text));
    const points = batch.map((row, i) => ({ id: pointId(row.chunk_id), vector: vectors[i], payload: toPayload(row) }));
    const response = await qdrant(`/collections/${encodeURIComponent(collection)}/points?wait=true`, 'PUT', { points });
    if (!response.ok) throw new Error(`Upsert schlug fehl (HTTP ${response.status}).`);
    upserted += points.length;
    process.stdout.write(`  upserted ${upserted}/${rows.length}\r`);
  }

  console.log(`\nIngestion abgeschlossen: ${upserted} Chunk(s) aus ${units.size} Unit(s) in Collection "${collection}" (Dim ${dimensions}).`);
}

main().catch(error => {
  console.error('Ingestion fehlgeschlagen:', error instanceof Error ? error.message : error);
  process.exitCode = 1;
});