Tenant-Importe per Stream in den Object Storage laden

backend/src/routes/admin.ts

@@ -1,7 +1,8 @@
 import { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
 import { and, eq, inArray, isNull, sql } from "drizzle-orm";
 import multipart from "@fastify/multipart";
-import { GetObjectCommand } from "@aws-sdk/client-s3";
+import { DeleteObjectCommand, GetObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3";
+import { randomUUID } from "node:crypto";
 
 import {
     authTenantUsers,
@@ -463,7 +464,7 @@ export default async function adminRoutes(server: FastifyInstance) {
     const startTenantImportJob = async (
         jobId: string,
         currentUser: { id: string; email: string },
-        source: { archiveBuffer: Buffer } | { exportData: TenantFullExport }
+        source: { archiveStoragePath: string } | { exportData: TenantFullExport }
     ) => {
         try {
             await server.db
@@ -487,9 +488,19 @@ export default async function adminRoutes(server: FastifyInstance) {
                     .where(eq(tenantExportJobs.id, jobId));
             };
 
-            const result = "archiveBuffer" in source
+            let result: Awaited<ReturnType<typeof importTenantFullExport>>;
-                ? await importTenantFullExportArchive(server, source.archiveBuffer, { onProgress })
+            if ("archiveStoragePath" in source) {
-                : await importTenantFullExport(server, source.exportData, { onProgress });
+                const object = await s3.send(new GetObjectCommand({
+                    Bucket: secrets.S3_BUCKET,
+                    Key: source.archiveStoragePath,
+                }));
+                if (!object.Body) throw new Error("Importarchiv konnte nicht aus dem Speicher gelesen werden");
+
+                const archiveBuffer = Buffer.from(await object.Body.transformToByteArray());
+                result = await importTenantFullExportArchive(server, archiveBuffer, { onProgress });
+            } else {
+                result = await importTenantFullExport(server, source.exportData, { onProgress });
+            }
             const access = await completeImportedTenantAccess(currentUser, result);
             const importResult = { success: true, ...access, ...result };
 
@@ -517,6 +528,15 @@ export default async function adminRoutes(server: FastifyInstance) {
                     updatedAt: new Date(),
                 })
                 .where(eq(tenantExportJobs.id, jobId));
+        } finally {
+            if ("archiveStoragePath" in source) {
+                await s3.send(new DeleteObjectCommand({
+                    Bucket: secrets.S3_BUCKET,
+                    Key: source.archiveStoragePath,
+                })).catch((cleanupError) => {
+                    console.error("ERROR cleanup tenant import archive:", cleanupError);
+                });
+            }
         }
     };
 
@@ -1425,12 +1445,13 @@ export default async function adminRoutes(server: FastifyInstance) {
     // POST /admin/tenant-imports
     // -------------------------------------------------------------
     server.post("/admin/tenant-imports", { bodyLimit: 1024 * 1024 * 1024 }, async (req, reply) => {
+        let archiveStoragePath: string | null = null;
         try {
             const currentUser = await requireAdmin(req, reply);
             if (!currentUser) return;
 
             const isMultipart = req.headers["content-type"]?.includes("multipart/form-data");
-            let source!: { archiveBuffer: Buffer } | { exportData: TenantFullExport };
+            let source!: { archiveStoragePath: string } | { exportData: TenantFullExport };
             let filename = "tenant-export.json";
             let fileSize: number | null = null;
             let targetTenantId: number | null = null;
@@ -1439,9 +1460,7 @@ export default async function adminRoutes(server: FastifyInstance) {
                 const data: any = await req.file();
                 if (!data?.file) return reply.code(400).send({ error: "export file required" });
 
-                const archiveBuffer = await data.toBuffer();
                 filename = data.filename || "tenant-export.fedeo-export.zip";
-                fileSize = archiveBuffer.length;
                 targetTenantId = Number(data.fields?.targetTenantId?.value || 0) || null;
                 if (targetTenantId) {
                     return reply.code(409).send({
@@ -1449,7 +1468,15 @@ export default async function adminRoutes(server: FastifyInstance) {
                     });
                 }
 
-                source = { archiveBuffer };
+                archiveStoragePath = `tenant-imports/${randomUUID()}-${filename}`;
+                await s3.send(new PutObjectCommand({
+                    Bucket: secrets.S3_BUCKET,
+                    Key: archiveStoragePath,
+                    Body: data.file,
+                    ContentType: data.mimetype || "application/zip",
+                }));
+                fileSize = data.file.bytesRead || null;
+                source = { archiveStoragePath };
             } else {
                 const body = req.body as TenantFullExport | { exportData?: TenantFullExport; targetTenantId?: number };
                 const exportData = "format" in body ? body : body.exportData;
@@ -1493,6 +1520,14 @@ export default async function adminRoutes(server: FastifyInstance) {
             });
         } catch (err: any) {
             console.error("ERROR /admin/tenant-imports:", err);
+            if (archiveStoragePath) {
+                await s3.send(new DeleteObjectCommand({
+                    Bucket: secrets.S3_BUCKET,
+                    Key: archiveStoragePath,
+                })).catch((cleanupError) => {
+                    console.error("ERROR cleanup tenant import archive:", cleanupError);
+                });
+            }
             const message = err?.message || "Internal Server Error";
             const statusCode = message.includes("Tenant mit dieser ID existiert bereits") ? 409 : 500;
             return reply.code(statusCode).send({ error: message });