Prevent tenant merge dry-run memory spikes
All checks were successful
Build and Push Docker Images / build-backend (push) Successful in 39s
Build and Push Docker Images / build-frontend (push) Successful in 24s
Build and Push Docker Images / build-website (push) Successful in 23s
Build and Push Docker Images / build-central-services-api (push) Successful in 24s
Build and Push Docker Images / build-central-services-admin (push) Successful in 24s
Build and Push Docker Images / build-docs (push) Successful in 24s
All checks were successful
Build and Push Docker Images / build-backend (push) Successful in 39s
Build and Push Docker Images / build-frontend (push) Successful in 24s
Build and Push Docker Images / build-website (push) Successful in 23s
Build and Push Docker Images / build-central-services-api (push) Successful in 24s
Build and Push Docker Images / build-central-services-admin (push) Successful in 24s
Build and Push Docker Images / build-docs (push) Successful in 24s
This commit is contained in:
@@ -401,6 +401,12 @@ export default async function adminRoutes(server: FastifyInstance) {
|
|||||||
contentType: string,
|
contentType: string,
|
||||||
filename: string
|
filename: string
|
||||||
) => {
|
) => {
|
||||||
|
const heartbeat = setInterval(() => {
|
||||||
|
void server.db
|
||||||
|
.update(tenantExportJobs)
|
||||||
|
.set({ updatedAt: new Date() })
|
||||||
|
.where(eq(tenantExportJobs.id, jobId));
|
||||||
|
}, 15_000);
|
||||||
try {
|
try {
|
||||||
const exportData = await parseMergeSource(source, contentType, filename);
|
const exportData = await parseMergeSource(source, contentType, filename);
|
||||||
const report = sanitizeTenantMergePlan(await createTenantMergeDryRun(exportData, targetTenantId));
|
const report = sanitizeTenantMergePlan(await createTenantMergeDryRun(exportData, targetTenantId));
|
||||||
@@ -434,6 +440,8 @@ export default async function adminRoutes(server: FastifyInstance) {
|
|||||||
updatedAt: new Date(),
|
updatedAt: new Date(),
|
||||||
})
|
})
|
||||||
.where(eq(tenantExportJobs.id, jobId));
|
.where(eq(tenantExportJobs.id, jobId));
|
||||||
|
} finally {
|
||||||
|
clearInterval(heartbeat);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -1466,11 +1474,25 @@ export default async function adminRoutes(server: FastifyInstance) {
|
|||||||
|
|
||||||
if (!job) return reply.code(404).send({ error: "Export not found" });
|
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 {
|
return {
|
||||||
exportId: job.id,
|
exportId: job.id,
|
||||||
tenantId: job.tenantId,
|
tenantId: job.tenantId,
|
||||||
operation: job.operation,
|
operation: job.operation,
|
||||||
status: job.status,
|
status: returnedStatus,
|
||||||
filename: job.filename,
|
filename: job.filename,
|
||||||
fileSize: job.fileSize,
|
fileSize: job.fileSize,
|
||||||
filesDone: job.filesDone,
|
filesDone: job.filesDone,
|
||||||
|
|||||||
@@ -53,9 +53,13 @@ const comparableRow = (row: Record<string, any>) => Object.fromEntries(
|
|||||||
.sort(([left], [right]) => left.localeCompare(right))
|
.sort(([left], [right]) => left.localeCompare(right))
|
||||||
)
|
)
|
||||||
|
|
||||||
const rowDifferences = (source: Record<string, any>, target: Record<string, any>) => {
|
const rowDifferences = (table: string, source: Record<string, any>, target: Record<string, any>) => {
|
||||||
const sourceComparable = comparableRow(source)
|
const sourceComparable = comparableRow(source)
|
||||||
const targetComparable = comparableRow(target)
|
const targetComparable = comparableRow(target)
|
||||||
|
if (table === "citys") {
|
||||||
|
delete sourceComparable.geometry
|
||||||
|
delete targetComparable.geometry
|
||||||
|
}
|
||||||
const columns = Array.from(new Set([
|
const columns = Array.from(new Set([
|
||||||
...Object.keys(sourceComparable),
|
...Object.keys(sourceComparable),
|
||||||
...Object.keys(targetComparable),
|
...Object.keys(targetComparable),
|
||||||
@@ -111,7 +115,7 @@ export const buildTenantMergePlan = (
|
|||||||
const naturalTarget = sourceNaturalKey ? targetByNaturalKey.get(sourceNaturalKey) : undefined
|
const naturalTarget = sourceNaturalKey ? targetByNaturalKey.get(sourceNaturalKey) : undefined
|
||||||
const primaryTarget = targetByPrimaryKey.get(sourceKey)
|
const primaryTarget = targetByPrimaryKey.get(sourceKey)
|
||||||
const targetRow = naturalTarget || primaryTarget || null
|
const targetRow = naturalTarget || primaryTarget || null
|
||||||
const differences = targetRow ? rowDifferences(sourceRow, targetRow) : []
|
const differences = targetRow ? rowDifferences(table, sourceRow, targetRow) : []
|
||||||
let kind: TenantMergeKind
|
let kind: TenantMergeKind
|
||||||
|
|
||||||
if (!targetRow) {
|
if (!targetRow) {
|
||||||
|
|||||||
@@ -161,16 +161,43 @@ const sensitiveColumns = new Set([
|
|||||||
"tokenHash",
|
"tokenHash",
|
||||||
])
|
])
|
||||||
|
|
||||||
const redactRow = (row: Record<string, any> | null) => row && Object.fromEntries(
|
const compactValue = (value: any) => {
|
||||||
Object.entries(row).map(([column, value]) => [column, sensitiveColumns.has(column) && value ? "***" : value])
|
if (value === null || typeof value !== "object") return value
|
||||||
)
|
const serialized = JSON.stringify(value)
|
||||||
|
return `[Struktur mit ${serialized.length} Zeichen]`
|
||||||
|
}
|
||||||
|
|
||||||
|
const compactReviewRow = (row: Record<string, any> | null, differences: string[]) => {
|
||||||
|
if (!row) return null
|
||||||
|
const displayColumns = new Set([
|
||||||
|
"id",
|
||||||
|
"tenant",
|
||||||
|
"tenant_id",
|
||||||
|
"name",
|
||||||
|
"label",
|
||||||
|
"title",
|
||||||
|
"number",
|
||||||
|
"email",
|
||||||
|
"zip",
|
||||||
|
"short",
|
||||||
|
"accountChart",
|
||||||
|
...differences,
|
||||||
|
])
|
||||||
|
|
||||||
|
return Object.fromEntries(Object.entries(row)
|
||||||
|
.filter(([column]) => displayColumns.has(column))
|
||||||
|
.map(([column, value]) => [
|
||||||
|
column,
|
||||||
|
sensitiveColumns.has(column) && value ? "***" : compactValue(value),
|
||||||
|
]))
|
||||||
|
}
|
||||||
|
|
||||||
export const sanitizeTenantMergePlan = (plan: TenantMergePlan): TenantMergePlan => ({
|
export const sanitizeTenantMergePlan = (plan: TenantMergePlan): TenantMergePlan => ({
|
||||||
...plan,
|
...plan,
|
||||||
items: plan.items.map((item) => ({
|
items: plan.items.filter((item) => item.kind !== "existing").map((item) => ({
|
||||||
...item,
|
...item,
|
||||||
sourceRow: redactRow(item.sourceRow) || {},
|
sourceRow: compactReviewRow(item.sourceRow, item.differences) || {},
|
||||||
targetRow: redactRow(item.targetRow),
|
targetRow: compactReviewRow(item.targetRow, item.differences),
|
||||||
})),
|
})),
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user