Compare commits
5 Commits
13936db100
...
291f278d99
| Author | SHA1 | Date | |
|---|---|---|---|
| 291f278d99 | |||
| e2fdfe9c0a | |||
| a2c5ad5dd9 | |||
| fd330b28c2 | |||
| f691e3db33 |
@@ -1,7 +1,7 @@
|
||||
import { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
|
||||
import { and, eq, inArray, isNull } from "drizzle-orm";
|
||||
import multipart from "@fastify/multipart";
|
||||
import { GetObjectCommand } from "@aws-sdk/client-s3";
|
||||
import { GetObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3";
|
||||
|
||||
import {
|
||||
authTenantUsers,
|
||||
@@ -22,12 +22,17 @@ import {
|
||||
createTenantFullExportArchive,
|
||||
importTenantFullExport,
|
||||
importTenantFullExportArchive,
|
||||
readTenantFullExportArchive,
|
||||
restoreTenantMergeArchiveFiles,
|
||||
restoreTenantMergeInlineFiles,
|
||||
} from "../utils/tenantFullExport";
|
||||
import type { TenantFullExport } from "../utils/tenantFullExport";
|
||||
import { buildSystemStatus } from "../modules/system-status.service";
|
||||
import { matrixService } from "../modules/matrix.service";
|
||||
import { s3 } from "../utils/s3";
|
||||
import { secrets } from "../utils/secrets";
|
||||
import { createTenantMergeDryRun, executeTenantMerge, sanitizeTenantMergePlan } from "../utils/tenantMergeService";
|
||||
import type { TenantMergePlan } from "../utils/tenantMergePlan";
|
||||
|
||||
export default async function adminRoutes(server: FastifyInstance) {
|
||||
await server.register(multipart, {
|
||||
@@ -340,6 +345,106 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
return job;
|
||||
};
|
||||
|
||||
const mergeArchivePath = (jobId: string) => `tenant-import-reviews/${jobId}/source`;
|
||||
const mergeReportPath = (jobId: string) => `tenant-import-reviews/${jobId}/report.json`;
|
||||
|
||||
const createTenantMergeReviewJob = async (
|
||||
tenantId: number,
|
||||
currentUserId: string,
|
||||
filename: string,
|
||||
contentType: string,
|
||||
source: Buffer
|
||||
) => {
|
||||
const [job] = await server.db
|
||||
.insert(tenantExportJobs)
|
||||
.values({
|
||||
tenantId,
|
||||
createdBy: currentUserId,
|
||||
operation: "merge",
|
||||
status: "running",
|
||||
filename,
|
||||
contentType,
|
||||
storagePath: "pending",
|
||||
fileSize: source.length,
|
||||
filesDone: 0,
|
||||
filesTotal: 1,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.returning();
|
||||
const storagePath = mergeArchivePath(job.id);
|
||||
|
||||
await s3.send(new PutObjectCommand({
|
||||
Bucket: secrets.S3_BUCKET,
|
||||
Key: storagePath,
|
||||
Body: source,
|
||||
ContentType: contentType,
|
||||
ContentLength: source.length,
|
||||
}));
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({ storagePath, updatedAt: new Date() })
|
||||
.where(eq(tenantExportJobs.id, job.id));
|
||||
|
||||
return { ...job, storagePath };
|
||||
};
|
||||
|
||||
const parseMergeSource = async (source: Buffer, contentType: string, filename: string) => {
|
||||
const isJson = contentType.includes("json") || filename.toLowerCase().endsWith(".json");
|
||||
if (isJson) return JSON.parse(source.toString("utf8")) as TenantFullExport;
|
||||
return await readTenantFullExportArchive(source);
|
||||
};
|
||||
|
||||
const startTenantMergeReviewJob = async (
|
||||
jobId: string,
|
||||
targetTenantId: number,
|
||||
source: Buffer,
|
||||
contentType: string,
|
||||
filename: string
|
||||
) => {
|
||||
try {
|
||||
const exportData = await parseMergeSource(source, contentType, filename);
|
||||
const report = sanitizeTenantMergePlan(await createTenantMergeDryRun(exportData, targetTenantId));
|
||||
const reportBuffer = Buffer.from(JSON.stringify(report), "utf8");
|
||||
|
||||
await s3.send(new PutObjectCommand({
|
||||
Bucket: secrets.S3_BUCKET,
|
||||
Key: mergeReportPath(jobId),
|
||||
Body: reportBuffer,
|
||||
ContentType: "application/json",
|
||||
ContentLength: reportBuffer.length,
|
||||
}));
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "review",
|
||||
filesDone: 1,
|
||||
filesTotal: 1,
|
||||
updatedAt: new Date(),
|
||||
error: null,
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, jobId));
|
||||
} catch (err: any) {
|
||||
server.log.error({ err, jobId }, "Tenant-Merge-Dry-Run fehlgeschlagen");
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "failed",
|
||||
error: err?.message || String(err),
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, jobId));
|
||||
}
|
||||
};
|
||||
|
||||
const readS3Buffer = async (key: string) => {
|
||||
const { Body } = await s3.send(new GetObjectCommand({ Bucket: secrets.S3_BUCKET, Key: key }));
|
||||
if (!Body) throw new Error("Gespeicherte Importdatei ist leer");
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of Body as any) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
return Buffer.concat(chunks);
|
||||
};
|
||||
|
||||
const completeImportedTenantAccess = async (
|
||||
currentUser: { id: string; email: string },
|
||||
result: { tenantId: number }
|
||||
@@ -1437,11 +1542,16 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
targetTenantId = Number(data.fields?.targetTenantId?.value || 0) || null;
|
||||
|
||||
if (targetTenantId) {
|
||||
const importJob = await createTenantImportJob(targetTenantId, currentUser.id, data.filename || "tenant-import.zip");
|
||||
void startTenantImportJob(importJob.id, targetTenantId, currentUser, {
|
||||
type: "archive",
|
||||
archiveBuffer,
|
||||
});
|
||||
const filename = data.filename || "tenant-import.zip";
|
||||
const contentType = data.mimetype || "application/zip";
|
||||
const importJob = await createTenantMergeReviewJob(
|
||||
targetTenantId,
|
||||
currentUser.id,
|
||||
filename,
|
||||
contentType,
|
||||
archiveBuffer
|
||||
);
|
||||
void startTenantMergeReviewJob(importJob.id, targetTenantId, archiveBuffer, contentType, filename);
|
||||
|
||||
return reply.code(202).send({
|
||||
importId: importJob.id,
|
||||
@@ -1450,6 +1560,7 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
status: importJob.status,
|
||||
filename: importJob.filename,
|
||||
statusUrl: `/api/admin/tenant-exports/${importJob.id}`,
|
||||
reviewUrl: `/administration/tenant-imports/${importJob.id}`,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1464,11 +1575,15 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
}
|
||||
|
||||
if (targetTenantId) {
|
||||
const importJob = await createTenantImportJob(targetTenantId, currentUser.id, "tenant-import.json");
|
||||
void startTenantImportJob(importJob.id, targetTenantId, currentUser, {
|
||||
type: "json",
|
||||
exportData,
|
||||
});
|
||||
const source = Buffer.from(JSON.stringify(exportData), "utf8");
|
||||
const importJob = await createTenantMergeReviewJob(
|
||||
targetTenantId,
|
||||
currentUser.id,
|
||||
"tenant-import.json",
|
||||
"application/json",
|
||||
source
|
||||
);
|
||||
void startTenantMergeReviewJob(importJob.id, targetTenantId, source, "application/json", "tenant-import.json");
|
||||
|
||||
return reply.code(202).send({
|
||||
importId: importJob.id,
|
||||
@@ -1477,6 +1592,7 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
status: importJob.status,
|
||||
filename: importJob.filename,
|
||||
statusUrl: `/api/admin/tenant-exports/${importJob.id}`,
|
||||
reviewUrl: `/administration/tenant-imports/${importJob.id}`,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1497,6 +1613,161 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
}
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------
|
||||
// GET /admin/tenant-imports/:import_id/review
|
||||
// -------------------------------------------------------------
|
||||
server.get("/admin/tenant-imports/:import_id/review", async (req, reply) => {
|
||||
try {
|
||||
const currentUser = await requireAdmin(req, reply);
|
||||
if (!currentUser) return;
|
||||
const { import_id } = req.params as { import_id: string };
|
||||
const query = req.query as { kind?: string; table?: string; offset?: string; limit?: string };
|
||||
const [job] = await server.db
|
||||
.select()
|
||||
.from(tenantExportJobs)
|
||||
.where(eq(tenantExportJobs.id, import_id))
|
||||
.limit(1);
|
||||
|
||||
if (!job || job.operation !== "merge") return reply.code(404).send({ error: "Merge-Import nicht gefunden" });
|
||||
if (job.status !== "review") {
|
||||
return reply.code(409).send({ error: "Dry-Run ist noch nicht verfügbar", status: job.status });
|
||||
}
|
||||
|
||||
const report = JSON.parse((await readS3Buffer(mergeReportPath(job.id))).toString("utf8")) as TenantMergePlan;
|
||||
const offset = Math.max(0, Number(query.offset || 0) || 0);
|
||||
const limit = Math.min(500, Math.max(1, Number(query.limit || 100) || 100));
|
||||
const filtered = report.items.filter((item) =>
|
||||
(!query.kind || item.kind === query.kind) && (!query.table || item.table === query.table)
|
||||
);
|
||||
|
||||
return {
|
||||
importId: job.id,
|
||||
tenantId: job.tenantId,
|
||||
status: job.status,
|
||||
filename: job.filename,
|
||||
summary: report.summary,
|
||||
tables: Array.from(new Set(report.items.map((item) => item.table))).sort(),
|
||||
total: filtered.length,
|
||||
offset,
|
||||
limit,
|
||||
items: filtered.slice(offset, offset + limit),
|
||||
};
|
||||
} catch (err: any) {
|
||||
console.error("ERROR /admin/tenant-imports/:import_id/review:", err);
|
||||
return reply.code(500).send({ error: err?.message || "Internal Server Error" });
|
||||
}
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------
|
||||
// POST /admin/tenant-imports/:import_id/execute
|
||||
// -------------------------------------------------------------
|
||||
server.post("/admin/tenant-imports/:import_id/execute", async (req, reply) => {
|
||||
const currentUser = await requireAdmin(req, reply);
|
||||
if (!currentUser) return;
|
||||
const { import_id } = req.params as { import_id: string };
|
||||
const body = req.body as { decisions?: Record<string, "source" | "target"> };
|
||||
const decisions = body?.decisions || {};
|
||||
const [job] = await server.db
|
||||
.select()
|
||||
.from(tenantExportJobs)
|
||||
.where(eq(tenantExportJobs.id, import_id))
|
||||
.limit(1);
|
||||
|
||||
if (!job || job.operation !== "merge") return reply.code(404).send({ error: "Merge-Import nicht gefunden" });
|
||||
if (job.status !== "review" || !job.storagePath) {
|
||||
return reply.code(409).send({ error: "Merge-Import ist nicht zur Ausführung bereit", status: job.status });
|
||||
}
|
||||
|
||||
let locked = false;
|
||||
try {
|
||||
const storedReport = JSON.parse((await readS3Buffer(mergeReportPath(job.id))).toString("utf8")) as TenantMergePlan;
|
||||
const unresolved = storedReport.items.filter((item) =>
|
||||
item.kind === "conflict" && !["source", "target"].includes(decisions[item.id])
|
||||
);
|
||||
if (unresolved.length) {
|
||||
return reply.code(400).send({
|
||||
error: `Für ${unresolved.length} Konflikte fehlt eine Entscheidung`,
|
||||
unresolved: unresolved.map((item) => item.id),
|
||||
});
|
||||
}
|
||||
|
||||
await lockTenantForJob(job.tenantId, job.id);
|
||||
locked = true;
|
||||
const source = await readS3Buffer(job.storagePath);
|
||||
const exportData = await parseMergeSource(source, job.contentType, job.filename);
|
||||
const currentReport = sanitizeTenantMergePlan(await createTenantMergeDryRun(exportData, job.tenantId));
|
||||
const comparable = (report: TenantMergePlan) => report.items.map((item) => ({
|
||||
id: item.id,
|
||||
kind: item.kind,
|
||||
targetRow: item.table === "tenants" && item.targetRow
|
||||
? Object.fromEntries(Object.entries(item.targetRow).filter(([column]) => ![
|
||||
"locked",
|
||||
"locked_by_export_job_id",
|
||||
"lockedByExportJobId",
|
||||
"updated_at",
|
||||
"updatedAt",
|
||||
].includes(column)))
|
||||
: item.targetRow,
|
||||
}));
|
||||
|
||||
if (JSON.stringify(comparable(currentReport)) !== JSON.stringify(comparable(storedReport))) {
|
||||
const reportBuffer = Buffer.from(JSON.stringify(currentReport), "utf8");
|
||||
await s3.send(new PutObjectCommand({
|
||||
Bucket: secrets.S3_BUCKET,
|
||||
Key: mergeReportPath(job.id),
|
||||
Body: reportBuffer,
|
||||
ContentType: "application/json",
|
||||
ContentLength: reportBuffer.length,
|
||||
}));
|
||||
return reply.code(409).send({
|
||||
error: "Der Ziel-Tenant hat sich seit dem Dry-Run geändert. Der Bericht wurde aktualisiert.",
|
||||
reviewRequired: true,
|
||||
});
|
||||
}
|
||||
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({ status: "running", error: null, updatedAt: new Date() })
|
||||
.where(eq(tenantExportJobs.id, job.id));
|
||||
|
||||
const result = await executeTenantMerge(exportData, job.tenantId, decisions);
|
||||
const selectedFileRefs = new Set<string>();
|
||||
for (const item of currentReport.items) {
|
||||
const decision = decisions[item.id] || item.defaultDecision;
|
||||
if (decision !== "source" || item.kind === "existing") continue;
|
||||
if (item.table === "files" && item.sourceRow.id) selectedFileRefs.add(String(item.sourceRow.id));
|
||||
if (item.table === "letterheads" && item.sourceRow.id) selectedFileRefs.add(`letterhead:${item.sourceRow.id}`);
|
||||
}
|
||||
const files = job.contentType.includes("json") || job.filename.toLowerCase().endsWith(".json")
|
||||
? await restoreTenantMergeInlineFiles(exportData, job.tenantId, selectedFileRefs)
|
||||
: await restoreTenantMergeArchiveFiles(source, exportData, job.tenantId, selectedFileRefs);
|
||||
|
||||
await completeImportedTenantAccess(currentUser, { tenantId: job.tenantId });
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "ready",
|
||||
filesDone: 1,
|
||||
filesTotal: 1,
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
error: null,
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, job.id));
|
||||
|
||||
return { success: true, importId: job.id, tenantId: job.tenantId, result, files };
|
||||
} catch (err: any) {
|
||||
server.log.error({ err, importId: job.id }, "Tenant-Merge-Ausführung fehlgeschlagen");
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({ status: "review", error: err?.message || String(err), updatedAt: new Date() })
|
||||
.where(eq(tenantExportJobs.id, job.id));
|
||||
return reply.code(500).send({ error: err?.message || "Merge-Import fehlgeschlagen" });
|
||||
} finally {
|
||||
if (locked) await unlockTenantForJob(job.tenantId, job.id);
|
||||
}
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------
|
||||
// PUT /admin/users/:user_id/access
|
||||
// -------------------------------------------------------------
|
||||
|
||||
@@ -712,6 +712,12 @@ const prepareCommunicationRoomsForImport = (exportData: TenantFullExport) => {
|
||||
}
|
||||
}
|
||||
|
||||
export const prepareTenantFullExportRowsForImport = (exportData: TenantFullExport) => {
|
||||
encryptEntityBankAccountRowsForImport(exportData)
|
||||
prepareCommunicationRoomsForImport(exportData)
|
||||
return exportData
|
||||
}
|
||||
|
||||
const cleanupImportedCommunicationRooms = async (client: any, exportData: TenantFullExport) => {
|
||||
const rows = exportData.tables.communication_rooms || []
|
||||
if (!rows.length) return 0
|
||||
@@ -900,8 +906,7 @@ export const importTenantFullExport = async (
|
||||
}
|
||||
|
||||
const exportData = remapTenantScopedExport(rawExportData, options.targetTenantId)
|
||||
encryptEntityBankAccountRowsForImport(exportData)
|
||||
prepareCommunicationRoomsForImport(exportData)
|
||||
prepareTenantFullExportRowsForImport(exportData)
|
||||
const client = await pool.connect()
|
||||
const importOrder = [
|
||||
"tenants",
|
||||
@@ -1025,41 +1030,14 @@ export const importTenantFullExportArchive = async (
|
||||
archiveBuffer: Buffer,
|
||||
options: ImportOptions = {}
|
||||
): Promise<ImportResult> => {
|
||||
const rawExportData = await readTenantFullExportArchive(archiveBuffer)
|
||||
const exportData = remapTenantScopedExport(rawExportData, options.targetTenantId)
|
||||
const reader = new ZipReader(new BlobReader(new Blob([archiveBuffer])))
|
||||
|
||||
try {
|
||||
const entries = await reader.getEntries()
|
||||
const entriesByName = new Map(entries.map((entry: any) => [entry.filename, entry]))
|
||||
const manifest = JSON.parse(await readZipTextEntry(entriesByName, "manifest.json")) as TenantArchiveManifest
|
||||
|
||||
if (manifest?.format !== "fedeo.tenant-archive-export" || manifest.version !== 1) {
|
||||
throw new Error("Ungültiges FEDEO Mandantenarchiv-Format")
|
||||
}
|
||||
|
||||
const tables: TableRows = {}
|
||||
for (const table of manifest.tables || []) {
|
||||
tables[table.name] = JSON.parse(await readZipTextEntry(entriesByName, table.path))
|
||||
}
|
||||
|
||||
const rawExportData: TenantFullExport = {
|
||||
format: "fedeo.tenant-full-export",
|
||||
version: 1,
|
||||
exportedAt: manifest.exportedAt,
|
||||
tenantId: manifest.tenantId,
|
||||
tables,
|
||||
files: (manifest.files || []).map((file) => ({
|
||||
id: file.id,
|
||||
path: file.path,
|
||||
name: file.name,
|
||||
mimeType: file.mimeType,
|
||||
size: file.size,
|
||||
contentBase64: null,
|
||||
missing: file.missing,
|
||||
error: file.error,
|
||||
})),
|
||||
}
|
||||
|
||||
const exportData = remapTenantScopedExport(rawExportData, options.targetTenantId)
|
||||
const sourceTenantId = rawExportData.tenantId
|
||||
const targetTenantId = exportData.tenantId
|
||||
const sourcePrefix = `${sourceTenantId}/`
|
||||
@@ -1086,3 +1064,93 @@ export const importTenantFullExportArchive = async (
|
||||
await reader.close()
|
||||
}
|
||||
}
|
||||
|
||||
export const readTenantFullExportArchive = async (archiveBuffer: Buffer): Promise<TenantFullExport> => {
|
||||
const reader = new ZipReader(new BlobReader(new Blob([archiveBuffer])))
|
||||
|
||||
try {
|
||||
const entries = await reader.getEntries()
|
||||
const entriesByName = new Map(entries.map((entry: any) => [entry.filename, entry]))
|
||||
const manifest = JSON.parse(await readZipTextEntry(entriesByName, "manifest.json")) as TenantArchiveManifest
|
||||
|
||||
if (manifest?.format !== "fedeo.tenant-archive-export" || manifest.version !== 1) {
|
||||
throw new Error("Ungültiges FEDEO Mandantenarchiv-Format")
|
||||
}
|
||||
|
||||
const tables: TableRows = {}
|
||||
for (const table of manifest.tables || []) {
|
||||
tables[table.name] = JSON.parse(await readZipTextEntry(entriesByName, table.path))
|
||||
}
|
||||
|
||||
return {
|
||||
format: "fedeo.tenant-full-export",
|
||||
version: 1,
|
||||
exportedAt: manifest.exportedAt,
|
||||
tenantId: manifest.tenantId,
|
||||
tables,
|
||||
files: (manifest.files || []).map((file) => ({
|
||||
id: file.id,
|
||||
path: file.path,
|
||||
name: file.name,
|
||||
mimeType: file.mimeType,
|
||||
size: file.size,
|
||||
contentBase64: null,
|
||||
missing: file.missing,
|
||||
error: file.error,
|
||||
})),
|
||||
}
|
||||
|
||||
} finally {
|
||||
await reader.close()
|
||||
}
|
||||
}
|
||||
|
||||
export const restoreTenantMergeArchiveFiles = async (
|
||||
archiveBuffer: Buffer,
|
||||
rawExportData: TenantFullExport,
|
||||
targetTenantId: number,
|
||||
selectedFileRefs: ReadonlySet<string>
|
||||
) => {
|
||||
const reader = new ZipReader(new BlobReader(new Blob([archiveBuffer])))
|
||||
try {
|
||||
const entries = await reader.getEntries()
|
||||
const entriesByName = new Map(entries.map((entry: any) => [entry.filename, entry]))
|
||||
const manifest = JSON.parse(await readZipTextEntry(entriesByName, "manifest.json")) as TenantArchiveManifest
|
||||
const exportData = remapTenantScopedExport(rawExportData, targetTenantId)
|
||||
const sourcePrefix = `${rawExportData.tenantId}/`
|
||||
const targetPrefix = `${targetTenantId}/`
|
||||
const filteredFiles = (manifest.files || [])
|
||||
.filter((file) => selectedFileRefs.has(String(file.id)))
|
||||
.map((file) => ({
|
||||
...file,
|
||||
path: file.path?.startsWith(sourcePrefix)
|
||||
? `${targetPrefix}${file.path.slice(sourcePrefix.length)}`
|
||||
: file.path,
|
||||
}))
|
||||
const filteredManifest: TenantArchiveManifest = {
|
||||
...manifest,
|
||||
tenantId: targetTenantId,
|
||||
files: filteredFiles,
|
||||
}
|
||||
const filteredExportData: TenantFullExport = {
|
||||
...exportData,
|
||||
files: exportData.files.filter((file) => selectedFileRefs.has(String(file.id))),
|
||||
}
|
||||
|
||||
return await restoreArchiveFiles(entriesByName, filteredExportData, filteredManifest)
|
||||
} finally {
|
||||
await reader.close()
|
||||
}
|
||||
}
|
||||
|
||||
export const restoreTenantMergeInlineFiles = async (
|
||||
rawExportData: TenantFullExport,
|
||||
targetTenantId: number,
|
||||
selectedFileRefs: ReadonlySet<string>
|
||||
) => {
|
||||
const exportData = remapTenantScopedExport(rawExportData, targetTenantId)
|
||||
return await restoreFiles({
|
||||
...exportData,
|
||||
files: exportData.files.filter((file) => selectedFileRefs.has(String(file.id))),
|
||||
})
|
||||
}
|
||||
|
||||
152
backend/src/utils/tenantMergePlan.ts
Normal file
152
backend/src/utils/tenantMergePlan.ts
Normal file
@@ -0,0 +1,152 @@
|
||||
import { createHash } from "crypto"
|
||||
|
||||
export type TenantMergeKind = "import" | "existing" | "conflict" | "id_collision"
|
||||
export type TenantMergeDecision = "source" | "target"
|
||||
|
||||
export type TenantMergeTableMetadata = {
|
||||
primaryKey: string[]
|
||||
}
|
||||
|
||||
export type TenantMergeItem = {
|
||||
id: string
|
||||
table: string
|
||||
kind: TenantMergeKind
|
||||
sourceKey: string
|
||||
targetKey: string | null
|
||||
label: string
|
||||
defaultDecision: TenantMergeDecision
|
||||
allowedDecisions: TenantMergeDecision[]
|
||||
differences: string[]
|
||||
sourceRow: Record<string, any>
|
||||
targetRow: Record<string, any> | null
|
||||
}
|
||||
|
||||
export type TenantMergePlan = {
|
||||
items: TenantMergeItem[]
|
||||
summary: Record<TenantMergeKind, number>
|
||||
}
|
||||
|
||||
const ignoredComparisonColumns = new Set([
|
||||
"created_at",
|
||||
"createdAt",
|
||||
"updated_at",
|
||||
"updatedAt",
|
||||
"updated_by",
|
||||
"updatedBy",
|
||||
"locked",
|
||||
"locked_by_export_job_id",
|
||||
"lockedByExportJobId",
|
||||
])
|
||||
|
||||
const normalizeText = (value: unknown) => String(value ?? "").trim().toLocaleLowerCase("de")
|
||||
|
||||
const naturalKeyColumns: Record<string, string[]> = {
|
||||
accounts: ["accountChart", "number"],
|
||||
units: ["name"],
|
||||
countrys: ["name"],
|
||||
citys: ["zip", "short", "districtCode"],
|
||||
}
|
||||
|
||||
const comparableRow = (row: Record<string, any>) => Object.fromEntries(
|
||||
Object.entries(row)
|
||||
.filter(([column]) => !ignoredComparisonColumns.has(column))
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
)
|
||||
|
||||
const rowDifferences = (source: Record<string, any>, target: Record<string, any>) => {
|
||||
const sourceComparable = comparableRow(source)
|
||||
const targetComparable = comparableRow(target)
|
||||
const columns = Array.from(new Set([
|
||||
...Object.keys(sourceComparable),
|
||||
...Object.keys(targetComparable),
|
||||
])).sort()
|
||||
|
||||
return columns.filter((column) =>
|
||||
JSON.stringify(sourceComparable[column]) !== JSON.stringify(targetComparable[column])
|
||||
)
|
||||
}
|
||||
|
||||
const rowKey = (row: Record<string, any>, columns: string[]) =>
|
||||
columns.map((column) => `${column}=${JSON.stringify(row[column] ?? null)}`).join("|")
|
||||
|
||||
const naturalKey = (table: string, row: Record<string, any>) => {
|
||||
const columns = naturalKeyColumns[table]
|
||||
if (!columns) return null
|
||||
|
||||
const values = columns.map((column) => row[column])
|
||||
if (values.every((value) => value === null || typeof value === "undefined" || value === "")) return null
|
||||
|
||||
return columns.map((column, index) => `${column}=${normalizeText(values[index])}`).join("|")
|
||||
}
|
||||
|
||||
const itemId = (table: string, sourceKey: string) =>
|
||||
createHash("sha256").update(`${table}\0${sourceKey}`).digest("hex").slice(0, 24)
|
||||
|
||||
const itemLabel = (table: string, row: Record<string, any>, fallback: string) => {
|
||||
const descriptive = row.name ?? row.label ?? row.title ?? row.number ?? row.email ?? row.filename
|
||||
return descriptive ? `${descriptive}` : `${table}: ${fallback}`
|
||||
}
|
||||
|
||||
export const buildTenantMergePlan = (
|
||||
sourceTables: Record<string, Record<string, any>[]>,
|
||||
targetTables: Record<string, Record<string, any>[]>,
|
||||
metadata: Record<string, TenantMergeTableMetadata>
|
||||
): TenantMergePlan => {
|
||||
const items: TenantMergeItem[] = []
|
||||
|
||||
for (const table of Object.keys(sourceTables).sort()) {
|
||||
const primaryKey = metadata[table]?.primaryKey?.length ? metadata[table].primaryKey : ["id"]
|
||||
const targetRows = targetTables[table] || []
|
||||
const targetByPrimaryKey = new Map(targetRows.map((row) => [rowKey(row, primaryKey), row]))
|
||||
const targetByNaturalKey = new Map<string, Record<string, any>>()
|
||||
|
||||
for (const targetRow of targetRows) {
|
||||
const key = naturalKey(table, targetRow)
|
||||
if (key) targetByNaturalKey.set(key, targetRow)
|
||||
}
|
||||
|
||||
for (const sourceRow of sourceTables[table] || []) {
|
||||
const sourceKey = rowKey(sourceRow, primaryKey)
|
||||
const sourceNaturalKey = naturalKey(table, sourceRow)
|
||||
const naturalTarget = sourceNaturalKey ? targetByNaturalKey.get(sourceNaturalKey) : undefined
|
||||
const primaryTarget = targetByPrimaryKey.get(sourceKey)
|
||||
const targetRow = naturalTarget || primaryTarget || null
|
||||
const differences = targetRow ? rowDifferences(sourceRow, targetRow) : []
|
||||
let kind: TenantMergeKind
|
||||
|
||||
if (!targetRow) {
|
||||
kind = "import"
|
||||
} else if (!differences.length) {
|
||||
kind = "existing"
|
||||
} else if (primaryTarget && sourceNaturalKey && naturalKey(table, primaryTarget) !== sourceNaturalKey) {
|
||||
kind = "id_collision"
|
||||
} else {
|
||||
kind = "conflict"
|
||||
}
|
||||
|
||||
items.push({
|
||||
id: itemId(table, sourceKey),
|
||||
table,
|
||||
kind,
|
||||
sourceKey,
|
||||
targetKey: targetRow ? rowKey(targetRow, primaryKey) : null,
|
||||
label: itemLabel(table, sourceRow, sourceKey),
|
||||
defaultDecision: kind === "import" || kind === "id_collision" ? "source" : "target",
|
||||
allowedDecisions: kind === "conflict" ? ["target", "source"] : [kind === "existing" ? "target" : "source"],
|
||||
differences,
|
||||
sourceRow,
|
||||
targetRow,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
items,
|
||||
summary: {
|
||||
import: items.filter((item) => item.kind === "import").length,
|
||||
existing: items.filter((item) => item.kind === "existing").length,
|
||||
conflict: items.filter((item) => item.kind === "conflict").length,
|
||||
id_collision: items.filter((item) => item.kind === "id_collision").length,
|
||||
},
|
||||
}
|
||||
}
|
||||
371
backend/src/utils/tenantMergeService.ts
Normal file
371
backend/src/utils/tenantMergeService.ts
Normal file
@@ -0,0 +1,371 @@
|
||||
import { pool } from "../../db"
|
||||
import { prepareTenantFullExportRowsForImport, type TenantFullExport } from "./tenantFullExport"
|
||||
import { buildTenantMergePlan, type TenantMergeDecision, type TenantMergePlan, type TenantMergeTableMetadata } from "./tenantMergePlan"
|
||||
|
||||
type MergeDatabaseMetadata = TenantMergeTableMetadata & {
|
||||
columns: string[]
|
||||
jsonColumns: Set<string>
|
||||
generatedColumns: Set<string>
|
||||
foreignKeys: { column: string, referencedTable: string, referencedColumn: string }[]
|
||||
}
|
||||
|
||||
const quoteIdent = (value: string) => `"${value.replace(/"/g, '""')}"`
|
||||
const globalNaturalKeyTables = new Set(["accounts", "units", "countrys", "citys"])
|
||||
|
||||
const loadMergeMetadata = async (client: any) => {
|
||||
const columnsResult = await client.query(`
|
||||
select table_name, column_name, data_type, is_generated
|
||||
from information_schema.columns
|
||||
where table_schema = 'public'
|
||||
order by table_name, ordinal_position
|
||||
`)
|
||||
const foreignKeyResult = await client.query(`
|
||||
select tc.table_name, kcu.column_name, ccu.table_name as referenced_table,
|
||||
ccu.column_name as referenced_column
|
||||
from information_schema.table_constraints tc
|
||||
join information_schema.key_column_usage kcu
|
||||
on tc.constraint_name = kcu.constraint_name
|
||||
and tc.constraint_schema = kcu.constraint_schema
|
||||
join information_schema.constraint_column_usage ccu
|
||||
on tc.constraint_name = ccu.constraint_name
|
||||
and tc.constraint_schema = ccu.constraint_schema
|
||||
where tc.table_schema = 'public' and tc.constraint_type = 'FOREIGN KEY'
|
||||
`)
|
||||
const primaryKeyResult = await client.query(`
|
||||
select tc.table_name, kcu.column_name
|
||||
from information_schema.table_constraints tc
|
||||
join information_schema.key_column_usage kcu
|
||||
on tc.constraint_name = kcu.constraint_name
|
||||
and tc.constraint_schema = kcu.constraint_schema
|
||||
where tc.table_schema = 'public' and tc.constraint_type = 'PRIMARY KEY'
|
||||
order by tc.table_name, kcu.ordinal_position
|
||||
`)
|
||||
const metadata: Record<string, MergeDatabaseMetadata> = {}
|
||||
|
||||
for (const row of columnsResult.rows) {
|
||||
metadata[row.table_name] ||= { columns: [], primaryKey: [], jsonColumns: new Set(), generatedColumns: new Set(), foreignKeys: [] }
|
||||
metadata[row.table_name].columns.push(row.column_name)
|
||||
if (row.data_type === "json" || row.data_type === "jsonb") metadata[row.table_name].jsonColumns.add(row.column_name)
|
||||
if (row.is_generated === "ALWAYS") metadata[row.table_name].generatedColumns.add(row.column_name)
|
||||
}
|
||||
for (const row of primaryKeyResult.rows) {
|
||||
metadata[row.table_name] ||= { columns: [], primaryKey: [], jsonColumns: new Set(), generatedColumns: new Set(), foreignKeys: [] }
|
||||
metadata[row.table_name].primaryKey.push(row.column_name)
|
||||
}
|
||||
for (const row of foreignKeyResult.rows) {
|
||||
metadata[row.table_name] ||= { columns: [], primaryKey: [], jsonColumns: new Set(), generatedColumns: new Set(), foreignKeys: [] }
|
||||
metadata[row.table_name].foreignKeys.push({
|
||||
column: row.column_name,
|
||||
referencedTable: row.referenced_table,
|
||||
referencedColumn: row.referenced_column,
|
||||
})
|
||||
}
|
||||
|
||||
return metadata
|
||||
}
|
||||
|
||||
const remapSourceTenant = (exportData: TenantFullExport, targetTenantId: number) => {
|
||||
const tables: Record<string, Record<string, any>[]> = {}
|
||||
|
||||
for (const [table, rows] of Object.entries(exportData.tables || {})) {
|
||||
tables[table] = rows.map((row) => {
|
||||
const next = { ...row }
|
||||
if (table === "tenants" && Number(next.id) === Number(exportData.tenantId)) next.id = targetTenantId
|
||||
if (Number(next.tenant) === Number(exportData.tenantId)) next.tenant = targetTenantId
|
||||
if (Number(next.tenant_id) === Number(exportData.tenantId)) next.tenant_id = targetTenantId
|
||||
return next
|
||||
})
|
||||
}
|
||||
|
||||
return tables
|
||||
}
|
||||
|
||||
const loadTargetRows = async (
|
||||
client: any,
|
||||
table: string,
|
||||
sourceRows: Record<string, any>[],
|
||||
metadata: MergeDatabaseMetadata,
|
||||
targetTenantId: number
|
||||
) => {
|
||||
if (!sourceRows.length) return []
|
||||
if (table === "tenants") {
|
||||
return (await client.query(`select * from ${quoteIdent(table)} where ${quoteIdent("id")} = $1`, [targetTenantId])).rows
|
||||
}
|
||||
|
||||
const tenantColumn = metadata.columns.includes("tenant")
|
||||
? "tenant"
|
||||
: metadata.columns.includes("tenant_id") ? "tenant_id" : null
|
||||
if (tenantColumn) {
|
||||
return (await client.query(
|
||||
`select * from ${quoteIdent(table)} where ${quoteIdent(tenantColumn)} = $1`,
|
||||
[targetTenantId]
|
||||
)).rows
|
||||
}
|
||||
if (globalNaturalKeyTables.has(table)) {
|
||||
return (await client.query(`select * from ${quoteIdent(table)}`)).rows
|
||||
}
|
||||
|
||||
if (metadata.primaryKey.length === 1) {
|
||||
const key = metadata.primaryKey[0]
|
||||
const values = Array.from(new Set(sourceRows.map((row) => row[key]).filter((value) => value !== null && typeof value !== "undefined").map(String)))
|
||||
if (!values.length) return []
|
||||
return (await client.query(
|
||||
`select * from ${quoteIdent(table)} where ${quoteIdent(key)}::text = any($1::text[])`,
|
||||
[values]
|
||||
)).rows
|
||||
}
|
||||
|
||||
return (await client.query(`select * from ${quoteIdent(table)}`)).rows
|
||||
}
|
||||
|
||||
export const createTenantMergeDryRunWithClient = async (
|
||||
client: any,
|
||||
exportData: TenantFullExport,
|
||||
targetTenantId: number
|
||||
): Promise<TenantMergePlan> => {
|
||||
const metadata = await loadMergeMetadata(client)
|
||||
const sourceTables = remapSourceTenant(exportData, targetTenantId)
|
||||
const targetTables: Record<string, Record<string, any>[]> = {}
|
||||
|
||||
for (const [table, sourceRows] of Object.entries(sourceTables)) {
|
||||
const tableMetadata = metadata[table]
|
||||
if (!tableMetadata) continue
|
||||
targetTables[table] = await loadTargetRows(client, table, sourceRows, tableMetadata, targetTenantId)
|
||||
}
|
||||
|
||||
return buildTenantMergePlan(sourceTables, targetTables, metadata)
|
||||
}
|
||||
|
||||
export const createTenantMergeDryRun = async (
|
||||
exportData: TenantFullExport,
|
||||
targetTenantId: number
|
||||
) => {
|
||||
const client = await pool.connect()
|
||||
try {
|
||||
return await createTenantMergeDryRunWithClient(client, exportData, targetTenantId)
|
||||
} finally {
|
||||
client.release()
|
||||
}
|
||||
}
|
||||
|
||||
const sensitiveColumns = new Set([
|
||||
"iban_encrypted",
|
||||
"bic_encrypted",
|
||||
"bank_name_encrypted",
|
||||
"__plainIban",
|
||||
"__plainBic",
|
||||
"__plainBankName",
|
||||
"password_hash",
|
||||
"passwordHash",
|
||||
"token_hash",
|
||||
"tokenHash",
|
||||
])
|
||||
|
||||
const redactRow = (row: Record<string, any> | null) => row && Object.fromEntries(
|
||||
Object.entries(row).map(([column, value]) => [column, sensitiveColumns.has(column) && value ? "***" : value])
|
||||
)
|
||||
|
||||
export const sanitizeTenantMergePlan = (plan: TenantMergePlan): TenantMergePlan => ({
|
||||
...plan,
|
||||
items: plan.items.map((item) => ({
|
||||
...item,
|
||||
sourceRow: redactRow(item.sourceRow) || {},
|
||||
targetRow: redactRow(item.targetRow),
|
||||
})),
|
||||
})
|
||||
|
||||
const prepareValue = (value: any, isJson: boolean) => {
|
||||
if (!isJson || value === null || typeof value === "undefined" || typeof value === "string") return value
|
||||
return JSON.stringify(value)
|
||||
}
|
||||
|
||||
const rowIdentity = (row: Record<string, any>, columns: string[]) =>
|
||||
columns.map((column) => row[column]).join("\0")
|
||||
|
||||
const topologicalTableOrder = (tables: string[], metadata: Record<string, MergeDatabaseMetadata>) => {
|
||||
const remaining = new Set(tables)
|
||||
const ordered: string[] = []
|
||||
while (remaining.size) {
|
||||
const ready = Array.from(remaining).filter((table) =>
|
||||
metadata[table].foreignKeys.every((foreignKey) => !remaining.has(foreignKey.referencedTable) || foreignKey.referencedTable === table)
|
||||
)
|
||||
const next = ready.length ? ready.sort() : [Array.from(remaining).sort()[0]]
|
||||
for (const table of next) {
|
||||
remaining.delete(table)
|
||||
ordered.push(table)
|
||||
}
|
||||
}
|
||||
return ordered
|
||||
}
|
||||
|
||||
export type TenantMergeExecutionResult = {
|
||||
imported: number
|
||||
updated: number
|
||||
retained: number
|
||||
remappedIds: number
|
||||
tables: Record<string, { imported: number, updated: number, retained: number }>
|
||||
}
|
||||
|
||||
export const executeTenantMergeWithClient = async (
|
||||
client: any,
|
||||
rawExportData: TenantFullExport,
|
||||
targetTenantId: number,
|
||||
decisions: Record<string, TenantMergeDecision>
|
||||
): Promise<TenantMergeExecutionResult> => {
|
||||
const metadata = await loadMergeMetadata(client)
|
||||
const sourceTables = remapSourceTenant(rawExportData, targetTenantId)
|
||||
const preparedExport: TenantFullExport = prepareTenantFullExportRowsForImport({
|
||||
...rawExportData,
|
||||
tenantId: targetTenantId,
|
||||
tables: Object.fromEntries(Object.entries(sourceTables).map(([table, rows]) => [table, rows.map((row) => ({ ...row }))])),
|
||||
})
|
||||
const targetTables: Record<string, Record<string, any>[]> = {}
|
||||
for (const [table, rows] of Object.entries(preparedExport.tables)) {
|
||||
if (metadata[table]) targetTables[table] = await loadTargetRows(client, table, rows, metadata[table], targetTenantId)
|
||||
}
|
||||
const plan = buildTenantMergePlan(preparedExport.tables, targetTables, metadata)
|
||||
const idMaps = new Map<string, Map<any, any>>()
|
||||
const selected = plan.items.filter((item) => {
|
||||
const decision = decisions[item.id] || item.defaultDecision
|
||||
return decision === "source" && item.kind !== "existing"
|
||||
})
|
||||
|
||||
for (const item of plan.items) {
|
||||
const tableMetadata = metadata[item.table]
|
||||
if (tableMetadata?.primaryKey.length !== 1 || !item.targetRow) continue
|
||||
const key = tableMetadata.primaryKey[0]
|
||||
const sourceId = item.sourceRow[key]
|
||||
const targetId = item.targetRow[key]
|
||||
if (sourceId !== null && typeof sourceId !== "undefined" && targetId !== null && typeof targetId !== "undefined") {
|
||||
if (!idMaps.has(item.table)) idMaps.set(item.table, new Map())
|
||||
idMaps.get(item.table)!.set(sourceId, targetId)
|
||||
}
|
||||
}
|
||||
|
||||
await client.query("begin")
|
||||
await client.query("set local session_replication_role = replica")
|
||||
try {
|
||||
for (const item of selected.filter((entry) => entry.kind === "id_collision")) {
|
||||
const tableMetadata = metadata[item.table]
|
||||
if (tableMetadata.primaryKey.length !== 1) throw new Error(`ID-Kollision in ${item.table} kann nicht automatisch aufgelöst werden`)
|
||||
const key = tableMetadata.primaryKey[0]
|
||||
const sequenceResult = await client.query("select pg_get_serial_sequence($1, $2) as sequence_name", [`public.${item.table}`, key])
|
||||
const sequenceName = sequenceResult.rows[0]?.sequence_name
|
||||
if (!sequenceName) throw new Error(`Keine Sequenz für ID-Kollision in ${item.table}.${key} gefunden`)
|
||||
const allocated = await client.query("select nextval($1::regclass) as id", [sequenceName])
|
||||
const sourceId = item.sourceRow[key]
|
||||
const targetId = allocated.rows[0].id
|
||||
item.sourceRow[key] = targetId
|
||||
if (!idMaps.has(item.table)) idMaps.set(item.table, new Map())
|
||||
idMaps.get(item.table)!.set(sourceId, targetId)
|
||||
}
|
||||
|
||||
for (const item of selected) {
|
||||
const tableMetadata = metadata[item.table]
|
||||
for (const foreignKey of tableMetadata.foreignKeys) {
|
||||
const mapping = idMaps.get(foreignKey.referencedTable)
|
||||
if (mapping?.has(item.sourceRow[foreignKey.column])) {
|
||||
item.sourceRow[foreignKey.column] = mapping.get(item.sourceRow[foreignKey.column])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const selectedByTable = new Map<string, typeof selected>()
|
||||
for (const item of selected) {
|
||||
const rows = selectedByTable.get(item.table) || []
|
||||
rows.push(item)
|
||||
selectedByTable.set(item.table, rows)
|
||||
}
|
||||
const result: TenantMergeExecutionResult = {
|
||||
imported: 0,
|
||||
updated: 0,
|
||||
retained: plan.items.length - selected.length,
|
||||
remappedIds: Array.from(idMaps.values()).reduce((sum, map) => sum + Array.from(map).filter(([source, target]) => source !== target).length, 0),
|
||||
tables: {},
|
||||
}
|
||||
|
||||
for (const table of topologicalTableOrder(Array.from(selectedByTable.keys()), metadata)) {
|
||||
const tableMetadata = metadata[table]
|
||||
result.tables[table] ||= { imported: 0, updated: 0, retained: plan.items.filter((item) => item.table === table && !selected.includes(item)).length }
|
||||
for (const item of selectedByTable.get(table) || []) {
|
||||
const row = { ...item.sourceRow }
|
||||
if (table === "tenants") {
|
||||
delete row.locked
|
||||
delete row.locked_by_export_job_id
|
||||
delete row.lockedByExportJobId
|
||||
}
|
||||
const columns = Object.keys(row).filter((column) =>
|
||||
tableMetadata.columns.includes(column) && !tableMetadata.generatedColumns.has(column)
|
||||
)
|
||||
const values = columns.map((column) => prepareValue(row[column], tableMetadata.jsonColumns.has(column)))
|
||||
const placeholders = columns.map((_, index) => `$${index + 1}`).join(", ")
|
||||
|
||||
if (item.kind === "conflict" && item.targetRow) {
|
||||
const primaryKey = tableMetadata.primaryKey
|
||||
const updateColumns = columns.filter((column) => !primaryKey.includes(column))
|
||||
if (!primaryKey.length || !updateColumns.length) continue
|
||||
const whereValues = primaryKey.map((column) => item.targetRow![column])
|
||||
const assignments = updateColumns.map((column) => `${quoteIdent(column)} = $${columns.indexOf(column) + 1}`).join(", ")
|
||||
const where = primaryKey.map((column, index) => `${quoteIdent(column)} = $${columns.length + index + 1}`).join(" and ")
|
||||
await client.query(`update ${quoteIdent(table)} set ${assignments} where ${where}`, [...values, ...whereValues])
|
||||
result.updated += 1
|
||||
result.tables[table].updated += 1
|
||||
} else {
|
||||
const inserted = await client.query(
|
||||
`insert into ${quoteIdent(table)} (${columns.map(quoteIdent).join(", ")}) values (${placeholders}) on conflict do nothing`,
|
||||
values
|
||||
)
|
||||
if (!inserted.rowCount) throw new Error(`Datensatz in ${table} konnte wegen eines neuen Konflikts nicht importiert werden`)
|
||||
result.imported += 1
|
||||
result.tables[table].imported += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const sourceTenant = preparedExport.tables.tenants?.find((row) => Number(row.id) === targetTenantId)
|
||||
if (sourceTenant?.numberRanges) {
|
||||
const targetTenantResult = await client.query(`select "numberRanges" from "tenants" where "id" = $1`, [targetTenantId])
|
||||
const targetRanges = targetTenantResult.rows[0]?.numberRanges || {}
|
||||
const mergedRanges = { ...targetRanges }
|
||||
for (const [key, sourceRange] of Object.entries(sourceTenant.numberRanges as Record<string, any>)) {
|
||||
const targetRange = targetRanges[key]
|
||||
mergedRanges[key] = targetRange
|
||||
? {
|
||||
...sourceRange,
|
||||
...targetRange,
|
||||
nextNumber: Math.max(Number(sourceRange?.nextNumber || 0), Number(targetRange?.nextNumber || 0)),
|
||||
}
|
||||
: sourceRange
|
||||
}
|
||||
await client.query(`update "tenants" set "numberRanges" = $1::jsonb where "id" = $2`, [JSON.stringify(mergedRanges), targetTenantId])
|
||||
}
|
||||
|
||||
for (const table of selectedByTable.keys()) {
|
||||
const tableMetadata = metadata[table]
|
||||
if (!tableMetadata.columns.includes("id")) continue
|
||||
const sequenceResult = await client.query("select pg_get_serial_sequence($1, $2) as sequence_name", [`public.${table}`, "id"])
|
||||
const sequenceName = sequenceResult.rows[0]?.sequence_name
|
||||
if (!sequenceName) continue
|
||||
await client.query(`select setval($1::regclass, greatest(coalesce((select max(id) from ${quoteIdent(table)}), 1), 1), true)`, [sequenceName])
|
||||
}
|
||||
|
||||
await client.query("commit")
|
||||
return result
|
||||
} catch (err) {
|
||||
await client.query("rollback")
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
export const executeTenantMerge = async (
|
||||
exportData: TenantFullExport,
|
||||
targetTenantId: number,
|
||||
decisions: Record<string, TenantMergeDecision>
|
||||
) => {
|
||||
const client = await pool.connect()
|
||||
try {
|
||||
return await executeTenantMergeWithClient(client, exportData, targetTenantId, decisions)
|
||||
} finally {
|
||||
client.release()
|
||||
}
|
||||
}
|
||||
68
backend/tests/tenantMergePlan.test.ts
Normal file
68
backend/tests/tenantMergePlan.test.ts
Normal file
@@ -0,0 +1,68 @@
|
||||
import assert from "node:assert/strict"
|
||||
import test from "node:test"
|
||||
|
||||
import { buildTenantMergePlan } from "../src/utils/tenantMergePlan"
|
||||
|
||||
test("classifies imports, existing rows, conflicts and global id collisions", () => {
|
||||
const plan = buildTenantMergePlan({
|
||||
units: [
|
||||
{ id: 1, name: "Stück", short: "Stk." },
|
||||
{ id: 2, name: "Monat", short: "Mon." },
|
||||
{ id: 3, name: "Stunde", short: "Std." },
|
||||
],
|
||||
customers: [
|
||||
{ id: 10, tenant: 42, name: "Neu" },
|
||||
{ id: 11, tenant: 42, name: "Geändert" },
|
||||
],
|
||||
}, {
|
||||
units: [
|
||||
{ id: 1, name: "Stück", short: "Stk.", updated_at: "later" },
|
||||
{ id: 2, name: "Kilometer", short: "km" },
|
||||
{ id: 9, name: "Stunde", short: "h" },
|
||||
],
|
||||
customers: [
|
||||
{ id: 11, tenant: 42, name: "Zieländerung" },
|
||||
],
|
||||
}, {
|
||||
units: { primaryKey: ["id"] },
|
||||
customers: { primaryKey: ["id"] },
|
||||
})
|
||||
|
||||
assert.deepEqual(plan.summary, {
|
||||
import: 1,
|
||||
existing: 1,
|
||||
conflict: 2,
|
||||
id_collision: 1,
|
||||
})
|
||||
|
||||
assert.equal(plan.items.find((item) => item.label === "Monat")?.kind, "id_collision")
|
||||
assert.equal(plan.items.find((item) => item.label === "Stunde")?.kind, "conflict")
|
||||
assert.equal(plan.items.find((item) => item.label === "Geändert")?.defaultDecision, "target")
|
||||
})
|
||||
|
||||
test("uses target as the safe default for two-way conflicts", () => {
|
||||
const plan = buildTenantMergePlan({
|
||||
accounts: [{ id: 5, accountChart: "skr03", number: "8400", label: "Quelle" }],
|
||||
}, {
|
||||
accounts: [{ id: 99, accountChart: "skr03", number: "8400", label: "Ziel" }],
|
||||
}, {
|
||||
accounts: { primaryKey: ["id"] },
|
||||
})
|
||||
|
||||
assert.equal(plan.items[0].kind, "conflict")
|
||||
assert.equal(plan.items[0].targetKey, "id=99")
|
||||
assert.equal(plan.items[0].defaultDecision, "target")
|
||||
assert.deepEqual(plan.items[0].differences.sort(), ["id", "label"])
|
||||
})
|
||||
|
||||
test("ignores maintenance locks and audit timestamps during comparison", () => {
|
||||
const plan = buildTenantMergePlan({
|
||||
tenants: [{ id: 42, name: "Tenant", locked: null, updatedAt: "before" }],
|
||||
}, {
|
||||
tenants: [{ id: 42, name: "Tenant", locked: "maintenance_tenant", updatedAt: "after" }],
|
||||
}, {
|
||||
tenants: { primaryKey: ["id"] },
|
||||
})
|
||||
|
||||
assert.equal(plan.items[0].kind, "existing")
|
||||
})
|
||||
@@ -61,6 +61,35 @@ export type TenantImportResult = {
|
||||
filesDone?: number
|
||||
filesTotal?: number
|
||||
error?: string | null
|
||||
reviewUrl?: string
|
||||
}
|
||||
|
||||
export type TenantMergeKind = "import" | "existing" | "conflict" | "id_collision"
|
||||
export type TenantMergeDecision = "source" | "target"
|
||||
export type TenantMergeItem = {
|
||||
id: string
|
||||
table: string
|
||||
kind: TenantMergeKind
|
||||
label: string
|
||||
sourceKey: string
|
||||
targetKey: string | null
|
||||
defaultDecision: TenantMergeDecision
|
||||
allowedDecisions: TenantMergeDecision[]
|
||||
differences: string[]
|
||||
sourceRow: Record<string, any>
|
||||
targetRow: Record<string, any> | null
|
||||
}
|
||||
export type TenantMergeReview = {
|
||||
importId: string
|
||||
tenantId: number
|
||||
status: string
|
||||
filename: string
|
||||
summary: Record<TenantMergeKind, number>
|
||||
tables: string[]
|
||||
total: number
|
||||
offset: number
|
||||
limit: number
|
||||
items: TenantMergeItem[]
|
||||
}
|
||||
|
||||
export type TenantExportJob = {
|
||||
@@ -222,6 +251,20 @@ export const useAdmin = () => {
|
||||
})
|
||||
}
|
||||
|
||||
const getTenantImportReview = async (
|
||||
importId: string,
|
||||
query: { kind?: string; table?: string; offset?: number; limit?: number } = {},
|
||||
): Promise<TenantMergeReview> => {
|
||||
return await $api(`/api/admin/tenant-imports/${importId}/review`, { query })
|
||||
}
|
||||
|
||||
const executeTenantImportMerge = async (importId: string, decisions: Record<string, TenantMergeDecision>) => {
|
||||
return await $api(`/api/admin/tenant-imports/${importId}/execute`, {
|
||||
method: "POST",
|
||||
body: { decisions },
|
||||
})
|
||||
}
|
||||
|
||||
const getSystemStatus = async (): Promise<SystemStatus> => {
|
||||
return await $api("/api/admin/system-status")
|
||||
}
|
||||
@@ -262,5 +305,7 @@ export const useAdmin = () => {
|
||||
getTenantExport,
|
||||
downloadTenantExport,
|
||||
importTenant,
|
||||
getTenantImportReview,
|
||||
executeTenantImportMerge,
|
||||
}
|
||||
}
|
||||
|
||||
248
frontend/pages/administration/tenant-imports/[importId].vue
Normal file
248
frontend/pages/administration/tenant-imports/[importId].vue
Normal file
@@ -0,0 +1,248 @@
|
||||
<script setup lang="ts">
|
||||
import type { TenantMergeDecision, TenantMergeItem, TenantMergeKind, TenantMergeReview } from "~/composables/useAdmin"
|
||||
|
||||
const route = useRoute()
|
||||
const router = useRouter()
|
||||
const toast = useToast()
|
||||
const admin = useAdmin()
|
||||
const importId = String(route.params.importId)
|
||||
|
||||
const loading = ref(true)
|
||||
const executing = ref(false)
|
||||
const executionResult = ref<any>(null)
|
||||
const review = ref<TenantMergeReview | null>(null)
|
||||
const tenantId = computed(() => review.value?.tenantId || 0)
|
||||
const decisions = reactive<Record<string, TenantMergeDecision>>({})
|
||||
const conflictItems = ref<TenantMergeItem[]>([])
|
||||
const kindFilter = ref<string>("")
|
||||
const tableFilter = ref<string>("")
|
||||
const offset = ref(0)
|
||||
const limit = 100
|
||||
|
||||
const kindOptions = [
|
||||
{ label: "Alle Arten", value: "" },
|
||||
{ label: "Wird importiert", value: "import" },
|
||||
{ label: "Bereits vorhanden", value: "existing" },
|
||||
{ label: "Konflikt", value: "conflict" },
|
||||
{ label: "ID wird remappt", value: "id_collision" },
|
||||
]
|
||||
const kindLabels: Record<TenantMergeKind, string> = {
|
||||
import: "Wird importiert",
|
||||
existing: "Bereits vorhanden",
|
||||
conflict: "Konflikt",
|
||||
id_collision: "ID wird remappt",
|
||||
}
|
||||
const kindColors: Record<TenantMergeKind, string> = {
|
||||
import: "success",
|
||||
existing: "neutral",
|
||||
conflict: "warning",
|
||||
id_collision: "info",
|
||||
}
|
||||
|
||||
const unresolvedConflicts = computed(() => conflictItems.value.filter((item) => !decisions[item.id]).length)
|
||||
const hasNextPage = computed(() => Boolean(review.value && review.value.offset + review.value.items.length < review.value.total))
|
||||
|
||||
const loadConflictDecisions = async () => {
|
||||
conflictItems.value = []
|
||||
let currentOffset = 0
|
||||
while (true) {
|
||||
const page = await admin.getTenantImportReview(importId, { kind: "conflict", offset: currentOffset, limit: 500 })
|
||||
conflictItems.value.push(...page.items)
|
||||
currentOffset += page.items.length
|
||||
if (!page.items.length || currentOffset >= page.total) break
|
||||
}
|
||||
}
|
||||
|
||||
const loadReview = async () => {
|
||||
loading.value = true
|
||||
try {
|
||||
review.value = await admin.getTenantImportReview(importId, {
|
||||
kind: kindFilter.value || undefined,
|
||||
table: tableFilter.value || undefined,
|
||||
offset: offset.value,
|
||||
limit,
|
||||
})
|
||||
} catch (err: any) {
|
||||
toast.add({
|
||||
title: "Merge-Bericht konnte nicht geladen werden",
|
||||
description: err?.data?.error || err?.message,
|
||||
color: "red",
|
||||
})
|
||||
} finally {
|
||||
loading.value = false
|
||||
}
|
||||
}
|
||||
|
||||
const resetAndLoad = async () => {
|
||||
offset.value = 0
|
||||
await loadReview()
|
||||
}
|
||||
|
||||
const setAllConflicts = (decision: TenantMergeDecision) => {
|
||||
for (const item of conflictItems.value) decisions[item.id] = decision
|
||||
}
|
||||
|
||||
const executeMerge = async () => {
|
||||
if (unresolvedConflicts.value || executing.value) return
|
||||
executing.value = true
|
||||
try {
|
||||
const result: any = await admin.executeTenantImportMerge(importId, decisions)
|
||||
executionResult.value = result
|
||||
toast.add({
|
||||
title: "Tenant-Merge abgeschlossen",
|
||||
description: `${result?.result?.imported || 0} importiert, ${result?.result?.updated || 0} aus der Quelle übernommen.`,
|
||||
color: "green",
|
||||
})
|
||||
} catch (err: any) {
|
||||
toast.add({
|
||||
title: err?.data?.reviewRequired ? "Dry-Run wurde aktualisiert" : "Tenant-Merge fehlgeschlagen",
|
||||
description: err?.data?.error || err?.message,
|
||||
color: "red",
|
||||
})
|
||||
if (err?.data?.reviewRequired) {
|
||||
Object.keys(decisions).forEach((key) => delete decisions[key])
|
||||
await loadConflictDecisions()
|
||||
await loadReview()
|
||||
}
|
||||
} finally {
|
||||
executing.value = false
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(async () => {
|
||||
await Promise.all([loadReview(), loadConflictDecisions()])
|
||||
})
|
||||
</script>
|
||||
|
||||
<template>
|
||||
<div class="space-y-4">
|
||||
<div class="flex flex-wrap items-center justify-between gap-3">
|
||||
<div>
|
||||
<h1 class="text-2xl font-semibold">Tenant-Import prüfen</h1>
|
||||
<p class="text-sm text-gray-500">Dry-Run {{ importId }}</p>
|
||||
</div>
|
||||
<UButton color="neutral" variant="soft" icon="i-heroicons-arrow-left" @click="router.push(`/administration/tenants/${tenantId}`)">
|
||||
Zurück zum Tenant
|
||||
</UButton>
|
||||
</div>
|
||||
|
||||
<UAlert
|
||||
title="Noch keine Daten wurden verändert"
|
||||
description="Importierte und bereits vorhandene Datensätze werden automatisch behandelt. Für jeden Konflikt ist vor der Ausführung eine Entscheidung erforderlich."
|
||||
color="info"
|
||||
variant="soft"
|
||||
/>
|
||||
|
||||
<UAlert
|
||||
v-if="executionResult"
|
||||
title="Tenant-Merge abgeschlossen"
|
||||
:description="`${executionResult.result.imported} Datensätze importiert, ${executionResult.result.updated} aus der Quelle übernommen, ${executionResult.result.retained} im Ziel beibehalten und ${executionResult.result.remappedIds} IDs remappt. Dateien: ${executionResult.files.restored} wiederhergestellt, ${executionResult.files.skipped} übersprungen.`"
|
||||
color="success"
|
||||
variant="soft"
|
||||
/>
|
||||
|
||||
<UCard v-if="executionResult">
|
||||
<h2 class="mb-3 text-lg font-semibold">Ausführung nach Tabelle</h2>
|
||||
<div class="overflow-x-auto">
|
||||
<table class="w-full text-left text-sm">
|
||||
<thead><tr><th class="py-2">Tabelle</th><th>Importiert</th><th>Quelle übernommen</th><th>Ziel behalten</th></tr></thead>
|
||||
<tbody>
|
||||
<tr v-for="(counts, table) in executionResult.result.tables" :key="table" class="border-t">
|
||||
<td class="py-2 font-medium">{{ table }}</td>
|
||||
<td>{{ counts.imported }}</td>
|
||||
<td>{{ counts.updated }}</td>
|
||||
<td>{{ counts.retained }}</td>
|
||||
</tr>
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</UCard>
|
||||
|
||||
<div v-if="review" class="grid gap-3 sm:grid-cols-2 lg:grid-cols-4">
|
||||
<UCard><p class="text-sm text-gray-500">Wird importiert</p><p class="text-2xl font-semibold">{{ review.summary.import }}</p></UCard>
|
||||
<UCard><p class="text-sm text-gray-500">Vorhanden</p><p class="text-2xl font-semibold">{{ review.summary.existing }}</p></UCard>
|
||||
<UCard><p class="text-sm text-gray-500">ID-Remapping</p><p class="text-2xl font-semibold">{{ review.summary.id_collision }}</p></UCard>
|
||||
<UCard><p class="text-sm text-gray-500">Konflikte</p><p class="text-2xl font-semibold">{{ review.summary.conflict }}</p></UCard>
|
||||
</div>
|
||||
|
||||
<UCard>
|
||||
<div class="flex flex-wrap items-end gap-3">
|
||||
<UFormField label="Art" class="min-w-48">
|
||||
<USelect v-model="kindFilter" :items="kindOptions" value-key="value" label-key="label" @update:model-value="resetAndLoad" />
|
||||
</UFormField>
|
||||
<UFormField label="Tabelle" class="min-w-56">
|
||||
<USelect
|
||||
v-model="tableFilter"
|
||||
:items="[{ label: 'Alle Tabellen', value: '' }, ...(review?.tables || []).map((table) => ({ label: table, value: table }))]"
|
||||
value-key="value"
|
||||
label-key="label"
|
||||
@update:model-value="resetAndLoad"
|
||||
/>
|
||||
</UFormField>
|
||||
<div class="ml-auto flex gap-2">
|
||||
<UButton color="neutral" variant="soft" @click="setAllConflicts('target')">Alle Konflikte: Ziel behalten</UButton>
|
||||
<UButton color="warning" variant="soft" @click="setAllConflicts('source')">Alle Konflikte: Quelle übernehmen</UButton>
|
||||
</div>
|
||||
</div>
|
||||
</UCard>
|
||||
|
||||
<UCard>
|
||||
<div v-if="loading" class="py-8 text-center text-gray-500">Dry-Run wird geladen …</div>
|
||||
<div v-else-if="!review?.items.length" class="py-8 text-center text-gray-500">Keine Einträge für diesen Filter.</div>
|
||||
<div v-else class="divide-y divide-gray-200 dark:divide-gray-800">
|
||||
<div v-for="item in review.items" :key="item.id" class="space-y-2 py-4">
|
||||
<div class="flex flex-wrap items-center gap-2">
|
||||
<UBadge :color="kindColors[item.kind] as any" variant="soft">{{ kindLabels[item.kind] }}</UBadge>
|
||||
<span class="font-medium">{{ item.label }}</span>
|
||||
<span class="text-xs text-gray-500">{{ item.table }} · {{ item.sourceKey }}</span>
|
||||
</div>
|
||||
<p v-if="item.differences.length" class="text-sm text-gray-600 dark:text-gray-300">
|
||||
Abweichende Felder: {{ item.differences.join(", ") }}
|
||||
</p>
|
||||
<div v-if="item.kind === 'conflict'" class="flex flex-wrap gap-2">
|
||||
<UButton
|
||||
size="sm"
|
||||
:color="decisions[item.id] === 'target' ? 'primary' : 'neutral'"
|
||||
:variant="decisions[item.id] === 'target' ? 'solid' : 'soft'"
|
||||
@click="decisions[item.id] = 'target'"
|
||||
>Ziel behalten</UButton>
|
||||
<UButton
|
||||
size="sm"
|
||||
:color="decisions[item.id] === 'source' ? 'warning' : 'neutral'"
|
||||
:variant="decisions[item.id] === 'source' ? 'solid' : 'soft'"
|
||||
@click="decisions[item.id] = 'source'"
|
||||
>Quelle übernehmen</UButton>
|
||||
</div>
|
||||
<details v-if="item.kind === 'conflict'" class="text-xs">
|
||||
<summary class="cursor-pointer text-gray-500">Quelldaten und Zieldaten anzeigen</summary>
|
||||
<div class="mt-2 grid gap-2 lg:grid-cols-2">
|
||||
<pre class="overflow-auto rounded bg-gray-100 p-3 dark:bg-gray-900">{{ JSON.stringify(item.sourceRow, null, 2) }}</pre>
|
||||
<pre class="overflow-auto rounded bg-gray-100 p-3 dark:bg-gray-900">{{ JSON.stringify(item.targetRow, null, 2) }}</pre>
|
||||
</div>
|
||||
</details>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div v-if="review" class="mt-4 flex items-center justify-between border-t pt-4">
|
||||
<UButton color="neutral" variant="soft" :disabled="offset === 0" @click="offset = Math.max(0, offset - limit); loadReview()">Zurück</UButton>
|
||||
<span class="text-sm text-gray-500">{{ offset + 1 }}–{{ Math.min(offset + review.items.length, review.total) }} von {{ review.total }}</span>
|
||||
<UButton color="neutral" variant="soft" :disabled="!hasNextPage" @click="offset += limit; loadReview()">Weiter</UButton>
|
||||
</div>
|
||||
</UCard>
|
||||
|
||||
<UCard>
|
||||
<div class="flex flex-wrap items-center justify-between gap-3">
|
||||
<p :class="unresolvedConflicts ? 'text-orange-600' : 'text-green-600'">
|
||||
{{ unresolvedConflicts ? `${unresolvedConflicts} Konflikte benötigen noch eine Entscheidung.` : "Alle Konflikte sind entschieden." }}
|
||||
</p>
|
||||
<UButton
|
||||
icon="i-heroicons-play"
|
||||
color="primary"
|
||||
:loading="executing"
|
||||
:disabled="loading || unresolvedConflicts > 0 || Boolean(executionResult)"
|
||||
@click="executeMerge"
|
||||
>Merge ausführen</UButton>
|
||||
</div>
|
||||
</UCard>
|
||||
</div>
|
||||
</template>
|
||||
@@ -206,7 +206,7 @@ const importTenantExport = async (event: Event) => {
|
||||
filename: job.filename || file.name,
|
||||
}
|
||||
|
||||
while (!["ready", "failed"].includes(job.status)) {
|
||||
while (!["ready", "review", "failed"].includes(job.status)) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500))
|
||||
job = await admin.getTenantExport(importJobId)
|
||||
tenantImportProgress.value = {
|
||||
@@ -221,6 +221,11 @@ const importTenantExport = async (event: Event) => {
|
||||
throw new Error(job.error || "Import konnte nicht abgeschlossen werden.")
|
||||
}
|
||||
|
||||
if (job.status === "review") {
|
||||
await router.push(`/administration/tenant-imports/${importJobId}`)
|
||||
return
|
||||
}
|
||||
|
||||
await fetchTenant()
|
||||
await auth.fetchMe()
|
||||
await auth.switchTenant(String(job.tenantId || targetTenantId))
|
||||
|
||||
Reference in New Issue
Block a user