Skip to content

Commit 269ba5f

Browse files
TheLastCicadaTheLastCicada
committed
fix: batch v1 staging diff lookups
Replace per-row staging diff hydration with batched project, unit, and issuance queries so large v1 staging responses avoid N+1 lookups. Keep the existing response shape while skipping the unneeded count query for unpaginated reads.
1 parent 8faec6d commit 269ba5f

3 files changed

Lines changed: 508 additions & 113 deletions

File tree

‎src/controllers/staging.controller.js‎

Lines changed: 33 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,8 @@ export const findAll = async (req, res) => {
3939
let { page, limit, type, table } = req.query;
4040

4141
let pagination = paginationParams(page, limit);
42+
const isPaginated = page !== undefined;
43+
const order = [['id', 'ASC']];
4244

4345
let where = {};
4446
if (type === 'staged') {
@@ -53,32 +55,37 @@ export const findAll = async (req, res) => {
5355
where.table = table;
5456
}
5557

56-
let stagingData = await Staging.findAndCountAll({
57-
distinct: true,
58-
where,
59-
...pagination,
60-
});
61-
62-
const results = await Promise.all(
63-
stagingData.rows.map(async (stagingRecord) => {
64-
const { uuid, table, action, data } = stagingRecord;
65-
const workingData = _.cloneDeep(stagingRecord.dataValues);
66-
workingData.diff = await Staging.getDiffObject(
67-
uuid,
68-
table,
69-
action,
70-
data,
71-
);
72-
73-
delete workingData.data;
74-
75-
return workingData;
76-
}),
77-
);
78-
79-
stagingData.rows = results;
80-
81-
const response = optionallyPaginatedResponse(stagingData, page, limit);
58+
const stagingData = isPaginated
59+
? await Staging.findAndCountAll({
60+
distinct: true,
61+
where,
62+
order,
63+
...pagination,
64+
})
65+
: {
66+
rows: await Staging.findAll({
67+
where,
68+
order,
69+
}),
70+
};
71+
72+
const diffs = await Staging.getDiffObjects(stagingData.rows);
73+
const results = stagingData.rows.map((stagingRecord, index) => {
74+
const workingData = _.cloneDeep(stagingRecord.dataValues);
75+
workingData.diff = diffs[index];
76+
77+
delete workingData.data;
78+
79+
return workingData;
80+
});
81+
82+
const response = isPaginated
83+
? optionallyPaginatedResponse(
84+
{ count: stagingData.count, rows: results },
85+
page,
86+
limit,
87+
)
88+
: results;
8289

8390
res.json(response);
8491
} catch (error) {

‎src/models/staging/staging.model.js‎

Lines changed: 138 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,16 @@ import {
2525
} from '../../utils/xls';
2626
import { updateNilVerificationBodyAsEmptyString } from '../../utils/helpers.js';
2727

28+
const buildAssociationIncludes = (ModelClass) =>
29+
ModelClass.getAssociatedModels().map((association) => {
30+
return {
31+
model: association.model,
32+
as: formatModelAssociationName(association),
33+
};
34+
});
35+
36+
const dedupe = (values) => _.uniq(values.filter(Boolean));
37+
2838
class Staging extends Model {
2939
static changes = new rxjs.Subject();
3040

@@ -316,100 +326,141 @@ class Staging extends Model {
316326
};
317327

318328
static getDiffObject = async (uuid, table, action, data) => {
319-
const diff = {};
320-
if (action === 'INSERT') {
321-
diff.original = {};
322-
diff.change = JSON.parse(data);
323-
}
329+
const [diff] = await Staging.getDiffObjects([{ uuid, table, action, data }]);
330+
return diff;
331+
};
324332

325-
if (action === 'UPDATE') {
326-
diff.change = JSON.parse(data);
327-
328-
let original;
329-
330-
if (table === 'Projects') {
331-
original = await Project.findOne({
332-
where: { warehouseProjectId: uuid },
333-
include: Project.getAssociatedModels().map((association) => {
334-
return {
335-
model: association.model,
336-
as: formatModelAssociationName(association),
337-
};
338-
}),
339-
});
340-
341-
// Show the issuance data if its being reused
342-
// this is just for view purposes onlys
343-
await Promise.all(
344-
diff.change.map(async (record) => {
345-
if (record.issuanceId) {
346-
const issuance = await Issuance.findOne({
347-
where: { id: record.issuanceId },
348-
});
349-
350-
record.issuance = issuance.dataValues;
351-
}
352-
}),
353-
);
354-
} else if (table === 'Units') {
355-
original = await Unit.findOne({
356-
where: { warehouseUnitId: uuid },
357-
include: Unit.getAssociatedModels().map((association) => {
358-
return {
359-
model: association.model,
360-
as: formatModelAssociationName(association),
361-
};
362-
}),
363-
});
364-
365-
// Show the issuance data if its being reused,
366-
// this is just for view purposes onlys
367-
await Promise.all(
368-
diff.change.map(async (record) => {
369-
if (record.issuanceId) {
370-
const issuance = await Issuance.findOne({
371-
where: { id: record.issuanceId },
372-
});
373-
374-
record.issuance = issuance.dataValues;
375-
}
376-
}),
377-
);
333+
static getDiffObjects = async (stagingRecords) => {
334+
const normalizedRecords = stagingRecords.map((record) => {
335+
if (!record) {
336+
return record;
378337
}
379338

380-
diff.original = original;
381-
}
339+
return record.dataValues ? record.dataValues : record;
340+
});
382341

383-
if (action === 'DELETE') {
384-
let original;
385-
386-
if (table === 'Projects') {
387-
original = await Project.findOne({
388-
where: { warehouseProjectId: uuid },
389-
include: Project.getAssociatedModels().map((association) => {
390-
return {
391-
model: association.model,
392-
as: formatModelAssociationName(association),
393-
};
394-
}),
395-
});
396-
} else if (table === 'Units') {
397-
original = await Unit.findOne({
398-
where: { warehouseUnitId: uuid },
399-
include: Unit.getAssociatedModels().map((association) => {
400-
return {
401-
model: association.model,
402-
as: formatModelAssociationName(association),
403-
};
404-
}),
405-
});
342+
const parsedChanges = normalizedRecords.map((record) => {
343+
if (!record || record.action !== 'UPDATE') {
344+
return null;
406345
}
407346

408-
diff.original = original;
409-
diff.change = {};
410-
}
347+
return JSON.parse(record.data);
348+
});
411349

412-
return diff;
350+
const unitIds = dedupe(
351+
normalizedRecords
352+
.filter(
353+
(record) =>
354+
record &&
355+
record.table === 'Units' &&
356+
['UPDATE', 'DELETE'].includes(record.action),
357+
)
358+
.map((record) => record.uuid),
359+
);
360+
const projectIds = dedupe(
361+
normalizedRecords
362+
.filter(
363+
(record) =>
364+
record &&
365+
record.table === 'Projects' &&
366+
['UPDATE', 'DELETE'].includes(record.action),
367+
)
368+
.map((record) => record.uuid),
369+
);
370+
const issuanceIds = dedupe(
371+
normalizedRecords.flatMap((record, index) => {
372+
if (
373+
!record ||
374+
record.action !== 'UPDATE' ||
375+
!['Projects', 'Units'].includes(record.table)
376+
) {
377+
return [];
378+
}
379+
380+
return (parsedChanges[index] || []).map((changeRecord) => changeRecord.issuanceId);
381+
}),
382+
);
383+
384+
const [units, projects, issuances] = await Promise.all([
385+
unitIds.length
386+
? Unit.findAll({
387+
where: { warehouseUnitId: { [Op.in]: unitIds } },
388+
include: buildAssociationIncludes(Unit),
389+
})
390+
: [],
391+
projectIds.length
392+
? Project.findAll({
393+
where: { warehouseProjectId: { [Op.in]: projectIds } },
394+
include: buildAssociationIncludes(Project),
395+
})
396+
: [],
397+
issuanceIds.length
398+
? Issuance.findAll({
399+
where: { id: { [Op.in]: issuanceIds } },
400+
})
401+
: [],
402+
]);
403+
404+
const unitsById = new Map(
405+
units.map((record) => [record.warehouseUnitId, record]),
406+
);
407+
const projectsById = new Map(
408+
projects.map((record) => [record.warehouseProjectId, record]),
409+
);
410+
const issuancesById = new Map(
411+
issuances.map((record) => [record.id, record.dataValues]),
412+
);
413+
414+
return normalizedRecords.map((record, index) => {
415+
const diff = {};
416+
417+
if (record.action === 'INSERT') {
418+
diff.original = {};
419+
diff.change = JSON.parse(record.data);
420+
return diff;
421+
}
422+
423+
if (record.action === 'UPDATE') {
424+
diff.change = parsedChanges[index];
425+
426+
if (['Projects', 'Units'].includes(record.table)) {
427+
diff.change.forEach((changeRecord) => {
428+
if (changeRecord.issuanceId) {
429+
if (!issuancesById.has(changeRecord.issuanceId)) {
430+
logger.warn(
431+
`[v1][staging:getDiffObjects] Missing issuance '${changeRecord.issuanceId}' for staged ${record.table} record '${record.uuid}'`,
432+
);
433+
throw new Error(
434+
`Could not find issuance '${changeRecord.issuanceId}' for staged ${record.table} record '${record.uuid}'`,
435+
);
436+
}
437+
438+
changeRecord.issuance = issuancesById.get(changeRecord.issuanceId);
439+
}
440+
});
441+
}
442+
443+
diff.original =
444+
record.table === 'Projects'
445+
? projectsById.get(record.uuid) ?? null
446+
: record.table === 'Units'
447+
? unitsById.get(record.uuid) ?? null
448+
: undefined;
449+
return diff;
450+
}
451+
452+
if (record.action === 'DELETE') {
453+
diff.original =
454+
record.table === 'Projects'
455+
? projectsById.get(record.uuid) ?? null
456+
: record.table === 'Units'
457+
? unitsById.get(record.uuid) ?? null
458+
: undefined;
459+
diff.change = {};
460+
}
461+
462+
return diff;
463+
});
413464
};
414465

415466
static seperateStagingDataIntoActionGroups = (stagedData, table) => {

0 commit comments

Comments
 (0)