Merge pull request #191 from lblod/feature/loket-sync

loket to op sync with reasoning

config/loket-consumer/leidinggevenden-dispatching/config.js

@@ -0,0 +1,29 @@
+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/organisatieportaal`;
+
+
+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,
+  FILE_SYNC_GRAPH
+};

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
+};

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
+};

config/loket-consumer/leidinggevenden-dispatching/util.js

@@ -0,0 +1,135 @@
+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 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(preProcess(fetch, triples), 0,
+    MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION)
+    .then(preprocessed => operationWithRetry(mainConversion(fetch, preprocessed), 0,
+      MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION));
+}
+
+
+function preProcess(fetch, triples) {
+  let formdata = new URLSearchParams();
+  formdata.append("data", triples);
+
+  let requestOptions = {
+    method: 'POST',
+    body: formdata,
+    redirect: 'follow'
+  };
+
+  return fetch("http://reasoner/reason/dl2op/preprocess", requestOptions)
+    .then(response => response.text());
+}
+
+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,
+  partition,
+  transformStatements
+};

config/loket-consumer/mandate-dispatching/config.js

@@ -0,0 +1,29 @@
+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/organisatieportaal`;
+
+
+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,
+  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,135 @@
+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 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(preProcess(fetch, triples), 0,
+    MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION)
+    .then(preprocessed => operationWithRetry(mainConversion(fetch, preprocessed), 0,
+      MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION));
+}
+
+
+function preProcess(fetch, triples) {
+  let formdata = new URLSearchParams();
+  formdata.append("data", triples);
+
+  let requestOptions = {
+    method: 'POST',
+    body: formdata,
+    redirect: 'follow'
+  };
+
+  return fetch("http://reasoner/reason/dl2op/preprocess", requestOptions)
+    .then(response => response.text());
+}
+
+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,
+  partition,
+  transformStatements
+};

config/reasoner/dl2op/main/mapping.n3

