Add tenant import progress polling
This commit is contained in:
@@ -340,6 +340,127 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
return job;
|
||||
};
|
||||
|
||||
const completeImportedTenantAccess = async (
|
||||
currentUser: { id: string; email: string },
|
||||
result: { tenantId: number }
|
||||
) => {
|
||||
const fallbackName = deriveNameFromEmail(currentUser.email);
|
||||
|
||||
await server.db
|
||||
.insert(authTenantUsers)
|
||||
.values({
|
||||
tenant_id: result.tenantId,
|
||||
user_id: currentUser.id,
|
||||
created_by: currentUser.id,
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
|
||||
const [existingAdminProfile] = await server.db
|
||||
.select({ id: authProfiles.id })
|
||||
.from(authProfiles)
|
||||
.where(and(
|
||||
eq(authProfiles.tenant_id, result.tenantId),
|
||||
eq(authProfiles.user_id, currentUser.id)
|
||||
))
|
||||
.limit(1);
|
||||
|
||||
if (!existingAdminProfile) {
|
||||
await server.db
|
||||
.insert(authProfiles)
|
||||
.values({
|
||||
tenant_id: result.tenantId,
|
||||
user_id: currentUser.id,
|
||||
first_name: fallbackName.first_name,
|
||||
last_name: fallbackName.last_name,
|
||||
email: currentUser.email,
|
||||
active: true,
|
||||
});
|
||||
}
|
||||
|
||||
let matrixProvisioned = false;
|
||||
let matrixProvisioningError: string | null = null;
|
||||
if (process.env.MATRIX_REGISTRATION_SHARED_SECRET) {
|
||||
try {
|
||||
const matrix = matrixService(server);
|
||||
await matrix.provisionTenantRoom(currentUser.id, result.tenantId, {
|
||||
key: "allgemein",
|
||||
name: "Allgemeiner Chat",
|
||||
type: "general",
|
||||
});
|
||||
matrixProvisioned = true;
|
||||
} catch (err: any) {
|
||||
matrixProvisioningError = err?.message || String(err);
|
||||
server.log.warn({ err }, "Matrix-Räume konnten nach Tenant-Import nicht neu provisioniert werden");
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
matrixProvisioned,
|
||||
matrixProvisioningError,
|
||||
};
|
||||
};
|
||||
|
||||
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);
|
||||
@@ -1237,6 +1358,7 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
return {
|
||||
exportId: job.id,
|
||||
tenantId: job.tenantId,
|
||||
operation: job.operation,
|
||||
status: job.status,
|
||||
filename: job.filename,
|
||||
fileSize: job.fileSize,
|
||||
@@ -1246,7 +1368,7 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
createdAt: job.createdAt,
|
||||
updatedAt: job.updatedAt,
|
||||
completedAt: job.completedAt,
|
||||
downloadUrl: job.status === "ready" ? `/api/admin/tenant-exports/${job.id}/download` : null,
|
||||
downloadUrl: job.operation === "export" && job.status === "ready" ? `/api/admin/tenant-exports/${job.id}/download` : null,
|
||||
};
|
||||
} catch (err) {
|
||||
console.error("ERROR /admin/tenant-exports/:export_id:", err);
|
||||
@@ -1293,15 +1415,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 importJob: any = null;
|
||||
let targetTenantId: number | null = null;
|
||||
|
||||
try {
|
||||
const currentUser = await requireAdmin(req, reply);
|
||||
if (!currentUser) return;
|
||||
|
||||
const isMultipart = req.headers["content-type"]?.includes("multipart/form-data");
|
||||
let result;
|
||||
let targetTenantId: number | null = null;
|
||||
|
||||
if (isMultipart) {
|
||||
const data: any = await req.file();
|
||||
@@ -1311,25 +1431,23 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
targetTenantId = Number(data.fields?.targetTenantId?.value || 0) || null;
|
||||
|
||||
if (targetTenantId) {
|
||||
importJob = await createTenantImportJob(targetTenantId, currentUser.id, data.filename || "tenant-import.zip");
|
||||
const importJob = await createTenantImportJob(targetTenantId, currentUser.id, data.filename || "tenant-import.zip");
|
||||
void startTenantImportJob(importJob.id, targetTenantId, currentUser, {
|
||||
type: "archive",
|
||||
archiveBuffer,
|
||||
});
|
||||
|
||||
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}`,
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
result = await importTenantFullExportArchive(server, archiveBuffer, { targetTenantId });
|
||||
} catch (err: any) {
|
||||
if (importJob) {
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "failed",
|
||||
error: err?.message || String(err),
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, importJob.id));
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
result = await importTenantFullExportArchive(server, archiveBuffer, { targetTenantId });
|
||||
} else {
|
||||
const body = req.body as TenantFullExport | { exportData?: TenantFullExport; targetTenantId?: number };
|
||||
const exportData = "format" in body ? body : body.exportData;
|
||||
@@ -1340,88 +1458,26 @@ export default async function adminRoutes(server: FastifyInstance) {
|
||||
}
|
||||
|
||||
if (targetTenantId) {
|
||||
importJob = await createTenantImportJob(targetTenantId, currentUser.id, "tenant-import.json");
|
||||
}
|
||||
|
||||
try {
|
||||
result = await importTenantFullExport(server, exportData, { targetTenantId });
|
||||
} catch (err: any) {
|
||||
if (importJob) {
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "failed",
|
||||
error: err?.message || String(err),
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, importJob.id));
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
if (importJob) {
|
||||
await server.db
|
||||
.update(tenantExportJobs)
|
||||
.set({
|
||||
status: "ready",
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
error: null,
|
||||
})
|
||||
.where(eq(tenantExportJobs.id, importJob.id));
|
||||
}
|
||||
|
||||
const fallbackName = deriveNameFromEmail(currentUser.email);
|
||||
|
||||
await server.db
|
||||
.insert(authTenantUsers)
|
||||
.values({
|
||||
tenant_id: result.tenantId,
|
||||
user_id: currentUser.id,
|
||||
created_by: currentUser.id,
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
|
||||
const [existingAdminProfile] = await server.db
|
||||
.select({ id: authProfiles.id })
|
||||
.from(authProfiles)
|
||||
.where(and(
|
||||
eq(authProfiles.tenant_id, result.tenantId),
|
||||
eq(authProfiles.user_id, currentUser.id)
|
||||
))
|
||||
.limit(1);
|
||||
|
||||
if (!existingAdminProfile) {
|
||||
await server.db
|
||||
.insert(authProfiles)
|
||||
.values({
|
||||
tenant_id: result.tenantId,
|
||||
user_id: currentUser.id,
|
||||
first_name: fallbackName.first_name,
|
||||
last_name: fallbackName.last_name,
|
||||
email: currentUser.email,
|
||||
active: true,
|
||||
const importJob = await createTenantImportJob(targetTenantId, currentUser.id, "tenant-import.json");
|
||||
void startTenantImportJob(importJob.id, targetTenantId, currentUser, {
|
||||
type: "json",
|
||||
exportData,
|
||||
});
|
||||
|
||||
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}`,
|
||||
});
|
||||
}
|
||||
|
||||
result = await importTenantFullExport(server, exportData, { targetTenantId });
|
||||
}
|
||||
|
||||
let matrixProvisioned = false;
|
||||
let matrixProvisioningError: string | null = null;
|
||||
if (process.env.MATRIX_REGISTRATION_SHARED_SECRET) {
|
||||
try {
|
||||
const matrix = matrixService(server);
|
||||
await matrix.provisionTenantRoom(currentUser.id, result.tenantId, {
|
||||
key: "allgemein",
|
||||
name: "Allgemeiner Chat",
|
||||
type: "general",
|
||||
});
|
||||
matrixProvisioned = true;
|
||||
} catch (err: any) {
|
||||
matrixProvisioningError = err?.message || String(err);
|
||||
req.log.warn({ err }, "Matrix-Räume konnten nach Tenant-Import nicht neu provisioniert werden");
|
||||
}
|
||||
}
|
||||
const { matrixProvisioned, matrixProvisioningError } = await completeImportedTenantAccess(currentUser, result);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
@@ -1432,14 +1488,6 @@ 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" });
|
||||
} finally {
|
||||
if (importJob && targetTenantId) {
|
||||
try {
|
||||
await unlockTenantForJob(targetTenantId, importJob.id);
|
||||
} catch (unlockErr) {
|
||||
req.log.error({ err: unlockErr, importJobId: importJob.id, targetTenantId }, "Tenant konnte nach Import nicht entsperrt werden");
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user