From a2c5ad5dd973e4cc4cf4888cbebc9dc601058eb0 Mon Sep 17 00:00:00 2001 From: flfeders Date: Thu, 6 Aug 2026 12:27:27 +0200 Subject: [PATCH] Create tenant merge review jobs --- backend/src/routes/admin.ts | 181 +++++++++++++++++++++++++++++++++--- 1 file changed, 170 insertions(+), 11 deletions(-) diff --git a/backend/src/routes/admin.ts b/backend/src/routes/admin.ts index 0dd7698..fdec24e 100644 --- a/backend/src/routes/admin.ts +++ b/backend/src/routes/admin.ts @@ -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,15 @@ import { createTenantFullExportArchive, importTenantFullExport, importTenantFullExportArchive, + readTenantFullExportArchive, } 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 } from "../utils/tenantMergeService"; +import type { TenantMergePlan } from "../utils/tenantMergePlan"; export default async function adminRoutes(server: FastifyInstance) { await server.register(multipart, { @@ -340,6 +343,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 = 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 +1540,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 +1558,7 @@ export default async function adminRoutes(server: FastifyInstance) { status: importJob.status, filename: importJob.filename, statusUrl: `/api/admin/tenant-exports/${importJob.id}`, + reviewUrl: `/administration/tenants/${targetTenantId}/imports/${importJob.id}`, }); } @@ -1464,11 +1573,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 +1590,7 @@ export default async function adminRoutes(server: FastifyInstance) { status: importJob.status, filename: importJob.filename, statusUrl: `/api/admin/tenant-exports/${importJob.id}`, + reviewUrl: `/administration/tenants/${targetTenantId}/imports/${importJob.id}`, }); } @@ -1497,6 +1611,51 @@ 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" }); + } + }); + // ------------------------------------------------------------- // PUT /admin/users/:user_id/access // -------------------------------------------------------------