Skip to content

Commit 9b6147a

Browse files
7sete77sete7
committed
feat: konsistent write ahead log
1 parent 573a5f4 commit 9b6147a

File tree

5 files changed

+143
-71
lines changed

5 files changed

+143
-71
lines changed

‎src/imports/data/data.js‎

Lines changed: 48 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -36,10 +36,7 @@ import { TRANSACTION_OPTIONS } from '@imports/consts';
3636
import { find } from "@imports/data/api";
3737
import { client } from '@imports/database';
3838
import { Konsistent } from '@imports/konsistent';
39-
import processIncomingChange from '@imports/konsistent/processIncomingChange';
4039
import eventManager from '@imports/lib/EventManager';
41-
import queueManager from '@imports/queue/QueueManager';
42-
import objectsDiff from '@imports/utils/objectsDiff';
4340
import { dateToString, stringToDate } from '../data/dateParser';
4441
import { populateLookupsData } from '../data/populateLookupsData';
4542
import { processCollectionLogin } from '../data/processCollectionLogin';
@@ -751,11 +748,19 @@ export async function create({ authTokenId, document, data, contextUser, upsert,
751748
}, dbSession);
752749

753750
if (loginFieldResult.success === false) {
751+
await dbSession.abortTransaction();
754752
return loginFieldResult;
755753
}
756754

