Как обновить большое количество документов mongodb в древовидной структуре? ⇐ Javascript

[Anonymous]

Я использую базу данных mongo с payloadcms, и у меня древовидная структура.

У нас есть коллекция под названием nodes. У узла есть дочерние элементы в виде массива и родительского идентификатора.
Теперь я хочу изменить статус узла, пометив его как релевантный, нерелевантный и т. д. Если я помечу узел верхнего уровня как нерелевантный, я хочу, чтобы все его дочерние элементы также были нерелевантными, и наоборот.
Как я могу это сделать эффективно?
На стороне интерфейса я храню карту идентификаторов узла и их статуса чтобы пользователь мог свободно пометить дерево и, наконец, нажать кнопку «Сохранить». Я отправляю эту карту на серверную часть для обработки изменений. Поскольку это затронет большую часть документов - около 4 00 000 - как я могу это эффективно обработать?
Кроме того, я использую API-интерфейс update-many для пакетного обновления узлов в зависимости от статуса.
Все узлы с одинаковым статусом обновляются вместе, поэтому всего у меня есть 4 запроса с множеством обновлений.
Я уже использую для этого задания полезной нагрузки, но есть ли что-нибудь еще что я могу здесь сделать?
Это код фонового задания, которое запускается, когда пользователь нажимает кнопку «Применить изменения»:
import { PaginatedDocs, PayloadRequest, TaskConfig } from 'payload';
import { Node } from '../payload-types';
import { NodeStatusFailureEmail } from '../../emails/NodeStatusFailureEmail';
import { env } from '../env';

type MarkNodeInput = {
input: {
changes: Array;
};
output: {
success: boolean;
updatedCount: number;
message: string;
errors?: string[];
};
};

