From fefb103cb5763bd0c61a362c2d5d5aed2a37063e Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Fri, 21 Apr 2023 11:57:15 +0200 Subject: [PATCH] Consume mandatarissen, leidinggevenden and worship positions from Loket (#239) * loket filter * cleanup previously migrated data before sync * increase mu-auth timouts * split up deletion migration to avoid timouts * identical utls added catch statements * deletes before inserts * Adapt migrations to splitted graphs * Adapt consumers to the delta-consumer with contexts * Bump consumers and cache * Resolve cache not getting cleared * Flush ministers of non active worship services too --------- Co-authored-by: Claire Lovisa --- .../leidinggevenden-dispatching/config.js | 17 +- .../delta-context-config.js | 11 + .../delta-sync-dispatching.js | 93 ++++---- .../initial-sync-dispatching.js | 68 ++++-- .../leidinggevenden-dispatching/util.js | 119 +++++----- .../config.js | 18 +- .../delta-context-config.js | 11 + .../delta-sync-dispatching.js | 85 ++++++++ .../initial-sync-dispatching.js | 85 ++++++++ .../util.js | 118 +++++----- .../delta-sync-dispatching.js | 70 ------ .../initial-sync-dispatching.js | 57 ----- .../mandate-dispatching/util.js | 153 ------------- .../positions-dispatching/config.js | 51 ----- .../delta-sync-dispatching.js | 88 -------- .../initial-sync-dispatching.js | 75 ------- .../config.js | 40 ++++ .../delta-context-config.js | 11 + .../delta-sync-dispatching.js | 114 ++++++++++ .../initial-sync-dispatching.js | 106 +++++++++ .../util.js | 204 ++++++++++++++++++ ...00-remove-loket-persons-before-sync.sparql | 54 +++++ ...e-loket-functionarissen-before-sync.sparql | 52 +++++ ...et-mandatarissen-before-sync-part-1.sparql | 53 +++++ ...et-mandatarissen-before-sync-part-2.sparql | 53 +++++ ...et-mandatarissen-before-sync-part-3.sparql | 52 +++++ ...move-ministers-that-will-get-synced.sparql | 21 ++ .../reasoner/dl2op/main/inverse_relations.n3 | 1 - config/reasoner/dl2op/main/mapping.n3 | 26 ++- .../dl2op/main/persoon_http_to_https.n3 | 1 - config/reasoner/dl2op/main/query.n3q | 63 ++++++ .../dl2op/main/retract_empty_strings.n3 | 1 - config/reasoner/dl2op/main/rrn.n3 | 2 - config/resources/domain.lisp | 1 + docker-compose.yml | 54 +++-- 35 files changed, 1329 insertions(+), 699 deletions(-) create mode 100644 config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js rename config/loket-consumer/{mandate-dispatching => mandatarissen-dispatching}/config.js (69%) create mode 100644 config/loket-consumer/mandatarissen-dispatching/delta-context-config.js create mode 100644 config/loket-consumer/mandatarissen-dispatching/delta-sync-dispatching.js create mode 100644 config/loket-consumer/mandatarissen-dispatching/initial-sync-dispatching.js rename config/loket-consumer/{positions-dispatching => mandatarissen-dispatching}/util.js (50%) delete mode 100644 config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js delete mode 100644 config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js delete mode 100644 config/loket-consumer/mandate-dispatching/util.js delete mode 100644 config/loket-consumer/positions-dispatching/config.js delete mode 100644 config/loket-consumer/positions-dispatching/delta-sync-dispatching.js delete mode 100644 config/loket-consumer/positions-dispatching/initial-sync-dispatching.js create mode 100644 config/loket-consumer/worship-services-sensitive-dispatching/config.js create mode 100644 config/loket-consumer/worship-services-sensitive-dispatching/delta-context-config.js create mode 100644 config/loket-consumer/worship-services-sensitive-dispatching/delta-sync-dispatching.js create mode 100644 config/loket-consumer/worship-services-sensitive-dispatching/initial-sync-dispatching.js create mode 100644 config/loket-consumer/worship-services-sensitive-dispatching/util.js create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164600-remove-loket-persons-before-sync.sparql create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164601-remove-loket-functionarissen-before-sync.sparql create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164602-remove-loket-mandatarissen-before-sync-part-1.sparql create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164603-remove-loket-mandatarissen-before-sync-part-2.sparql create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164604-remove-loket-mandatarissen-before-sync-part-3.sparql create mode 100644 config/migrations/2023/20230330164600-sync-loket-to-op/20230330164605-remove-ministers-that-will-get-synced.sparql create mode 100644 config/reasoner/dl2op/main/query.n3q diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js index 2afb525..b3ebc2a 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/config.js +++ b/config/loket-consumer/leidinggevenden-dispatching/config.js @@ -1,18 +1,20 @@ 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://virtuoso:8890/sparql'; +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); -const INGEST_GRAPH = process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/administrative-unit`; +// 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`; -if (!process.env.FILE_SYNC_GRAPH) - throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; -const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; +// 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, @@ -24,6 +26,7 @@ module.exports = { SLEEP_BETWEEN_BATCHES, SLEEP_TIME_AFTER_FAILED_DB_OPERATION, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, - INGEST_GRAPH, - FILE_SYNC_GRAPH + 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 new file mode 100644 index 0000000..a8df81c --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/delta-context-config.js @@ -0,0 +1,11 @@ +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: [] +} + +module.exports = { + contextConfig +}; diff --git a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js index 669895b..b309f35 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js +++ b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js @@ -1,11 +1,8 @@ -const { transformStatements, batchedDbUpdate, deleteFromAllGraphs } = require('./util'); const { - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - INGEST_GRAPH -} = require('./config'); + transformStatements, + deleteFromTargetGraph, + insertIntoTargetGraph, +} = require('./util'); /** * Dispatch the fetched information to a target graph. @@ -23,43 +20,61 @@ const { * @return {void} Nothing */ async function dispatch(lib, data) { - const { mu, muAuthSudo, fetch } = lib; - const { termObjectChangeSets } = data; + const { mu, fetch } = lib; + let { termObjectChangeSets, termObjectChangeSetsWithContext } = data; - for (let { deletes, inserts } of termObjectChangeSets) { - const deleteStatements = deletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - const insertStatements = inserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + // 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] + })); - if (deleteStatements.length) { - const transformedStatementsToDelete = await transformStatements(fetch, deleteStatements); - if (transformedStatementsToDelete.length) { - await deleteFromAllGraphs( - muAuthSudo.updateSudo, - transformedStatementsToDelete, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - ); + 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 (insertStatements.length) { - const transformedStatementsToInsert = await transformStatements(fetch, insertStatements); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - "INSERT" - ); + 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); } } } diff --git a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js index 7967073..d5d672f 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js +++ b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js @@ -1,13 +1,17 @@ -const { transformStatements, batchedDbUpdate } = require('./util'); -const { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, +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, - INGEST_GRAPH + 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; /** @@ -24,34 +28,58 @@ const endpoint = BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES ? DIRECT_DATABASE_ENDPOINT * @return {void} Nothing */ async function dispatch(lib, data) { - const { mu, muAuthSudo, fetch } = lib; + 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) { - originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`) + const originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - const transformedStatementsToInsert = await transformStatements(fetch, originalInsertTriples); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { '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 - ); - } + 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 + dispatch, + onFinishInitialIngest }; diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js index bdf477b..5eced4a 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/util.js +++ b/config/loket-consumer/leidinggevenden-dispatching/util.js @@ -1,9 +1,16 @@ const { MAX_REASONING_RETRY_ATTEMPTS, - SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION + 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, +async function batchedDbUpdate( + muUpdate, graph, triples, extraHeaders, @@ -37,39 +44,6 @@ ${batch} } } -async function deleteFromAllGraphs(muUpdate, - triples, - extraHeaders, - endpoint, - maxAttempts, - sleepBetweenBatches = 1000, - sleepTimeOnFail = 1000, -) { - - for (const triple of triples) { - - console.log(`Deleting a triple from all graphs in triplestore`); - - const deleteCall = async () => { - await muUpdate(` - DELETE { - GRAPH ?g { - ${triple} - } - } WHERE { - GRAPH ?g { - ${triple} - } - } - `, extraHeaders, endpoint); - }; - - await operationWithRetry(deleteCall, 0, maxAttempts, sleepTimeOnFail); - console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); - await new Promise(r => setTimeout(r, sleepBetweenBatches)); - } -} - async function operationWithRetry(callback, attempt, maxAttempts, @@ -110,18 +84,16 @@ function partition(arr, fn) { return { passes, fails }; } - /** * Send triples to reasoning service for conversion * */ -function transformTriples(fetch, triples) { - return operationWithRetry(mainConversion(fetch, triples), 0, +async function transformTriples(fetch, triples, mapping) { + return await operationWithRetry(mainConversion(fetch, triples, mapping), 0, MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION); } - -function mainConversion(fetch, triples) { +async function mainConversion(fetch, triples, mapping) { let formdata = new URLSearchParams(); formdata.append("data", triples); @@ -131,23 +103,66 @@ function mainConversion(fetch, triples) { redirect: 'follow' }; - return fetch("http://reasoner/reason/dl2op/main", requestOptions) - .then(response => response.text()); + const response = await fetch(`http://reasoner/reason/dl2op/${mapping}`, requestOptions) + return await response.text(); } -function transformStatements(fetch, triples) { - return transformTriples(fetch, triples.join('\n')).then( - graph => { - statements = graph.replace(/\n{2,}/g, '').split('\n') - console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) - return statements - } - ) +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, - deleteFromAllGraphs, partition, - transformStatements + transformStatements, + transformLandingZoneGraph, + deleteFromTargetGraph, + insertIntoTargetGraph }; diff --git a/config/loket-consumer/mandate-dispatching/config.js b/config/loket-consumer/mandatarissen-dispatching/config.js similarity index 69% rename from config/loket-consumer/mandate-dispatching/config.js rename to config/loket-consumer/mandatarissen-dispatching/config.js index 2afb525..13291cf 100644 --- a/config/loket-consumer/mandate-dispatching/config.js +++ b/config/loket-consumer/mandatarissen-dispatching/config.js @@ -1,18 +1,20 @@ 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://virtuoso:8890/sparql'; +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); -const INGEST_GRAPH = process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/administrative-unit`; +// 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`; -if (!process.env.FILE_SYNC_GRAPH) - throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; -const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; +// 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, @@ -24,6 +26,8 @@ module.exports = { SLEEP_BETWEEN_BATCHES, SLEEP_TIME_AFTER_FAILED_DB_OPERATION, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, - INGEST_GRAPH, - FILE_SYNC_GRAPH + LANDING_ZONE_GRAPH, + LANDING_ZONE_DATABASE_ENDPOINT, + TARGET_GRAPH }; + diff --git a/config/loket-consumer/mandatarissen-dispatching/delta-context-config.js b/config/loket-consumer/mandatarissen-dispatching/delta-context-config.js new file mode 100644 index 0000000..a8df81c --- /dev/null +++ b/config/loket-consumer/mandatarissen-dispatching/delta-context-config.js @@ -0,0 +1,11 @@ +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: [] +} + +module.exports = { + contextConfig +}; diff --git a/config/loket-consumer/mandatarissen-dispatching/delta-sync-dispatching.js b/config/loket-consumer/mandatarissen-dispatching/delta-sync-dispatching.js new file mode 100644 index 0000000..b309f35 --- /dev/null +++ b/config/loket-consumer/mandatarissen-dispatching/delta-sync-dispatching.js @@ -0,0 +1,85 @@ +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/mandatarissen-dispatching/initial-sync-dispatching.js b/config/loket-consumer/mandatarissen-dispatching/initial-sync-dispatching.js new file mode 100644 index 0000000..d5d672f --- /dev/null +++ b/config/loket-consumer/mandatarissen-dispatching/initial-sync-dispatching.js @@ -0,0 +1,85 @@ +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/positions-dispatching/util.js b/config/loket-consumer/mandatarissen-dispatching/util.js similarity index 50% rename from config/loket-consumer/positions-dispatching/util.js rename to config/loket-consumer/mandatarissen-dispatching/util.js index f6411a2..524fe45 100644 --- a/config/loket-consumer/positions-dispatching/util.js +++ b/config/loket-consumer/mandatarissen-dispatching/util.js @@ -1,9 +1,16 @@ const { MAX_REASONING_RETRY_ATTEMPTS, - SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION + 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, +async function batchedDbUpdate( + muUpdate, graph, triples, extraHeaders, @@ -37,39 +44,6 @@ ${batch} } } -async function deleteFromAllGraphs(muUpdate, - triples, - extraHeaders, - endpoint, - maxAttempts, - sleepBetweenBatches = 1000, - sleepTimeOnFail = 1000, -) { - - for (const triple of triples) { - - console.log(`Deleting a triple from all graphs in triplestore`); - - const deleteCall = async () => { - await muUpdate(` - DELETE { - GRAPH ?g { - ${triple} - } - } WHERE { - GRAPH ?g { - ${triple} - } - } - `, extraHeaders, endpoint); - }; - - await operationWithRetry(deleteCall, 0, maxAttempts, sleepTimeOnFail); - console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); - await new Promise(r => setTimeout(r, sleepBetweenBatches)); - } -} - async function operationWithRetry(callback, attempt, maxAttempts, @@ -110,18 +84,16 @@ function partition(arr, fn) { return { passes, fails }; } - /** * Send triples to reasoning service for conversion * */ -function transformTriples(fetch, triples, mapping) { - return operationWithRetry(mainConversion(fetch, triples, mapping), 0, +async function transformTriples(fetch, triples, mapping) { + return await operationWithRetry(mainConversion(fetch, triples, mapping), 0, MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION); } - -function mainConversion(fetch, triples, mapping) { +async function mainConversion(fetch, triples, mapping) { let formdata = new URLSearchParams(); formdata.append("data", triples); @@ -131,24 +103,66 @@ function mainConversion(fetch, triples, mapping) { redirect: 'follow' }; - return fetch(`http://reasoner/reason/dl2op/${mapping}`, requestOptions) - .then(response => response.text()); + 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; +} -function transformStatements(fetch, triples, mapping) { - return transformTriples(fetch, triples.join('\n'), mapping).then( - graph => { - statements = graph ? graph.split('\n') : []; - console.log(`CONVERSION ${mapping}: 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, - deleteFromAllGraphs, partition, - transformStatements + transformStatements, + transformLandingZoneGraph, + deleteFromTargetGraph, + insertIntoTargetGraph }; diff --git a/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js b/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js deleted file mode 100644 index 669895b..0000000 --- a/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js +++ /dev/null @@ -1,70 +0,0 @@ -const { transformStatements, batchedDbUpdate, deleteFromAllGraphs } = require('./util'); -const { - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - INGEST_GRAPH -} = require('./config'); - -/** -* 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, muAuthSudo, fetch } = lib; - const { termObjectChangeSets } = data; - - for (let { deletes, inserts } of termObjectChangeSets) { - const deleteStatements = deletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - const insertStatements = inserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - - if (deleteStatements.length) { - const transformedStatementsToDelete = await transformStatements(fetch, deleteStatements); - if (transformedStatementsToDelete.length) { - await deleteFromAllGraphs( - muAuthSudo.updateSudo, - transformedStatementsToDelete, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - ); - } - } - - if (insertStatements.length) { - const transformedStatementsToInsert = await transformStatements(fetch, insertStatements); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - "INSERT" - ); - } - } - } -} - -module.exports = { - dispatch -}; diff --git a/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js b/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js deleted file mode 100644 index 152abda..0000000 --- a/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js +++ /dev/null @@ -1,57 +0,0 @@ -const { transformStatements, batchedDbUpdate } = 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, - INGEST_GRAPH -} = 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, fetch } = lib; - const { termObjects } = data; - - 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) { - originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`) - - const transformedStatementsToInsert = await transformStatements(fetch, originalInsertTriples); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { '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 -}; diff --git a/config/loket-consumer/mandate-dispatching/util.js b/config/loket-consumer/mandate-dispatching/util.js deleted file mode 100644 index bdf477b..0000000 --- a/config/loket-consumer/mandate-dispatching/util.js +++ /dev/null @@ -1,153 +0,0 @@ -const { - MAX_REASONING_RETRY_ATTEMPTS, - SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION -} = 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 deleteFromAllGraphs(muUpdate, - triples, - extraHeaders, - endpoint, - maxAttempts, - sleepBetweenBatches = 1000, - sleepTimeOnFail = 1000, -) { - - for (const triple of triples) { - - console.log(`Deleting a triple from all graphs in triplestore`); - - const deleteCall = async () => { - await muUpdate(` - DELETE { - GRAPH ?g { - ${triple} - } - } WHERE { - GRAPH ?g { - ${triple} - } - } - `, extraHeaders, endpoint); - }; - - await operationWithRetry(deleteCall, 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 - * - */ -function transformTriples(fetch, triples) { - return operationWithRetry(mainConversion(fetch, triples), 0, - MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION); -} - - -function mainConversion(fetch, triples) { - let formdata = new URLSearchParams(); - formdata.append("data", triples); - - let requestOptions = { - method: 'POST', - body: formdata, - redirect: 'follow' - }; - - return fetch("http://reasoner/reason/dl2op/main", requestOptions) - .then(response => response.text()); -} - -function transformStatements(fetch, triples) { - return transformTriples(fetch, triples.join('\n')).then( - graph => { - statements = graph.replace(/\n{2,}/g, '').split('\n') - console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) - return statements - } - ) -} - -module.exports = { - batchedDbUpdate, - deleteFromAllGraphs, - partition, - transformStatements -}; diff --git a/config/loket-consumer/positions-dispatching/config.js b/config/loket-consumer/positions-dispatching/config.js deleted file mode 100644 index ab34eca..0000000 --- a/config/loket-consumer/positions-dispatching/config.js +++ /dev/null @@ -1,51 +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://virtuoso: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 -); -const INGEST_GRAPH = - process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/worship-service`; -const PRIVACY_SENSITIVE_GRAPH = - process.env.PRIVACY_SENSITIVE_GRAPH || - `http://mu.semte.ch/graphs/worship-privacy-centric-graph`; -const UNFILTERED_MAPPING = "main"; -const MAIN_INFO_MAPPING = "main-info"; -const PRIVATE_INFO_MAPPING = "private-info"; - -if (!process.env.FILE_SYNC_GRAPH) - throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; -const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; - -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, - INGEST_GRAPH, - PRIVACY_SENSITIVE_GRAPH, - UNFILTERED_MAPPING, - MAIN_INFO_MAPPING, - PRIVATE_INFO_MAPPING, - FILE_SYNC_GRAPH, -}; diff --git a/config/loket-consumer/positions-dispatching/delta-sync-dispatching.js b/config/loket-consumer/positions-dispatching/delta-sync-dispatching.js deleted file mode 100644 index 5384005..0000000 --- a/config/loket-consumer/positions-dispatching/delta-sync-dispatching.js +++ /dev/null @@ -1,88 +0,0 @@ -const { transformStatements, batchedDbUpdate, deleteFromAllGraphs } = require('./util'); -const { - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - INGEST_GRAPH, - PRIVACY_SENSITIVE_GRAPH, - UNFILTERED_MAPPING, - MAIN_INFO_MAPPING, - PRIVATE_INFO_MAPPING, -} = require('./config'); - -/** -* 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, muAuthSudo, fetch } = lib; - const { termObjectChangeSets } = data; - - for (let { deletes, inserts } of termObjectChangeSets) { - const deleteStatements = deletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - const insertStatements = inserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); - - if (deleteStatements.length) { - const transformedStatementsToDelete = await transformStatements(fetch, deleteStatements, UNFILTERED_MAPPING); - if (transformedStatementsToDelete.length) { - await deleteFromAllGraphs( - muAuthSudo.updateSudo, - transformedStatementsToDelete, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION, - ); - } - } - - if (insertStatements.length) { - const transformedStatementsToInsert = await transformStatements(fetch, insertStatements, MAIN_INFO_MAPPING); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION - ); - } - - const transformedStatementsToInsertPrivate = await transformStatements(fetch, insertStatements, PRIVATE_INFO_MAPPING); - if (transformedStatementsToInsertPrivate.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - PRIVACY_SENSITIVE_GRAPH, - transformedStatementsToInsertPrivate, - { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, - process.env.MU_SPARQL_ENDPOINT, // Note: this is the default endpoint through auth - BATCH_SIZE, - MAX_DB_RETRY_ATTEMPTS, - SLEEP_BETWEEN_BATCHES, - SLEEP_TIME_AFTER_FAILED_DB_OPERATION - ); - } - } - } -} - -module.exports = { - dispatch -}; diff --git a/config/loket-consumer/positions-dispatching/initial-sync-dispatching.js b/config/loket-consumer/positions-dispatching/initial-sync-dispatching.js deleted file mode 100644 index 86f34b7..0000000 --- a/config/loket-consumer/positions-dispatching/initial-sync-dispatching.js +++ /dev/null @@ -1,75 +0,0 @@ -const { transformStatements, batchedDbUpdate, } = 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, - INGEST_GRAPH, - PRIVACY_SENSITIVE_GRAPH, - MAIN_INFO_MAPPING, - PRIVATE_INFO_MAPPING -} = 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, fetch } = lib; - const { termObjects } = data; - - 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) { - originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`) - - const transformedStatementsToInsert = await transformStatements(fetch, originalInsertTriples, MAIN_INFO_MAPPING); - if (transformedStatementsToInsert.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - INGEST_GRAPH, - transformedStatementsToInsert, - { '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 - ); - } - - const transformedStatementsToInsertPrivate = await transformStatements(fetch, originalInsertTriples, PRIVATE_INFO_MAPPING); - if (transformedStatementsToInsertPrivate.length) { - await batchedDbUpdate( - muAuthSudo.updateSudo, - PRIVACY_SENSITIVE_GRAPH, - transformedStatementsToInsertPrivate, - { '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 -}; diff --git a/config/loket-consumer/worship-services-sensitive-dispatching/config.js b/config/loket-consumer/worship-services-sensitive-dispatching/config.js new file mode 100644 index 0000000..e89bca3 --- /dev/null +++ b/config/loket-consumer/worship-services-sensitive-dispatching/config.js @@ -0,0 +1,40 @@ +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/worship-service`; +const PRIVACY_SENSITIVE_GRAPH = process.env.PRIVACY_SENSITIVE_GRAPH || `http://mu.semte.ch/graphs/worship-privacy-centric-graph`; +const LANDING_ZONE_GRAPH = process.env.DCR_LANDING_ZONE_GRAPH || `http://mu.semte.ch/graphs/landing-zone/leidinggevenden`; + +// Mappings +const MAIN_INFO_MAPPING = "main-info"; +const PRIVATE_INFO_MAPPING = "private-info"; + +// 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, + LANDING_ZONE_DATABASE_ENDPOINT, + TARGET_GRAPH, + PRIVACY_SENSITIVE_GRAPH, + MAIN_INFO_MAPPING, + PRIVATE_INFO_MAPPING +}; diff --git a/config/loket-consumer/worship-services-sensitive-dispatching/delta-context-config.js b/config/loket-consumer/worship-services-sensitive-dispatching/delta-context-config.js new file mode 100644 index 0000000..a8df81c --- /dev/null +++ b/config/loket-consumer/worship-services-sensitive-dispatching/delta-context-config.js @@ -0,0 +1,11 @@ +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: [] +} + +module.exports = { + contextConfig +}; diff --git a/config/loket-consumer/worship-services-sensitive-dispatching/delta-sync-dispatching.js b/config/loket-consumer/worship-services-sensitive-dispatching/delta-sync-dispatching.js new file mode 100644 index 0000000..ef18e2e --- /dev/null +++ b/config/loket-consumer/worship-services-sensitive-dispatching/delta-sync-dispatching.js @@ -0,0 +1,114 @@ +const { + transformStatements, + deleteFromTargetGraph, + insertIntoTargetGraph, + deleteFromPrivateGraph, + insertIntoPrivateGraph +} = require('./util'); + +const { + MAIN_INFO_MAPPING, + PRIVATE_INFO_MAPPING +} = require('./config'); + +/** +* 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 main deletes from DL to OP model + const transformedMainDeletes = await transformStatements(fetch, deletesWithContext, MAIN_INFO_MAPPING); + + if (!transformedMainDeletes.length) { + console.log(`Warn: Delete main statements mapped to empty result.`); + console.log(`Input: ${deletesWithContext}`); + console.log(`Output: ${transformedMainDeletes}`); + } else { + await deleteFromTargetGraph(lib, transformedMainDeletes); + } + + // Map private deletes from DL to OP model + const transformedPrivateDeletes = await transformStatements(fetch, deletesWithContext, PRIVATE_INFO_MAPPING); + + if (!transformedPrivateDeletes.length) { + console.log(`Warn: Delete private statements mapped to empty result.`); + console.log(`Input: ${deletesWithContext}`); + console.log(`Output: ${transformedPrivateDeletes}`); + } else { + await deleteFromPrivateGraph(lib, transformedDeletes); + } + } + + if (originalInserts.length) { + // Map main inserts from DL to OP model + const transformedMainInserts = await transformStatements(fetch, insertsWithContext, MAIN_INFO_MAPPING); + + if (!transformedMainInserts.length) { + console.log(`Warn: Insert statements mapped to empty result.`); + console.log(`Input: ${insertsWithContext}`); + console.log(`Output: ${transformedMainInserts}`); + } else { + await insertIntoTargetGraph(lib, transformedMainInserts); + } + + // Map private inserts from DL to OP model + const transformedPrivateInserts = await transformStatements(fetch, insertsWithContext, PRIVATE_INFO_MAPPING); + + if (!transformedPrivateInserts.length) { + console.log(`Warn: Insert statements mapped to empty result.`); + console.log(`Input: ${insertsWithContext}`); + console.log(`Output: ${transformedPrivateInserts}`); + } else { + await insertIntoPrivateGraph(lib, transformedPrivateInserts); + } + } + } +} + +module.exports = { + dispatch +}; diff --git a/config/loket-consumer/worship-services-sensitive-dispatching/initial-sync-dispatching.js b/config/loket-consumer/worship-services-sensitive-dispatching/initial-sync-dispatching.js new file mode 100644 index 0000000..559250f --- /dev/null +++ b/config/loket-consumer/worship-services-sensitive-dispatching/initial-sync-dispatching.js @@ -0,0 +1,106 @@ +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, + PRIVACY_SENSITIVE_GRAPH, + LANDING_ZONE_GRAPH, + LANDING_ZONE_DATABASE_ENDPOINT, + MAIN_INFO_MAPPING, + PRIVATE_INFO_MAPPING +} = 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 transformedMainInsertTriples = await transformLandingZoneGraph(fetch, endpoint, MAIN_INFO_MAPPING); + console.log(`Transformed ${transformedMainInsertTriples.length} main triples`); + + if (transformedMainInsertTriples.length) { + await batchedDbUpdate( + muAuthSudo.updateSudo, + TARGET_GRAPH, + transformedMainInsertTriples, + { '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 + ); + } + + const transformedPrivateInsertTriples = await transformLandingZoneGraph(fetch, endpoint, PRIVATE_INFO_MAPPING); + console.log(`Transformed ${transformedPrivateInsertTriples.length} private triples`); + + if (transformedPrivateInsertTriples.length) { + await batchedDbUpdate( + muAuthSudo.updateSudo, + PRIVACY_SENSITIVE_GRAPH, + transformedPrivateInsertTriples, + { '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/worship-services-sensitive-dispatching/util.js b/config/loket-consumer/worship-services-sensitive-dispatching/util.js new file mode 100644 index 0000000..4b17b75 --- /dev/null +++ b/config/loket-consumer/worship-services-sensitive-dispatching/util.js @@ -0,0 +1,204 @@ +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, + PRIVACY_SENSITIVE_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 deleteFromPrivateGraph(lib, statements) { + console.log(`Deleting ${statements.length} statements from target graph`); + console.log(`Statements: ${JSON.stringify(statements)}`) + await batchedDbUpdate( + lib.muAuthSudo.updateSudo, + PRIVACY_SENSITIVE_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 insertIntoPrivateGraph(lib, statements) { + console.log(`Inserting ${statements.length} statements into target graph`); + console.log(`Statements: ${JSON.stringify(statements)}`) + + await batchedDbUpdate( + lib.muAuthSudo.updateSudo, + PRIVACY_SENSITIVE_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, + deleteFromPrivateGraph, + insertIntoPrivateGraph +}; diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164600-remove-loket-persons-before-sync.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164600-remove-loket-persons-before-sync.sparql new file mode 100644 index 0000000..7f6b7b7 --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164600-remove-loket-persons-before-sync.sparql @@ -0,0 +1,54 @@ +PREFIX adms: +PREFIX adres: +PREFIX besluit: +PREFIX ch: +PREFIX code: +PREFIX dbpedia: +PREFIX dc_terms: +PREFIX ere: +PREFIX euro: +PREFIX euvoc: +PREFIX ext: +PREFIX foaf: +PREFIX generiek: +PREFIX lblodlg: +PREFIX locn: +PREFIX mandaat: +PREFIX org: +PREFIX organisatie: +PREFIX person: +PREFIX persoon: +PREFIX prov: +PREFIX rdfs: +PREFIX regorg: +PREFIX schema: +PREFIX skos: +PREFIX vcard: +PREFIX xsd: + +DELETE { + GRAPH { + ?person + ?p ?o + } +} WHERE { + GRAPH { + ?person a person:Person ; + ?p ?o. + } + GRAPH { + VALUES ?typePosition { + + + } + + ?position mandaat:isBestuurlijkeAliasVan ?person ; + a ?typePosition . + } + + FILTER NOT EXISTS { + GRAPH { + ?worshipPosition ?person . + } + } +} diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164601-remove-loket-functionarissen-before-sync.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164601-remove-loket-functionarissen-before-sync.sparql new file mode 100644 index 0000000..55e716b --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164601-remove-loket-functionarissen-before-sync.sparql @@ -0,0 +1,52 @@ +PREFIX adms: +PREFIX adres: +PREFIX besluit: +PREFIX ch: +PREFIX code: +PREFIX dbpedia: +PREFIX dc_terms: +PREFIX ere: +PREFIX euro: +PREFIX euvoc: +PREFIX ext: +PREFIX foaf: +PREFIX generiek: +PREFIX lblodlg: +PREFIX locn: +PREFIX mandaat: +PREFIX org: +PREFIX organisatie: +PREFIX person: +PREFIX persoon: +PREFIX prov: +PREFIX rdfs: +PREFIX regorg: +PREFIX schema: +PREFIX skos: +PREFIX vcard: +PREFIX xsd: + +DELETE { + GRAPH { + ?functionaris ?p ?o. + } +} WHERE { + GRAPH { + ?functionaris a ; + org:holds ?bestuursFunctie ; + ?p ?o . + } + + ?bestuursorgaanInTime lblodlg:heeftBestuursfunctie ?bestuursFunctie ; + generiek:isTijdspecialisatieVan ?bestuursorgaan . + + ?bestuursorgaan besluit:bestuurt ?bestuurseenheid . + + FILTER NOT EXISTS { + VALUES ?typeBestuur { + + + } + ?bestuurseenheid a ?typeBestuur . + } +} diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164602-remove-loket-mandatarissen-before-sync-part-1.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164602-remove-loket-mandatarissen-before-sync-part-1.sparql new file mode 100644 index 0000000..3fc6737 --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164602-remove-loket-mandatarissen-before-sync-part-1.sparql @@ -0,0 +1,53 @@ +PREFIX adms: +PREFIX adres: +PREFIX besluit: +PREFIX ch: +PREFIX code: +PREFIX dbpedia: +PREFIX dc_terms: +PREFIX ere: +PREFIX euro: +PREFIX euvoc: +PREFIX ext: +PREFIX foaf: +PREFIX generiek: +PREFIX lblodlg: +PREFIX locn: +PREFIX mandaat: +PREFIX org: +PREFIX organisatie: +PREFIX person: +PREFIX persoon: +PREFIX prov: +PREFIX rdfs: +PREFIX regorg: +PREFIX schema: +PREFIX skos: +PREFIX vcard: +PREFIX xsd: + +DELETE { + GRAPH { + ?mandataris ?p ?o. + } +} WHERE { + GRAPH { + ?mandataris a ; + org:holds ?mandaat ; + ?p ?o . + } + + ?bestuursorgaanInTime org:hasPost ?mandaat ; + generiek:isTijdspecialisatieVan ?bestuursorgaan . + + ?bestuursorgaan besluit:bestuurt ?bestuurseenheid . + + FILTER NOT EXISTS { + VALUES ?typeBestuur { + + + } + ?bestuurseenheid a ?typeBestuur . + } +} +LIMIT 1000000 \ No newline at end of file diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164603-remove-loket-mandatarissen-before-sync-part-2.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164603-remove-loket-mandatarissen-before-sync-part-2.sparql new file mode 100644 index 0000000..59d01c2 --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164603-remove-loket-mandatarissen-before-sync-part-2.sparql @@ -0,0 +1,53 @@ +PREFIX adms: +PREFIX adres: +PREFIX besluit: +PREFIX ch: +PREFIX code: +PREFIX dbpedia: +PREFIX dc_terms: +PREFIX ere: +PREFIX euro: +PREFIX euvoc: +PREFIX ext: +PREFIX foaf: +PREFIX generiek: +PREFIX lblodlg: +PREFIX locn: +PREFIX mandaat: +PREFIX org: +PREFIX organisatie: +PREFIX person: +PREFIX persoon: +PREFIX prov: +PREFIX rdfs: +PREFIX regorg: +PREFIX schema: +PREFIX skos: +PREFIX vcard: +PREFIX xsd: + +DELETE { + GRAPH { + ?mandataris ?p ?o. + } +} WHERE { + GRAPH { + ?mandataris a ; + org:holds ?mandaat ; + ?p ?o . + } + + ?bestuursorgaanInTime org:hasPost ?mandaat ; + generiek:isTijdspecialisatieVan ?bestuursorgaan . + + ?bestuursorgaan besluit:bestuurt ?bestuurseenheid . + + FILTER NOT EXISTS { + VALUES ?typeBestuur { + + + } + ?bestuurseenheid a ?typeBestuur . + } +} +LIMIT 1000000 diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164604-remove-loket-mandatarissen-before-sync-part-3.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164604-remove-loket-mandatarissen-before-sync-part-3.sparql new file mode 100644 index 0000000..e7d6d00 --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164604-remove-loket-mandatarissen-before-sync-part-3.sparql @@ -0,0 +1,52 @@ +PREFIX adms: +PREFIX adres: +PREFIX besluit: +PREFIX ch: +PREFIX code: +PREFIX dbpedia: +PREFIX dc_terms: +PREFIX ere: +PREFIX euro: +PREFIX euvoc: +PREFIX ext: +PREFIX foaf: +PREFIX generiek: +PREFIX lblodlg: +PREFIX locn: +PREFIX mandaat: +PREFIX org: +PREFIX organisatie: +PREFIX person: +PREFIX persoon: +PREFIX prov: +PREFIX rdfs: +PREFIX regorg: +PREFIX schema: +PREFIX skos: +PREFIX vcard: +PREFIX xsd: + +DELETE { + GRAPH { + ?mandataris ?p ?o. + } +} WHERE { + GRAPH { + ?mandataris a ; + org:holds ?mandaat ; + ?p ?o . + } + + ?bestuursorgaanInTime org:hasPost ?mandaat ; + generiek:isTijdspecialisatieVan ?bestuursorgaan . + + ?bestuursorgaan besluit:bestuurt ?bestuurseenheid . + + FILTER NOT EXISTS { + VALUES ?typeBestuur { + + + } + ?bestuurseenheid a ?typeBestuur . + } +} diff --git a/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164605-remove-ministers-that-will-get-synced.sparql b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164605-remove-ministers-that-will-get-synced.sparql new file mode 100644 index 0000000..76dfb29 --- /dev/null +++ b/config/migrations/2023/20230330164600-sync-loket-to-op/20230330164605-remove-ministers-that-will-get-synced.sparql @@ -0,0 +1,21 @@ +PREFIX ere: + +DELETE { + GRAPH { + ?ministerPosition ?p ?o . + ?worshipService ?ministerPosition . + } +} WHERE { + GRAPH { + ?ministerPosition a ere:PositieBedienaar ; + ?p ?o . + + VALUES ?status { + # Actief + # Niet Actief + } + + ?worshipService ?ministerPosition ; + ?status . + } +} diff --git a/config/reasoner/dl2op/main/inverse_relations.n3 b/config/reasoner/dl2op/main/inverse_relations.n3 index bfeda31..ea25c90 100644 --- a/config/reasoner/dl2op/main/inverse_relations.n3 +++ b/config/reasoner/dl2op/main/inverse_relations.n3 @@ -18,7 +18,6 @@ PREFIX skos: PREFIX string: PREFIX xsd: - # Inverse properties { ?s org:holds ?o. diff --git a/config/reasoner/dl2op/main/mapping.n3 b/config/reasoner/dl2op/main/mapping.n3 index 82bebd9..1838441 100644 --- a/config/reasoner/dl2op/main/mapping.n3 +++ b/config/reasoner/dl2op/main/mapping.n3 @@ -18,7 +18,6 @@ PREFIX skos: PREFIX string: PREFIX xsd: - # DL and OP both use persoon:gebruikteVoornaam but with different semantics (!) # Hence the mapping is in two steps. # The first step maps the DL persoon:gebruikteVoornaam to the OP counterpart @@ -92,6 +91,31 @@ PREFIX xsd: ?s mandaat:einde ?eindeDate }. +{ + ?s mandaat:bindingStart ?bindingStartDateTime. + # use backward arrow rule defined below + ?bindingStartDateTime :dateFromDateTime ?bindingStartDate. + { + ?s mandaat:bindingStart ?bindingStartDateTime + } e:becomes { + ?s mandaat:bindingStart ?bindingStartDate + }. +} => { + ?s mandaat:bindingStart ?bindingStartDate +}. + +{ + ?s mandaat:bindingEinde ?bindingEindeDateTime. + # use backward arrow rule defined below + ?bindingEindeDateTime :dateFromDateTime ?bindingEindeDate. + { + ?s mandaat:bindingEinde ?bindingEindeDateTime + } e:becomes { + ?s mandaat:bindingEinde ?bindingEindeDate + }. +} => { + ?s mandaat:bindingEinde ?bindingEindeDate +}. # Backward arrow rule is like a function call. # Removes time info in the returned date diff --git a/config/reasoner/dl2op/main/persoon_http_to_https.n3 b/config/reasoner/dl2op/main/persoon_http_to_https.n3 index 6ced405..3658d1d 100644 --- a/config/reasoner/dl2op/main/persoon_http_to_https.n3 +++ b/config/reasoner/dl2op/main/persoon_http_to_https.n3 @@ -18,7 +18,6 @@ PREFIX skos: PREFIX string: PREFIX xsd: - # Map from http persoon to https persoon namespace. ## classes: diff --git a/config/reasoner/dl2op/main/query.n3q b/config/reasoner/dl2op/main/query.n3q new file mode 100644 index 0000000..9ff9d10 --- /dev/null +++ b/config/reasoner/dl2op/main/query.n3q @@ -0,0 +1,63 @@ +# Filtering classes to ingest + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. + +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. +{ + ?s a ; + ?p ?o . + } => { + ?s a ; + ?p ?o . +}. diff --git a/config/reasoner/dl2op/main/retract_empty_strings.n3 b/config/reasoner/dl2op/main/retract_empty_strings.n3 index f8204a7..dfe2ddd 100644 --- a/config/reasoner/dl2op/main/retract_empty_strings.n3 +++ b/config/reasoner/dl2op/main/retract_empty_strings.n3 @@ -18,7 +18,6 @@ PREFIX skos: PREFIX string: PREFIX xsd: - # Retract empty strings { { diff --git a/config/reasoner/dl2op/main/rrn.n3 b/config/reasoner/dl2op/main/rrn.n3 index b45c688..a87e31d 100644 --- a/config/reasoner/dl2op/main/rrn.n3 +++ b/config/reasoner/dl2op/main/rrn.n3 @@ -18,8 +18,6 @@ PREFIX skos: PREFIX string: PREFIX xsd: - - # Rijksregisternummer # Extend identifiers to match OP's model # skos:notation "00010100105". diff --git a/config/resources/domain.lisp b/config/resources/domain.lisp index 2f6b2a8..8b83654 100644 --- a/config/resources/domain.lisp +++ b/config/resources/domain.lisp @@ -9,6 +9,7 @@ (defparameter *cache-model-properties* t) (defparameter *cache-count-queries* t) (defparameter *supply-cache-headers-p* t) +(defparameter sparql:*experimental-no-application-graph-for-sudo-select-queries* t) (read-domain-file "domain.json") (read-domain-file "auth.json") diff --git a/docker-compose.yml b/docker-compose.yml index 7290967..32e0c9c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -35,7 +35,7 @@ services: restart: always logging: *default-logging db: - image: semtech/mu-authorization:feature-service-roam-r1 + image: semtech/mu-authorization:feature-service-roam-r1.1 environment: MU_SPARQL_ENDPOINT: "http://triplestore:8890/sparql" LOG_OUTGOING_SPARQL_QUERIES: "true" @@ -43,6 +43,8 @@ services: LOG_ACCESS_RIGHTS: "true" LOG_OUTGOING_SPARQL_QUERY_ROUNDTRIP: "true" ERROR_ON_UNWRITTEN_DATA: "true" + QUERY_MAX_PROCESSING_TIME: "300000" + QUERY_MAX_EXECUTION_TIME: "300000" volumes: - ./config/authorization:/config labels: @@ -70,7 +72,7 @@ services: restart: always logging: *default-logging resource: - image: semtech/mu-cl-resources:1.21.1 + image: semtech/mu-cl-resources:1.22.0 environment: CACHE_CLEAR_PATH: "http://cache/.mu/clear-keys" links: @@ -82,7 +84,7 @@ services: restart: always logging: *default-logging cache: - image: semtech/mu-cache:2.0.1 + image: semtech/mu-cache:2.0.2 links: - resource:backend labels: @@ -226,7 +228,7 @@ services: restart: always logging: *default-logging reasoner: - image: eyereasoner/reasoning-service:1.0.1 + image: eyereasoner/reasoning-service:1.0.2 volumes: - ./config/reasoner:/config ################################################################################ @@ -418,25 +420,28 @@ services: # DELTA CONSUMERS ################################################################################ ################################################################################ - # LOKET MANDATEN CONSUMERS + # LOKET MANDATARISSEN CONSUMERS ################################################################################ - mandaten-consumer: - image: lblod/delta-consumer:0.0.14 + mandatarissen-consumer: + image: lblod/delta-consumer:0.0.17-rc.7 environment: - DCR_SERVICE_NAME: "mandatendatabank-consumer" + DCR_SERVICE_NAME: "mandatarissen-consumer" DCR_SYNC_BASE_URL: "https://loket.lblod.info" # replace with link to Loket API DCR_SYNC_DATASET_SUBJECT: "http://data.lblod.info/datasets/delta-producer/dumps/MandatarissenCacheGraphDump" DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" - DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" + DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatarissen-consumer" DCR_DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" + BATCH_SIZE: 1000 DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" - FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" - KEEP_DELTA_FILES: "true" + DCR_KEEP_DELTA_FILES: "true" + DCR_ENABLE_DELTA_CONTEXT: "true" + DCR_LANDING_ZONE_DATABASE: "triplestore" + DCR_LANDING_ZONE_GRAPH: "http://mu.semte.ch/graphs/landing-zone/mandatarissen" volumes: - - ./config/loket-consumer/mandate-dispatching:/config/triples-dispatching/custom-dispatching - - ./data/files/consumer-files/mandaten:/consumer-files/ + - ./config/loket-consumer/mandatarissen-dispatching:/config/triples-dispatching/custom-dispatching + - ./data/files/consumer-files/mandatarissen:/consumer-files/ restart: always labels: - "logging=true" @@ -449,7 +454,7 @@ services: # LOKET LEIDINGGEVENDEN CONSUMERS ################################################################################ leidinggevenden-consumer: - image: lblod/delta-consumer:0.0.14 + image: lblod/delta-consumer:0.0.17-rc.7 environment: DCR_SERVICE_NAME: "leidinggevenden-consumer" DCR_SYNC_BASE_URL: "https://loket.lblod.info" # replace with link to Loket API @@ -460,8 +465,11 @@ services: DCR_DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/leidinggevenden/files" - FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" - KEEP_DELTA_FILES: "true" + DCR_KEEP_DELTA_FILES: "true" + BATCH_SIZE: 1000 + DCR_ENABLE_DELTA_CONTEXT: "true" + DCR_LANDING_ZONE_DATABASE: "triplestore" + DCR_LANDING_ZONE_GRAPH: "http://mu.semte.ch/graphs/landing-zone/leidinggevenden" volumes: - ./config/loket-consumer/leidinggevenden-dispatching:/config/triples-dispatching/custom-dispatching - ./data/files/consumer-files/leidinggevenden:/consumer-files/ @@ -477,22 +485,24 @@ services: # LOKET WORSHIP SENSITIVE CONSUMERS ################################################################################ worship-services-sensitive-consumer: - image: lblod/delta-consumer:0.0.14 + image: lblod/delta-consumer:0.0.17-rc.7 environment: DCR_SYNC_BASE_URL: "https://loket.lblod.info/" DCR_SERVICE_NAME: "worship-services-sensitive-consumer" - DCR_SYNC_FILES_PATH: "/sync/worship-services-sensitive/files" DCR_SYNC_DATASET_SUBJECT: "http://data.lblod.info/datasets/delta-producer/dumps/WorshipServicesSensitiveCacheGraphDump" DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/wsSensitive" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/wsSensitiveDeltaFileSyncing" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/worship-services-sensitive-consumer" DCR_DISABLE_INITIAL_SYNC: "true" - FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" + BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" + DCR_SYNC_FILES_PATH: "/sync/worship-services-sensitive/files" + BATCH_SIZE: 1000 DCR_KEEP_DELTA_FILES: "true" - DCR_DELTA_FILE_FOLDER: "/consumer-files" - BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "false" + DCR_ENABLE_DELTA_CONTEXT: "true" + DCR_LANDING_ZONE_DATABASE: "triplestore" + DCR_LANDING_ZONE_GRAPH: "http://mu.semte.ch/graphs/landing-zone/worship-services-sensitive" volumes: - - ./config/loket-consumer/positions-dispatching:/config/triples-dispatching/custom-dispatching + - ./config/loket-consumer/worship-services-sensitive-dispatching:/config/triples-dispatching/custom-dispatching - ./data/files/consumer-files/worship-services-sensitive:/consumer-files/ restart: always labels: