feat: complete catalog fleet management workflow

This commit is contained in:
Codex
2026-08-31 15:58:43 +02:00
parent 866aee4249
commit 9c697dc062
56 changed files with 6692 additions and 611 deletions
+7 -1
View File
@@ -57,6 +57,7 @@ import { catalogDescriptionConsolidationRoutes } from "./routes/catalog-descript
import { PythonModelCompleter, type ModelCompleter } from "./catalog/model-completer.js";
import { DescriptionGenerationWorker } from "./catalog/description-generation-worker.js";
import { SensitiveDataSuggester } from "./catalog/sensitive-data-suggester.js";
import { SensitiveDataSuggestionRunner } from "./catalog/sensitive-data-suggestion-runner.js";
import {
PostgresDescriptionSourceSampler,
type DescriptionSourceSampler,
@@ -185,6 +186,10 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
metadataGenerationModels,
modelCompleter,
);
const sensitiveDataSuggestionRunner = new SensitiveDataSuggestionRunner(
catalogRepository,
sensitiveDataSuggester,
);
const catalogService = deps?.catalogService ?? new CatalogService(
catalogRepository,
workspaceRegistry,
@@ -207,6 +212,7 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
);
app.addHook("onReady", async () => { await catalogSyncWorker.initialize(); });
app.addHook("onReady", async () => { await descriptionGenerationWorker.initialize(); });
app.addHook("onReady", async () => { await sensitiveDataSuggestionRunner.initialize(); });
if (!deps?.catalogRepository && catalogRepository.close) {
app.addHook("onClose", async () => { await catalogRepository.close?.(); });
}
@@ -444,7 +450,7 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
catalogDescriptionGenerationRoutes(app, {
repository: catalogRepository,
worker: descriptionGenerationWorker,
sensitiveDataSuggester,
sensitiveDataSuggestionRunner,
});
settingsRoutes(app, { cfg: config, listModels, getSettings });
piManagementRoutes(app, { service: piManagement });
+149
View File
@@ -1,4 +1,9 @@
import { randomUUID } from "node:crypto";
import {
createCatalogMetrics,
hasCatalogDescription,
latestCatalogTimestamp,
} from "./metrics.js";
import {
CatalogConflictError,
CatalogConnectorError,
@@ -8,6 +13,7 @@ import {
type CatalogDescriptionTarget,
type CatalogDatabaseMetadataDeleteTarget,
type CatalogMetadataDeleteCounts,
type CatalogMetrics,
type CatalogRelationship,
type CatalogSchemaDiff,
type CatalogSyncCounts,
@@ -26,6 +32,10 @@ import {
type DescriptionGenerationRun,
type DescriptionGenerationRunUpdate,
type DescriptionGenerationScope,
type SensitiveDataSuggestionEvent,
type SensitiveDataSuggestionRun,
type SensitiveDataSuggestionRunUpdate,
type SensitiveDataSuggestionScope,
type TableSyncRepositoryResult,
type WorkspaceDatabase,
} from "./types.js";
@@ -42,6 +52,8 @@ export class MemoryCatalogRepository implements CatalogRepository {
private readonly relationships = new Map<string, CatalogRelationship>();
private readonly descriptionGenerationRuns = new Map<string, DescriptionGenerationRun>();
private readonly descriptionGenerationEvents = new Map<string, DescriptionGenerationEvent[]>();
private readonly sensitiveDataSuggestionRuns = new Map<string, SensitiveDataSuggestionRun>();
private readonly sensitiveDataSuggestionEvents = new Map<string, SensitiveDataSuggestionEvent[]>();
private readonly syncRuns = new Map<string, CatalogSyncRun>();
private readonly syncEvents = new Map<string, CatalogSyncEvent[]>();
@@ -56,6 +68,34 @@ export class MemoryCatalogRepository implements CatalogRepository {
const value = [...this.records.values()].find((record) => record.workspaceId === workspaceId);
return value ? clone(value) : undefined;
}
async getCatalogMetrics(databaseId?: string): Promise<CatalogMetrics | undefined> {
if (databaseId !== undefined && !this.records.has(databaseId)) return undefined;
const selectedDatabaseIds = new Set(
databaseId === undefined ? this.records.keys() : [databaseId],
);
const tables = [...this.tables.values()]
.filter((table) => selectedDatabaseIds.has(table.databaseId));
const tableIds = new Set(tables.map((table) => table.id));
const columns = [...this.columns.values()]
.filter((column) => tableIds.has(column.tableId));
const relationships = [...this.relationships.values()]
.filter((relationship) => selectedDatabaseIds.has(relationship.databaseId));
return createCatalogMetrics(databaseId, {
tables: tables.length,
columns: columns.length,
sensitiveColumns: columns.filter((column) => column.sensitive).length,
relationships: relationships.length,
describedTables: tables.filter(hasCatalogDescription).length,
describedColumns: columns.filter(hasCatalogDescription).length,
}, latestCatalogTimestamp([
...[...selectedDatabaseIds].map((id) => this.records.get(id)?.schemaSyncedAt),
...tables.map((table) => table.updatedAt),
...columns.map((column) => column.updatedAt),
...relationships.map((relationship) => relationship.updatedAt),
]));
}
async create(input: DatabaseConfigurationInput): Promise<WorkspaceDatabase> {
if ([...this.records.values()].some((record) => record.workspaceId === input.workspaceId)) {
throw new CatalogConflictError("Workspace database already exists");
@@ -136,6 +176,11 @@ export class MemoryCatalogRepository implements CatalogRepository {
this.descriptionGenerationRuns.delete(runId);
this.descriptionGenerationEvents.delete(runId);
}
for (const [runId, run] of this.sensitiveDataSuggestionRuns) {
if (run.databaseId !== id) continue;
this.sensitiveDataSuggestionRuns.delete(runId);
this.sensitiveDataSuggestionEvents.delete(runId);
}
return this.records.delete(id);
}
async listTables(databaseId: string): Promise<CatalogTable[]> {
@@ -378,6 +423,110 @@ export class MemoryCatalogRepository implements CatalogRepository {
.map((event) => structuredClone(event));
}
async createSensitiveDataSuggestionRun(
databaseId: string,
scope: SensitiveDataSuggestionScope,
modelId: string,
): Promise<SensitiveDataSuggestionRun> {
const now = new Date().toISOString();
const run: SensitiveDataSuggestionRun = {
id: randomUUID(),
databaseId,
scope,
modelId,
status: "running",
total: 0,
suggestedSensitive: 0,
suggestedNonSensitive: 0,
createdAt: now,
startedAt: now,
updatedAt: now,
finishedAt: null,
errorSummary: null,
};
this.sensitiveDataSuggestionRuns.set(run.id, run);
return structuredClone(run);
}
async getSensitiveDataSuggestionRun(
runId: string,
): Promise<SensitiveDataSuggestionRun | undefined> {
const run = this.sensitiveDataSuggestionRuns.get(runId);
return run ? structuredClone(run) : undefined;
}
async listSensitiveDataSuggestionRuns(limit = 50): Promise<SensitiveDataSuggestionRun[]> {
return [...this.sensitiveDataSuggestionRuns.values()]
.sort((a, b) => b.createdAt.localeCompare(a.createdAt) || b.id.localeCompare(a.id))
.slice(0, limit)
.map((run) => structuredClone(run));
}
async interruptActiveSensitiveDataSuggestionRuns(
errorSummary: string,
): Promise<SensitiveDataSuggestionRun[]> {
const interrupted: SensitiveDataSuggestionRun[] = [];
for (const run of this.sensitiveDataSuggestionRuns.values()) {
if (run.status !== "running") continue;
const now = new Date().toISOString();
const updated: SensitiveDataSuggestionRun = {
...run,
status: "interrupted",
updatedAt: now,
finishedAt: now,
errorSummary,
};
this.sensitiveDataSuggestionRuns.set(run.id, updated);
interrupted.push(structuredClone(updated));
}
return interrupted;
}
async updateSensitiveDataSuggestionRun(
runId: string,
update: SensitiveDataSuggestionRunUpdate,
): Promise<SensitiveDataSuggestionRun | undefined> {
const current = this.sensitiveDataSuggestionRuns.get(runId);
if (!current) return undefined;
const updated = {
...current,
...structuredClone(update),
updatedAt: new Date().toISOString(),
};
this.sensitiveDataSuggestionRuns.set(runId, updated);
return structuredClone(updated);
}
async appendSensitiveDataSuggestionEvent(
runId: string,
level: SensitiveDataSuggestionEvent["level"],
message: string,
): Promise<SensitiveDataSuggestionEvent> {
if (!this.sensitiveDataSuggestionRuns.has(runId)) {
throw new CatalogConflictError("Sensitive Data Suggestion Run does not exist");
}
const events = this.sensitiveDataSuggestionEvents.get(runId) ?? [];
const event: SensitiveDataSuggestionEvent = {
runId,
sequence: events.length + 1,
level,
message,
createdAt: new Date().toISOString(),
};
events.push(event);
this.sensitiveDataSuggestionEvents.set(runId, events);
return structuredClone(event);
}
async listSensitiveDataSuggestionEvents(
runId: string,
afterSequence = 0,
): Promise<SensitiveDataSuggestionEvent[]> {
return (this.sensitiveDataSuggestionEvents.get(runId) ?? [])
.filter((event) => event.sequence > afterSequence)
.map((event) => structuredClone(event));
}
async listRelationships(databaseId: string): Promise<CatalogRelationship[]> {
return [...this.relationships.values()].filter((relationship) => relationship.databaseId === databaseId)
.sort((a, b) => `${a.sourceTableName}.${a.constraintName}`.localeCompare(`${b.sourceTableName}.${b.constraintName}`))
+53
View File
@@ -0,0 +1,53 @@
import type { CatalogMetrics } from "./types.js";
export interface CatalogMetricCounts {
tables: number;
columns: number;
sensitiveColumns: number;
relationships: number;
describedTables: number;
describedColumns: number;
}
export function hasCatalogDescription(target: {
description: string | null;
generatedDescription: string | null;
}): boolean {
return Boolean(target.description?.trim() || target.generatedDescription?.trim());
}
export function latestCatalogTimestamp(
values: readonly (string | null | undefined)[],
): string | null {
let latest: number | undefined;
for (const value of values) {
if (!value) continue;
const timestamp = Date.parse(value);
if (!Number.isFinite(timestamp)) continue;
latest = latest === undefined ? timestamp : Math.max(latest, timestamp);
}
return latest === undefined ? null : new Date(latest).toISOString();
}
export function createCatalogMetrics(
databaseId: string | undefined,
counts: CatalogMetricCounts,
updatedAt: string | null,
): CatalogMetrics {
const descriptionTargets = counts.tables + counts.columns;
const describedTargets = counts.describedTables + counts.describedColumns;
return {
scope: databaseId === undefined ? "global" : "database",
databaseId: databaseId ?? null,
tables: counts.tables,
columns: counts.columns,
sensitiveColumns: counts.sensitiveColumns,
relationships: counts.relationships,
descriptionTargets,
describedTargets,
descriptionCoverage: descriptionTargets === 0
? 0
: Math.round((describedTargets / descriptionTargets) * 100),
updatedAt,
};
}
+2
View File
@@ -9,6 +9,7 @@ import * as catalogSchemaSyncMigration from "./migrations/003_catalog_schema_syn
import * as catalogRuntimeSequencePrivilegesMigration from "./migrations/004_catalog_runtime_sequence_privileges.js";
import * as descriptionGenerationRunsMigration from "./migrations/005_description_generation_runs.js";
import * as sensitiveDataFlagMigration from "./migrations/006_sensitive_data_flag.js";
import * as sensitiveDataSuggestionRunsMigration from "./migrations/007_sensitive_data_suggestion_runs.js";
const connectionString = process.env.THT_CATALOG_MIGRATOR_DATABASE_URL;
const host = process.env.THT_CATALOG_DB_HOST;
@@ -40,6 +41,7 @@ const provider: MigrationProvider = {
"004_catalog_runtime_sequence_privileges": catalogRuntimeSequencePrivilegesMigration,
"005_description_generation_runs": descriptionGenerationRunsMigration,
"006_sensitive_data_flag": sensitiveDataFlagMigration,
"007_sensitive_data_suggestion_runs": sensitiveDataSuggestionRunsMigration,
};
},
};
@@ -0,0 +1,73 @@
import { type Kysely, sql } from "kysely";
import type { CatalogDatabase } from "../repository.js";
export async function up(db: Kysely<CatalogDatabase>): Promise<void> {
await db.schema.createTable("sensitive_data_suggestion_runs")
.addColumn("id", "uuid", (column) => column.primaryKey())
.addColumn("database_id", "uuid", (column) => column.notNull()
.references("workspace_databases.id").onDelete("cascade"))
.addColumn("scope", "text", (column) => column.notNull())
.addColumn("model_id", "text", (column) => column.notNull())
.addColumn("status", "text", (column) => column.notNull())
.addColumn("total", "integer", (column) => column.notNull().defaultTo(0))
.addColumn("suggested_sensitive", "integer", (column) => column.notNull().defaultTo(0))
.addColumn("suggested_non_sensitive", "integer", (column) => column.notNull().defaultTo(0))
.addColumn("created_at", "timestamptz", (column) => column.notNull().defaultTo(sql`now()`))
.addColumn("started_at", "timestamptz", (column) => column.notNull().defaultTo(sql`now()`))
.addColumn("updated_at", "timestamptz", (column) => column.notNull().defaultTo(sql`now()`))
.addColumn("finished_at", "timestamptz")
.addColumn("error_summary", "text")
.addCheckConstraint(
"sensitive_data_suggestion_runs_scope_check",
sql`scope in ('selected_columns', 'selected_tables', 'all')`,
)
.addCheckConstraint(
"sensitive_data_suggestion_runs_model_id_check",
sql`model_id ~ '^[a-z][a-z0-9._-]{0,63}$'`,
)
.addCheckConstraint(
"sensitive_data_suggestion_runs_status_check",
sql`status in ('running', 'completed', 'failed', 'interrupted')`,
)
.addCheckConstraint(
"sensitive_data_suggestion_runs_counters_check",
sql`total >= 0
and suggested_sensitive >= 0
and suggested_non_sensitive >= 0
and suggested_sensitive + suggested_non_sensitive <= total`,
)
.addCheckConstraint(
"sensitive_data_suggestion_runs_error_summary_check",
sql`error_summary is null or char_length(error_summary) between 1 and 2000`,
)
.execute();
await db.schema.createIndex("sensitive_data_suggestion_runs_database_created_idx")
.on("sensitive_data_suggestion_runs")
.columns(["database_id", "created_at"])
.execute();
await db.schema.createTable("sensitive_data_suggestion_events")
.addColumn("run_id", "uuid", (column) => column.notNull()
.references("sensitive_data_suggestion_runs.id").onDelete("cascade"))
.addColumn("sequence", "integer", (column) => column.notNull())
.addColumn("level", "text", (column) => column.notNull())
.addColumn("message", "text", (column) => column.notNull())
.addColumn("created_at", "timestamptz", (column) => column.notNull().defaultTo(sql`now()`))
.addPrimaryKeyConstraint("sensitive_data_suggestion_events_pkey", ["run_id", "sequence"])
.addCheckConstraint("sensitive_data_suggestion_events_sequence_check", sql`sequence > 0`)
.addCheckConstraint(
"sensitive_data_suggestion_events_level_check",
sql`level in ('info', 'warning', 'error')`,
)
.addCheckConstraint(
"sensitive_data_suggestion_events_message_check",
sql`char_length(message) between 1 and 2000`,
)
.execute();
}
export async function down(db: Kysely<CatalogDatabase>): Promise<void> {
await db.schema.dropTable("sensitive_data_suggestion_events").execute();
await db.schema.dropTable("sensitive_data_suggestion_runs").execute();
}
+255
View File
@@ -11,6 +11,7 @@ import {
type Transaction,
} from "kysely";
import { Pool } from "pg";
import { createCatalogMetrics } from "./metrics.js";
import {
CatalogConflictError,
CatalogConnectorError,
@@ -21,6 +22,7 @@ import {
type CatalogDescriptionTarget,
type CatalogDatabaseMetadataDeleteTarget,
type CatalogMetadataDeleteCounts,
type CatalogMetrics,
type CatalogRelationship,
type CatalogSchemaDiff,
type CatalogSyncCounts,
@@ -40,6 +42,10 @@ import {
type DescriptionGenerationScope,
type ObservedCatalogTable,
type ObservedSchemaSnapshot,
type SensitiveDataSuggestionEvent,
type SensitiveDataSuggestionRun,
type SensitiveDataSuggestionRunUpdate,
type SensitiveDataSuggestionScope,
type TableSyncRepositoryResult,
type WorkspaceDatabase,
} from "./types.js";
@@ -166,6 +172,30 @@ interface DescriptionGenerationEventTable {
createdAt: Timestamp;
}
interface SensitiveDataSuggestionRunTable {
id: string;
databaseId: string;
scope: SensitiveDataSuggestionScope;
modelId: string;
status: SensitiveDataSuggestionRun["status"];
total: number;
suggestedSensitive: number;
suggestedNonSensitive: number;
createdAt: Timestamp;
startedAt: Timestamp;
updatedAt: Timestamp;
finishedAt: Timestamp | null;
errorSummary: string | null;
}
interface SensitiveDataSuggestionEventTable {
runId: string;
sequence: number;
level: SensitiveDataSuggestionEvent["level"];
message: string;
createdAt: Timestamp;
}
interface CatalogSyncRunTable {
id: string;
databaseId: string;
@@ -213,6 +243,8 @@ export interface CatalogDatabase {
catalogRelationshipColumns: CatalogRelationshipColumnTable;
descriptionGenerationRuns: DescriptionGenerationRunTable;
descriptionGenerationEvents: DescriptionGenerationEventTable;
sensitiveDataSuggestionRuns: SensitiveDataSuggestionRunTable;
sensitiveDataSuggestionEvents: SensitiveDataSuggestionEventTable;
catalogSyncRuns: CatalogSyncRunTable;
catalogSyncEvents: CatalogSyncEventTable;
}
@@ -220,6 +252,17 @@ export interface CatalogDatabase {
type DbOrTransaction = Kysely<CatalogDatabase> | Transaction<CatalogDatabase>;
type JoinedRow = Selectable<WorkspaceDatabaseTable> & Selectable<DatabaseBindingTable>;
interface CatalogMetricsRow {
databaseCount: number;
tables: number;
columns: number;
sensitiveColumns: number;
relationships: number;
describedTables: number;
describedColumns: number;
updatedAt: Date | string | null;
}
function present<T>(value: T | null): T | undefined {
return value === null ? undefined : value;
}
@@ -338,6 +381,25 @@ function serializeDescriptionGenerationEvent(
return { ...row, createdAt: new Date(row.createdAt).toISOString() };
}
function serializeSensitiveDataSuggestionRun(
row: Selectable<SensitiveDataSuggestionRunTable>,
): SensitiveDataSuggestionRun {
const stamp = (value: Date | string | null) => value === null ? null : new Date(value).toISOString();
return {
...row,
createdAt: new Date(row.createdAt).toISOString(),
startedAt: new Date(row.startedAt).toISOString(),
updatedAt: new Date(row.updatedAt).toISOString(),
finishedAt: stamp(row.finishedAt),
};
}
function serializeSensitiveDataSuggestionEvent(
row: Selectable<SensitiveDataSuggestionEventTable>,
): SensitiveDataSuggestionEvent {
return { ...row, createdAt: new Date(row.createdAt).toISOString() };
}
function bindingValues(databaseId: string, binding: DatabaseBinding) {
return {
databaseId,
@@ -388,6 +450,83 @@ export class KyselyCatalogRepository implements CatalogRepository {
return id ? await this.get(id.id) : undefined;
}
async getCatalogMetrics(databaseId?: string): Promise<CatalogMetrics | undefined> {
const result = await sql<CatalogMetricsRow>`
WITH requested_database AS (
SELECT ${databaseId ?? null}::uuid AS id
),
selected_databases AS (
SELECT workspace_databases.id, workspace_databases.schema_synced_at
FROM workspace_databases
CROSS JOIN requested_database
WHERE requested_database.id IS NULL
OR workspace_databases.id = requested_database.id
),
table_metrics AS (
SELECT
count(*)::int AS tables,
count(*) FILTER (
WHERE nullif(btrim(catalog_tables.description), '') IS NOT NULL
OR nullif(btrim(catalog_tables.generated_description), '') IS NOT NULL
)::int AS "describedTables",
max(catalog_tables.updated_at) AS updated_at
FROM catalog_tables
INNER JOIN selected_databases
ON selected_databases.id = catalog_tables.database_id
),
column_metrics AS (
SELECT
count(*)::int AS columns,
count(*) FILTER (WHERE catalog_columns.sensitive)::int AS "sensitiveColumns",
count(*) FILTER (
WHERE nullif(btrim(catalog_columns.description), '') IS NOT NULL
OR nullif(btrim(catalog_columns.generated_description), '') IS NOT NULL
)::int AS "describedColumns",
max(catalog_columns.updated_at) AS updated_at
FROM catalog_columns
INNER JOIN catalog_tables ON catalog_tables.id = catalog_columns.table_id
INNER JOIN selected_databases
ON selected_databases.id = catalog_tables.database_id
),
relationship_metrics AS (
SELECT
count(*)::int AS relationships,
max(catalog_relationships.updated_at) AS updated_at
FROM catalog_relationships
INNER JOIN selected_databases
ON selected_databases.id = catalog_relationships.database_id
)
SELECT
(SELECT count(*)::int FROM selected_databases) AS "databaseCount",
table_metrics.tables,
column_metrics.columns,
column_metrics."sensitiveColumns",
relationship_metrics.relationships,
table_metrics."describedTables",
column_metrics."describedColumns",
greatest(
(SELECT max(schema_synced_at) FROM selected_databases),
table_metrics.updated_at,
column_metrics.updated_at,
relationship_metrics.updated_at
) AS "updatedAt"
FROM table_metrics
CROSS JOIN column_metrics
CROSS JOIN relationship_metrics
`.execute(this.db);
const row = result.rows[0];
if (!row || (databaseId !== undefined && Number(row.databaseCount) === 0)) return undefined;
return createCatalogMetrics(databaseId, {
tables: Number(row.tables),
columns: Number(row.columns),
sensitiveColumns: Number(row.sensitiveColumns),
relationships: Number(row.relationships),
describedTables: Number(row.describedTables),
describedColumns: Number(row.describedColumns),
}, row.updatedAt === null ? null : new Date(row.updatedAt).toISOString());
}
async create(input: DatabaseConfigurationInput): Promise<WorkspaceDatabase> {
try {
return await this.db.transaction().execute(async (trx) => {
@@ -775,6 +914,114 @@ export class KyselyCatalogRepository implements CatalogRepository {
return rows.map(serializeDescriptionGenerationEvent);
}
async createSensitiveDataSuggestionRun(
databaseId: string,
scope: SensitiveDataSuggestionScope,
modelId: string,
): Promise<SensitiveDataSuggestionRun> {
const row = await this.db.insertInto("sensitiveDataSuggestionRuns").values({
id: randomUUID(),
databaseId,
scope,
modelId,
status: "running",
total: 0,
suggestedSensitive: 0,
suggestedNonSensitive: 0,
finishedAt: null,
errorSummary: null,
}).returningAll().executeTakeFirstOrThrow();
return serializeSensitiveDataSuggestionRun(row);
}
async getSensitiveDataSuggestionRun(
runId: string,
): Promise<SensitiveDataSuggestionRun | undefined> {
const row = await this.db.selectFrom("sensitiveDataSuggestionRuns")
.selectAll()
.where("id", "=", runId)
.executeTakeFirst();
return row ? serializeSensitiveDataSuggestionRun(row) : undefined;
}
async listSensitiveDataSuggestionRuns(limit = 50): Promise<SensitiveDataSuggestionRun[]> {
const rows = await this.db.selectFrom("sensitiveDataSuggestionRuns")
.selectAll()
.orderBy("createdAt", "desc")
.orderBy("id", "desc")
.limit(limit)
.execute();
return rows.map(serializeSensitiveDataSuggestionRun);
}
async interruptActiveSensitiveDataSuggestionRuns(
errorSummary: string,
): Promise<SensitiveDataSuggestionRun[]> {
const rows = await this.db.updateTable("sensitiveDataSuggestionRuns")
.set({
status: "interrupted",
finishedAt: sql`now()`,
updatedAt: sql`now()`,
errorSummary,
})
.where("status", "=", "running")
.returningAll()
.execute();
return rows.map(serializeSensitiveDataSuggestionRun);
}
async updateSensitiveDataSuggestionRun(
runId: string,
update: SensitiveDataSuggestionRunUpdate,
): Promise<SensitiveDataSuggestionRun | undefined> {
const values: any = { ...update, updatedAt: sql`now()` };
const row = await this.db.updateTable("sensitiveDataSuggestionRuns")
.set(values)
.where("id", "=", runId)
.returningAll()
.executeTakeFirst();
return row ? serializeSensitiveDataSuggestionRun(row) : undefined;
}
async appendSensitiveDataSuggestionEvent(
runId: string,
level: SensitiveDataSuggestionEvent["level"],
message: string,
): Promise<SensitiveDataSuggestionEvent> {
return await this.db.transaction().execute(async (trx) => {
const run = await trx.selectFrom("sensitiveDataSuggestionRuns")
.select("id")
.where("id", "=", runId)
.forUpdate()
.executeTakeFirst();
if (!run) throw new CatalogConflictError("Sensitive Data Suggestion Run does not exist");
const current = await trx.selectFrom("sensitiveDataSuggestionEvents")
.select(sql<number>`coalesce(max(sequence), 0)::int`.as("sequence"))
.where("runId", "=", runId)
.executeTakeFirst();
const row = await trx.insertInto("sensitiveDataSuggestionEvents").values({
runId,
sequence: Number(current?.sequence ?? 0) + 1,
level,
message,
}).returningAll().executeTakeFirstOrThrow();
return serializeSensitiveDataSuggestionEvent(row);
});
}
async listSensitiveDataSuggestionEvents(
runId: string,
afterSequence = 0,
): Promise<SensitiveDataSuggestionEvent[]> {
const rows = await this.db.selectFrom("sensitiveDataSuggestionEvents")
.selectAll()
.where("runId", "=", runId)
.where("sequence", ">", afterSequence)
.orderBy("sequence")
.execute();
return rows.map(serializeSensitiveDataSuggestionEvent);
}
async listRelationships(databaseId: string): Promise<CatalogRelationship[]> {
const rows = await this.db.selectFrom("catalogRelationships as relationship")
.innerJoin("catalogTables as sourceTable", "sourceTable.id", "relationship.sourceTableId")
@@ -1345,6 +1592,7 @@ export class UnavailableCatalogRepository implements CatalogRepository {
async list(): Promise<WorkspaceDatabase[]> { return this.fail(); }
async get(): Promise<WorkspaceDatabase | undefined> { return this.fail(); }
async getByWorkspace(): Promise<WorkspaceDatabase | undefined> { return this.fail(); }
async getCatalogMetrics(): Promise<CatalogMetrics | undefined> { return this.fail(); }
async create(): Promise<WorkspaceDatabase> { return this.fail(); }
async update(): Promise<WorkspaceDatabase | undefined> { return this.fail(); }
async recordTest(): Promise<WorkspaceDatabase | undefined> { return this.fail(); }
@@ -1366,6 +1614,13 @@ export class UnavailableCatalogRepository implements CatalogRepository {
async updateDescriptionGenerationRun(): Promise<DescriptionGenerationRun | undefined> { return this.fail(); }
async appendDescriptionGenerationEvent(): Promise<DescriptionGenerationEvent> { return this.fail(); }
async listDescriptionGenerationEvents(): Promise<DescriptionGenerationEvent[]> { return this.fail(); }
async createSensitiveDataSuggestionRun(): Promise<SensitiveDataSuggestionRun> { return this.fail(); }
async getSensitiveDataSuggestionRun(): Promise<SensitiveDataSuggestionRun | undefined> { return this.fail(); }
async listSensitiveDataSuggestionRuns(): Promise<SensitiveDataSuggestionRun[]> { return this.fail(); }
async interruptActiveSensitiveDataSuggestionRuns(): Promise<SensitiveDataSuggestionRun[]> { return this.fail(); }
async updateSensitiveDataSuggestionRun(): Promise<SensitiveDataSuggestionRun | undefined> { return this.fail(); }
async appendSensitiveDataSuggestionEvent(): Promise<SensitiveDataSuggestionEvent> { return this.fail(); }
async listSensitiveDataSuggestionEvents(): Promise<SensitiveDataSuggestionEvent[]> { return this.fail(); }
async listRelationships(): Promise<CatalogRelationship[]> { return this.fail(); }
async deleteDatabaseMetadata(): Promise<CatalogMetadataDeleteCounts | undefined> { return this.fail(); }
async deleteTableMetadata(): Promise<CatalogMetadataDeleteCounts | undefined> { return this.fail(); }
@@ -1,7 +1,14 @@
import { z } from "zod";
import type { MetadataGenerationModels } from "./metadata-generation-models.js";
import type { ModelCompleter, ModelCompletionMessage } from "./model-completer.js";
import type { CatalogColumn, CatalogRepository, CatalogTable } from "./types.js";
import type {
CatalogColumn,
CatalogRepository,
CatalogTable,
SensitiveDataSuggestionScope,
} from "./types.js";
export type { SensitiveDataSuggestionScope } from "./types.js";
// The helper accepts at most 64 KiB per message. Keep the same safety margin used by
// Description Generation so UTF-8 structural metadata never reaches that hard limit.
@@ -16,8 +23,6 @@ const responseSchema = z.object({
}).strict()),
}).strict();
export type SensitiveDataSuggestionScope = "all" | "selected_tables" | "selected_columns";
interface StructuralColumn {
columnId: string;
tableId: string;
@@ -195,10 +200,12 @@ export class SensitiveDataSuggester {
scope: SensitiveDataSuggestionScope,
targetIds: readonly string[],
signal: AbortSignal,
onPrepared?: (total: number) => void | Promise<void>,
): Promise<readonly SensitiveDataSuggestion[]> {
const database = await this.repository.get(databaseId);
if (!database) throw new SensitiveDataSuggestionTargetNotFoundError("database");
const columns = await this.selectColumns(databaseId, scope, targetIds);
await onPrepared?.(columns.length);
const model = this.models.resolve(modelId);
const suggestions: SensitiveDataSuggestion[] = [];
@@ -0,0 +1,110 @@
import type {
SensitiveDataSuggestion,
} from "./sensitive-data-suggester.js";
import {
SensitiveDataSuggester,
SensitiveDataSuggestionTargetNotFoundError,
} from "./sensitive-data-suggester.js";
import type {
CatalogRepository,
SensitiveDataSuggestionRun,
SensitiveDataSuggestionScope,
} from "./types.js";
const interruptedMessage = "Sensitive-field suggestion generation was interrupted by backend restart.";
const failedMessage = "Sensitive-field suggestion generation failed.";
export interface SensitiveDataSuggestionRunResult {
suggestions: readonly SensitiveDataSuggestion[];
run: SensitiveDataSuggestionRun;
}
export class SensitiveDataSuggestionRunner {
constructor(
private readonly repository: CatalogRepository,
private readonly suggester: SensitiveDataSuggester,
) {}
async initialize(): Promise<void> {
if (!(await this.repository.available())) return;
const interrupted = await this.repository.interruptActiveSensitiveDataSuggestionRuns(
interruptedMessage,
);
for (const run of interrupted) {
await this.repository.appendSensitiveDataSuggestionEvent(
run.id,
"warning",
interruptedMessage,
);
}
}
async run(
databaseId: string,
modelId: string,
scope: SensitiveDataSuggestionScope,
targetIds: readonly string[],
signal: AbortSignal,
): Promise<SensitiveDataSuggestionRunResult> {
if (!(await this.repository.get(databaseId))) {
throw new SensitiveDataSuggestionTargetNotFoundError("database");
}
const started = await this.repository.createSensitiveDataSuggestionRun(
databaseId,
scope,
modelId,
);
try {
await this.repository.appendSensitiveDataSuggestionEvent(
started.id,
"info",
"Sensitive-field suggestion generation started.",
);
const suggestions = await this.suggester.suggest(
databaseId,
modelId,
scope,
targetIds,
signal,
async (total) => {
const prepared = await this.repository.updateSensitiveDataSuggestionRun(started.id, {
total,
});
if (!prepared) throw new Error("Sensitive Data Suggestion Run disappeared");
},
);
const suggestedSensitive = suggestions.filter((suggestion) => suggestion.sensitive).length;
const suggestedNonSensitive = suggestions.length - suggestedSensitive;
await this.repository.appendSensitiveDataSuggestionEvent(
started.id,
"info",
`Sensitive-field suggestion generation completed for ${suggestions.length} column${
suggestions.length === 1 ? "" : "s"
}.`,
);
const completed = await this.repository.updateSensitiveDataSuggestionRun(started.id, {
status: "completed",
total: suggestions.length,
suggestedSensitive,
suggestedNonSensitive,
finishedAt: new Date().toISOString(),
errorSummary: null,
});
if (!completed) throw new Error("Sensitive Data Suggestion Run disappeared");
return { suggestions, run: completed };
} catch (error) {
await this.repository.updateSensitiveDataSuggestionRun(started.id, {
status: "failed",
finishedAt: new Date().toISOString(),
errorSummary: failedMessage,
}).catch(() => undefined);
await this.repository.appendSensitiveDataSuggestionEvent(
started.id,
"error",
failedMessage,
).catch(() => undefined);
throw error;
}
}
}
+73
View File
@@ -38,6 +38,19 @@ export interface WorkspaceDatabase {
schemaSyncedAt?: string;
}
export interface CatalogMetrics {
scope: "global" | "database";
databaseId: string | null;
tables: number;
columns: number;
sensitiveColumns: number;
relationships: number;
descriptionTargets: number;
describedTargets: number;
descriptionCoverage: number;
updatedAt: string | null;
}
export interface DatabaseConfigurationInput {
workspaceId: string;
engine: "postgres";
@@ -201,6 +214,42 @@ export interface DescriptionGenerationEvent {
createdAt: string;
}
export type SensitiveDataSuggestionScope = "all" | "selected_tables" | "selected_columns";
export type SensitiveDataSuggestionStatus = "running" | "completed" | "failed" | "interrupted";
export interface SensitiveDataSuggestionRun {
id: string;
databaseId: string;
scope: SensitiveDataSuggestionScope;
modelId: string;
status: SensitiveDataSuggestionStatus;
total: number;
suggestedSensitive: number;
suggestedNonSensitive: number;
createdAt: string;
startedAt: string;
updatedAt: string;
finishedAt: string | null;
errorSummary: string | null;
}
export interface SensitiveDataSuggestionRunUpdate {
status?: SensitiveDataSuggestionStatus;
total?: number;
suggestedSensitive?: number;
suggestedNonSensitive?: number;
finishedAt?: string | null;
errorSummary?: string | null;
}
export interface SensitiveDataSuggestionEvent {
runId: string;
sequence: number;
level: "info" | "warning" | "error";
message: string;
createdAt: string;
}
export interface ObservedRelationshipColumn {
position: number;
sourceColumnName: string;
@@ -321,6 +370,7 @@ export interface CatalogRepository {
list(): Promise<WorkspaceDatabase[]>;
get(id: string): Promise<WorkspaceDatabase | undefined>;
getByWorkspace(workspaceId: string): Promise<WorkspaceDatabase | undefined>;
getCatalogMetrics(databaseId?: string): Promise<CatalogMetrics | undefined>;
create(input: DatabaseConfigurationInput): Promise<WorkspaceDatabase>;
update(id: string, expectedVersion: number, input: DatabaseConfigurationInput): Promise<WorkspaceDatabase | undefined>;
recordTest(id: string, expectedVersion: number, result: DatabaseTestResult): Promise<WorkspaceDatabase | undefined>;
@@ -383,6 +433,29 @@ export interface CatalogRepository {
runId: string,
afterSequence?: number,
): Promise<DescriptionGenerationEvent[]>;
createSensitiveDataSuggestionRun(
databaseId: string,
scope: SensitiveDataSuggestionScope,
modelId: string,
): Promise<SensitiveDataSuggestionRun>;
getSensitiveDataSuggestionRun(runId: string): Promise<SensitiveDataSuggestionRun | undefined>;
listSensitiveDataSuggestionRuns(limit?: number): Promise<SensitiveDataSuggestionRun[]>;
interruptActiveSensitiveDataSuggestionRuns(
errorSummary: string,
): Promise<SensitiveDataSuggestionRun[]>;
updateSensitiveDataSuggestionRun(
runId: string,
update: SensitiveDataSuggestionRunUpdate,
): Promise<SensitiveDataSuggestionRun | undefined>;
appendSensitiveDataSuggestionEvent(
runId: string,
level: SensitiveDataSuggestionEvent["level"],
message: string,
): Promise<SensitiveDataSuggestionEvent>;
listSensitiveDataSuggestionEvents(
runId: string,
afterSequence?: number,
): Promise<SensitiveDataSuggestionEvent[]>;
listRelationships(databaseId: string): Promise<CatalogRelationship[]>;
deleteDatabaseMetadata(
databaseIds: readonly string[],
+16
View File
@@ -66,6 +66,7 @@ const secretsSchema = z.object({
values: z.partialRecord(z.enum(secretNames), z.string().min(1).max(65_536)).refine((values) => Object.keys(values).length > 0),
}).strict();
const versionQuery = z.object({ version: z.coerce.number().int().positive() });
const metricsQuery = z.object({ databaseId: z.uuid().optional() }).strict();
function safeError(reply: FastifyReply, error: unknown) {
if (error instanceof CatalogUnavailableError) {
@@ -114,6 +115,21 @@ export function catalogDatabaseRoutes(
return { available: await deps.repository.available() };
});
app.get("/catalog/metrics", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
const { databaseId } = metricsQuery.parse(request.query);
const metrics = await deps.repository.getCatalogMetrics(databaseId);
if (!metrics) {
return reply.code(404).send({
code: "database_not_found",
message: "Database configuration was not found.",
});
}
return metrics;
} catch (error) { return safeError(reply, error); }
});
app.get("/catalog/databases", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
@@ -13,13 +13,13 @@ import {
import { MetadataGenerationModelUnavailableError } from "../catalog/metadata-generation-models.js";
import { ModelCompletionProviderError } from "../catalog/model-completer.js";
import {
SensitiveDataSuggester,
SensitiveDataSuggestionDuplicateTargetIdsError,
SensitiveDataSuggestionInvalidResponseError,
SensitiveDataSuggestionNoEligibleColumnsError,
SensitiveDataSuggestionPayloadTooLargeError,
SensitiveDataSuggestionTargetNotFoundError,
} from "../catalog/sensitive-data-suggester.js";
import type { SensitiveDataSuggestionRunner } from "../catalog/sensitive-data-suggestion-runner.js";
import {
CatalogOperationInProgressError,
CatalogUnavailableError,
@@ -27,6 +27,8 @@ import {
type CatalogRepository,
type DescriptionGenerationEvent,
type DescriptionGenerationRun,
type SensitiveDataSuggestionEvent,
type SensitiveDataSuggestionRun,
} from "../catalog/types.js";
const idSchema = z.uuid();
@@ -107,6 +109,34 @@ function publicRun(run: DescriptionGenerationRun) {
};
}
function publicSensitiveDataSuggestionEvent(event: SensitiveDataSuggestionEvent) {
return {
runId: event.runId,
sequence: event.sequence,
level: event.level,
message: event.message,
createdAt: event.createdAt,
};
}
function publicSensitiveDataSuggestionRun(run: SensitiveDataSuggestionRun) {
return {
id: run.id,
databaseId: run.databaseId,
scope: run.scope,
modelId: run.modelId,
status: run.status,
total: run.total,
suggestedSensitive: run.suggestedSensitive,
suggestedNonSensitive: run.suggestedNonSensitive,
createdAt: run.createdAt,
startedAt: run.startedAt,
updatedAt: run.updatedAt,
finishedAt: run.finishedAt,
errorSummary: run.errorSummary,
};
}
function safeError(reply: FastifyReply, error: unknown) {
if (error instanceof CatalogUnavailableError) {
return reply.code(503).send({
@@ -260,12 +290,31 @@ function safeSuggestionError(reply: FastifyReply, error: unknown) {
});
}
function safeSuggestionHistoryError(reply: FastifyReply, error: unknown) {
if (error instanceof CatalogUnavailableError) {
return reply.code(503).send({
code: "catalog_unavailable",
message: "Sensitive Data Suggestion history is unavailable because the database catalog is unavailable.",
});
}
if (error instanceof z.ZodError) {
return reply.code(400).send({
code: "sensitive_data_suggestion_history_request_invalid",
message: "Sensitive Data Suggestion history parameters are invalid.",
});
}
return reply.code(500).send({
code: "sensitive_data_suggestion_history_failed",
message: "Sensitive Data Suggestion history could not be loaded.",
});
}
export function catalogDescriptionGenerationRoutes(
app: FastifyInstance,
deps: {
repository: CatalogRepository;
worker: DescriptionGenerationWorker;
sensitiveDataSuggester: SensitiveDataSuggester;
sensitiveDataSuggestionRunner: SensitiveDataSuggestionRunner;
},
): void {
app.post("/catalog/databases/:databaseId/sensitive-data-suggestions", async (request, reply) => {
@@ -273,19 +322,66 @@ export function catalogDescriptionGenerationRoutes(
try {
const databaseId = idSchema.parse((request.params as { databaseId?: unknown }).databaseId);
const input = suggestionSchema.parse(request.body);
const suggestions = await deps.sensitiveDataSuggester.suggest(
const result = await deps.sensitiveDataSuggestionRunner.run(
databaseId,
input.modelId,
input.scope,
"targetIds" in input ? input.targetIds : [],
new AbortController().signal,
);
return { suggestions };
return {
suggestions: result.suggestions,
run: publicSensitiveDataSuggestionRun(result.run),
};
} catch (error) {
return safeSuggestionError(reply, error);
}
});
app.get("/catalog/sensitive-data-suggestion-runs", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
const { limit } = historyQuerySchema.parse(request.query);
return (await deps.repository.listSensitiveDataSuggestionRuns(limit))
.map(publicSensitiveDataSuggestionRun);
} catch (error) {
return safeSuggestionHistoryError(reply, error);
}
});
app.get("/catalog/sensitive-data-suggestion-runs/:runId", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
const run = await deps.repository.getSensitiveDataSuggestionRun(runId);
if (!run) return reply.code(404).send({
code: "sensitive_data_suggestion_run_not_found",
message: "Sensitive Data Suggestion Run was not found.",
});
return publicSensitiveDataSuggestionRun(run);
} catch (error) {
return safeSuggestionHistoryError(reply, error);
}
});
app.get("/catalog/sensitive-data-suggestion-runs/:runId/events-list", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
const { after } = eventQuerySchema.parse(request.query);
if (!(await deps.repository.getSensitiveDataSuggestionRun(runId))) {
return reply.code(404).send({
code: "sensitive_data_suggestion_run_not_found",
message: "Sensitive Data Suggestion Run was not found.",
});
}
return (await deps.repository.listSensitiveDataSuggestionEvents(runId, after))
.map(publicSensitiveDataSuggestionEvent);
} catch (error) {
return safeSuggestionHistoryError(reply, error);
}
});
app.post("/catalog/databases/:databaseId/description-generation-runs", async (request, reply) => {
if (!manage(request, reply)) return reply;
try {
+168 -2
View File
@@ -5,6 +5,7 @@ import { afterEach, expect, test, vi } from "vitest";
import { buildApp } from "../src/app.js";
import { loadConfig } from "../src/config.js";
import { MemoryCatalogRepository } from "../src/catalog/memory-repository.js";
import type { ObservedSchemaSnapshot } from "../src/catalog/types.js";
import { WorkspaceSecretStore } from "../src/workspaces/secret-store.js";
import type { WorkspaceRegistry, WorkspaceRevision } from "../src/workspaces/registry.js";
import type { WorkspaceDescriptor } from "../src/workspaces/schema.js";
@@ -27,7 +28,7 @@ const workspace: WorkspaceDescriptor = {
};
const revision: WorkspaceRevision = { id: "psd-clinical", commit: "a".repeat(40), blob: "b".repeat(40), snapshotPath: "/tmp/psd.yaml" };
function setup() {
function setup(environment: Record<string, string> = {}) {
const secretRoot = mkdtempSync(join(tmpdir(), "catalog-secret-"));
const runtimeRoot = mkdtempSync(join(tmpdir(), "catalog-secret-runtime-"));
roots.push(secretRoot, runtimeRoot);
@@ -38,7 +39,11 @@ function setup() {
listCatalog: vi.fn(async () => [{ id: "psd-clinical", name: "Policlinico San Donato", configurationState: "ready", revision }]),
read: vi.fn(async () => ({ workspace, revision })),
} as unknown as WorkspaceRegistry;
const app = buildApp(loadConfig({ THT_HARNESS_DIR: "/missing", NODE_ENV: "test" }), {
const app = buildApp(loadConfig({
THT_HARNESS_DIR: "/missing",
NODE_ENV: "test",
...environment,
}), {
thtRunner: {} as never,
workspaceRegistry: registry,
workspaceSecretStore: secretStore,
@@ -56,6 +61,31 @@ const direct = {
binding: { transport: "postgres_direct", host: "db.internal", port: 5432, username: "reader" },
};
const fleetSnapshot: ObservedSchemaSnapshot = {
schemaVersion: 1,
capabilities: { tables: "available", columns: "available", relationships: "available" },
tables: [
{ name: "patients", sourceComment: "Clinical patients" },
{ name: "visits", sourceComment: null },
],
columns: [
{ tableName: "patients", name: "id", ordinalPosition: 1, dataType: "bigint", isNullable: false, defaultExpression: null, primaryKeyPosition: 1, sourceComment: null },
{ tableName: "patients", name: "name", ordinalPosition: 2, dataType: "text", isNullable: true, defaultExpression: null, primaryKeyPosition: null, sourceComment: null },
{ tableName: "visits", name: "id", ordinalPosition: 1, dataType: "bigint", isNullable: false, defaultExpression: null, primaryKeyPosition: 1, sourceComment: null },
{ tableName: "visits", name: "patient_id", ordinalPosition: 2, dataType: "bigint", isNullable: false, defaultExpression: null, primaryKeyPosition: null, sourceComment: null },
],
relationships: [{
constraintName: "visits_patient_id_fkey",
sourceTableName: "visits",
targetTableName: "patients",
updateRule: "NO ACTION",
deleteRule: "CASCADE",
deferrable: false,
initiallyDeferred: false,
columns: [{ position: 1, sourceColumnName: "patient_id", targetColumnName: "id" }],
}],
};
test("lists every YAML workspace and creates its one database configuration", async () => {
const { app } = setup();
const initial = await app.inject({ method: "GET", url: "/catalog/databases" });
@@ -120,3 +150,139 @@ test("uses optimistic versions, keeps secrets write-only, and hard-deletes only
expect(secretStore.has("psd-clinical", "catalog.dwh.password")).toBe(false);
expect((await app.inject({ method: "GET", url: "/catalog/databases" })).json()).toMatchObject([{ configured: false }]);
});
test("returns exact global and per-database fleet metrics", async () => {
const { app, repository } = setup();
const database = await repository.create(direct);
await repository.applySchemaSync(database.id, database.version, "all", [], fleetSnapshot);
const patients = (await repository.listTables(database.id))
.find((table) => table.name === "patients")!;
await repository.updateTableDescription(
database.id,
patients.id,
patients.version,
"Curated patients",
);
const patientName = (await repository.listColumns(database.id, patients.id))
.find((column) => column.name === "name")!;
await repository.updateColumnMetadata(
database.id,
patients.id,
patientName.id,
patientName.version,
null,
"Generated patient name",
true,
);
const archive = await repository.create({
workspaceId: "removed-workspace",
engine: "postgres",
databaseName: "archive",
schema: "public",
binding: {
transport: "postgres_direct",
host: "archive.internal",
port: 5432,
username: "reader",
},
});
await repository.applySchemaSync(archive.id, archive.version, "all", [], {
schemaVersion: 1,
capabilities: { tables: "available", columns: "available", relationships: "available" },
tables: [{ name: "events", sourceComment: null }],
columns: [{
tableName: "events",
name: "id",
ordinalPosition: 1,
dataType: "bigint",
isNullable: false,
defaultExpression: null,
primaryKeyPosition: 1,
sourceComment: null,
}],
relationships: [],
});
const events = (await repository.listTables(archive.id))[0]!;
await repository.updateTableMetadata(
archive.id,
events.id,
events.version,
null,
"Generated archive events",
);
const scoped = await app.inject({
method: "GET",
url: `/catalog/metrics?databaseId=${database.id}`,
});
expect(scoped.statusCode).toBe(200);
expect(scoped.json()).toEqual({
scope: "database",
databaseId: database.id,
tables: 2,
columns: 4,
sensitiveColumns: 1,
relationships: 1,
descriptionTargets: 6,
describedTargets: 2,
descriptionCoverage: 33,
updatedAt: expect.any(String),
});
const global = await app.inject({ method: "GET", url: "/catalog/metrics" });
expect(global.statusCode).toBe(200);
expect(global.json()).toEqual({
scope: "global",
databaseId: null,
tables: 3,
columns: 5,
sensitiveColumns: 1,
relationships: 1,
descriptionTargets: 8,
describedTargets: 3,
descriptionCoverage: 38,
updatedAt: expect.any(String),
});
expect(Number.isNaN(Date.parse(global.json().updatedAt))).toBe(false);
});
test("validates fleet metric scope and requires database.manage", async () => {
const { app } = setup();
const unknown = await app.inject({
method: "GET",
url: "/catalog/metrics?databaseId=99999999-9999-4999-8999-999999999999",
});
expect(unknown.statusCode).toBe(404);
expect(unknown.json()).toEqual({
code: "database_not_found",
message: "Database configuration was not found.",
});
const invalid = await app.inject({
method: "GET",
url: "/catalog/metrics?databaseId=not-a-uuid",
});
expect(invalid.statusCode).toBe(400);
expect(invalid.json()).toEqual({
code: "database_invalid",
message: "Database configuration is invalid.",
});
const { app: restrictedApp } = setup({ AUTH_MODE: "upstream" });
const forbidden = await restrictedApp.inject({
method: "GET",
url: "/catalog/metrics",
headers: {
"x-thoth-principal-issuer": "portal",
"x-thoth-principal-subject": "catalog-reader",
"x-thoth-is-admin": "0",
},
});
expect(forbidden.statusCode).toBe(403);
expect(forbidden.json()).toEqual({
code: "auth_forbidden",
error: "This operation is not permitted",
});
});
@@ -160,7 +160,18 @@ test("suggests sensitive flags from structural metadata without persisting them"
});
expect(response.statusCode).toBe(200);
expect(response.json()).toEqual({
const responseBody = response.json();
expect(responseBody).toMatchObject({
run: {
databaseId: database.id,
scope: "all",
modelId: configuredModel.id,
status: "completed",
total: 1,
suggestedSensitive: 1,
suggestedNonSensitive: 0,
errorSummary: null,
},
suggestions: [{
columnId: column.id,
tableId: table.id,
@@ -173,6 +184,43 @@ test("suggests sensitive flags from structural metadata without persisting them"
});
expect(await repository.getColumn(database.id, column.tableId, column.id))
.toMatchObject({ sensitive: false });
expect(responseBody.run).not.toHaveProperty("suggestions");
const history = await app.inject({
method: "GET",
url: "/catalog/sensitive-data-suggestion-runs?limit=1",
});
expect(history.statusCode).toBe(200);
expect(history.json()).toEqual([responseBody.run]);
expect(history.body).not.toContain(column.id);
const detail = await app.inject({
method: "GET",
url: `/catalog/sensitive-data-suggestion-runs/${responseBody.run.id}`,
});
expect(detail.statusCode).toBe(200);
expect(detail.json()).toEqual(responseBody.run);
const events = await app.inject({
method: "GET",
url: `/catalog/sensitive-data-suggestion-runs/${responseBody.run.id}/events-list`,
});
expect(events.statusCode).toBe(200);
expect(events.body).not.toContain(column.id);
expect(events.json()).toMatchObject([
{
runId: responseBody.run.id,
sequence: 1,
level: "info",
message: "Sensitive-field suggestion generation started.",
},
{
runId: responseBody.run.id,
sequence: 2,
level: "info",
message: "Sensitive-field suggestion generation completed for 1 column.",
},
]);
const request = modelCompleter.complete.mock.calls[0]![0] as ModelCompletionRequest;
const prompt = request.messages.map((message) => message.content).join("\n");
@@ -505,6 +553,42 @@ test("explains a sensitive-data suggestion provider failure without exposing pro
expect(response.body).not.toContain("model completion failed");
expect(await repository.getColumn(database.id, column.tableId, column.id))
.toMatchObject({ sensitive: false });
const history = await app.inject({
method: "GET",
url: "/catalog/sensitive-data-suggestion-runs",
});
expect(history.statusCode).toBe(200);
const [failedRun] = history.json();
expect(failedRun).toMatchObject({
databaseId: database.id,
status: "failed",
total: 1,
suggestedSensitive: 0,
suggestedNonSensitive: 0,
errorSummary: "Sensitive-field suggestion generation failed.",
});
const events = await app.inject({
method: "GET",
url: `/catalog/sensitive-data-suggestion-runs/${failedRun.id}/events-list`,
});
expect(events.statusCode).toBe(200);
expect(events.json()).toMatchObject([
{
runId: failedRun.id,
sequence: 1,
level: "info",
message: "Sensitive-field suggestion generation started.",
},
{
runId: failedRun.id,
sequence: 2,
level: "error",
message: "Sensitive-field suggestion generation failed.",
},
]);
expect(events.body).not.toContain("model completion failed");
} finally {
await app.close();
}
@@ -2676,10 +2760,25 @@ test("requires database.manage for every Description Generation route", async ()
url: "/catalog/description-generation-runs/99999999-9999-4999-8999-999999999999/events",
headers,
}),
app.inject({
method: "GET",
url: "/catalog/sensitive-data-suggestion-runs",
headers,
}),
app.inject({
method: "GET",
url: "/catalog/sensitive-data-suggestion-runs/99999999-9999-4999-8999-999999999999",
headers,
}),
app.inject({
method: "GET",
url: "/catalog/sensitive-data-suggestion-runs/99999999-9999-4999-8999-999999999999/events-list",
headers,
}),
]);
expect(responses.map((response) => response.statusCode)).toEqual([
403, 403, 403, 403, 403, 403, 403,
403, 403, 403, 403, 403, 403, 403, 403, 403, 403,
]);
expect(modelCompleter.complete).not.toHaveBeenCalled();
} finally {
@@ -12,6 +12,7 @@ import { up as upTables } from "../src/catalog/migrations/002_catalog_tables.js"
import { up as upSchemaSync } from "../src/catalog/migrations/003_catalog_schema_sync.js";
import { up as upDescriptionGeneration } from "../src/catalog/migrations/005_description_generation_runs.js";
import { up as upSensitiveDataFlag } from "../src/catalog/migrations/006_sensitive_data_flag.js";
import { up as upSensitiveSuggestionRuns } from "../src/catalog/migrations/007_sensitive_data_suggestion_runs.js";
import { KyselyCatalogRepository, type CatalogDatabase } from "../src/catalog/repository.js";
import { loadConfig } from "../src/config.js";
import type { WorkspaceRegistry } from "../src/workspaces/registry.js";
@@ -46,6 +47,7 @@ test.skipIf(!dockerAvailable)("Fastify persists Description Generation success a
await upSchemaSync(db);
await upSensitiveDataFlag(db);
await upDescriptionGeneration(db);
await upSensitiveSuggestionRuns(db);
const repository = new KyselyCatalogRepository(db);
const database = await repository.create({
workspaceId: "psd-clinical",
@@ -11,6 +11,7 @@ import { up as upSchemaSync } from "../src/catalog/migrations/003_catalog_schema
import { up as upRuntimeSequencePrivileges } from "../src/catalog/migrations/004_catalog_runtime_sequence_privileges.js";
import { up as upDescriptionGeneration } from "../src/catalog/migrations/005_description_generation_runs.js";
import { up as upSensitiveDataFlag } from "../src/catalog/migrations/006_sensitive_data_flag.js";
import { up as upSensitiveSuggestionRuns } from "../src/catalog/migrations/007_sensitive_data_suggestion_runs.js";
const dockerAvailable = spawnSync("docker", ["info"], { stdio: "ignore" }).status === 0;
@@ -98,6 +99,33 @@ test.skipIf(!dockerAvailable)("PostgreSQL migration enforces one database per wo
patientName.generatedDescription,
true,
)).toMatchObject({ sensitive: true });
expect(await repository.getCatalogMetrics(created.id)).toEqual({
scope: "database",
databaseId: created.id,
tables: 2,
columns: 4,
sensitiveColumns: 1,
relationships: 0,
descriptionTargets: 6,
describedTargets: 1,
descriptionCoverage: 17,
updatedAt: expect.any(String),
});
expect(await repository.getCatalogMetrics()).toEqual({
scope: "global",
databaseId: null,
tables: 2,
columns: 4,
sensitiveColumns: 1,
relationships: 0,
descriptionTargets: 6,
describedTargets: 1,
descriptionCoverage: 17,
updatedAt: expect.any(String),
});
expect(await repository.getCatalogMetrics(
"99999999-9999-4999-8999-999999999999",
)).toBeUndefined();
const refreshedColumnsSnapshot: ObservedSchemaSnapshot = {
...fullColumnsSnapshot,
schemaVersion: 2,
@@ -340,7 +368,7 @@ test.skipIf(!dockerAvailable)("PostgreSQL repository atomically consolidates sel
}
}, 60_000);
test.skipIf(!dockerAvailable)("PostgreSQL repository persists globally exclusive Description Generation Runs and ordered events", async () => {
test.skipIf(!dockerAvailable)("PostgreSQL repository persists description and sensitive suggestion run histories", async () => {
const container = await new PostgreSqlContainer("postgres:17.6-bookworm").start();
const db = new Kysely<CatalogDatabase>({
dialect: new PostgresDialect({ pool: new Pool({ connectionString: container.getConnectionUri() }) }),
@@ -352,6 +380,7 @@ test.skipIf(!dockerAvailable)("PostgreSQL repository persists globally exclusive
await upSchemaSync(db);
await upSensitiveDataFlag(db);
await upDescriptionGeneration(db);
await upSensitiveSuggestionRuns(db);
const repository = new KyselyCatalogRepository(db);
const firstDatabase = await repository.create({
workspaceId: "generation-one",
@@ -519,6 +548,63 @@ test.skipIf(!dockerAvailable)("PostgreSQL repository persists globally exclusive
}),
]);
expect(await repository.getActiveDescriptionGenerationRun()).toBeUndefined();
const suggestionRun = await repository.createSensitiveDataSuggestionRun(
firstDatabase.id,
"selected_columns",
"openai-mini",
);
expect(suggestionRun).toMatchObject({
databaseId: firstDatabase.id,
status: "running",
total: 0,
suggestedSensitive: 0,
suggestedNonSensitive: 0,
startedAt: expect.any(String),
});
await repository.appendSensitiveDataSuggestionEvent(
suggestionRun.id,
"info",
"Sensitive-field suggestion generation started.",
);
await repository.appendSensitiveDataSuggestionEvent(
suggestionRun.id,
"info",
"Sensitive-field suggestion generation completed for 2 columns.",
);
expect(await repository.updateSensitiveDataSuggestionRun(suggestionRun.id, {
status: "completed",
total: 2,
suggestedSensitive: 1,
suggestedNonSensitive: 1,
finishedAt: new Date().toISOString(),
})).toMatchObject({
status: "completed",
total: 2,
suggestedSensitive: 1,
suggestedNonSensitive: 1,
});
expect(await repository.listSensitiveDataSuggestionEvents(suggestionRun.id, 1)).toEqual([
expect.objectContaining({ sequence: 2, level: "info" }),
]);
expect((await repository.listSensitiveDataSuggestionRuns(1))[0]).toMatchObject({
id: suggestionRun.id,
});
const interruptedSuggestionRun = await repository.createSensitiveDataSuggestionRun(
secondDatabase.id,
"all",
"openai-mini",
);
expect(await repository.interruptActiveSensitiveDataSuggestionRuns(
"Sensitive-field suggestion generation was interrupted by backend restart.",
)).toEqual([
expect.objectContaining({
id: interruptedSuggestionRun.id,
status: "interrupted",
finishedAt: expect.any(String),
}),
]);
} finally {
await db.destroy();
await container.stop();