From a001f6b9e4a0ec4d8473a7aaa1e15267c8a9440a Mon Sep 17 00:00:00 2001 From: felix Date: Fri, 17 Jan 2025 15:12:52 +0100 Subject: [PATCH] Remove old dispatching rules --- .../leidinggevenden-dispatching/config.js | 32 ---- .../delta-context-config.js | 72 -------- .../delta-sync-dispatching.js | 85 --------- .../initial-sync-dispatching.js | 85 --------- .../leidinggevenden-dispatching/util.js | 168 ------------------ 5 files changed, 442 deletions(-) delete mode 100644 config/loket-consumer/leidinggevenden-dispatching/config.js delete mode 100644 config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js delete mode 100644 config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js delete mode 100644 config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js delete mode 100644 config/loket-consumer/leidinggevenden-dispatching/util.js diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js deleted file mode 100644 index b3ebc2a..0000000 --- a/config/loket-consumer/leidinggevenden-dispatching/config.js +++ /dev/null @@ -1,32 +0,0 @@ -const BATCH_SIZE = parseInt(process.env.BATCH_SIZE) || 100; -const MU_CALL_SCOPE_ID_INITIAL_SYNC = process.env.MU_CALL_SCOPE_ID_INITIAL_SYNC || 'http://redpencil.data.gift/id/concept/muScope/deltas/consumer/initialSync'; -const BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES = process.env.BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES == 'true' ? true : false; -const DIRECT_DATABASE_ENDPOINT = process.env.DIRECT_DATABASE_ENDPOINT || 'http://triplestore:8890/sparql'; -const MAX_DB_RETRY_ATTEMPTS = parseInt(process.env.MAX_DB_RETRY_ATTEMPTS || 5); -const MAX_REASONING_RETRY_ATTEMPTS = parseInt(process.env.MAX_REASONING_RETRY_ATTEMPTS || 5); -const SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); -const SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION || 10000); -const SLEEP_TIME_AFTER_FAILED_DB_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_DB_OPERATION || 60000); - -// Graphs -const TARGET_GRAPH = process.env.TARGET_GRAPH || `http://mu.semte.ch/graphs/administrative-unit`; -const LANDING_ZONE_GRAPH = process.env.DCR_LANDING_ZONE_GRAPH || `http://mu.semte.ch/graphs/landing-zone/leidinggevenden`; - -// ENDPOINTS -const LANDING_ZONE_DATABASE = process.env.DCR_LANDING_ZONE_DATABASE || 'db'; -const LANDING_ZONE_DATABASE_ENDPOINT = process.env.DCR_LANDING_ZONE_DATABASE_ENDPOINT || `http://${LANDING_ZONE_DATABASE}:8890/sparql`; - -module.exports = { - BATCH_SIZE, - MU_CALL_SCOPE_ID_INITIAL_SYNC, - BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, - DIRECT_DATABASE_ENDPOINT, - MAX_DB_RETRY_ATTEMPTS, - MAX_REASONING_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, - LANDING_ZONE_GRAPH, - TARGET_GRAPH, - LANDING_ZONE_DATABASE_ENDPOINT -}; diff --git a/config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js b/config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js deleted file mode 100644 index 973462f..0000000 --- a/config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js +++ /dev/null @@ -1,72 +0,0 @@ - -const PREFIXES = ` -PREFIX org: -PREFIX mandaat: -PREFIX lblodlg: -PREFIX besluit: ` - -const contextConfig = { - addTypes: { - scope: 'all', // 'inserts, 'deletes', 'all' or none. To add rdf:type to subjects of inserts, deletes or both - exhausitive: false, // true or false: find all types for a subject, even if one is already present in delta - }, - contextQueries: [ - // Case Leidnggevenden - // They can be triggered when being inserted or when their link to a bestuursfunctie is added - { - trigger: { // subjectType or predicateValue - subjectType: "lblodlg:Functionaris" - }, - queryTemplate: (subject) => ` - ${PREFIXES} - CONSTRUCT { - ?functionaris org:holds ?bestuursfunctie ; - ?p ?o . - ?orgaanInTime lblodlg:heeftBestuursfunctie ?bestuursfunctie ; - mandaat:isTijdspecialisatieVan ?orgaan . - ?orgaan besluit:bestuurt ?eenheid . - ?eenheid besluit:classificatie ?classification . - } WHERE { - GRAPH { - VALUES ?functionaris { ${subject} } - ?functionaris org:holds ?bestuursfunctie ; - ?p ?o . - ?orgaanInTime lblodlg:heeftBestuursfunctie ?bestuursfunctie ; - mandaat:isTijdspecialisatieVan ?orgaan . - ?orgaan besluit:bestuurt ?eenheid . - ?eenheid besluit:classificatie ?classification . - } - }` - }, - { - trigger: { - predicateValue: "org:holds" - }, - queryTemplate: (subject) => ` - ${PREFIXES} - CONSTRUCT { - ?functionaris org:holds ?bestuursfunctie ; - ?p ?o . - ?orgaanInTime lblodlg:heeftBestuursfunctie ?bestuursfunctie ; - mandaat:isTijdspecialisatieVan ?orgaan . - ?orgaan besluit:bestuurt ?eenheid . - ?eenheid besluit:classificatie ?classification . - } WHERE { - GRAPH { - VALUES ?functionaris { ${subject} } - ?functionaris org:holds ?bestuursfunctie ; - ?p ?o . - ?orgaanInTime lblodlg:heeftBestuursfunctie ?bestuursfunctie ; - mandaat:isTijdspecialisatieVan ?orgaan . - ?orgaan besluit:bestuurt ?eenheid . - ?eenheid besluit:classificatie ?classification . - } - }` - } - ] -} - -module.exports = { - contextConfig, - PREFIXES -}; diff --git a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js deleted file mode 100644 index b309f35..0000000 --- a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js +++ /dev/null @@ -1,85 +0,0 @@ -const { - transformStatements, - deleteFromTargetGraph, - insertIntoTargetGraph, -} = require('./util'); - -/** -* Dispatch the fetched information to a target graph. -* Note: will be ADDED to it's own graph. -* We take only care of adding them, not updating triples, this is a TODO -* @param { mu, muAuthSudo } lib - The provided libraries from the host service. -* @param { termObjectChangeSets: { deletes, inserts } } data - The fetched changes sets, which objects of serialized Terms -* [ { -* graph: "", -* subject: "", -* predicate: "", -* object: "^^" -* } -* ] -* @return {void} Nothing -*/ -async function dispatch(lib, data) { - const { mu, fetch } = lib; - let { termObjectChangeSets, termObjectChangeSetsWithContext } = data; - - // Both arrays are the same length, so we can zip them together for easier processing - zippedChangeSets = termObjectChangeSets.map((o, i) => ({ - original: o, - withContext: termObjectChangeSetsWithContext[i] - })); - - console.log(`Received ${zippedChangeSets.length} change sets`) - - for (let { original, withContext } of zippedChangeSets) { - let originalDeletes, originalInserts, insertsWithContext, deletesWithContext; - try { - originalInserts = original.inserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - originalDeletes = original.deletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - } catch (e) { - console.log(`Error while processing change set: ${e}`); - console.log(`Original change set: ${JSON.stringify(original)}`); - throw e; - } - - // Extra context needed for mapping from DL to OP model and filtering based on type. - try { - insertsWithContext = withContext.inserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - deletesWithContext = withContext.deletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - } catch (e) { - console.log(`Error while processing change set with context: ${e}`); - console.log(`Change set with context: ${JSON.stringify(withContext)}`); - throw e; - } - - if (originalDeletes.length) { - // Map deletes from DL to OP model - const transformedDeletes = await transformStatements(fetch, deletesWithContext); - - if (!transformedDeletes.length) { - console.log(`Warn: Delete statements mapped to empty result.`); - console.log(`Input: ${deletesWithContext}`); - console.log(`Output: ${transformedDeletes}`); - } else { - await deleteFromTargetGraph(lib, transformedDeletes); - } - } - - if (originalInserts.length) { - // Map inserts from DL to OP model - const transformedInserts = await transformStatements(fetch, insertsWithContext); - - if (!transformedInserts.length) { - console.log(`Warn: Insert statements mapped to empty result.`); - console.log(`Input: ${insertsWithContext}`); - console.log(`Output: ${transformedInserts}`); - } else { - await insertIntoTargetGraph(lib, transformedInserts); - } - } - } -} - -module.exports = { - dispatch -}; diff --git a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js deleted file mode 100644 index d5d672f..0000000 --- a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js +++ /dev/null @@ -1,85 +0,0 @@ -const { batchedDbUpdate, transformLandingZoneGraph } = require('./util'); -const { - BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, - DIRECT_DATABASE_ENDPOINT, - MU_CALL_SCOPE_ID_INITIAL_SYNC, - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - TARGET_GRAPH, - LANDING_ZONE_GRAPH, - LANDING_ZONE_DATABASE_ENDPOINT -} = require('./config'); - -const endpoint = BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES ? DIRECT_DATABASE_ENDPOINT : process.env.MU_SPARQL_ENDPOINT; - -/** -* Dispatch the fetched information to a target graph. -* @param { mu, muAuthSudo } lib - The provided libraries from the host service. -* @param { termObjects } data - The fetched quad information, which objects of serialized Terms -* [ { -* graph: "", -* subject: "", -* predicate: "", -* object: "^^" -* } -* ] -* @return {void} Nothing -*/ -async function dispatch(lib, data) { - const { mu, muAuthSudo } = lib; - const { termObjects } = data; - - // Steps: - // 1. Write the triples to the landing zone graph - withouth any mapping or filtering - // 2. One-time reasoning run with the full landing zone graph and write the results to the target graph (on-finish) - - if (BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES) { - console.warn(`Service configured to skip MU_AUTH!`); - } - console.log(`Using ${endpoint} to insert triples`); - - if (termObjects.length) { - const originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - - await batchedDbUpdate( - muAuthSudo.updateSudo, - LANDING_ZONE_GRAPH, - originalInsertTriples, - { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, - LANDING_ZONE_DATABASE_ENDPOINT, - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION - ); - } -} - -async function onFinishInitialIngest(lib) { - const { muAuthSudo, fetch } = lib; - - console.log(`!! On-finish triggered !!`); - - const transformedInsertTriples = await transformLandingZoneGraph(fetch, endpoint); - - console.log(`Transformed ${transformedInsertTriples.length} triples`); - - await batchedDbUpdate( - muAuthSudo.updateSudo, - TARGET_GRAPH, - transformedInsertTriples, - { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, - endpoint, - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION - ); -} - -module.exports = { - dispatch, - onFinishInitialIngest -}; diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js deleted file mode 100644 index 5eced4a..0000000 --- a/config/loket-consumer/leidinggevenden-dispatching/util.js +++ /dev/null @@ -1,168 +0,0 @@ -const { - MAX_REASONING_RETRY_ATTEMPTS, - SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - TARGET_GRAPH, - LANDING_ZONE_GRAPH, - BATCH_SIZE, -} = require('./config'); - -async function batchedDbUpdate( - muUpdate, - graph, - triples, - extraHeaders, - endpoint, - batchSize, - maxAttempts, - sleepBetweenBatches = 1000, - sleepTimeOnFail = 1000, - operation = 'INSERT' -) { - - for (let i = 0; i < triples.length; i += batchSize) { - console.log(`Inserting triples in batch: ${i}-${i + batchSize}`); - - const batch = triples.slice(i, i + batchSize).join('\n'); - - const insertCall = async () => { - await muUpdate(` -${operation} DATA { -GRAPH <${graph}> { -${batch} -} -} -`, extraHeaders, endpoint); - }; - - await operationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); - - console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); - await new Promise(r => setTimeout(r, sleepBetweenBatches)); - } -} - -async function operationWithRetry(callback, - attempt, - maxAttempts, - sleepTimeOnFail) { - try { - if (typeof callback === "function") - return await callback(); - else // Catch error from promise - not how I would do it normally, but allows re use of existing code. - return await callback; - } - catch (e) { - console.log(`Operation failed for ${callback.toString()}, attempt: ${attempt} of ${maxAttempts}`); - console.log(`Error: ${e}`); - console.log(`Sleeping ${sleepTimeOnFail} ms`); - - if (attempt >= maxAttempts) { - console.log(`Max attempts reached for ${callback.toString()}, giving up`); - throw e; - } - - await new Promise(r => setTimeout(r, sleepTimeOnFail)); - return operationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); - } -} - -/** - * Splits an array into two parts, a part that passes and a part that fails a predicate function. - * Credits: https://github.com/benjay10 - * @public - * @function partition - * @param {Array} arr - Array to be partitioned - * @param {Function} fn - Function that accepts single argument: an element of the array, and should return a truthy or falsy value. - * @returns {Object} Object that contains keys passes and fails, each representing an array with elemets that pass or fail the predicate respectively - */ -function partition(arr, fn) { - let passes = [], fails = []; - arr.forEach((item) => (fn(item) ? passes : fails).push(item)); - return { passes, fails }; -} - -/** - * Send triples to reasoning service for conversion - * - */ -async function transformTriples(fetch, triples, mapping) { - return await operationWithRetry(mainConversion(fetch, triples, mapping), 0, - MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION); -} - -async function mainConversion(fetch, triples, mapping) { - let formdata = new URLSearchParams(); - formdata.append("data", triples); - - let requestOptions = { - method: 'POST', - body: formdata, - redirect: 'follow' - }; - - const response = await fetch(`http://reasoner/reason/dl2op/${mapping}`, requestOptions) - return await response.text(); -} - -async function transformStatements(fetch, triples, mapping = 'main') { - console.log(`Received ${JSON.stringify(triples)} to transform`); - const transformedTriples = await transformTriples(fetch, triples.join('\n'), mapping); - statements = transformedTriples ? transformedTriples.replace(/\n{2,}/g, '').split('\n') : []; - console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`); - return statements; -} - -async function deleteFromTargetGraph(lib, statements) { - console.log(`Deleting ${statements.length} statements from target graph`); - console.log(`Statements: ${JSON.stringify(statements)}`) - await batchedDbUpdate( - lib.muAuthSudo.updateSudo, - TARGET_GRAPH, - statements, - {}, - process.env.MU_SPARQL_ENDPOINT, - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - 'DELETE'); -} - -async function insertIntoTargetGraph(lib, statements) { - console.log(`Inserting ${statements.length} statements into target graph`); - console.log(`Statements: ${JSON.stringify(statements)}`) - - await batchedDbUpdate( - lib.muAuthSudo.updateSudo, - TARGET_GRAPH, - statements, - {}, - process.env.MU_SPARQL_ENDPOINT, - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - 'INSERT'); -} - -async function transformLandingZoneGraph(fetch, endpoint, mapping = 'main') { - console.log(`Transforming landing zone graph: ${LANDING_ZONE_GRAPH}`); - - const response = await fetch(`http://reasoner/reason/dl2op/${mapping}?data=${encodeURIComponent(`${endpoint}?default-graph-uri=&query=CONSTRUCT+%7B%0D%0A%3Fs+%3Fp+%3Fo%0D%0A%7D+WHERE+%7B%0D%0A+GRAPH+<${LANDING_ZONE_GRAPH}>+%7B%0D%0A%3Fs+%3Fp+%3Fo%0D%0A%7D%0D%0A%7D&should-sponge=&format=text%2Fplain&timeout=0&run=+Run+Query`)}`); - const text = await response.text(); - const statements = text.replace(/\n{2,}/g, '').split('\n'); - - return statements; -} - -module.exports = { - batchedDbUpdate, - partition, - transformStatements, - transformLandingZoneGraph, - deleteFromTargetGraph, - insertIntoTargetGraph -};