farmcontrol/farmcontrol-api
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor aggregateRollupsHistory function to improve rollup snapshotting
All checks were successful
farmcontrol/farmcontrol-api/pipeline/head This commit looks good
Details

Browse Source 
This commit enhances the `aggregateRollupsHistory` function by restructuring the logic for filtering and processing audit logs, improving the accuracy of rollup snapshots over time. New utility functions are introduced to streamline the matching of objects against filters and to handle the existence of objects at specific timestamps. The overall flow is optimized to ensure that rollup calculations reflect the correct state of objects based on historical data, enhancing the reliability of data analysis and reporting.
This commit is contained in:
Tom Butcher 2026-08-24 23:14:43 +01:00
parent 9435d5c64c
commit f5551f4f03
1 changed files with 94 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

205
src/database/database.js
View File

@ -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();
};