303 lines
14 KiB
TypeScript
303 lines
14 KiB
TypeScript
import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
|
|
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,
|
|
CatalogOperationInProgressError,
|
|
CatalogUnavailableError,
|
|
type CatalogRepository,
|
|
type CatalogSyncRun,
|
|
} from "../catalog/types.js";
|
|
|
|
const idSchema = z.uuid();
|
|
const metadataSchema = z.object({
|
|
version: z.number().int().positive(),
|
|
description: z.string().max(20_000).nullable().optional(),
|
|
generatedDescription: z.string().max(20_000).nullable().optional(),
|
|
sensitive: z.boolean().optional(),
|
|
sensitivityReason: z.string().max(2_000).nullable().optional(),
|
|
}).strict().refine((value) => (
|
|
"description" in value
|
|
|| "generatedDescription" in value
|
|
|| "sensitive" in value
|
|
|| "sensitivityReason" in value
|
|
)).refine((value) => (
|
|
value.sensitivityReason == null || value.sensitive === true
|
|
));
|
|
const createRunSchema = z.object({
|
|
version: z.number().int().positive(),
|
|
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) });
|
|
|
|
function manage(request: FastifyRequest, reply: FastifyReply) {
|
|
return isPrincipalContext(requirePermission(request, reply, "database.manage"));
|
|
}
|
|
|
|
function safeError(reply: FastifyReply, error: unknown) {
|
|
if (error instanceof CatalogUnavailableError) {
|
|
return reply.code(503).send({ code: "catalog_unavailable", message: "Database catalog is unavailable." });
|
|
}
|
|
if (error instanceof CatalogOperationInProgressError) {
|
|
return reply.code(409).send({
|
|
code: "database_operation_in_progress",
|
|
message: "A database operation is already in progress.",
|
|
});
|
|
}
|
|
if (error instanceof CatalogConflictError) {
|
|
return reply.code(409).send({ code: "schema_sync_conflict", message: error.message });
|
|
}
|
|
if (error instanceof CatalogConnectorError) {
|
|
return reply.code(502).send({
|
|
code: "schema_introspection_failed",
|
|
message: "The database schema could not be read. Check the connection and credentials, then try again.",
|
|
});
|
|
}
|
|
if (error instanceof z.ZodError) {
|
|
return reply.code(400).send({ code: "schema_request_invalid", message: "Schema request is invalid." });
|
|
}
|
|
return reply.code(500).send({ code: "schema_operation_failed", message: "Schema operation failed." });
|
|
}
|
|
|
|
function normalized(value: string | null): string | null {
|
|
return value?.trim() || null;
|
|
}
|
|
|
|
function publicRun(run: CatalogSyncRun) {
|
|
const {
|
|
observedSnapshot: _snapshot,
|
|
leaseOwner: _leaseOwner,
|
|
leaseExpiresAt: _leaseExpiresAt,
|
|
...result
|
|
} = run;
|
|
return result;
|
|
}
|
|
|
|
export function catalogSchemaRoutes(
|
|
app: FastifyInstance,
|
|
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;
|
|
try {
|
|
const params = request.params as { databaseId?: unknown; tableId?: unknown };
|
|
const databaseId = idSchema.parse(params.databaseId);
|
|
const tableId = idSchema.parse(params.tableId);
|
|
if (!(await deps.repository.getTable(databaseId, tableId))) {
|
|
return reply.code(404).send({ code: "table_not_found", message: "Catalog table was not found." });
|
|
}
|
|
return await deps.repository.listColumns(databaseId, tableId);
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.patch("/catalog/databases/:databaseId/tables/:tableId/columns/:columnId", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const params = request.params as { databaseId?: unknown; tableId?: unknown; columnId?: unknown };
|
|
const databaseId = idSchema.parse(params.databaseId);
|
|
const tableId = idSchema.parse(params.tableId);
|
|
const columnId = idSchema.parse(params.columnId);
|
|
const input = metadataSchema.parse(request.body);
|
|
const operation = async () => {
|
|
const current = await deps.repository.getColumn(databaseId, tableId, columnId);
|
|
if (!current) return reply.code(404).send({ code: "column_not_found", message: "Catalog column was not found." });
|
|
if (current.version !== input.version) {
|
|
return reply.code(409).send({ code: "column_stale", message: "Column metadata changed. Reload and try again." });
|
|
}
|
|
const nextSensitive = input.sensitive ?? current.sensitive;
|
|
const nextSensitivityReason = nextSensitive
|
|
? ("sensitivityReason" in input
|
|
? normalized(input.sensitivityReason ?? null)
|
|
: current.sensitivityReason)
|
|
: null;
|
|
const updated = await deps.repository.updateColumnMetadata(
|
|
databaseId,
|
|
tableId,
|
|
columnId,
|
|
input.version,
|
|
"description" in input ? normalized(input.description ?? null) : current.description,
|
|
"generatedDescription" in input
|
|
? normalized(input.generatedDescription ?? null)
|
|
: current.generatedDescription,
|
|
input.sensitive,
|
|
nextSensitivityReason,
|
|
);
|
|
if (!updated) return reply.code(409).send({ code: "column_stale", message: "Column metadata changed. Reload and try again." });
|
|
return updated;
|
|
};
|
|
return deps.operations ? await deps.operations.run(databaseId, operation) : await operation();
|
|
} 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 {
|
|
const databaseId = idSchema.parse((request.params as { databaseId?: unknown }).databaseId);
|
|
const input = createRunSchema.parse(request.body);
|
|
const database = await deps.repository.get(databaseId);
|
|
if (!database) return reply.code(404).send({ code: "database_not_found", message: "Database configuration was not found." });
|
|
if (database.version !== input.version) {
|
|
return reply.code(409).send({ code: "database_stale", message: "Database configuration changed. Reload and try again." });
|
|
}
|
|
return reply.code(202).send(publicRun(await deps.worker.start(database, input.scope, input.tableIds)));
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.get("/catalog/databases/:databaseId/sync-runs", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const databaseId = idSchema.parse((request.params as { databaseId?: unknown }).databaseId);
|
|
if (!(await deps.repository.get(databaseId))) {
|
|
return reply.code(404).send({ code: "database_not_found", message: "Database configuration was not found." });
|
|
}
|
|
return (await deps.repository.listSyncRuns(databaseId)).map(publicRun);
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.get("/catalog/sync-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.getSyncRun(runId);
|
|
return run ? publicRun(run) : reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.post("/catalog/sync-runs/:runId/confirm", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
|
|
const { confirmationToken } = confirmationSchema.parse(request.body);
|
|
const run = await deps.worker.confirm(runId, confirmationToken);
|
|
return run ? publicRun(run) : reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.post("/catalog/sync-runs/:runId/cancel", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
|
|
const run = await deps.worker.cancel(runId);
|
|
return run ? publicRun(run) : reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.post("/catalog/sync-runs/:runId/retry", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
|
|
const run = await deps.worker.retry(runId);
|
|
return run ? reply.code(202).send(publicRun(run)) : reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.get("/catalog/sync-runs/:runId/events", async (request, reply) => {
|
|
if (!manage(request, reply)) return reply;
|
|
try {
|
|
const runId = idSchema.parse((request.params as { runId?: unknown }).runId);
|
|
let after = eventQuerySchema.parse(request.query).after;
|
|
const headerCursor = Number(request.headers["last-event-id"]);
|
|
if (Number.isInteger(headerCursor) && headerCursor >= 0) after = Math.max(after, headerCursor);
|
|
if (!(await deps.repository.getSyncRun(runId))) {
|
|
return reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
}
|
|
reply.hijack();
|
|
reply.raw.writeHead(200, {
|
|
"content-type": "text/event-stream; charset=utf-8",
|
|
"cache-control": "no-cache, no-transform",
|
|
connection: "keep-alive",
|
|
"x-accel-buffering": "no",
|
|
});
|
|
const controller = new AbortController();
|
|
request.raw.once("close", () => controller.abort());
|
|
let lastRunUpdate = "";
|
|
while (!controller.signal.aborted) {
|
|
const events = await deps.repository.listSyncEvents(runId, after);
|
|
for (const event of events) {
|
|
after = event.sequence;
|
|
reply.raw.write(`id: ${event.sequence}\nevent: log\ndata: ${JSON.stringify(event)}\n\n`);
|
|
}
|
|
const run = await deps.repository.getSyncRun(runId);
|
|
if (!run) break;
|
|
if (run.updatedAt !== lastRunUpdate) {
|
|
lastRunUpdate = run.updatedAt;
|
|
reply.raw.write(`event: run\ndata: ${JSON.stringify(publicRun(run))}\n\n`);
|
|
}
|
|
if (["succeeded", "failed", "cancelled", "interrupted"].includes(run.state)) break;
|
|
await delay(500, undefined, { signal: controller.signal }).catch(() => undefined);
|
|
}
|
|
reply.raw.end();
|
|
return reply;
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
|
|
app.get("/catalog/sync-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).after;
|
|
if (!(await deps.repository.getSyncRun(runId))) {
|
|
return reply.code(404).send({ code: "sync_run_not_found", message: "Synchronization run was not found." });
|
|
}
|
|
return await deps.repository.listSyncEvents(runId, after);
|
|
} catch (error) { return safeError(reply, error); }
|
|
});
|
|
}
|