const changeNodeStatusTask: TaskConfig = {
slug: 'changeNodeStatusTask',
handler: async ({ input, req }) => {
const startTime = new Date().toISOString();

try {
console.log('\n\nStarting changeNodeStatusTask with input:');
const changes = input.changes;
const nodeIDs: string[] = changes.map((change) => change.nodeId);
const validChanges: Array = [];
const allNodes: PaginatedDocs = await req.payload.find({
collection: 'nodes',
where: {
id: {
in: nodeIDs,
},
},
pagination: false,
});

for (const node of allNodes.docs) {
try {
if (node.linkedWithID) {
console.warn(`Skipping linked/wrapper node: ${node.id}`);
continue;
}

const change = changes.find((c) => c.nodeId === node.id);
if (change) {
validChanges.push(change);
}
} catch (error: any) {
console.error(`Failed to validate node ${node.id}:`, error);

continue;
}
}

if (validChanges.length === 0) {
throw new Error('No valid nodes to update (all were linked/wrapper nodes or invalid)');
}

const statusMap: Record = {
decision: 'decision_node',
relevant: 'relevant',
irrelevant: 'irrelevant',
archived: 'archived',
};

const nodesToUpdate = new Map();
const cascadeNodes = new Set();
const errors: string[] = [];

for (const change of validChanges) {
const dbStatus = statusMap[change.status] || change.status;
nodesToUpdate.set(change.nodeId, dbStatus);

if (change.status === 'relevant' || change.status === 'irrelevant') {
cascadeNodes.add(change.nodeId);
}
}

for (const nodeId of cascadeNodes) {
try {
const status = nodesToUpdate.get(nodeId)!;
const descendants = await getAllDescendants(req, nodeId);

// Add all descendants to update list with the same status
for (const descendantId of descendants) {
nodesToUpdate.set(descendantId, status);
}
} catch (error: any) {
console.error(`Failed to get descendants for node ${nodeId}:`, error);
errors.push(`Node ${nodeId} descendants: ${error?.message || 'Unknown error'}`);
}
}

// Group nodes by status for batch updates
const nodesByStatus = new Map();
for (const [nodeId, status] of nodesToUpdate.entries()) {
if (!nodesByStatus.has(status)) {
nodesByStatus.set(status, []);
}
nodesByStatus.get(status)!.push(nodeId);
}

let updatedCount = 0;

// Batch update nodes by status using updateMany (via where clause)
const updatePromises = Array.from(nodesByStatus.entries()).map(async ([status, nodeIds]) => {
try {
console.log(`Updating ${nodeIds.length} nodes to status: ${status}`);
const result = await req.payload.update({
collection: 'nodes',
where: {
id: {
in: nodeIds,
},
},
data: {
status: status as 'relevant' | 'irrelevant' | 'archived' | 'decision_node',
},
});

updatedCount += result.docs.length;

// Track any errors returned by the batch update
if (result.errors && result.errors.length > 0) {
for (const error of result.errors) {
console.error(`Failed to update node ${error.id}:`, error);
errors.push(`Node ${error.id}: ${error.message || 'Unknown error'}`);
}
}
} catch (error: any) {
console.error(`Failed to batch update nodes with status ${status}:`, error);
errors.push(`Batch update for status ${status}: ${error?.message || 'Unknown error'}`);
}
});

await Promise.all(updatePromises);

if (errors.length > 0) {
try {
console.log('Sending node status failure email to admin...');
const adminEmail = env.SMTP_ADMIN_EMAIL || 'info@oncoproai.com';
const fromEmail = 'info@notifications.oncoproai.com';
const fromName = 'OncoproAI System';

await NodeStatusFailureEmail({
updatedCount,
totalNodes: nodesToUpdate.size,
errors,
timestamp: startTime,
}).send(req.payload.sendEmail, {
to: adminEmail,
from: {
name: fromName,
address: fromEmail,
},
});
console.log('Node status failure email sent successfully');
} catch (emailError) {
console.error('Failed to send node status failure email:', emailError);
}
}

// Return output property to match TaskHandlerResult type
return {
output: {
success: errors.length === 0,
updatedCount,
message: `Updated ${updatedCount} nodes.`,
...(errors.length > 0 && { errors }),
},
};
} catch (error: any) {
console.error('Critical error in changeNodeStatusTask:', error);

// Send failure email for critical errors
try {
const adminEmail = env.SMTP_ADMIN_EMAIL || 'info@oncoproai.com';
const fromEmail = 'info@oncoproai.com';
const fromName = 'OncoproAI';

await NodeStatusFailureEmail({
updatedCount: 0,
totalNodes: input.changes.length,
errors: [error?.message || 'Unknown critical error occurred'],
timestamp: startTime,
}).send(req.payload.sendEmail, {
to: adminEmail,
from: {
name: fromName,
address: fromEmail,
},
});
console.log('Critical error email sent successfully');
} catch (emailError) {
console.error('Failed to send critical error email:', emailError);
}

// Re-throw the error or return failure output
return {
output: {
success: false,
updatedCount: 0,
message: 'Task failed with critical error',
errors: [error?.message || 'Unknown error'],
},
};
}
},
};

export default changeNodeStatusTask;

/**
* Recursively get all descendant node IDs using BFS
* Similar to StatusTracker.getAllDescendantIds but works with database queries
*/
async function getAllDescendants(req: PayloadRequest, nodeId: string): Promise {
const descendants: string[] = [];
const queue: string[] = [nodeId];
const visited = new Set([nodeId]);

while (queue.length > 0) {
const currentId = queue.shift()!;

const children = await req.payload.find({
collection: 'nodes',
where: {
parentID: {
equals: currentId,
},
linkedWithID: { exists: false },
},
limit: 1000,
});

for (const child of children.docs) {
if (!visited.has(child.id)) {
visited.add(child.id);
descendants.push(child.id);
queue.push(child.id);
}
}
}

return descendants;
}


Подробнее здесь: https://stackoverflow.com/questions/798 ... -structure