757755
set(newRecord, get(metaObject, 'login.field', 'login'), loginFieldResult.data);
758756
}
757+
758+
const walResult = await Konsistent.writeAheadLog(document, 'create', newRecord, user, dbSession);
759+
if (walResult.success === false) {
760+
await dbSession.abortTransaction();
761+
return walResult;
762+
}
763+
759764
if (upsert != null && isObject(upsert)) {
760765
const updateOperation = {
761766
$setOnInsert: {},
@@ -881,17 +886,13 @@ export async function create({ authTokenId, document, data, contextUser, upsert,
881886
);
882887
}
883888

884-
// Process sync Konsistent
885-
if (MetaObject.Namespace.plan?.useExternalKonsistent !== true || Konsistent.isQueueEnabled === false) {
886-
try {
887-
tracingSpan?.addEvent('Processing sync Konsistent');
888-
await processIncomingChange(document, resultRecord, 'create', user, resultRecord, dbSession);
889-
} catch (e) {
890-
tracingSpan?.addEvent('Error on Konsistent', { error: e.message });
891-
logger.error(e, `Error on processIncomingChange ${document}: ${e.message}`);
892-
await dbSession.abortTransaction();
893-
return errorReturn(`[${document}] Error on Konsistent: ${e.message}`);
894-
}
889+
try {
890+
await Konsistent.processChangeSync(document, 'create', user, { newRecord: resultRecord }, dbSession);
891+
} catch (e) {
892+
tracingSpan?.addEvent('Error on sync Konsistent', { error: e.message });
893+
logger.error(e, `Error on sync Konsistent ${document}: ${e.message}`);
894+
await dbSession.abortTransaction();
895+
return errorReturn(`[${document}] Error on sync Konsistent: ${e.message}`);
895896
}
896897

897898
return successReturn([dateToString(resultRecord)]);
@@ -907,18 +908,10 @@ export async function create({ authTokenId, document, data, contextUser, upsert,
907908
if (transactionResult.success === true && transactionResult.data?.[0] != null) {
908909
const record = transactionResult.data[0];
909910

910-
// Process Konsistent
911-
if (MetaObject.Namespace.plan?.useExternalKonsistent === true && Konsistent.isQueueEnabled) {
912-
tracingSpan?.addEvent('Sending Konsistent message');
913-
try {
914-
await queueManager.sendMessage(Konsistent.queue.resource, Konsistent.queue.name, {
915-
metaName: document,
916-
operation: 'create',
917-
data: record
918-
});
919-
} catch (e) {
920-
logger.error(e, `Error sending Konsistent message: ${e.message}`);
921-
}
911+
try {
912+
await Konsistent.processChangeAsync(record);
913+
} catch (e) {
914+
logger.error(e, `Error sending Konsistent message: ${e.message}`);
922915
}
923916

924917
// Send events
@@ -1325,6 +1318,21 @@ export async function update({ authTokenId, document, data, contextUser, tracing
13251318
);
13261319
}
13271320

1321+
const walResults = await BluebirdPromise.map(
1322+
updateResults,
1323+
async result => await Konsistent.writeAheadLog(document, 'update', result.data, user, dbSession),
1324+
{ concurrency: 5 },
1325+
);
1326+
1327+
if (walResults.some(result => result.success === false)) {
1328+
await dbSession.abortTransaction();
1329+
return errorReturn(
1330+
walResults
1331+
.filter(result => result.success === false)
1332+
.map(result => result.errors)
1333+
.flat(),
1334+
);
1335+
}
13281336
const updatedIs = updateResults.map(result => result.data._id);
13291337

13301338
if (updatedIs.length > 0) {
@@ -1380,23 +1388,17 @@ export async function update({ authTokenId, document, data, contextUser, tracing
13801388
}
13811389

13821390
// Process sync Konsistent
1383-
if (MetaObject.Namespace.plan?.useExternalKonsistent !== true || Konsistent.isQueueEnabled === false) {
1384-
logger.debug('Processing Konsistent');
1385-
for await (const record of updatedRecords) {
1386-
const original = existsRecords.find(r => r._id === record._id);
1387-
const newRecord = omit(record, ['_id', '_createdAt', '_createdBy', '_updatedAt', '_updatedBy']);
1388-
const changedProps = objectsDiff(original, newRecord);
1391+
for await (const newRecord of updatedRecords) {
1392+
const originalRecord = existsRecords.find(r => r._id === newRecord._id);
13891393

1390-
try {
1391-
tracingSpan?.addEvent('Processing sync Konsistent');
1392-
await processIncomingChange(document, record, 'update', user, changedProps, dbSession);
1393-
} catch (e) {
1394-
logger.error(e, `Error on processIncomingChange ${document}: ${e.message}`);
1395-
tracingSpan?.addEvent('Error on Konsistent', { error: e.message });
1396-
await dbSession.abortTransaction();
1394+
try {
1395+
await Konsistent.processChangeSync(document, 'update', user, { originalRecord, newRecord }, dbSession);
1396+
} catch (e) {
1397+
logger.error(e, `Error on processIncomingChange ${document}: ${e.message}`);
1398+
tracingSpan?.addEvent('Error on Konsistent', { error: e.message });
1399+
await dbSession.abortTransaction();
13971400

1398-
return errorReturn(`[${document}] Error on Konsistent: ${e.message}`);
1399-
}
1401+
return errorReturn(`[${document}] Error on Konsistent: ${e.message}`);
14001402
}
14011403
}
14021404

@@ -1436,23 +1438,11 @@ export async function update({ authTokenId, document, data, contextUser, tracing
14361438
if (transactionResult.success === true && transactionResult.data?.length > 0) {
14371439
const updatedRecords = transactionResult.data;
14381440

1439-
// Process Konsistent
1440-
if (MetaObject.Namespace.plan?.useExternalKonsistent === true && Konsistent.isQueueEnabled) {
1441-
tracingSpan?.addEvent('Sending Konsistent messages');
1442-
for (const record of updatedRecords) {
1443-
try {
1444-
const original = existsRecords.find(r => r._id === record._id);
1445-
const newRecord = omit(record, ['_id', '_createdAt', '_createdBy', '_updatedAt', '_updatedBy']);
1446-
const changedProps = objectsDiff(original, newRecord);
1447-
1448-
await queueManager.sendMessage(Konsistent.queue.resource, Konsistent.queue.name, {
1449-
metaName: document,
1450-
operation: 'update',
1451-
data: changedProps
1452-
});
1453-
} catch (e) {
1454-
logger.error(e, `Error sending Konsistent message: ${e.message}`);
1455-
}
1441+
for (const record of updatedRecords) {
1442+
try {
1443+
await Konsistent.processChangeAsync(record);
1444+
} catch (e) {
1445+
logger.error(e, `Error sending Konsistent message: ${e.message}`);
14561446
}
14571447
}
14581448

‎src/imports/konsistent/index.ts‎

Lines changed: 77 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,38 @@
11
import { MetaObject } from '@imports/model/MetaObject';
22

3+
import { db } from '@imports/database';
4+
import { User } from '@imports/model/User';
5+
import queueManager from '@imports/queue/QueueManager';
6+
import { DataDocument } from '@imports/types/data';
7+
import { LogDocument } from '@imports/types/Konsistent';
38
import getMissingParams from '@imports/utils/getMissingParams';
9+
import objectsDiff from '@imports/utils/objectsDiff';
10+
import { errorReturn, successReturn } from '@imports/utils/return';
11+
import omit from 'lodash/omit';
12+
import { ClientSession, Collection } from 'mongodb';
413
import { logger } from '../utils/logger';
14+
import processIncomingChange from './processIncomingChange';
515

616
type RunningKonsistent = {
717
isQueueEnabled: boolean;
818
queue?: {
919
resource: string | undefined;
1020
name: string | undefined;
1121
};
22+
LogCollection: Collection<LogDocument>;
23+
processChangeSync: typeof processChangeSync;
24+
processChangeAsync: typeof processChangeAsync;
25+
writeAheadLog: typeof writeAheadLog;
26+
removeWAL: typeof removeWAL;
1227
};
1328

1429
export const Konsistent: RunningKonsistent = {
1530
isQueueEnabled: false,
31+
LogCollection: db.collection('Konsistent'),
32+
processChangeSync,
33+
processChangeAsync,
34+
writeAheadLog,
35+
removeWAL,
1636
};
1737

1838
export async function setupKonsistent() {
@@ -28,6 +48,62 @@ export async function setupKonsistent() {
2848
}
2949

3050
if (usingExternalKonsistent && !Konsistent.isQueueEnabled) {
31-
logger.warn('[konsistent] is set to external but no config found');
51+
logger.warn('[konsistent] is set to external but no config found - default to using sync');
52+
}
53+
}
54+
55+
async function processChangeAsync(data: DataDocument) {
56+
if (MetaObject.Namespace.plan?.useExternalKonsistent === true && Konsistent.isQueueEnabled) {
57+
await queueManager.sendMessage(Konsistent.queue!.resource!, Konsistent.queue!.name!, {
58+
_id: data._id,
59+
});
60+
}
61+
}
62+
63+
async function processChangeSync(metaName: string, operation: string, user: object, data: { originalRecord?: DataDocument; newRecord: DataDocument }, dbSession?: ClientSession) {
64+
if (MetaObject.Namespace.plan?.useExternalKonsistent !== true || Konsistent.isQueueEnabled === false) {
65+
logger.debug('Processing sync Konsistent');
66+
67+
const changedProps = data.originalRecord
68+
? objectsDiff(data.originalRecord, omit(data.newRecord, ['_id', '_createdAt', '_createdBy', '_updatedAt', '_updatedBy']))
69+
: omit(data.newRecord, ['_id', '_createdAt', '_createdBy', '_updatedAt', '_updatedBy']);
70+
return processIncomingChange(metaName, data.newRecord, operation, user, changedProps, dbSession);
71+
}
72+
}
73+
74+
async function writeAheadLog(metaName: string, operation: string, data: DataDocument, user: User, dbSession: ClientSession) {
75+
if (MetaObject.Namespace.plan?.useExternalKonsistent === true && Konsistent.isQueueEnabled) {
76+
try {
77+
const result = await Konsistent.LogCollection.insertOne(
78+
{
79+
_id: `${metaName}-${data._id}-${Date.now()}`,
80+
dataId: data._id,
81+
metaName: metaName,
82+
operation: operation,
83+
data: data,
84+
userId: user._id,
85+
ts: new Date(),
86+
},
87+
{ session: dbSession },
88+
);
89+
90+
return result.insertedId ? successReturn(result.insertedId) : errorReturn('Error on writeAheadLog');
91+
} catch (e) {
92+
const message = `Error on writeAheadLog ${metaName}: ${(e as Error).message}`;
93+
logger.error(e, message);
94+
return errorReturn(message);
95+
}
96+
}
97+
98+
return successReturn(null);
99+
}
100+
101+
async function removeWAL(payload: Awaited<ReturnType<typeof writeAheadLog>>) {
102+
if (payload.success === false || payload.data === null) {
103+
return;
104+
}
105+
106+
if (MetaObject.Namespace.plan?.useExternalKonsistent === true && Konsistent.isQueueEnabled) {
107+
await Konsistent.LogCollection.deleteOne({ _id: payload.data });
32108
}
33109
}

‎src/imports/konsistent/processIncomingChange.ts‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,6 @@ import * as References from './updateReferences';
66
import { DataDocument } from '@imports/types/data';
77
import { ClientSession, MongoServerError } from 'mongodb';
88

9-
type Action = 'create' | 'update' | 'delete';
10-
119
const logTimeSpent = (startTime: [number, number], message: string) => {
1210
const totalTime = process.hrtime(startTime);
1311
logger.debug(`${totalTime[0]}s ${totalTime[1] / 1000000}ms => ${message}`);
@@ -16,26 +14,26 @@ const logTimeSpent = (startTime: [number, number], message: string) => {
1614
export default async function processIncomingChange(
1715
metaName: string,
1816
incomingChange: DataDocument,
19-
action: Action,
17+
operation: string,
2018
user: object,
2119
changedProps: Record<string, any>,
2220
dbSession?: ClientSession,
2321
) {
2422
let startTime = process.hrtime();
2523

2624
try {
27-
if (action === 'update') {
25+
if (operation === 'update') {
2826
await References.updateLookups(metaName, incomingChange._id, changedProps, dbSession);
2927
logTimeSpent(startTime, `Updated lookup references for ${metaName}`);
3028
}
3129

32-
await processReverseLookups(metaName, incomingChange._id, incomingChange, action);
30+
await processReverseLookups(metaName, incomingChange._id, incomingChange, operation);
3331
logTimeSpent(startTime, `Process'd reverse lookups for ${metaName}`);
3432

35-
await References.updateRelations(metaName, action, incomingChange._id, incomingChange, dbSession);
33+
await References.updateRelations(metaName, operation, incomingChange._id, incomingChange, dbSession);
3634
logTimeSpent(startTime, `Updated relation references for ${metaName}`);
3735

38-
await createHistory(metaName, action, incomingChange._id, user, new Date(), changedProps, dbSession);
36+
await createHistory(metaName, operation, incomingChange._id, user, new Date(), changedProps, dbSession);
3937
logTimeSpent(startTime, `Created history for ${metaName}`);
4038
} catch (e) {
4139
if ((e as MongoServerError).codeName === 'NoSuchTransaction') {

‎src/imports/konsistent/updateReferences/relationReferences.ts‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,9 @@ import { logger } from '@imports/utils/logger';
1616
import { ClientSession, Collection, FindOptions } from 'mongodb';
1717
import updateRelationReference from './relationReference';
1818

19-
type Action = 'update' | 'create' | 'delete';
2019
const CONCURRENCY = 5;
2120

22-
export default async function updateRelationReferences(metaName: string, action: Action, id: string, data: Record<string, any>, dbSession?: ClientSession) {
21+
export default async function updateRelationReferences(metaName: string, operation: string, id: string, data: Record<string, any>, dbSession?: ClientSession) {
2322
// Get references from meta
2423
let relation, relations, relationsFromDocumentName;
2524
const references = MetaObject.References[metaName];
@@ -33,14 +32,14 @@ export default async function updateRelationReferences(metaName: string, action:
3332
let collection = MetaObject.Collections[metaName];
3433

3534
// If action is delete then get collection trash
36-
if (action === 'delete') {
35+
if (operation === 'delete') {
3736
collection = MetaObject.Collections[`${metaName}.Trash`];
3837
}
3938

4039
const referencesToUpdate: Record<string, Relation[]> = {};
4140

4241
// If action is create or delete then update all records with data related in this record
43-
if (action !== 'update') {
42+
if (operation !== 'update') {
4443
for (relationsFromDocumentName in references.relationsFrom) {
4544
relations = references.relationsFrom[relationsFromDocumentName];
4645
referencesToUpdate[relationsFromDocumentName] = relations;
@@ -121,7 +120,7 @@ export default async function updateRelationReferences(metaName: string, action:
121120
}
122121

123122
// If action is update and the lookup field of relation was updated go to hitory to update old relation
124-
if (lookupId.length > 0 && action === 'update' && has(data, `${relation.lookup}._id`)) {
123+
if (lookupId.length > 0 && operation === 'update' && has(data, `${relation.lookup}._id`)) {
125124
// Try to get history model
126125
const historyCollection = MetaObject.Collections[`${metaName}.History`];
127126

‎src/imports/types/Konsistent.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
export type LogDocument = {
2+
_id: string;
3+
dataId: string;
4+
metaName: string;
5+
operation: string;
6+
data: Record<string, any>;
7+
userId: string;
8+
ts: Date;
9+
};

0 commit comments

Comments
 (0)