// The reconciliation product's events and computed request servers, as its // scripts at the version the showcase pins run them (reconciliation v8.15.4, // scripts/platform-preset/seeds/reconciliation@8.15.4/scripts: // reconciliation-eventhandler.kts below as EH, reconciliation-reqrep.kts as // RR), with what they call in reconciliation-app and -core 8.15.4 // (ReconciliationManager, ReconStrategyUtils, FieldToleranceUtils, read from // their bytecode). // // A run is the product's lifecycle without its matching: validated and // started as ReconciliationManager starts one, IN_PROGRESS for a while, then // COMPLETED with no record results (the mock reads no data to match). import type { AuditContext } from '../db/audit.ts'; import { primaryKeyIndexName } from '../db/schema.ts'; import { checkUnique } from '../handlers/crudEvents.ts'; import { NackError } from '../protocol/errors.ts'; import type { EventHandler, EventHandlerCtx, EventSchemaOverride, FieldDef, RequestReplyDef, Row, } from '../types.ts'; import { kotlinString } from './javaTime.ts'; import { dto, nullable } from './notify.ts'; // --- schemas --------------------------------------------------------------------- const MESSAGES = 'global.genesis.recon.messages'; const enumOf = ( table: string, name: string, className: string, values: string[], more: Partial = {}, ) => dto(name, 'ENUM', { values, enumClass: `global.genesis.gen.dao.enums.reconciliation.${table}.${className}`, ...more, }); const FIELD_TYPES = [ 'STRING', 'SHORT', 'INT', 'LONG', 'DOUBLE', 'RAW', 'BOOLEAN', 'BIGDECIMAL', 'DATETIME', 'DATE', 'NANO_TIMESTAMP', ]; const TOLERANCE_TYPES = ['EQUALITY', 'PERCENTAGE', 'ABSOLUTE', 'DAYS', 'MINUTES']; const STRATEGIES = ['ONE_TO_ONE', 'ONE_TO_MANY', 'MANY_TO_MANY']; const fieldType = (more: Partial = {}) => enumOf('data_dictionary_field', 'FIELD_TYPE', 'FieldType', FIELD_TYPES, more); const strategy = (more: Partial = {}) => enumOf( 'reconciliation_config', 'RECONCILIATION_STRATEGY', 'ReconciliationStrategy', STRATEGIES, more, ); const listOf = (name: string, className: string, fields: FieldDef[]) => dto(name, 'ARRAY', { items: { type: 'OBJECT', className, nullable: false, fields } }); // ReconciliationRuleData: no defaults, so a nullable TOLERANCE_VALUE is sent too. const RULES = listOf('RULES', `${MESSAGES}.ReconciliationRuleData`, [ dto('FIELD_NAME', 'STRING'), fieldType(), dto('IS_MATCH_KEY', 'BOOLEAN'), enumOf('reconciliation_rule', 'TOLERANCE_TYPE', 'ToleranceType', TOLERANCE_TYPES), dto('TOLERANCE_VALUE', 'STRING', { nullable: true }), ]); const DATASOURCES = listOf('DATASOURCES', `${MESSAGES}.DatasourceConfigData`, [ dto('NAME', 'STRING'), enumOf('datasource_config', 'TYPE', 'Type', ['FILE', 'DATABASE', 'PIPELINE']), dto('INDEX', 'INT'), dto('CONFIG_DATA', 'STRING', nullable), ]); const SCHEMAS: Record = { EVENT_DATA_DICTIONARY_INSERT: { className: `${MESSAGES}.ReconDictionaryInsert`, fields: [ dto('DICTIONARY_NAME', 'STRING'), listOf('FIELDS', `${MESSAGES}.Field`, [dto('FIELD_NAME', 'STRING'), fieldType()]), ], }, // DataDictionary unwrapped, its fields as table entities. EVENT_DATA_DICTIONARY_UPDATE: { className: `${MESSAGES}.ReconDictionaryUpdate`, fields: [ dto('DICTIONARY_ID', 'STRING'), dto('DICTIONARY_NAME', 'STRING'), listOf('FIELDS', 'global.genesis.gen.dao.DataDictionaryField', [ dto('DICTIONARY_ID', 'STRING'), dto('FIELD_NAME', 'STRING'), fieldType({ optional: true }), ]), ], }, EVENT_DATA_DICTIONARY_DELETE: { table: 'DATA_DICTIONARY', pkOnly: true, className: 'global.genesis.gen.dao.DataDictionary.ByDictionaryId', }, EVENT_RECONCILIATION_CONFIG_INSERT: { className: `${MESSAGES}.ReconciliationConfigInsert`, fields: [ dto('NAME', 'STRING'), dto('DICTIONARY_ID', 'STRING'), strategy(), RULES, DATASOURCES, dto('MATCH_NULL', 'BOOLEAN', { optional: true }), dto('PRIMARY_DATASOURCE_NAME', 'STRING', nullable), ], }, // ReconciliationConfig unwrapped: its key and defaulted columns optional, // as the generated entity has them. EVENT_RECONCILIATION_CONFIG_UPDATE: { className: `${MESSAGES}.ReconciliationConfigUpdate`, fields: [ RULES, DATASOURCES, dto('RECONCILIATION_CONFIG_ID', 'STRING', { optional: true }), dto('NAME', 'STRING'), dto('DICTIONARY_ID', 'STRING'), strategy({ optional: true }), dto('PRIMARY_DATASOURCE', 'STRING', nullable), dto('MATCH_NULL', 'BOOLEAN', { optional: true }), ], }, EVENT_RECONCILIATION_CONFIG_DELETE: { table: 'RECONCILIATION_CONFIG', pkOnly: true, className: 'global.genesis.gen.dao.ReconciliationConfig.ById', }, EVENT_RUN_RECONCILIATION: { className: `${MESSAGES}.ReconciliationRun`, fields: [ dto('RECONCILIATION_CONFIG_NAME', 'STRING', nullable), dto('RECONCILIATION_CONFIG_ID', 'STRING', nullable), dto('RECONCILIATION_NAME', 'STRING'), dto('INPUT_PARAMETERS', 'MAP', { items: { type: 'MAP' } }), ], }, EVENT_CANCEL_RECONCILIATION: { table: 'RECONCILIATION', pkOnly: true, className: 'global.genesis.gen.dao.Reconciliation.ById', }, EVENT_UPDATE_RESULT_COMMENT: { className: `${MESSAGES}.RecordComment`, fields: [dto('TEXT', 'STRING'), dto('RECORD_RESULT_ID', 'STRING')], }, }; // --- validators ---------------------------------------------------------------------- // ReconStrategyUtils.validateStrategyElseThrow. const MANY = 3; function validateStrategy(count: number, reconStrategy: unknown, primary: unknown): void { if (reconStrategy === 'ONE_TO_ONE' && count !== 2) { throw new NackError('Reconciliation strategy of ONE_TO_ONE must have two data sources'); } if (reconStrategy === 'ONE_TO_MANY') { if (count < MANY) { throw new NackError( 'Reconciliation strategy of ONE_TO_MANY must have three or more data sources', ); } if (primary === null || primary === undefined || String(primary).trim() === '') { throw new NackError('Reconciliation strategy of ONE_TO_MANY must have a primary datasource'); } } if (reconStrategy === 'MANY_TO_MANY' && count < MANY) { throw new NackError( 'Reconciliation strategy of MANY_TO_MANY must have at least three data sources', ); } } // ReconStrategyUtils.createDataSourcePairs: which datasources are compared. function dataSourcePairs(reconStrategy: unknown, names: string[], primary: unknown): Row[] { validateStrategy(names.length, reconStrategy, primary); const pair = (FIRST_DATA_SOURCE_NAME: string, SECOND_DATA_SOURCE_NAME: string) => ({ FIRST_DATA_SOURCE_NAME, SECOND_DATA_SOURCE_NAME, }); if (reconStrategy === 'ONE_TO_ONE') return [pair(names[0], names.at(-1)!)]; if (reconStrategy === 'ONE_TO_MANY') { const at = names.indexOf(String(primary)); return names.flatMap((name, i) => { if (name === primary) return []; return [i < at ? pair(name, String(primary)) : pair(String(primary), name)]; }); } return names.flatMap((name, i) => names .slice(i) .filter((other) => other !== name) .map((other) => pair(name, other)), ); } // Kotlin's String.toLongOrNull, toDoubleOrNull and toBigDecimalOrNull. const LONG_MAX = 9_223_372_036_854_775_807n; const isLong = (s: string) => /^[+-]?\d+$/.test(s) && BigInt(s) <= LONG_MAX && BigInt(s) >= -LONG_MAX - 1n; // Java's Double grammar, the characters up to a space trimmed off each end. const SPACE = 0x20; const trimToSpace = (s: string) => { let start = 0; let end = s.length; while (start < end && s.charCodeAt(start) <= SPACE) start += 1; while (end > start && s.charCodeAt(end - 1) <= SPACE) end -= 1; return s.slice(start, end); }; const isDouble = (s: string) => /^[+-]?(NaN|Infinity|((\d+\.?\d*|\.\d+)([eE][+-]?\d+)?[fFdD]?))$/.test(trimToSpace(s)); const isBigDecimal = (s: string) => /^[+-]?(\d+\.?\d*|\.\d+)([eE][+-]?\d+)?$/.test(s); type ValueCheck = (value: string | null) => boolean; const anyValue: ValueCheck = () => true; // A whole number: one with only zeros after its point is too. const whole: ValueCheck = (value) => { if (value === null) return false; if (!value.includes('.')) return isLong(value); return isDouble(value) && /^0*$/.test(value.slice(value.lastIndexOf('.') + 1)); }; const number: ValueCheck = (value) => value !== null && isBigDecimal(value); const EQUALITY_ONLY = { EQUALITY: anyValue }; const NUMERIC = { EQUALITY: anyValue, ABSOLUTE: whole, PERCENTAGE: whole }; // FieldToleranceUtils' table: each field type's tolerance types, in order, // and the check on a tolerance value. const TOLERANCES: Record> = { STRING: EQUALITY_ONLY, SHORT: NUMERIC, INT: NUMERIC, LONG: NUMERIC, DOUBLE: NUMERIC, RAW: EQUALITY_ONLY, BOOLEAN: EQUALITY_ONLY, BIGDECIMAL: { EQUALITY: anyValue, ABSOLUTE: number, PERCENTAGE: whole }, DATETIME: { EQUALITY: anyValue, MINUTES: whole }, DATE: { EQUALITY: anyValue, DAYS: whole }, NANO_TIMESTAMP: EQUALITY_ONLY, }; // FieldToleranceUtils.isValidFieldTolerancesElseThrow. function validateTolerances(rules: Row[]): void { for (const rule of rules) { const tolerances = TOLERANCES[rule.FIELD_TYPE]; if (!tolerances) throw new NackError(`Unsupported Field Type '${rule.FIELD_TYPE}'`); const check = tolerances[rule.TOLERANCE_TYPE]; if (!check) { throw new NackError( `Field Type '${rule.FIELD_TYPE}' does not support Tolerance Type '${rule.TOLERANCE_TYPE}'`, ); } const value = rule.TOLERANCE_VALUE ?? null; if (!check(value)) { throw new NackError( `Tolerance value '${value}' is not supported for ${rule.FIELD_TYPE} -> ${rule.TOLERANCE_TYPE} pair`, ); } } } // Jackson's default pretty printer: objects a field a line, `"key" : value`, // arrays inline. function prettyJson(value: unknown, indent = ''): string { if (Array.isArray(value)) { return value.length ? `[ ${value.map((item) => prettyJson(item, indent)).join(', ')} ]` : '[ ]'; } if (value !== null && typeof value === 'object') { const entries = Object.entries(value); if (!entries.length) return '{ }'; const inner = `${indent} `; return `{\n${entries .map(([key, item]) => `${inner}${JSON.stringify(key)} : ${prettyJson(item, inner)}`) .join(',\n')}\n${indent}}`; } return JSON.stringify(value ?? null); } // --- writes ------------------------------------------------------------------------ function insert(ctx: EventHandlerCtx, table: string, row: Row): Row { checkUnique(table, ctx.config.tables![table], row, ctx.store); const written = ctx.store.insertRow(table, row); ctx.broadcastTableChange(table, 'INSERT', ctx.store.rowKey(table, written)); return written; } function modify(ctx: EventHandlerCtx, table: string, key: string, patch: Row): Row { const def = ctx.config.tables![table]; const existing = ctx.store.getRow(table, key); if (!existing) throw NackError.recordNotFound(table, primaryKeyIndexName(table, def)); checkUnique(table, def, { ...existing, ...patch }, ctx.store, key); const written = ctx.store.updateRow(table, key, patch)!; ctx.broadcastTableChange(table, 'MODIFY', key); return written; } function remove(ctx: EventHandlerCtx, table: string, key: string): void { if (!ctx.store.deleteRow(table, key)) throw NackError.recordNotFound(table); ctx.broadcastTableChange(table, 'DELETE', key); } const rowsWhere = ( store: EventHandlerCtx['store'], table: string, column: string, value: unknown, ) => store.getAllRows(table).filter((row) => row[column] === value); const removeWhere = (ctx: EventHandlerCtx, table: string, column: string, value: unknown) => { for (const row of rowsWhere(ctx.store, table, column, value)) remove(ctx, table, ctx.store.rowKey(table, row)); }; // A configuration's datasources, by INDEX (a nullable INT, 0 when unset). const datasourcesOf = (store: EventHandlerCtx['store'], configId: unknown) => rowsWhere(store, 'DATASOURCE_CONFIG', 'RECONCILIATION_CONFIG_ID', configId).sort( (a, b) => Number(a.INDEX ?? 0) - Number(b.INDEX ?? 0), ); // ReconciliationManager's live runs: the IN_PROGRESS ones, here. const activeConfigIds = (store: EventHandlerCtx['store']) => rowsWhere(store, 'RECONCILIATION', 'STATUS', 'IN_PROGRESS').map( (row) => row.RECONCILIATION_CONFIG_ID, ); const LIVE_RUN = 'Reconciliation Config ID is currently being used by a live reconciliation run'; const isBlank = (value: unknown) => value === undefined || value === null || String(value).trim() === ''; // The longest a Node timer waits (2^31 - 1 ms); a longer one fires at once. const MAX_TIMER_MS = 2_147_483_647; // Writes outside the event, audited as `audit` when given: what a // non-transactional handler keeps though its event NACKs, and what the run // writes once it's over. function writeAfter( ctx: EventHandlerCtx, audit: AuditContext | undefined, write: () => void, retry = false, ) { const { store } = ctx; // Inside the event (its transaction still open): once it's over, before // the next message; then, should a transaction stay open, on a timer that // neither spins nor keeps the process up. if (store.inTransaction) { const again = () => writeAfter(ctx, audit, write, true); if (retry) setTimeout(again, 1).unref(); else setImmediate(again); return; } if (!audit) { try { write(); } catch (error) { console.warn('[mock-server] reconciliation: a write after the event failed:', error); } return; } store.begin(audit); try { write(); } catch (error) { store.rollback(); console.warn('[mock-server] reconciliation: a write after the event failed:', error); return; } const audits = store.takeAuditWrites(); store.commit(); for (const { tableName, key } of audits) ctx.broadcastTableChange(tableName, 'INSERT', key); } // --- the backing ------------------------------------------------------------------- // What a FILE datasource's file is to the run: its FILE_STORAGE row's name, or // the message the content fetch fails with. export type FileContent = ( store: EventHandlerCtx['store'], fileStorageId: string, ) => { fileName: string } | { error: string }; export interface ReconciliationOptions { userOf(ctx: EventHandlerCtx): string; // How long a run stays IN_PROGRESS before it completes. runMs: number; // The batch pipelines the RECONCILIATION_MANAGER process defines. pipelines?: string[]; // File-server's content fetch, when file-server is in the app. fileContent?: FileContent; } export interface ReconciliationBacking { eventHandlers: Record; eventSchemas: Record; requestReplies: Record; } export function reconciliationBacking(options: ReconciliationOptions): ReconciliationBacking { const { userOf } = options; // DataSourceFactory.getDataSourceDetails: what each datasource needs before // the run can start. A pipeline is only read once the run has started. function checkDatasources(store: EventHandlerCtx['store'], sources: Row[], inputs: Row): void { for (const source of sources) { if (source.TYPE === 'DATABASE') throw new NackError('DataSource Type DATABASE has not been implemented yet!'); if (source.TYPE === 'PIPELINE') continue; const input = inputs[source.NAME]; if (input === undefined || input === null) { throw new NackError( `Details for datasource: ${source.NAME} can't be found in the input parameters!`, ); } if (isBlank(input.FILE_STORAGE_ID)) { throw new NackError( `File Storage Id has not been provided for input parameter ${kotlinString(input)}`, ); } const content = options.fileContent?.(store, String(input.FILE_STORAGE_ID)); if (!content) continue; if ('error' in content) throw new NackError(content.error); const extension = content.fileName.slice(content.fileName.lastIndexOf('.') + 1); // Named for the datasource, not the extension. if (extension !== 'csv') throw new NackError(`File type for ${source.NAME} has not been implemented yet!`); } } // The run over: COMPLETED, unless it was cancelled meanwhile, or FAILED // when it reads a pipeline that has no name (onReconciliationFailed: the // message as thrown). function complete(ctx: EventHandlerCtx, id: string, sources: Row[]) { const unnamed = sources.some( (source) => source.TYPE === 'PIPELINE' && isBlank(source.CONFIG_DATA), ); // A run that never ends: Infinity, or past what a timer can wait. if (!(options.runMs <= MAX_TIMER_MS)) return; setTimeout(() => { writeAfter(ctx, undefined, () => { const run = ctx.store.getRow('RECONCILIATION', id); if (run?.STATUS !== 'IN_PROGRESS') return; ctx.store.updateRow('RECONCILIATION', id, { STATUS: unnamed ? 'FAILED' : 'COMPLETED', ...(unnamed ? { MESSAGE: 'Unable to invoke pipeline as no pipeline name was specified' } : {}), END_TIMESTAMP: Date.now(), }); ctx.broadcastTableChange('RECONCILIATION', 'MODIFY', id); }); }, options.runMs).unref(); } const eventHandlers: Record = { // EH :35-62 EVENT_DATA_DICTIONARY_INSERT: (details, ctx) => { if (String(details.DICTIONARY_NAME).trim() === '') throw new NackError('Dictionary name must not be empty'); if (!details.FIELDS.length) { throw new NackError('One or more fields are required to create a dictionary'); } if (ctx.validate) return {}; const dictionary = insert(ctx, 'DATA_DICTIONARY', { DICTIONARY_NAME: details.DICTIONARY_NAME, }); for (const field of details.FIELDS as Row[]) { insert(ctx, 'DATA_DICTIONARY_FIELD', { DICTIONARY_ID: dictionary.DICTIONARY_ID, FIELD_NAME: field.FIELD_NAME, FIELD_TYPE: field.FIELD_TYPE, }); } return { generated: [{ DICTIONARY_ID: dictionary.DICTIONARY_ID }] }; }, // EH :64-87: the fields replaced by those sent, each with its own // DICTIONARY_ID. EVENT_DATA_DICTIONARY_UPDATE: (details, ctx) => { if (isBlank(details.DICTIONARY_ID)) throw new NackError('Dictionary ID must not be empty'); if (isBlank(details.DICTIONARY_NAME)) throw new NackError('Dictionary name must not be empty'); if (!details.FIELDS.length) throw new NackError('Dictionary fields must not be empty'); if (ctx.validate) return {}; const id = details.DICTIONARY_ID; modify(ctx, 'DATA_DICTIONARY', id, { DICTIONARY_NAME: details.DICTIONARY_NAME }); removeWhere(ctx, 'DATA_DICTIONARY_FIELD', 'DICTIONARY_ID', id); for (const field of details.FIELDS as Row[]) { insert(ctx, 'DATA_DICTIONARY_FIELD', { DICTIONARY_ID: field.DICTIONARY_ID, FIELD_NAME: field.FIELD_NAME, FIELD_TYPE: field.FIELD_TYPE ?? 'STRING', }); } return {}; }, // EH :89-105: a configuration using it is left pointing at nothing. EVENT_DATA_DICTIONARY_DELETE: (details, ctx) => { if (isBlank(details.DICTIONARY_ID)) throw new NackError('Dictionary ID must not be empty'); if (ctx.validate) return {}; remove(ctx, 'DATA_DICTIONARY', details.DICTIONARY_ID); removeWhere(ctx, 'DATA_DICTIONARY_FIELD', 'DICTIONARY_ID', details.DICTIONARY_ID); return {}; }, // EH :107-157 EVENT_RECONCILIATION_CONFIG_INSERT: (details, ctx) => { validateStrategy( details.DATASOURCES.length, details.RECONCILIATION_STRATEGY, details.PRIMARY_DATASOURCE_NAME, ); validateTolerances(details.RULES); if (!ctx.store.getRow('DATA_DICTIONARY', details.DICTIONARY_ID)) { throw new NackError(`Data Dictionary of ID ${details.DICTIONARY_ID} doesn't exist!`); } if (ctx.validate) return {}; const config = insert(ctx, 'RECONCILIATION_CONFIG', { NAME: details.NAME, DICTIONARY_ID: details.DICTIONARY_ID, RECONCILIATION_STRATEGY: details.RECONCILIATION_STRATEGY, MATCH_NULL: details.MATCH_NULL ?? false, PRIMARY_DATASOURCE: details.PRIMARY_DATASOURCE_NAME ?? null, }); writeRules(ctx, config.RECONCILIATION_CONFIG_ID, details); return { generated: [{ RECONCILIATION_CONFIG_ID: config.RECONCILIATION_CONFIG_ID }] }; }, // EH :160-212: the whole row replaced, rules and datasources too. EVENT_RECONCILIATION_CONFIG_UPDATE: (details, ctx) => { validateStrategy( details.DATASOURCES.length, details.RECONCILIATION_STRATEGY ?? 'ONE_TO_ONE', details.PRIMARY_DATASOURCE, ); validateTolerances(details.RULES); if (!ctx.store.getRow('DATA_DICTIONARY', details.DICTIONARY_ID)) { throw new NackError(`Data Dictionary of ID ${details.DICTIONARY_ID} doesn't exist!`); } if (activeConfigIds(ctx.store).includes(details.RECONCILIATION_CONFIG_ID)) throw new NackError(LIVE_RUN); if (ctx.validate) return {}; const id = details.RECONCILIATION_CONFIG_ID; modify(ctx, 'RECONCILIATION_CONFIG', id, { NAME: details.NAME, DICTIONARY_ID: details.DICTIONARY_ID, RECONCILIATION_STRATEGY: details.RECONCILIATION_STRATEGY ?? 'ONE_TO_ONE', PRIMARY_DATASOURCE: details.PRIMARY_DATASOURCE ?? null, MATCH_NULL: details.MATCH_NULL ?? false, }); removeWhere(ctx, 'RECONCILIATION_RULE', 'RECONCILIATION_CONFIG_ID', id); removeWhere(ctx, 'DATASOURCE_CONFIG', 'RECONCILIATION_CONFIG_ID', id); writeRules(ctx, id, details); return {}; }, // EH :215-235: its past runs are kept. EVENT_RECONCILIATION_CONFIG_DELETE: (details, ctx) => { const id = details.RECONCILIATION_CONFIG_ID; if (!ctx.store.getRow('RECONCILIATION_CONFIG', id)) { throw new NackError(`Reconciliation Configuration of ID ${id} doesn't exist!`); } if (activeConfigIds(ctx.store).includes(id)) throw new NackError(LIVE_RUN); if (ctx.validate) return {}; remove(ctx, 'RECONCILIATION_CONFIG', id); removeWhere(ctx, 'RECONCILIATION_RULE', 'RECONCILIATION_CONFIG_ID', id); removeWhere(ctx, 'DATASOURCE_CONFIG', 'RECONCILIATION_CONFIG_ID', id); return {}; }, // EH :238-282, ReconciliationManager.runReconciliation. Not transactional: // a run that fails to start NACKs, and its row stays FAILED. EVENT_RUN_RECONCILIATION: (details, ctx) => { const id = details.RECONCILIATION_CONFIG_ID ?? null; const name = details.RECONCILIATION_CONFIG_NAME ?? null; if (id === null && name === null) { throw new NackError('Either configuration ID or configuration name must be provided'); } let config: Row | undefined; if (id !== null) config = ctx.store.getRow('RECONCILIATION_CONFIG', id); else { // getRange(ByName).toList().first(): NoSuchElementException. [config] = rowsWhere(ctx.store, 'RECONCILIATION_CONFIG', 'NAME', name); if (!config) throw NackError.nack('List is empty.'); } if (!config) throw new NackError(`Reconciliation Config for ID ${id} can't be found!`); const sources = datasourcesOf(ctx.store, config.RECONCILIATION_CONFIG_ID); if (!sources.length) { throw new NackError( `No datasource config has been set up for Reconciliation Config ID: ${id}`, ); } validateStrategy(sources.length, config.RECONCILIATION_STRATEGY, config.PRIMARY_DATASOURCE); if ( !rowsWhere(ctx.store, 'DATA_DICTIONARY_FIELD', 'DICTIONARY_ID', config.DICTIONARY_ID).length ) { throw new NackError( `No fields have been set up for Data Dictionary ID: ${config.DICTIONARY_ID}`, ); } if (ctx.validate) return {}; const run = insert(ctx, 'RECONCILIATION', { RECONCILIATION_CONFIG_ID: config.RECONCILIATION_CONFIG_ID, RECONCILIATION_NAME: isBlank(details.RECONCILIATION_NAME) ? config.NAME : details.RECONCILIATION_NAME, STATUS: 'IN_PROGRESS', INPUT_PARAMETERS: prettyJson(details.INPUT_PARAMETERS), START_TIMESTAMP: Date.now(), MESSAGE: null, CANCELED_BY: null, END_TIMESTAMP: null, }); try { checkDatasources(ctx.store, sources, details.INPUT_PARAMETERS); } catch (error) { const message = error instanceof NackError ? error.errors[0]?.TEXT : String(error); const { RECONCILIATION_ID, RECORD_ID, TIMESTAMP, ...columns } = run; const audit = { eventType: ctx.message.MESSAGE_TYPE, user: userOf(ctx), text: null }; // Under fidelity: 'legacy' no transaction undoes the IN_PROGRESS row: // that one fails; else it is written again once the event rolls back. const kept = !ctx.store.inTransaction; writeAfter(ctx, audit, () => { const id = kept ? RECONCILIATION_ID : ctx.store.insertRow('RECONCILIATION', columns).RECONCILIATION_ID; if (!kept) ctx.broadcastTableChange('RECONCILIATION', 'INSERT', id); ctx.store.updateRow('RECONCILIATION', id, { STATUS: 'FAILED', MESSAGE: `Error caught while attempting to start reconciliation task, ${message}`, }); ctx.broadcastTableChange('RECONCILIATION', 'MODIFY', id); }); throw error; } complete(ctx, run.RECONCILIATION_ID, sources); return { generated: [{ RECONCILIATION_ID: run.RECONCILIATION_ID }] }; }, // EH :285-303: cancelled once the event is over, as the workflow does. EVENT_CANCEL_RECONCILIATION: (details, ctx) => { const user = userOf(ctx); const id = details.RECONCILIATION_ID; if (isBlank(user)) throw new NackError('No User name has been provided'); const run = ctx.store.getRow('RECONCILIATION', id); if (!run) throw new NackError(`Reconciliation record does not exist for ID: ${id}`); if (run.STATUS !== 'IN_PROGRESS') { throw new NackError( `Reconciliation of ID: ${id} cannot be cancelled as it is not running. The current status is: ${run.STATUS}`, ); } if (ctx.validate) return {}; setImmediate(() => writeAfter(ctx, undefined, () => { // Unless it has ended since. if (ctx.store.getRow('RECONCILIATION', id)?.STATUS !== 'IN_PROGRESS') return; ctx.store.updateRow('RECONCILIATION', id, { STATUS: 'CANCELLED', CANCELED_BY: user, END_TIMESTAMP: Date.now(), }); ctx.broadcastTableChange('RECONCILIATION', 'MODIFY', id); }), ); return {}; }, // EH :306-326: one comment a result, replaced. EVENT_UPDATE_RESULT_COMMENT: (details, ctx) => { const id = details.RECORD_RESULT_ID; if (isBlank(id)) throw new NackError('Comment can not be added as record result id is missing'); if (!ctx.store.getRow('RECONCILIATION_RECORD_RESULT', id)) { throw new NackError( `Comment could not be added as record result for ID '${id}' does not exist`, ); } if (ctx.validate) return {}; const comment = { RECONCILIATION_RECORD_RESULT_ID: id, TEXT: details.TEXT, USER_NAME: userOf(ctx), }; if (ctx.store.getRow('RECONCILIATION_RECORD_RESULT_COMMENT', id)) { modify(ctx, 'RECONCILIATION_RECORD_RESULT_COMMENT', id, comment); } else { insert(ctx, 'RECONCILIATION_RECORD_RESULT_COMMENT', comment); } return {}; }, }; // A configuration's rules and datasources, under its id. function writeRules(ctx: EventHandlerCtx, configId: string, details: Row) { for (const rule of details.RULES as Row[]) { insert(ctx, 'RECONCILIATION_RULE', { RECONCILIATION_CONFIG_ID: configId, FIELD_NAME: rule.FIELD_NAME, IS_MATCH_KEY: rule.IS_MATCH_KEY, TOLERANCE_TYPE: rule.TOLERANCE_TYPE, TOLERANCE_VALUE: rule.TOLERANCE_VALUE ?? null, }); } for (const source of details.DATASOURCES as Row[]) { insert(ctx, 'DATASOURCE_CONFIG', { RECONCILIATION_CONFIG_ID: configId, NAME: source.NAME, TYPE: source.TYPE, INDEX: source.INDEX, CONFIG_DATA: source.CONFIG_DATA ?? null, }); } } // --- request servers -------------------------------------------------------------- // RECONCILIATION_DATA_SOURCE_PAIRS (RR :69-89). const pairs: RequestReplyDef['resolver'] = ({ request, store }) => { const id = request.RECONCILIATION_ID; if (isBlank(id)) throw new Error('Reconciliation ID is missing'); const run = store.getRow('RECONCILIATION', id); if (!run) throw new Error(`Reconciliation with ID '${id}' does not exist`); const config = store.getRow('RECONCILIATION_CONFIG', run.RECONCILIATION_CONFIG_ID); if (!config) { throw new Error( `Reconciliation Config with ID '${run.RECONCILIATION_CONFIG_ID}' does not exist`, ); } const names = datasourcesOf(store, config.RECONCILIATION_CONFIG_ID).map( (row) => row.NAME as string, ); // A strategy the datasources don't fit is an exception like any other: // GENERIC_ERROR, with its text. try { return [ { PAIRS: dataSourcePairs(config.RECONCILIATION_STRATEGY, names, config.PRIMARY_DATASOURCE), }, ]; } catch (error) { if (error instanceof NackError) throw new Error(error.errors[0]?.TEXT); throw error; } }; // RECONCILIATION_RECORD_PAIR_RESULT (RR :92-135): each side's rule values. const pairResult: RequestReplyDef['resolver'] = ({ request, store }) => { const id = request.RECONCILIATION_RECORD_RESULT_ID; if (isBlank(id)) throw new Error('Reconciliation Record Result ID is missing'); const result = store.getRow('RECONCILIATION_RECORD_RESULT', id); if (!result) throw new Error(`Reconciliation Record Result of ID '${id}' does not exist`); const rules = rowsWhere( store, 'RECONCILIATION_RULE_RESULT', 'RECONCILIATION_RECORD_RESULT_ID', id, ); const side = (name: unknown, value: string) => ({ DATA_SOURCE_NAME: name ?? null, RULES_INFO: rules.map((rule) => ({ FIELD_NAME: rule.FIELD_NAME, FIELD_VALUE: rule[value] ?? null, RESULT: rule.RESULT, })), }); return [ { KEY_RESULT: result.KEY_RESULT, COMPARISON_RESULT: result.COMPARISON_RESULT, FIRST_DATA_SOURCE_DATA: side(result.FIRST_DATASOURCE_NAME, 'FIRST_VALUE'), SECOND_DATA_SOURCE_DATA: side(result.SECOND_DATASOURCE_NAME, 'SECOND_VALUE'), }, ]; }; return { eventHandlers, eventSchemas: SCHEMAS, requestReplies: { // RR :148-156: every running run's configuration, repeats kept. ACTIVE_RECONCILIATION_CONFIG_IDS: { requestFields: [], resolver: ({ store }) => activeConfigIds(store).map((RECONCILIATION_CONFIG_ID) => ({ RECONCILIATION_CONFIG_ID })), metadataRows: [{ RECONCILIATION_CONFIG_ID: '' }], }, // RR :138-146: a field type's tolerance types; REQUEST is the type. FIELD_TYPE_TOLERANCES: { requestFields: [], resolver: ({ message }) => Object.keys(TOLERANCES[String(message.REQUEST)] ?? {}).map((TOLERANCE_TYPE) => ({ TOLERANCE_TYPE, })), metadataRows: [{ TOLERANCE_TYPE: 'EQUALITY' }], }, // reconciliation-data-pipelines-reqrep.kts :8-18. RECONCILIATION_DATA_PIPELINES: { requestFields: [], resolver: () => [ { BATCH_DATA_SOURCE_NAMES: [...(options.pipelines ?? [])], REAL_TIME_DATA_SOURCE_NAMES: [], }, ], metadataRows: [{ BATCH_DATA_SOURCE_NAMES: [], REAL_TIME_DATA_SOURCE_NAMES: [] }], }, RECONCILIATION_DATA_SOURCE_PAIRS: { requestFields: ['RECONCILIATION_ID'], resolver: pairs, metadataRows: [{ PAIRS: [] }], }, RECONCILIATION_RECORD_PAIR_RESULT: { requestFields: ['RECONCILIATION_RECORD_RESULT_ID'], resolver: pairResult, metadataRows: [ { KEY_RESULT: 'MATCHED', COMPARISON_RESULT: 'MATCHED', FIRST_DATA_SOURCE_DATA: {}, SECOND_DATA_SOURCE_DATA: {}, }, ], }, }, }; }