flfeders/FEDEO
2
1
Fork 0
Code Issues 106 Pull Requests Actions Packages Projects 22 Releases Wiki Activity

Restrict tenant imports to new tenants

Browse Source
This commit is contained in:
Florian Federspiel
2026-08-06 15:19:11 +02:00
parent c66d04f0b5
commit cb37970074
12 changed files with 38 additions and 1867 deletions
Download Patch File Download Diff File Expand all files Collapse all files

415
backend/src/routes/admin.ts
View File

@@ -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, PutObjectCommand } from "@aws-sdk/client-s3";
import { GetObjectCommand } from "@aws-sdk/client-s3";
import {
authTenantUsers,
@@ -22,17 +22,12 @@ 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, {
@@ -327,132 +322,6 @@ export default async function adminRoutes(server: FastifyInstance) {
));
};
const createTenantImportJob = async (tenantId: number, currentUserId: string, filename: string) => {
const [job] = await server.db
.insert(tenantExportJobs)
.values({
tenantId,
createdBy: currentUserId,
operation: "import",
status: "running",
filename,
updatedAt: new Date(),
})
.returning();
await lockTenantForJob(tenantId, job.id);
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
) => {
const heartbeat = setInterval(() => {
void server.db
.update(tenantExportJobs)
.set({ updatedAt: new Date() })
.where(eq(tenantExportJobs.id, jobId));
}, 15_000);
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));
} finally {
clearInterval(heartbeat);
}
};
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 }
@@ -513,67 +382,6 @@ export default async function adminRoutes(server: FastifyInstance) {
};
};
const startTenantImportJob = async (
jobId: string,
targetTenantId: number,
currentUser: { id: string; email: string },
importData: { type: "archive"; archiveBuffer: Buffer } | { type: "json"; exportData: TenantFullExport }
) => {
try {
await server.db
.update(tenantExportJobs)
.set({
status: "running",
filesDone: 0,
filesTotal: 1,
updatedAt: new Date(),
})
.where(eq(tenantExportJobs.id, jobId));
const onProgress = async ({ done, total }: { done: number; total: number }) => {
await server.db
.update(tenantExportJobs)
.set({
filesDone: done,
filesTotal: total,
updatedAt: new Date(),
})
.where(eq(tenantExportJobs.id, jobId));
};
const result = importData.type === "archive"
? await importTenantFullExportArchive(server, importData.archiveBuffer, { targetTenantId, onProgress })
: await importTenantFullExport(server, importData.exportData, { targetTenantId, onProgress });
await completeImportedTenantAccess(currentUser, result);
await server.db
.update(tenantExportJobs)
.set({
status: "ready",
filesDone: 1,
filesTotal: 1,
completedAt: new Date(),
updatedAt: new Date(),
error: null,
})
.where(eq(tenantExportJobs.id, jobId));
} catch (err: any) {
console.error("ERROR tenant import job:", err);
await server.db
.update(tenantExportJobs)
.set({
status: "failed",
error: err?.message || String(err),
completedAt: new Date(),
updatedAt: new Date(),
})
.where(eq(tenantExportJobs.id, jobId));
} finally {
await unlockTenantForJob(targetTenantId, jobId);
}
};
const startTenantExportJob = async (jobId: string, tenantId: number, filename: string) => {
try {
await lockTenantForJob(tenantId, jobId);
@@ -1474,25 +1282,11 @@ export default async function adminRoutes(server: FastifyInstance) {
if (!job) return reply.code(404).send({ error: "Export not found" });
let returnedStatus = job.status;
const staleMergeReview = job.operation === "merge"
&& ["running", "recovering"].includes(job.status)
&& (!job.updatedAt || Date.now() - new Date(job.updatedAt).getTime() > 60_000);
if (staleMergeReview && job.storagePath) {
const source = await readS3Buffer(job.storagePath);
await server.db
.update(tenantExportJobs)
.set({ status: "recovering", updatedAt: new Date(), error: null })
.where(eq(tenantExportJobs.id, job.id));
void startTenantMergeReviewJob(job.id, job.tenantId, source, job.contentType, job.filename);
returnedStatus = "recovering";
}
return {
exportId: job.id,
tenantId: job.tenantId,
operation: job.operation,
status: returnedStatus,
status: job.status,
filename: job.filename,
fileSize: job.fileSize,
filesDone: job.filesDone,
@@ -1562,31 +1356,13 @@ export default async function adminRoutes(server: FastifyInstance) {
const archiveBuffer = await data.toBuffer();
targetTenantId = Number(data.fields?.targetTenantId?.value || 0) || null;
if (targetTenantId) {
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,
exportId: importJob.id,
tenantId: targetTenantId,
status: importJob.status,
filename: importJob.filename,
statusUrl: `/api/admin/tenant-exports/${importJob.id}`,
reviewUrl: `/administration/tenant-imports/${importJob.id}`,
return reply.code(409).send({
error: "Ein Tenant-Export kann nur als neuer Tenant importiert werden. Der Import in einen bestehenden Tenant ist deaktiviert.",
});
}
result = await importTenantFullExportArchive(server, archiveBuffer, { targetTenantId });
result = await importTenantFullExportArchive(server, archiveBuffer);
} else {
const body = req.body as TenantFullExport | { exportData?: TenantFullExport; targetTenantId?: number };
const exportData = "format" in body ? body : body.exportData;
@@ -1597,28 +1373,12 @@ export default async function adminRoutes(server: FastifyInstance) {
}
if (targetTenantId) {
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,
exportId: importJob.id,
tenantId: targetTenantId,
status: importJob.status,
filename: importJob.filename,
statusUrl: `/api/admin/tenant-exports/${importJob.id}`,
reviewUrl: `/administration/tenant-imports/${importJob.id}`,
return reply.code(409).send({
error: "Ein Tenant-Export kann nur als neuer Tenant importiert werden. Der Import in einen bestehenden Tenant ist deaktiviert.",
});
}
result = await importTenantFullExport(server, exportData, { targetTenantId });
result = await importTenantFullExport(server, exportData);
}
const { matrixProvisioned, matrixProvisioningError } = await completeImportedTenantAccess(currentUser, result);
@@ -1631,162 +1391,9 @@ export default async function adminRoutes(server: FastifyInstance) {
};
} catch (err: any) {
console.error("ERROR /admin/tenant-imports:", err);
return reply.code(500).send({ error: err?.message || "Internal Server Error" });
}
});
// -------------------------------------------------------------
// 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);
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 });
}
});