diff --git a/src/database/database.js b/src/database/database.js index ed6f78a..78312c3 100644 --- a/src/database/database.js +++ b/src/database/database.js @@ -336,7 +336,8 @@ export const aggregateRollups = async ({ model, baseFilter = {}, rollupConfigs = }, {}); }; -// Reusable function to aggregate rollups over history using state reconstruction +// Snapshot absolute rollup values at each point in time by reconstructing +// object state from the current documents plus audit logs. export const aggregateRollupsHistory = async ({ model, baseFilter = {}, @@ -348,57 +349,15 @@ export const aggregateRollupsHistory = async ({ return []; } - // Set default dates if not provided const end = endDate ? new Date(endDate) : new Date(); const start = startDate ? new Date(startDate) : new Date(end.getTime() - 24 * 60 * 60 * 1000); - - // Get model name for filtering audit logs const parentType = model.modelName ? model.modelName : 'unknown'; - // 1. Fetch all audit logs for this model type from start date to now - // Filter by parentType instead of fetching object IDs first - const auditLogs = await auditLogModel - .find({ - parentType, - createdAt: { $gte: start }, - }) - .sort({ createdAt: -1 }) // Newest first - .lean(); - - // 2. Extract unique parent IDs from audit logs - const parentIds = [...new Set(auditLogs.map((log) => log.parent.toString()))]; - - if (parentIds.length === 0) { - return []; - } - - // 3. Fetch current state of relevant objects that match baseFilter - // Note: This only includes objects that CURRENTLY match the baseFilter. - // Objects that matched in the past but don't match now are excluded. - const currentObjects = await model - .find({ - _id: { $in: parentIds }, - ...baseFilter, - }) - .lean(); - const objectMap = new Map(); - currentObjects.forEach((obj) => { - // Ensure _id is a string for map keys - objectMap.set(obj._id.toString(), expandObjectIds(obj)); - }); - - if (objectMap.size === 0) { - return []; - } - - // Helper to check if object matches filter const matchesFilter = (obj, filter) => { if (!filter || Object.keys(filter).length === 0) return true; for (const [key, expectedValue] of Object.entries(filter)) { const actualValue = _.get(obj, key); - - // Handle simple equality if (actualValue != expectedValue) { return false; } @@ -406,79 +365,25 @@ export const aggregateRollupsHistory = async ({ return true; }; - // 3. Generate time buckets (1 minute intervals) - const buckets = []; - let currentTime = new Date(end); - // Round down to nearest minute - currentTime.setSeconds(0, 0); + const existedAt = (obj, bucketDate) => { + if (!obj?.createdAt) return true; + return new Date(obj.createdAt) <= bucketDate; + }; - while (currentTime >= start) { - buckets.push(new Date(currentTime)); - currentTime = new Date(currentTime.getTime() - 60000); // -1 minute - } - - // 4. Rewind state and snapshot - const results = []; - let logIndex = 0; - - // Create a working copy of objects to mutate during rewind - // (deep clone to avoid issues if we need original later, though expandObjectIds creates new objs) - const workingObjects = new Map(); - objectMap.forEach((val, key) => workingObjects.set(key, _.cloneDeep(val))); - - // Iterate backwards through time - for (const bucketDate of buckets) { - // Apply all logs that happened AFTER this bucket time (between last bucket and this one) - // Since we iterate backwards, these are logs with createdAt > bucketDate - while (logIndex < auditLogs.length) { - const log = auditLogs[logIndex]; - const logDate = new Date(log.createdAt); - - if (logDate <= bucketDate) { - // This log happened at or before the current bucket time, - // so its effects are already present (or rather, will be handled in a future/earlier bucket). - // Stop processing logs for this step. - break; - } - - // Revert this change - const objectId = log.parent.toString(); - const object = workingObjects.get(objectId); - - if (object) { - if (log.operation === 'new') { - // Object didn't exist before this creation event - workingObjects.delete(objectId); - } else if (log.changes && log.changes.old) { - // Apply old values to revert state - _.merge(object, log.changes.old); - } - } - - logIndex++; - } - - // Snapshot: Calculate rollups for current state of all objects + const snapshotRollups = (objects, bucketDate) => { const bucketResult = { date: bucketDate.toISOString(), }; - const activeObjects = Array.from(workingObjects.values()); - rollupConfigs.forEach((config) => { - const configName = config.name; - - // Filter objects for this config - // Note: We also check baseFilter here in case the object state reverted to something - // that no longer matches baseFilter (e.g. active: false) - const matchingObjects = activeObjects.filter( - (obj) => matchesFilter(obj, baseFilter) && matchesFilter(obj, config.filter) + const matchingObjects = objects.filter( + (obj) => + existedAt(obj, bucketDate) && + matchesFilter(obj, baseFilter) && + matchesFilter(obj, config.filter) ); - // Calculate rollups (config.rollups || []).forEach((rollup) => { - const rollupName = rollup.name; - let value = 0; if (rollup.operation === 'count') { value = matchingObjects.length; @@ -489,15 +394,93 @@ export const aggregateRollupsHistory = async ({ value = matchingObjects.length ? sum / matchingObjects.length : 0; } - // Nest the value under the operation type - bucketResult[rollupName] = { [rollup.operation]: value }; + bucketResult[rollup.name] = { [rollup.operation]: value }; }); }); - results.push(bucketResult); + return bucketResult; + }; + + const auditLogs = await auditLogModel + .find({ + parentType, + createdAt: { $gte: start }, + }) + .sort({ createdAt: -1 }) + .lean(); + + const currentObjects = await model.find(baseFilter).lean(); + const objectMap = new Map(); + currentObjects.forEach((obj) => { + objectMap.set(obj._id.toString(), expandObjectIds(obj)); + }); + + const extraIds = [ + ...new Set( + auditLogs.map((log) => log.parent?.toString()).filter((id) => id && !objectMap.has(id)) + ), + ]; + if (extraIds.length) { + const extraObjects = await model.find({ _id: { $in: extraIds } }).lean(); + extraObjects.forEach((obj) => { + objectMap.set(obj._id.toString(), expandObjectIds(obj)); + }); + } + + if (objectMap.size === 0 && auditLogs.length === 0) { + return []; + } + + const buckets = []; + let currentTime = new Date(end); + currentTime.setSeconds(0, 0); + + while (currentTime >= start) { + buckets.push(new Date(currentTime)); + currentTime = new Date(currentTime.getTime() - 60000); + } + + if (!buckets.length) { + return []; + } + + const workingObjects = new Map(); + objectMap.forEach((val, key) => workingObjects.set(key, _.cloneDeep(val))); + + const results = []; + let logIndex = 0; + + for (const bucketDate of buckets) { + while (logIndex < auditLogs.length) { + const log = auditLogs[logIndex]; + const logDate = new Date(log.createdAt); + + if (logDate <= bucketDate) { + break; + } + + const objectId = log.parent.toString(); + const object = workingObjects.get(objectId); + + if (log.operation === 'new') { + workingObjects.delete(objectId); + } else if (log.operation === 'delete' && log.changes?.old) { + if (!workingObjects.has(objectId)) { + workingObjects.set( + objectId, + expandObjectIds({ ...log.changes.old, _id: log.parent }) + ); + } + } else if (object && log.changes?.old) { + mergeObjectUpdates(object, log.changes.old); + } + + logIndex++; + } + + results.push(snapshotRollups(Array.from(workingObjects.values()), bucketDate)); } - // Reverse results to be chronological return results.reverse(); };