@@ -0,0 +1,92 @@
+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 info in the returned date
+# The user in DL selects a date via the UI.
+# These dates are persisted as midnight dateTimes in UTC.
+# Meaning that 1st of July is stored as
+# "2022-06-30T22:00:00Z"^^xsd:dateTime
+# and 1 December
+# "2022-11-30T23:00:00Z"^^xsd:dateTime
+# So, the following rule adds 2 hours to the datetime, and then takes the date part of the UTC date time
+# to come to the date that the DL user selected (in CE(S)T)
+{
+   ?dateTime :dateFromDateTime ?date
+} <= {
+  (?dateTimeLiteral xsd:dateTime) log:dtlit ?dateTime.
+  (?dateTime "PT2H"^^xsd:dayTimeDuration) func:add-dayTimeDuration-to-dateTime ?localDateTime.
+  (?localDateTimeLiteral xsd:dateTime) log:dtlit ?localDateTime.
+  (?localDateTimeLiteral 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
+}.
+

docker-compose.yml

@@ -1,7 +1,6 @@
-version: '3.4'
+version: "3.4"
 
-x-logging:
-  &default-logging
+x-logging: &default-logging
   driver: "json-file"
   options:
     max-size: "10m"
@@ -11,7 +10,7 @@ services:
   identifier:
     image: semtech/mu-identifier:1.9.1
     environment:
-      DEFAULT_MU_AUTH_ALLOWED_GROUPS_HEADER: "[{\"variables\":[],\"name\":\"public\"}]"
+      DEFAULT_MU_AUTH_ALLOWED_GROUPS_HEADER: '[{"variables":[],"name":"public"}]'
       SESSION_COOKIE_SECURE: "on"
     labels:
       - "logging=true"
@@ -201,34 +200,38 @@ services:
       - db:database
     restart: always
     logging: *default-logging
-################################################################################
-# DELTA GENERAL
-################################################################################
+  reasoner:
+    image: bdevloed/mu-reasoner
+    volumes:
+      - ./config/reasoner:/config
+  ################################################################################
+  # DELTA GENERAL
+  ################################################################################
   delta-producer-dump-file-publisher:
     image: bdevloed/delta-producer-dump-file-publisher:0.8.0-beta
     volumes:
       - ./config/delta-producer/dump-file-publisher:/config
       - ./data/files:/share
     environment:
-      FILES_GRAPH: 'http://mu.semte.ch/graphs/public' # Note: this will be extracted in a config per job one day
-      DCAT_DATASET_GRAPH: 'http://mu.semte.ch/graphs/public'
+      FILES_GRAPH: "http://mu.semte.ch/graphs/public" # Note: this will be extracted in a config per job one day
+      DCAT_DATASET_GRAPH: "http://mu.semte.ch/graphs/public"
     links:
       - db:database
     restart: always
     logging: *default-logging
-################################################################################
-# DELTA ADMINISTRATIVE-UNITS: START
-################################################################################
+  ################################################################################
+  # DELTA ADMINISTRATIVE-UNITS: START
+  ################################################################################
   delta-producer-bg-jobs-initiator-administrative-units:
     image: lblod/delta-producer-background-jobs-initiator:0.3.1
     environment:
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
-      DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/administrative-units'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units'
-      CRON_PATTERN_DUMP_JOB: '0 0 4 * * 6' # weekly on saturday
-      CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM
-      START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts.
+      DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/administrative-units"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units"
+      CRON_PATTERN_DUMP_JOB: "0 0 4 * * 6" # weekly on saturday
+      CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM
+      START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts.
     labels:
       - "logging=true"
     links:
@@ -241,12 +244,12 @@ services:
       MAX_BODY_SIZE: "50mb"
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/administrative-units"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/administrative-units'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/administrative-units"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units"
       QUEUE_POLL_INTERVAL: 3000
       USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive
-      SKIP_MU_AUTH_INITIAL_SYNC: 'true'
+      SKIP_MU_AUTH_INITIAL_SYNC: "true"
       HEALING_PATCH_GRAPH_BATCH_SIZE: 500
     volumes:
       - ./config/delta-producer/administrative-units:/config
@@ -263,7 +266,7 @@ services:
       ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/administrative-units"
       PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/administrative-units"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/administrative-units'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/administrative-units"
       DELTA_INTERVAL_MS: 10000
       PRETTY_PRINT_DIFF_JSON: "true"
     volumes:
@@ -274,22 +277,22 @@ services:
       - db:database
     restart: always
     logging: *default-logging
-################################################################################
-# DELTA ADMINISTRATIVE-UNITS: END
-################################################################################
-################################################################################
-# DELTA ORGANIZATIONS: START
-################################################################################
+  ################################################################################
+  # DELTA ADMINISTRATIVE-UNITS: END
+  ################################################################################
+  ################################################################################
+  # DELTA ORGANIZATIONS: START
+  ################################################################################
   delta-producer-bg-jobs-initiator-organizations:
     image: lblod/delta-producer-background-jobs-initiator:0.3.1
     environment:
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
-      DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/organizations'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations'
-      CRON_PATTERN_DUMP_JOB: '42 * * * *' # At minute 42 every hour
-      CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM
-      START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts.
+      DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/organizations"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations"
+      CRON_PATTERN_DUMP_JOB: "42 * * * *" # At minute 42 every hour
+      CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM
+      START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts.
     labels:
       - "logging=true"
     links:
@@ -302,12 +305,12 @@ services:
       MAX_BODY_SIZE: "50mb"
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/organizations"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/organizations'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/organizations"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations"
       QUEUE_POLL_INTERVAL: 3000
       USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive
-      SKIP_MU_AUTH_INITIAL_SYNC: 'true'
+      SKIP_MU_AUTH_INITIAL_SYNC: "true"
       HEALING_PATCH_GRAPH_BATCH_SIZE: 500
     volumes:
       - ./config/delta-producer/organizations:/config
@@ -324,7 +327,7 @@ services:
       ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/organizations"
       PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/organizations"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/organizations'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/organizations"
       DELTA_INTERVAL_MS: 10000
       PRETTY_PRINT_DIFF_JSON: "true"
     volumes:
@@ -335,24 +338,23 @@ services:
       - db:database
     restart: always
     logging: *default-logging
-################################################################################
-# DELTA ORGANIZATIONS: END
-################################################################################
+  ################################################################################
+  # DELTA ORGANIZATIONS: END
+  ################################################################################
 
-
-################################################################################
-# DELTA PUBLIC INFORMATION: START
-################################################################################
+  ################################################################################
+  # DELTA PUBLIC INFORMATION: START
+  ################################################################################
   delta-producer-bg-jobs-initiator-public:
     image: lblod/delta-producer-background-jobs-initiator:0.3.1
     environment:
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
-      DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/public'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public'
-      CRON_PATTERN_DUMP_JOB: '11 */4 * * *' # At minute 11 every 4 hours
-      CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM
-      START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts.
+      DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/public"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public"
+      CRON_PATTERN_DUMP_JOB: "11 */4 * * *" # At minute 11 every 4 hours
+      CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM
+      START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts.
     labels:
       - "logging=true"
     links:
@@ -365,12 +367,12 @@ services:
       MAX_BODY_SIZE: "50mb"
       JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/public"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/public'
-      HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public'
-      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/public"
+      HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public"
+      INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public"
       QUEUE_POLL_INTERVAL: 3000
       USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive
-      SKIP_MU_AUTH_INITIAL_SYNC: 'true'
+      SKIP_MU_AUTH_INITIAL_SYNC: "true"
       HEALING_PATCH_GRAPH_BATCH_SIZE: 500
     volumes:
       - ./config/delta-producer/public:/config
@@ -387,7 +389,7 @@ services:
       ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal"
       RELATIVE_FILE_PATH: "deltas/public"
       PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/public"
-      PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/public'
+      PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/public"
       DELTA_INTERVAL_MS: 10000
       PRETTY_PRINT_DIFF_JSON: "true"
     volumes:
@@ -398,14 +400,74 @@ services:
       - db:database
     restart: always
     logging: *default-logging
-################################################################################
-# DELTA PUBLIC INFORMATION: END
-################################################################################
+  ################################################################################
+  # DELTA PUBLIC INFORMATION: END
+  ################################################################################
 
+  ################################################################################
+  # DELTA CONSUMERS
+  ################################################################################
+  ################################################################################
+  # LOKET MANDATEN CONSUMERS
+  ################################################################################
+  mandaten-consumer:
+    image: bdevloed/delta-consumer:fetch
+    environment:
+      DCR_SERVICE_NAME: "mandatendatabank-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_DISABLE_INITIAL_SYNC: "true"
+      BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true"
+      DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files"
+      FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data"
+      KEEP_DELTA_FILES: "true"
+    volumes:
+      - ./config/loket-consumer/mandate-dispatching:/config/triples-dispatching/custom-dispatching
+    restart: always
+    labels:
+      - "logging=true"
+    links:
+      - db:database
+      - triplestore:virtuoso
+    logging: *default-logging
 
-################################################################################
-# MU SEARCH - TODO USE THE CORRECT IMAGE ONCE INHERITANCE IS IN MU SEARCH
-################################################################################
+  ################################################################################
+  # 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"
+      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"
+    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
+  ################################################################################
+
+  ################################################################################
+  # MU SEARCH - TODO USE THE CORRECT IMAGE ONCE INHERITANCE IS IN MU SEARCH
+  ################################################################################
   search:
     image: nbittich/mu-search:wip-inheritance-alpha2 # Bump to fixed version when merged and release to the original repo
     links:
@@ -418,7 +480,7 @@ services:
     volumes:
       - ./data/elasticsearch/:/usr/share/elasticsearch/data
     environment:
-      - discovery.type=single-node 
+      - discovery.type=single-node
     restart: always
 
   error-report-service: