diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js new file mode 100644 index 0000000..0c97ddd --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/config.js @@ -0,0 +1,25 @@ +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 SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); +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/public`; + + +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, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js new file mode 100644 index 0000000..9f936df --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js @@ -0,0 +1,126 @@ +const { transformStatements, batchedDbUpdate, partition } = 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, + FILE_SYNC_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 deletePartitions = partition(deletes, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const deleteStatements = regularDeletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + if (deleteStatements.length) { + await transformStatements(fetch, deleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + { '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, + "DELETE" + ); + } + ) + } + + if (fileDeleteStatements.length) { + await transformStatements(fetch, fileDeleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + { '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, + "DELETE" + ); + } + ) + } + + const insertPartitions = partition(inserts, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const fileInsertStatements = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + + if (regularInsertStatements.length) { + await transformStatements(fetch, regularInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + {}, + 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 (fileInsertStatements.length) { + await transformStatements(fetch, fileInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + {}, + process.env.MU_SPARQL_ENDPOINT, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "INSERT" + ); + } + ) + } + } +} + +module.exports = { + dispatch +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js new file mode 100644 index 0000000..69bcdb4 --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js @@ -0,0 +1,84 @@ +const { transformStatements, batchedDbUpdate, partition } = 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, + FILE_SYNC_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; + + const partitions = partition(termObjects, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`) + + await transformStatements(fetch, originalRegularInsertTriples).then( + transformedRegularInsertTriples => { + // console.log(transformedRegularInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedRegularInsertTriples, + { '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 + ) + } + ) + } + + if (fileInserts.length) { + originalFileInsertTriples = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`) + await transformStatements(fetch, originalFileInsertTriples).then( + transformedFileInsertTriples => { + // console.log(transformedFileInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedFileInsertTriples, + { '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 +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js new file mode 100644 index 0000000..496cda5 --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/util.js @@ -0,0 +1,137 @@ +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 dbOperationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); + + console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); + await new Promise(r => setTimeout(r, sleepBetweenBatches)); + } +} + +async function dbOperationWithRetry(callback, + attempt, + maxAttempts, + sleepTimeOnFail) { + try { + 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 dbOperationWithRetry(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 transformGraph(fetch, triples) { + return preProcess(fetch, triples) + .then(preprocessed => mainConversion(fetch, preprocessed)); +} + + +function preProcess(fetch, triples) { + let formdata = new URLSearchParams(); + formdata.append("data", triples); + + let requestOptions = { + method: 'POST', + body: formdata, + redirect: 'follow' + }; + + // console.log('!!! Input\n' + triples) + + return fetch("http://reasoner/reason/dl2op/preprocess", requestOptions) + .then(response => response.text()) + // .then(result => { + // console.log(`!!! PREPROCESS\n${result}`) + // return result + // }) + .catch(error => console.log('error', error)); +} + +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()) + // .then(result => { + // console.log(`!!! MAIN\n${result}`) + // return result + // }) + .catch(error => console.log('error', error)); +} + +function transformStatements(fetch, triples) { + return transformGraph(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, + partition, + transformStatements +}; \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index df09169..911ec3f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -407,7 +407,10 @@ services: ################################################################################ # DELTA CONSUMERS ################################################################################ - mandatendatabank-consumer: + ################################################################################ + # LOKET MANDATEN CONSUMERS + ################################################################################ + mandaten-consumer: image: bdevloed/delta-consumer:fetch environment: DCR_SERVICE_NAME: "mandatendatabank-consumer" @@ -432,6 +435,34 @@ services: - triplestore:virtuoso logging: *default-logging + ################################################################################ + # LOKET LEIDINGGEVENDEN CONSUMERS + ################################################################################ + leidinggevenden-consumer: + image: bdevloed/delta-consumer:fetch + environment: + DCR_SERVICE_NAME: "leidinggevenden-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/LeidinggevendenCacheGraphDump" + DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/leidinggevenden" + DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/leidinggevenden" + DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/leidinggevenden-consumer" + INGEST_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" + # 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" + volumes: + - ./config/loket-consumer/leidinggevenden-dispatching:/config/triples-dispatching/custom-dispatching + restart: always + labels: + - "logging=true" + links: + - db:database + - triplestore:virtuoso + logging: *default-logging + ################################################################################ # DELTA CONSUMERS: END ################################################################################