loket to op sync with reasoning

config/loket-consumer/mandate-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
+};

config/loket-consumer/mandate-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
+};

config/loket-consumer/mandate-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
+};

config/loket-consumer/mandate-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
+};

config/reasoner/dl2op/main/mapping.n3

@@ -0,0 +1,82 @@
+PREFIX : <#>
+PREFIX besluit: <http://data.vlaanderen.be/ns/besluit#>
+PREFIX e: <http://eulersharp.sourceforge.net/2003/03swap/log-rules#>
+PREFIX ex: <http://example.org/test#>
+PREFIX foaf: <http://xmlns.com/foaf/0.1/>
+PREFIX func: <http://www.w3.org/2007/rif-builtin-function#>
+PREFIX log: <http://www.w3.org/2000/10/swap/log#>
+PREFIX mandaat: <http://data.vlaanderen.be/ns/mandaat#>
+PREFIX org: <http://www.w3.org/ns/org#>
+PREFIX persoon: <http://data.vlaanderen.be/ns/persoon#>
+PREFIX string: <http://www.w3.org/2000/10/swap/string#>
+PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
+
+
+# First name
+# persoon:gebruikteVoornaam => foaf:givenName (done in pre-processing)
+# Alternatieve voornaam
+# foaf:name => persoon:gebruikteVoornaam
+
+# Test Mock LLB
+# <http://example.org/person> foaf:name "Didier".
+{
+  {
+    ?s foaf:name ?o
+  } e:becomes {
+    ?s persoon:gebruikteVoornaam ?o
+  }
+} => {
+  ?s persoon:gebruikteVoornaam ?o
+}.
+
+# Test
+# ex:s <http://data.vlaanderen.be/ns/mandaat#start> "2019-01-01T19:32:52+00:00"^^xsd:dateTime.
+{
+  ?s mandaat:start ?startDateTime.
+  # use backward arrow rule defined below
+  ?startDateTime :dateFromDateTime ?startDate.
+  {
+    ?s mandaat:start ?startDateTime
+  } e:becomes {
+    ?s mandaat:start ?startDate
+  }.
+} => {
+  ?s mandaat:start ?startDate
+}.
+
+# ex:s <http://data.vlaanderen.be/ns/mandaat#einde> "2019-02-01T10:32:52+00:00"^^xsd:dateTime.
+{
+  ?s mandaat:einde ?eindeDateTime.
+  # use backward arrow rule defined below
+  ?eindeDateTime :dateFromDateTime ?eindeDate.
+  {
+    ?s mandaat:einde ?eindeDateTime
+  } e:becomes {
+    ?s mandaat:einde ?eindeDate
+  }.
+} => {
+  ?s mandaat:einde ?eindeDate
+}.
+
+
+# Backward arrow rule is like a function call.
+# Removes time and timezone info in the returned date
+{
+   ?dateTime :dateFromDateTime ?date
+} <= {
+  (?dateTimeLiteral xsd:dateTime) log:dtlit ?dateTime.
+  (?dateTimeLiteral 0 11) func:substring ?dateLiteral.
+  (?dateLiteral xsd:date) log:dtlit ?date.
+}.
+
+
+# ex:orgA besluit:classificatie <http://data.vlaanderen.be/id/concept/BestuurseenheidClassificatieCode/5ab0e9b8a3b2ca7c5e000002>.
+{
+  {
+    ?s besluit:classificatie ?o
+  } e:becomes {
+    ?s org:classification ?o
+  }
+} => {
+  ?s org:classification ?o
+}.

config/reasoner/dl2op/preprocess/mapping.n3

@@ -0,0 +1,31 @@
+PREFIX e: <http://eulersharp.sourceforge.net/2003/03swap/log-rules#>
+PREFIX foaf: <http://xmlns.com/foaf/0.1/>
+PREFIX persoon: <http://data.vlaanderen.be/ns/persoon#>
+
+
+# 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
+# and drops the DL persoon:gebruikteVoornaam statements with e:becomes
+# A second reasoning will map DL foaf:name to OP persoon:gebruikteVoornaam
+# Doing both at once would lead all DL foaf:names to incorrectly become foaf:givennames
+# in OP
+
+# First name
+# persoon:gebruikteVoornaam => foaf:givenName
+# Alternatieve voornaam
+# foaf:name => persoon:gebruikteVoornaam
+
+# Test Mock DL
+# <http://example.org/person> persoon:gebruikteVoornaam "Didier".
+
+{
+  {
+    ?s persoon:gebruikteVoornaam ?o
+  } e:becomes {
+    ?s foaf:givenName ?o
+  }
+} => {
+  ?s foaf:givenName ?o
+}.
+