Create tenant merge review jobs
This commit is contained in:
@@ -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
|
||||
// -------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user