feat: add metadata catalog cleanup commands
This commit is contained in:
+5
-1
@@ -385,7 +385,11 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
|
||||
});
|
||||
catalogDatabaseRoutes(app, { repository: catalogRepository, service: catalogService, operations: catalogOperationCoordinator });
|
||||
catalogTableRoutes(app, { repository: catalogRepository, service: catalogTableService });
|
||||
catalogSchemaRoutes(app, { repository: catalogRepository, worker: catalogSyncWorker });
|
||||
catalogSchemaRoutes(app, {
|
||||
repository: catalogRepository,
|
||||
worker: catalogSyncWorker,
|
||||
operations: catalogOperationCoordinator,
|
||||
});
|
||||
settingsRoutes(app, { cfg: config, listModels, getSettings });
|
||||
piManagementRoutes(app, { service: piManagement });
|
||||
|
||||
|
||||
@@ -3,6 +3,8 @@ import {
|
||||
CatalogConflictError,
|
||||
CatalogConnectorError,
|
||||
type CatalogColumn,
|
||||
type CatalogDatabaseMetadataDeleteTarget,
|
||||
type CatalogMetadataDeleteCounts,
|
||||
type CatalogRelationship,
|
||||
type CatalogSchemaDiff,
|
||||
type CatalogSyncCounts,
|
||||
@@ -11,6 +13,7 @@ import {
|
||||
type CatalogSyncRunUpdate,
|
||||
type CatalogSyncScope,
|
||||
type CatalogTable,
|
||||
type CatalogTableMetadataDeleteTarget,
|
||||
type CatalogRepository,
|
||||
type DatabaseConfigurationInput,
|
||||
type DatabaseTestResult,
|
||||
@@ -199,6 +202,71 @@ export class MemoryCatalogRepository implements CatalogRepository {
|
||||
.map((relationship) => structuredClone(relationship));
|
||||
}
|
||||
|
||||
async deleteDatabaseMetadata(
|
||||
databaseIds: readonly string[],
|
||||
target: CatalogDatabaseMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined> {
|
||||
const selectedDatabaseIds = [...new Set(databaseIds)];
|
||||
if (selectedDatabaseIds.length === 0 || selectedDatabaseIds.some((id) => !this.records.has(id))) {
|
||||
return undefined;
|
||||
}
|
||||
const selected = new Set(selectedDatabaseIds);
|
||||
const tables = [...this.tables.values()].filter((table) => selected.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) => selected.has(relationship.databaseId));
|
||||
|
||||
if (target === "tables") {
|
||||
for (const table of tables) this.deleteTable(table.id);
|
||||
this.markCatalogIncomplete(selectedDatabaseIds);
|
||||
return { tables: tables.length, columns: columns.length, relationships: relationships.length };
|
||||
}
|
||||
for (const relationship of relationships) this.relationships.delete(relationship.id);
|
||||
for (const databaseId of selectedDatabaseIds) this.refreshForeignKeyFlags(databaseId);
|
||||
this.markCatalogIncomplete(selectedDatabaseIds);
|
||||
return { tables: 0, columns: 0, relationships: relationships.length };
|
||||
}
|
||||
|
||||
async deleteTableMetadata(
|
||||
databaseId: string,
|
||||
tableIds: readonly string[],
|
||||
target: CatalogTableMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined> {
|
||||
const selectedTableIds = [...new Set(tableIds)];
|
||||
if (!this.records.has(databaseId) || selectedTableIds.length === 0) return undefined;
|
||||
const tables = selectedTableIds.map((id) => this.tables.get(id));
|
||||
if (tables.some((table) => !table || table.databaseId !== databaseId)) return undefined;
|
||||
const selected = new Set(selectedTableIds);
|
||||
|
||||
if (target === "columns") {
|
||||
const columns = [...this.columns.values()].filter((column) => selected.has(column.tableId));
|
||||
const deletedColumnIds = new Set(columns.map((column) => column.id));
|
||||
for (const column of columns) this.columns.delete(column.id);
|
||||
for (const [relationshipId, relationship] of this.relationships) {
|
||||
if (relationship.databaseId !== databaseId) continue;
|
||||
this.relationships.set(relationshipId, {
|
||||
...relationship,
|
||||
columns: relationship.columns.filter((pair) => (
|
||||
!deletedColumnIds.has(pair.sourceColumnId) && !deletedColumnIds.has(pair.targetColumnId)
|
||||
)),
|
||||
});
|
||||
}
|
||||
this.refreshForeignKeyFlags(databaseId);
|
||||
this.markCatalogIncomplete([databaseId]);
|
||||
return { tables: 0, columns: columns.length, relationships: 0 };
|
||||
}
|
||||
|
||||
const relationships = [...this.relationships.values()].filter((relationship) => (
|
||||
relationship.databaseId === databaseId
|
||||
&& (selected.has(relationship.sourceTableId) || selected.has(relationship.targetTableId))
|
||||
));
|
||||
for (const relationship of relationships) this.relationships.delete(relationship.id);
|
||||
this.refreshForeignKeyFlags(databaseId);
|
||||
this.markCatalogIncomplete([databaseId]);
|
||||
return { tables: 0, columns: 0, relationships: relationships.length };
|
||||
}
|
||||
|
||||
async planSchemaSync(
|
||||
databaseId: string,
|
||||
scope: CatalogSyncScope,
|
||||
@@ -675,6 +743,18 @@ export class MemoryCatalogRepository implements CatalogRepository {
|
||||
}
|
||||
}
|
||||
|
||||
private markCatalogIncomplete(databaseIds: readonly string[]): void {
|
||||
for (const databaseId of databaseIds) {
|
||||
const database = this.records.get(databaseId);
|
||||
if (!database) continue;
|
||||
this.records.set(databaseId, {
|
||||
...database,
|
||||
schemaSyncedVersion: undefined,
|
||||
schemaSyncedAt: undefined,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private refreshForeignKeyFlags(databaseId: string): void {
|
||||
const counts = new Map<string, number>();
|
||||
for (const relationship of this.relationships.values()) {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { CatalogOperationInProgressError } from "./types.js";
|
||||
|
||||
/** Serializes connection tests and schema synchronization for each catalog database. */
|
||||
/** Serializes connection tests, synchronization, and metadata cleanup for each catalog database. */
|
||||
export class CatalogOperationCoordinator {
|
||||
private readonly active = new Set<string>();
|
||||
|
||||
@@ -25,4 +25,16 @@ export class CatalogOperationCoordinator {
|
||||
release();
|
||||
}
|
||||
}
|
||||
|
||||
async runMany<T>(databaseIds: readonly string[], operation: () => Promise<T>): Promise<T> {
|
||||
const releases: Array<() => void> = [];
|
||||
try {
|
||||
for (const databaseId of [...new Set(databaseIds)].sort()) {
|
||||
releases.push(this.reserve(databaseId));
|
||||
}
|
||||
return await operation();
|
||||
} finally {
|
||||
for (const release of releases.reverse()) release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,8 @@ import {
|
||||
CatalogConnectorError,
|
||||
CatalogUnavailableError,
|
||||
type CatalogColumn,
|
||||
type CatalogDatabaseMetadataDeleteTarget,
|
||||
type CatalogMetadataDeleteCounts,
|
||||
type CatalogRelationship,
|
||||
type CatalogSchemaDiff,
|
||||
type CatalogSyncCounts,
|
||||
@@ -24,6 +26,7 @@ import {
|
||||
type CatalogSyncRunUpdate,
|
||||
type CatalogSyncScope,
|
||||
type CatalogTable,
|
||||
type CatalogTableMetadataDeleteTarget,
|
||||
type CatalogRepository,
|
||||
type DatabaseBinding,
|
||||
type DatabaseConfigurationInput,
|
||||
@@ -559,6 +562,104 @@ export class KyselyCatalogRepository implements CatalogRepository {
|
||||
}));
|
||||
}
|
||||
|
||||
async deleteDatabaseMetadata(
|
||||
databaseIds: readonly string[],
|
||||
target: CatalogDatabaseMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined> {
|
||||
const selectedDatabaseIds = [...new Set(databaseIds)];
|
||||
if (selectedDatabaseIds.length === 0) return undefined;
|
||||
return await this.db.transaction().execute(async (trx) => {
|
||||
const databases = await trx.selectFrom("workspaceDatabases").select("id")
|
||||
.where("id", "in", selectedDatabaseIds).orderBy("id").forUpdate().execute();
|
||||
if (databases.length !== selectedDatabaseIds.length) return undefined;
|
||||
|
||||
if (target === "relationships") {
|
||||
const count = await trx.selectFrom("catalogRelationships")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("databaseId", "in", selectedDatabaseIds).executeTakeFirst();
|
||||
await trx.deleteFrom("catalogRelationships")
|
||||
.where("databaseId", "in", selectedDatabaseIds).execute();
|
||||
await trx.updateTable("workspaceDatabases").set({
|
||||
schemaSyncedVersion: null,
|
||||
schemaSyncedAt: null,
|
||||
}).where("id", "in", selectedDatabaseIds).execute();
|
||||
return { tables: 0, columns: 0, relationships: Number(count?.count ?? 0) };
|
||||
}
|
||||
|
||||
const tableCount = await trx.selectFrom("catalogTables")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("databaseId", "in", selectedDatabaseIds).executeTakeFirst();
|
||||
const columnCount = await trx.selectFrom("catalogColumns")
|
||||
.innerJoin("catalogTables", "catalogTables.id", "catalogColumns.tableId")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("catalogTables.databaseId", "in", selectedDatabaseIds).executeTakeFirst();
|
||||
const relationshipCount = await trx.selectFrom("catalogRelationships")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("databaseId", "in", selectedDatabaseIds).executeTakeFirst();
|
||||
await trx.deleteFrom("catalogTables")
|
||||
.where("databaseId", "in", selectedDatabaseIds).execute();
|
||||
await trx.updateTable("workspaceDatabases").set({
|
||||
schemaSyncedVersion: null,
|
||||
schemaSyncedAt: null,
|
||||
}).where("id", "in", selectedDatabaseIds).execute();
|
||||
return {
|
||||
tables: Number(tableCount?.count ?? 0),
|
||||
columns: Number(columnCount?.count ?? 0),
|
||||
relationships: Number(relationshipCount?.count ?? 0),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
async deleteTableMetadata(
|
||||
databaseId: string,
|
||||
tableIds: readonly string[],
|
||||
target: CatalogTableMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined> {
|
||||
const selectedTableIds = [...new Set(tableIds)];
|
||||
if (selectedTableIds.length === 0) return undefined;
|
||||
return await this.db.transaction().execute(async (trx) => {
|
||||
const database = await trx.selectFrom("workspaceDatabases").select("id")
|
||||
.where("id", "=", databaseId).forUpdate().executeTakeFirst();
|
||||
if (!database) return undefined;
|
||||
const tables = await trx.selectFrom("catalogTables").select("id")
|
||||
.where("databaseId", "=", databaseId)
|
||||
.where("id", "in", selectedTableIds).execute();
|
||||
if (tables.length !== selectedTableIds.length) return undefined;
|
||||
|
||||
if (target === "columns") {
|
||||
const count = await trx.selectFrom("catalogColumns")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("tableId", "in", selectedTableIds).executeTakeFirst();
|
||||
await trx.deleteFrom("catalogColumns")
|
||||
.where("tableId", "in", selectedTableIds).execute();
|
||||
await trx.updateTable("workspaceDatabases").set({
|
||||
schemaSyncedVersion: null,
|
||||
schemaSyncedAt: null,
|
||||
}).where("id", "=", databaseId).execute();
|
||||
return { tables: 0, columns: Number(count?.count ?? 0), relationships: 0 };
|
||||
}
|
||||
|
||||
const count = await trx.selectFrom("catalogRelationships")
|
||||
.select(sql<number>`count(*)::int`.as("count"))
|
||||
.where("databaseId", "=", databaseId)
|
||||
.where((eb) => eb.or([
|
||||
eb("sourceTableId", "in", selectedTableIds),
|
||||
eb("targetTableId", "in", selectedTableIds),
|
||||
])).executeTakeFirst();
|
||||
await trx.deleteFrom("catalogRelationships")
|
||||
.where("databaseId", "=", databaseId)
|
||||
.where((eb) => eb.or([
|
||||
eb("sourceTableId", "in", selectedTableIds),
|
||||
eb("targetTableId", "in", selectedTableIds),
|
||||
])).execute();
|
||||
await trx.updateTable("workspaceDatabases").set({
|
||||
schemaSyncedVersion: null,
|
||||
schemaSyncedAt: null,
|
||||
}).where("id", "=", databaseId).execute();
|
||||
return { tables: 0, columns: 0, relationships: Number(count?.count ?? 0) };
|
||||
});
|
||||
}
|
||||
|
||||
async planSchemaSync(
|
||||
databaseId: string,
|
||||
scope: CatalogSyncScope,
|
||||
@@ -1002,6 +1103,8 @@ export class UnavailableCatalogRepository implements CatalogRepository {
|
||||
async getColumn(): Promise<CatalogColumn | undefined> { return this.fail(); }
|
||||
async updateColumnMetadata(): Promise<CatalogColumn | undefined> { 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(); }
|
||||
async planSchemaSync(): Promise<CatalogSchemaDiff> { return this.fail(); }
|
||||
async applySchemaSync(): Promise<CatalogSyncCounts | undefined> { return this.fail(); }
|
||||
async createSyncRun(): Promise<CatalogSyncRun> { return this.fail(); }
|
||||
|
||||
@@ -133,6 +133,15 @@ export interface CatalogRelationship {
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export type CatalogDatabaseMetadataDeleteTarget = "tables" | "relationships";
|
||||
export type CatalogTableMetadataDeleteTarget = "columns" | "relationships";
|
||||
|
||||
export interface CatalogMetadataDeleteCounts {
|
||||
tables: number;
|
||||
columns: number;
|
||||
relationships: number;
|
||||
}
|
||||
|
||||
export interface ObservedRelationshipColumn {
|
||||
position: number;
|
||||
sourceColumnName: string;
|
||||
@@ -284,6 +293,15 @@ export interface CatalogRepository {
|
||||
generatedDescription: string | null,
|
||||
): Promise<CatalogColumn | undefined>;
|
||||
listRelationships(databaseId: string): Promise<CatalogRelationship[]>;
|
||||
deleteDatabaseMetadata(
|
||||
databaseIds: readonly string[],
|
||||
target: CatalogDatabaseMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined>;
|
||||
deleteTableMetadata(
|
||||
databaseId: string,
|
||||
tableIds: readonly string[],
|
||||
target: CatalogTableMetadataDeleteTarget,
|
||||
): Promise<CatalogMetadataDeleteCounts | undefined>;
|
||||
planSchemaSync(
|
||||
databaseId: string,
|
||||
scope: CatalogSyncScope,
|
||||
|
||||
@@ -3,6 +3,7 @@ import { setTimeout as delay } from "node:timers/promises";
|
||||
import { z } from "zod";
|
||||
import { isPrincipalContext, requirePermission } from "../auth/authorization.js";
|
||||
import type { CatalogSyncWorker } from "../catalog/sync-worker.js";
|
||||
import type { CatalogOperationCoordinator } from "../catalog/operation-coordinator.js";
|
||||
import {
|
||||
CatalogConflictError,
|
||||
CatalogConnectorError,
|
||||
@@ -23,6 +24,14 @@ const createRunSchema = z.object({
|
||||
scope: z.enum(["tables", "columns", "relationships", "all"]),
|
||||
tableIds: z.array(idSchema).max(10_000).default([]),
|
||||
}).strict();
|
||||
const databaseCleanupSchema = z.object({
|
||||
databaseIds: z.array(idSchema).min(1).max(1_000),
|
||||
target: z.enum(["tables", "relationships"]),
|
||||
}).strict();
|
||||
const tableCleanupSchema = z.object({
|
||||
tableIds: z.array(idSchema).min(1).max(10_000),
|
||||
target: z.enum(["columns", "relationships"]),
|
||||
}).strict();
|
||||
const confirmationSchema = z.object({ confirmationToken: z.string().uuid() }).strict();
|
||||
const eventQuerySchema = z.object({ after: z.coerce.number().int().nonnegative().default(0) });
|
||||
|
||||
@@ -62,7 +71,11 @@ function publicRun(run: CatalogSyncRun) {
|
||||
|
||||
export function catalogSchemaRoutes(
|
||||
app: FastifyInstance,
|
||||
deps: { repository: CatalogRepository; worker: CatalogSyncWorker },
|
||||
deps: {
|
||||
repository: CatalogRepository;
|
||||
worker: CatalogSyncWorker;
|
||||
operations?: CatalogOperationCoordinator;
|
||||
},
|
||||
): void {
|
||||
app.get("/catalog/databases/:databaseId/tables/:tableId/columns", async (request, reply) => {
|
||||
if (!manage(request, reply)) return reply;
|
||||
@@ -114,6 +127,45 @@ export function catalogSchemaRoutes(
|
||||
} catch (error) { return safeError(reply, error); }
|
||||
});
|
||||
|
||||
app.post("/catalog/databases/metadata-cleanup", async (request, reply) => {
|
||||
if (!manage(request, reply)) return reply;
|
||||
try {
|
||||
const input = databaseCleanupSchema.parse(request.body);
|
||||
const databaseIds = [...new Set(input.databaseIds)];
|
||||
const operation = async () => await deps.repository.deleteDatabaseMetadata(databaseIds, input.target);
|
||||
const deleted = deps.operations
|
||||
? await deps.operations.runMany(databaseIds, operation)
|
||||
: await operation();
|
||||
if (!deleted) {
|
||||
return reply.code(404).send({
|
||||
code: "database_not_found",
|
||||
message: "One or more database configurations were not found.",
|
||||
});
|
||||
}
|
||||
return deleted;
|
||||
} catch (error) { return safeError(reply, error); }
|
||||
});
|
||||
|
||||
app.post("/catalog/databases/:databaseId/tables/metadata-cleanup", async (request, reply) => {
|
||||
if (!manage(request, reply)) return reply;
|
||||
try {
|
||||
const databaseId = idSchema.parse((request.params as { databaseId?: unknown }).databaseId);
|
||||
const input = tableCleanupSchema.parse(request.body);
|
||||
const tableIds = [...new Set(input.tableIds)];
|
||||
const operation = async () => await deps.repository.deleteTableMetadata(databaseId, tableIds, input.target);
|
||||
const deleted = deps.operations
|
||||
? await deps.operations.run(databaseId, operation)
|
||||
: await operation();
|
||||
if (!deleted) {
|
||||
return reply.code(404).send({
|
||||
code: "table_not_found",
|
||||
message: "One or more catalog tables were not found.",
|
||||
});
|
||||
}
|
||||
return deleted;
|
||||
} catch (error) { return safeError(reply, error); }
|
||||
});
|
||||
|
||||
app.post("/catalog/databases/:databaseId/sync-runs", async (request, reply) => {
|
||||
if (!manage(request, reply)) return reply;
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
import { expect, test } from "vitest";
|
||||
import { CatalogOperationCoordinator } from "../src/catalog/operation-coordinator.js";
|
||||
|
||||
test("releases earlier database reservations when a batch reservation cannot be completed", async () => {
|
||||
const coordinator = new CatalogOperationCoordinator();
|
||||
const releaseBusyDatabase = coordinator.reserve("database-b");
|
||||
|
||||
await expect(coordinator.runMany(["database-a", "database-b"], async () => "unreachable"))
|
||||
.rejects.toThrow("A database operation is already in progress");
|
||||
|
||||
expect(await coordinator.run("database-a", async () => "available")).toBe("available");
|
||||
releaseBusyDatabase();
|
||||
});
|
||||
|
||||
test("reserves duplicate database ids only once for a batch operation", async () => {
|
||||
const coordinator = new CatalogOperationCoordinator();
|
||||
|
||||
expect(await coordinator.runMany(["database-a", "database-a"], async () => "completed"))
|
||||
.toBe("completed");
|
||||
});
|
||||
@@ -123,3 +123,80 @@ test.skipIf(!dockerAvailable)("PostgreSQL migration enforces one database per wo
|
||||
await container.stop();
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
test.skipIf(!dockerAvailable)("PostgreSQL repository performs scoped metadata cleanup with catalog cascades", 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() }) }),
|
||||
plugins: [new CamelCasePlugin()],
|
||||
});
|
||||
try {
|
||||
await upDatabases(db);
|
||||
await upTables(db);
|
||||
await upSchemaSync(db);
|
||||
const repository = new KyselyCatalogRepository(db);
|
||||
const database = await repository.create({
|
||||
workspaceId: "cleanup-test",
|
||||
engine: "postgres",
|
||||
databaseName: "warehouse",
|
||||
schema: "public",
|
||||
binding: { transport: "postgres_direct", host: "db.internal", port: 5432, username: "reader" },
|
||||
});
|
||||
const snapshot: ObservedSchemaSnapshot = {
|
||||
schemaVersion: 1,
|
||||
capabilities: { tables: "available", columns: "available", relationships: "available" },
|
||||
tables: [
|
||||
{ name: "patients", sourceComment: null },
|
||||
{ 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" }],
|
||||
}],
|
||||
};
|
||||
await repository.applySchemaSync(database.id, database.version, "all", [], snapshot);
|
||||
const patients = (await repository.listTables(database.id)).find((table) => table.name === "patients")!;
|
||||
|
||||
expect(await repository.deleteTableMetadata(database.id, [patients.id], "relationships"))
|
||||
.toEqual({ tables: 0, columns: 0, relationships: 1 });
|
||||
expect(await repository.listRelationships(database.id)).toEqual([]);
|
||||
expect(await repository.listColumns(database.id, patients.id)).toHaveLength(2);
|
||||
expect((await repository.get(database.id))?.schemaSyncedVersion).toBeUndefined();
|
||||
|
||||
await repository.applySchemaSync(database.id, database.version, "relationships", [], snapshot);
|
||||
expect(await repository.deleteTableMetadata(database.id, [patients.id], "columns"))
|
||||
.toEqual({ tables: 0, columns: 2, relationships: 0 });
|
||||
expect(await repository.listColumns(database.id, patients.id)).toEqual([]);
|
||||
expect(await repository.listRelationships(database.id)).toMatchObject([
|
||||
{ constraintName: "visits_patient_id_fkey", columns: [] },
|
||||
]);
|
||||
|
||||
await repository.applySchemaSync(database.id, database.version, "columns", [patients.id], snapshot);
|
||||
await repository.applySchemaSync(database.id, database.version, "relationships", [], snapshot);
|
||||
expect(await repository.deleteDatabaseMetadata([
|
||||
database.id,
|
||||
"99999999-9999-4999-8999-999999999999",
|
||||
], "tables")).toBeUndefined();
|
||||
expect(await repository.listTables(database.id)).toHaveLength(2);
|
||||
expect(await repository.deleteDatabaseMetadata([database.id], "tables"))
|
||||
.toEqual({ tables: 2, columns: 4, relationships: 1 });
|
||||
expect(await repository.get(database.id)).toBeDefined();
|
||||
expect(await repository.listTables(database.id)).toEqual([]);
|
||||
expect(await repository.listRelationships(database.id)).toEqual([]);
|
||||
} finally {
|
||||
await db.destroy();
|
||||
await container.stop();
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
@@ -51,6 +51,41 @@ function snapshot(): ObservedSchemaSnapshot {
|
||||
};
|
||||
}
|
||||
|
||||
function expandedSnapshot(): ObservedSchemaSnapshot {
|
||||
const expanded = snapshot();
|
||||
expanded.tables.push(
|
||||
{ name: "encounters", sourceComment: null },
|
||||
{ name: "departments", sourceComment: null },
|
||||
);
|
||||
expanded.columns.push(
|
||||
{ tableName: "encounters", name: "visit_id", ordinalPosition: 1, dataType: "bigint", isNullable: false, defaultExpression: null, primaryKeyPosition: null, sourceComment: null },
|
||||
{ tableName: "departments", name: "patient_id", ordinalPosition: 1, dataType: "bigint", isNullable: false, defaultExpression: null, primaryKeyPosition: null, sourceComment: null },
|
||||
);
|
||||
expanded.relationships.push(
|
||||
{
|
||||
constraintName: "encounters_visit_id_fkey",
|
||||
sourceTableName: "encounters",
|
||||
targetTableName: "visits",
|
||||
updateRule: "NO ACTION",
|
||||
deleteRule: "CASCADE",
|
||||
deferrable: false,
|
||||
initiallyDeferred: false,
|
||||
columns: [{ position: 1, sourceColumnName: "visit_id", targetColumnName: "id" }],
|
||||
},
|
||||
{
|
||||
constraintName: "departments_patient_id_fkey",
|
||||
sourceTableName: "departments",
|
||||
targetTableName: "patients",
|
||||
updateRule: "NO ACTION",
|
||||
deleteRule: "CASCADE",
|
||||
deferrable: false,
|
||||
initiallyDeferred: false,
|
||||
columns: [{ position: 1, sourceColumnName: "patient_id", targetColumnName: "id" }],
|
||||
},
|
||||
);
|
||||
return expanded;
|
||||
}
|
||||
|
||||
async function waitFor(repository: MemoryCatalogRepository, runId: string, state: CatalogSyncRun["state"]): Promise<CatalogSyncRun> {
|
||||
for (let attempt = 0; attempt < 100; attempt += 1) {
|
||||
const run = await repository.getSyncRun(runId);
|
||||
@@ -100,6 +135,14 @@ async function setup() {
|
||||
};
|
||||
}
|
||||
|
||||
async function seedCatalog(
|
||||
repository: MemoryCatalogRepository,
|
||||
database: { id: string; version: number },
|
||||
observed: ObservedSchemaSnapshot = snapshot(),
|
||||
): Promise<void> {
|
||||
await repository.applySchemaSync(database.id, database.version, "all", [], observed);
|
||||
}
|
||||
|
||||
test("synchronizes a full physical schema and derives primary and foreign key flags", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
const started = await app.inject({
|
||||
@@ -222,3 +265,110 @@ test("waits for confirmation and rescans before applying destructive changes", a
|
||||
expect((await repository.listTables(database.id)).map((table) => table.name)).toEqual(["patients"]);
|
||||
expect(scan).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
test("deletes every catalog table for multiple selected databases and cascades dependent metadata", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
const second = await repository.create({
|
||||
workspaceId: "radiology", engine: "postgres", databaseName: "radiology", schema: "public",
|
||||
binding: { transport: "postgres_direct", host: "db.internal", port: 5432, username: "reader" },
|
||||
});
|
||||
await seedCatalog(repository, database);
|
||||
await seedCatalog(repository, second);
|
||||
|
||||
const response = await app.inject({
|
||||
method: "POST",
|
||||
url: "/catalog/databases/metadata-cleanup",
|
||||
payload: { databaseIds: [database.id, second.id], target: "tables" },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.json()).toEqual({ tables: 4, columns: 6, relationships: 2 });
|
||||
expect(await repository.get(database.id)).toBeDefined();
|
||||
expect(await repository.get(second.id)).toBeDefined();
|
||||
expect(await repository.listTables(database.id)).toEqual([]);
|
||||
expect(await repository.listTables(second.id)).toEqual([]);
|
||||
expect(await repository.listRelationships(database.id)).toEqual([]);
|
||||
expect(await repository.listRelationships(second.id)).toEqual([]);
|
||||
});
|
||||
|
||||
test("deletes relationships for selected databases without deleting their tables or columns", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
await seedCatalog(repository, database);
|
||||
const tables = await repository.listTables(database.id);
|
||||
|
||||
const response = await app.inject({
|
||||
method: "POST",
|
||||
url: "/catalog/databases/metadata-cleanup",
|
||||
payload: { databaseIds: [database.id], target: "relationships" },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.json()).toEqual({ tables: 0, columns: 0, relationships: 1 });
|
||||
expect(await repository.listTables(database.id)).toHaveLength(2);
|
||||
expect(await repository.listColumns(database.id, tables[0]!.id)).not.toEqual([]);
|
||||
expect(await repository.listRelationships(database.id)).toEqual([]);
|
||||
expect((await repository.get(database.id))?.schemaSyncedVersion).toBeUndefined();
|
||||
});
|
||||
|
||||
test("deletes columns from multiple selected catalog tables only", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
await seedCatalog(repository, database, expandedSnapshot());
|
||||
const tables = await repository.listTables(database.id);
|
||||
const patients = tables.find((table) => table.name === "patients")!;
|
||||
const visits = tables.find((table) => table.name === "visits")!;
|
||||
const encounters = tables.find((table) => table.name === "encounters")!;
|
||||
const departments = tables.find((table) => table.name === "departments")!;
|
||||
|
||||
const response = await app.inject({
|
||||
method: "POST",
|
||||
url: `/catalog/databases/${database.id}/tables/metadata-cleanup`,
|
||||
payload: { tableIds: [patients.id, visits.id], target: "columns" },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.json()).toEqual({ tables: 0, columns: 3, relationships: 0 });
|
||||
expect(await repository.listColumns(database.id, patients.id)).toEqual([]);
|
||||
expect(await repository.listColumns(database.id, visits.id)).toEqual([]);
|
||||
expect((await repository.listColumns(database.id, encounters.id)).map((column) => column.name))
|
||||
.toEqual(["visit_id"]);
|
||||
expect((await repository.listColumns(database.id, departments.id)).map((column) => column.name))
|
||||
.toEqual(["patient_id"]);
|
||||
expect((await repository.get(database.id))?.schemaSyncedVersion).toBeUndefined();
|
||||
});
|
||||
|
||||
test("deletes incoming and outgoing relationships for multiple selected catalog tables only", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
await seedCatalog(repository, database, expandedSnapshot());
|
||||
const tables = await repository.listTables(database.id);
|
||||
const visits = tables.find((table) => table.name === "visits")!;
|
||||
const encounters = tables.find((table) => table.name === "encounters")!;
|
||||
|
||||
const response = await app.inject({
|
||||
method: "POST",
|
||||
url: `/catalog/databases/${database.id}/tables/metadata-cleanup`,
|
||||
payload: { tableIds: [visits.id, encounters.id], target: "relationships" },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.json()).toEqual({ tables: 0, columns: 0, relationships: 2 });
|
||||
expect((await repository.listRelationships(database.id)).map((relationship) => relationship.constraintName))
|
||||
.toEqual(["departments_patient_id_fkey"]);
|
||||
expect(await repository.listTables(database.id)).toHaveLength(4);
|
||||
});
|
||||
|
||||
test("rejects a missing database without partially cleaning valid selections", async () => {
|
||||
const { app, repository, database } = await setup();
|
||||
await seedCatalog(repository, database);
|
||||
|
||||
const response = await app.inject({
|
||||
method: "POST",
|
||||
url: "/catalog/databases/metadata-cleanup",
|
||||
payload: {
|
||||
databaseIds: [database.id, "99999999-9999-4999-8999-999999999999"],
|
||||
target: "tables",
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(404);
|
||||
expect(await repository.listTables(database.id)).toHaveLength(2);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user