add leidinggevenden consumer

2022-11-30 14:56:36 +00:00
parent d2f49fef61
commit e1fdde7010
5 changed files with 404 additions and 1 deletions
--- a/config/loket-consumer/leidinggevenden-dispatching/config.js
+++ 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
+};
--- a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js
+++ 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: <share://file/data> 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: "<http://foo>",
+*              subject: "<http://bar>",
+*              predicate: "<http://baz>",
+*              object: "<http://boom>^^<http://datatype>"
+*            }
+*         ]
+* @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('<share://'));
+    const regularDeletes = deletePartitions.fails;
+    const fileDeletes = deletePartitions.passes;
+    const fileDeleteStatements = fileDeletes.map(o => `${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('<share://'));
+    const regularInserts = insertPartitions.fails;
+    const fileInserts = insertPartitions.passes;
+    const regularInsertStatements = regularInserts.map(o => `${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
+};
--- a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js
+++ 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: "<http://foo>",
+*              subject: "<http://bar>",
+*              predicate: "<http://baz>",
+*              object: "<http://boom>^^<http://datatype>"
+*            }
+*         ]
+* @return {void} Nothing
+*/
+async function dispatch(lib, data) {
+  const { mu, muAuthSudo, fetch } = lib;
+  const { termObjects } = data;
+
+  const partitions = partition(termObjects, o => o.subject.startsWith('<share://'));
+  const regularInserts = partitions.fails;
+  const fileInserts = partitions.passes;
+
+  if (BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES) {
+    console.warn(`Service configured to skip MU_AUTH!`);
+  }
+  console.log(`Using ${endpoint} to insert triples`);
+
+  if (regularInserts.length) {
+    originalRegularInsertTriples = regularInserts.map(o => `${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
+};
--- a/config/loket-consumer/leidinggevenden-dispatching/util.js
+++ 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
+};
--- 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
  ################################################################################