ThothII/backend/test/workspace-preprocessing-service.test.ts

import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, expect, test, vi } from "vitest";
import { validateWorkspaceDescriptor } from "../src/workspaces/schema.js";
import {
  WorkspacePreprocessingService,
  type ChildProcessRequest,
} from "../src/workspaces/preprocessing-service.js";

const roots: string[] = [];

afterEach(() => {
  roots.splice(0).forEach((root) => rmSync(root, { recursive: true, force: true }));
});

const workspace = validateWorkspaceDescriptor({
  workspace: {
    schema_version: 4,
    id: "catalog-workspace",
    name: "Catalog only",
    language: "en",
  },
});

function runtime() {
  return {
    workspace,
    workspaceId: "catalog-workspace",
    workspaceRevision: "a".repeat(40),
    descriptorBlob: "b".repeat(40),
    catalogBlob: "c".repeat(40),
    configLease: {
      path: `/data/sessions/catalog-workspace/preprocessing/runtime-config/${"a".repeat(40)}.yaml`,
      workspaceId: "catalog-workspace",
      workspaceRevision: "a".repeat(40),
      descriptorBlob: "b".repeat(40),
      catalogBlob: "c".repeat(40),
      configDigest: "sha256:config",
      bindingDigest: "sha256:bindings",
      semanticQdrantUrl: "http://qdrant:6333",
      effectiveConfig: {
        schemaVersion: 3,
        dwh: {
          engine: "postgres",
          database: "warehouse",
          schema: "analytics",
          transport: "postgres_direct",
          host: "db.internal",
          port: 5432,
          user: "reader",
        },
        vector: {
          collections: {
            reference: "catalog-workspace-reference",
            memory: "catalog-workspace-memory",
          },
          dimensions: 1024,
          distance: "cosine",
        },
        embedding: {
          id: "ollama/qwen3-embedding:0.6b",
          model: "qwen3-embedding:0.6b",
          dimensions: 1024,
        },
        roots: { artifacts: "/data/artifacts", indexes: "/data/indexes" },
      },
      effectiveConfigIdentity: "workspace://catalog-workspace@v1:" + "d".repeat(64),
      configFingerprint: "sha256:" + "e".repeat(64),
      inputFingerprint: "sha256:" + "f".repeat(64),
      release: () => undefined,
    },
  };
}

const database = {
  id: "database-1",
  workspaceId: "catalog-workspace",
  engine: "postgres",
  databaseName: "warehouse",
  schema: "analytics",
  version: 4,
  createdAt: "2026-01-01T00:00:00Z",
  updatedAt: "2026-01-01T00:00:00Z",
  binding: { transport: "postgres_direct", host: "db", port: 5432, username: "reader" },
  connectionStatus: "reachable",
  schemaSyncedVersion: 4,
  metadataContentRevision: 9,
  preprocessingStatus: "running",
} as const;

function repository(overrides: Record<string, unknown> = {}) {
  const finishPreprocessing = vi.fn(async () => ({
    ...database,
    preprocessingStatus: "succeeded" as const,
    preprocessedMetadataRevision: 9,
  }));
  return {
    beginPreprocessing: vi.fn(async () => ({ kind: "started" as const, database })),
    finishPreprocessing,
    clearPreprocessing: vi.fn(async () => ({ kind: "cleared" as const, database })),
    getByWorkspace: vi.fn(async () => database),
    listTables: vi.fn(async () => [{
      id: "table-1", databaseId: database.id, name: "orders",
      description: "Curated orders", generatedDescription: "Generated orders",
      sourceComment: "PostgreSQL orders", version: 1,
      createdAt: database.createdAt, updatedAt: database.updatedAt,
      lastSyncedDatabaseVersion: 4, lastSyncedAt: database.updatedAt,
    }]),
    listColumns: vi.fn(async () => [{
      id: "column-1", tableId: "table-1", name: "customer_id", ordinalPosition: 1,
      dataType: "uuid", isNullable: false, defaultExpression: null,
      primaryKeyPosition: null, isPrimaryKey: false, isForeignKey: true, foreignKeyCount: 1,
      sourceComment: "PostgreSQL customer", description: null,
      generatedDescription: "Generated customer", sensitive: true,
      sensitivityReason: "identifier", lastSyncedDatabaseVersion: 4,
      lastSyncedAt: database.updatedAt, version: 1,
      createdAt: database.createdAt, updatedAt: database.updatedAt,
    }]),
    listRelationships: vi.fn(async () => []),
    listLogicalRelationships: vi.fn(async () => []),
    ...overrides,
  } as any;
}

function service(options: {
  repository?: any;
  runChild?: (request: ChildProcessRequest) => Promise<{
    exitCode: number; stdout: string; stderr: string;
  }>;
  semanticPreflight?: () => Promise<
    { ok: true } | { ok: false; code: "semantic_index_incompatible" }
  >;
} = {}) {
  const dataRoot = mkdtempSync(join(tmpdir(), "tht-catalog-preprocessing-"));
  roots.push(dataRoot);
  return new WorkspacePreprocessingService({
    dataRoot,
    acquireActiveRuntime: async () => runtime(),
    runChild: options.runChild ?? (async () => ({
      exitCode: 0,
      stdout: JSON.stringify({ counts: { tables: 1, columns: 1, relationships: 0 } }),
      stderr: "",
    })),
    semanticPreflight: options.semanticPreflight ?? (async () => ({ ok: true })),
    evidencePreflight: async () => ({ ok: true }),
    catalogRepository: options.repository,
  });
}

test("complete preprocessing snapshots Catalog metadata and commits its PostgreSQL state", async () => {
  const catalog = repository();
  const runChild = vi.fn(async (request: ChildProcessRequest) => {
    const snapshotPath = request.argv[request.argv.indexOf("--catalog-metadata") + 1]!;
    const snapshot = JSON.parse(readFileSync(snapshotPath, "utf8"));
    expect(snapshot).toMatchObject({
      workspaceId: "catalog-workspace",
      databaseName: "warehouse",
      metadataContentRevision: 9,
      tables: [{
        name: "orders",
        description: "Curated orders",
        descriptionSource: "curated",
        columns: [{
          name: "customer_id",
          description: "Generated customer",
          descriptionSource: "generated",
          sensitive: true,
        }],
      }],
    });
    return {
      exitCode: 0,
      stdout: JSON.stringify({ counts: { tables: 1, columns: 1, relationships: 0 } }),
      stderr: "",
    };
  });

  const result = await service({ repository: catalog, runChild }).run({
    workspaceId: "catalog-workspace",
  });

  expect(result).toMatchObject({
    status: "succeeded",
    code: "ok",
    completedStages: ["catalog_snapshot", "catalog_metadata", "lsh", "schema_index"],
  });
  expect(runChild).toHaveBeenCalledWith(expect.objectContaining({
    argv: [
      "preprocess", "catalog", "--catalog-metadata",
      expect.stringMatching(/\/catalog-metadata\.json$/),
      "--json", "-c", "/dev/fd/3",
    ],
  }));
  expect(catalog.finishPreprocessing).toHaveBeenCalledWith(
    "catalog-workspace",
    9,
    "sha256:" + "f".repeat(64),
    { status: "succeeded" },
  );
});

test("preprocessing fails closed when the PostgreSQL Catalog is unavailable", async () => {
  const result = await service().run({ workspaceId: "catalog-workspace" });
  expect(result).toMatchObject({ status: "failed", code: "catalog_not_ready" });
});

test("preprocessing refuses a concurrent Catalog run without touching derived data", async () => {
  const catalog = repository({
    beginPreprocessing: vi.fn(async () => ({ kind: "already_running" as const })),
  });
  const runChild = vi.fn();

  const result = await service({ repository: catalog, runChild }).run({
    workspaceId: "catalog-workspace",
  });

  expect(result).toMatchObject({ status: "failed", code: "preprocessing_conflict" });
  expect(runChild).not.toHaveBeenCalled();
  expect(catalog.finishPreprocessing).not.toHaveBeenCalled();
});

test("preprocessing records a failed PostgreSQL state when its hidden worker fails", async () => {
  const catalog = repository();
  const instance = service({
    repository: catalog,
    runChild: async () => ({ exitCode: 1, stdout: "", stderr: "private failure" }),
  });

  await expect(instance.run({ workspaceId: "catalog-workspace" })).rejects.toThrow(
    "workspace child failed",
  );
  expect(catalog.finishPreprocessing).toHaveBeenCalledWith(
    "catalog-workspace",
    9,
    "sha256:" + "f".repeat(64),
    { status: "failed", errorCode: "schema_index_failed" },
  );
});

test("semantic preflight failure is persisted before returning", async () => {
  const catalog = repository();
  const result = await service({
    repository: catalog,
    semanticPreflight: async () => ({ ok: false, code: "semantic_index_incompatible" }),
  }).run({ workspaceId: "catalog-workspace" });

  expect(result).toMatchObject({ status: "failed", code: "semantic_index_incompatible" });
  expect(catalog.finishPreprocessing).toHaveBeenCalledWith(
    "catalog-workspace",
    9,
    "sha256:" + "f".repeat(64),
    { status: "failed", errorCode: "semantic_index_incompatible" },
  );
});

test("clear invalidates Catalog readiness before clearing only derived worker data", async () => {
  const catalog = repository();
  const runChild = vi.fn(async () => ({
    exitCode: 0,
    stdout: JSON.stringify({ counts: { referenceCollections: 1, derivedPaths: 3 } }),
    stderr: "",
  }));

  const result = await service({ repository: catalog, runChild }).clear({
    workspaceId: "catalog-workspace",
  });

  expect(catalog.clearPreprocessing).toHaveBeenCalledWith("catalog-workspace");
  expect(runChild).toHaveBeenCalledWith({
    argv: ["preprocess", "clear", "--json", "-c", "/dev/fd/3"],
    configPath: expect.any(String),
  });
  expect(result).toMatchObject({
    status: "succeeded",
    code: "ok",
    operation: "preprocess clear",
    counts: { referenceCollections: 1, derivedPaths: 3 },
  });
});