Merge branch 'development' into improvement/update-change-event-dates

2024-04-17 11:24:25 +02:00
parent 1a0987afeb 30504c9135
commit 6305343672
13 changed files with 887 additions and 3 deletions
--- a/config/clb-consumer/contact-data/config.js
+++ b/config/clb-consumer/contact-data/config.js
@@ -0,0 +1,30 @@
+const BATCH_SIZE = parseInt(process.env.BATCH_SIZE) || 1000;
+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 || 100);
+const SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION || 10000);
+const SLEEP_TIME_AFTER_FAILED_DB_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_DB_OPERATION || 60000);
+
+// Graphs
+const LANDING_ZONE_GRAPH = process.env.DCR_LANDING_ZONE_GRAPH || `http://mu.semte.ch/graphs/landing-zone/clb-contact-data`;
+
+// ENDPOINTS
+const LANDING_ZONE_DATABASE = process.env.DCR_LANDING_ZONE_DATABASE || 'database';
+const LANDING_ZONE_DATABASE_ENDPOINT = process.env.DCR_LANDING_ZONE_DATABASE_ENDPOINT || `http://${LANDING_ZONE_DATABASE}:8890/sparql`;
+
+module.exports = {
+  BATCH_SIZE,
+  MU_CALL_SCOPE_ID_INITIAL_SYNC,
+  BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES,
+  DIRECT_DATABASE_ENDPOINT,
+  MAX_DB_RETRY_ATTEMPTS,
+  MAX_REASONING_RETRY_ATTEMPTS,
+  SLEEP_BETWEEN_BATCHES,
+  SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+  SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION,
+  LANDING_ZONE_GRAPH,
+  LANDING_ZONE_DATABASE_ENDPOINT
+};
--- a/config/clb-consumer/contact-data/delta-context-config.js
+++ b/config/clb-consumer/contact-data/delta-context-config.js
@@ -0,0 +1,254 @@
+
+const PREFIXES = `
+PREFIX besluit: <http://data.vlaanderen.be/ns/besluit#>
+  PREFIX skos: <http://www.w3.org/2004/02/skos/core#>
+  PREFIX lblodgeneriek: <https://data.lblod.info/vocabularies/generiek/>
+  PREFIX org: <http://www.w3.org/ns/org#>
+  PREFIX code: <http://lblod.data.gift/vocabularies/organisatie/>
+  PREFIX adms: <http://www.w3.org/ns/adms#>
+  PREFIX generiek: <https://data.vlaanderen.be/ns/generiek#>
+  PREFIX ere: <http://data.lblod.info/vocabularies/erediensten/>
+  PREFIX organisatie: <https://data.vlaanderen.be/ns/organisatie#>
+  PREFIX mu: <http://mu.semte.ch/vocabularies/core/>
+  PREFIX euvoc: <http://publications.europa.eu/ontology/euvoc#>
+  PREFIX prov: <http://www.w3.org/ns/prov#>
+  PREFIX schema: <http://schema.org/>
+  PREFIX locn: <http://www.w3.org/ns/locn#>
+  PREFIX ext: <http://mu.semte.ch/vocabularies/ext/>`
+
+const contextConfig = {
+  addTypes: {
+    scope: 'all', // 'inserts, 'deletes', 'all' or none. To add rdf:type to subjects of inserts, deletes or both
+    exhausitive: false, // true or false: find all types for a subject, even if one is already present in delta
+  },
+  contextQueries: [
+    {
+      trigger: { // subjectType or predicateValue
+        subjectType: "org:Site"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ${subject} ext:goesInGraph ?graphOne.
+          ${subject} ext:goesInGraph ?graphTwo.
+          ${subject} ext:goesInGraph ?graphThree.
+          ${subject} ext:goesInGraph ?graphFour.
+        } WHERE {
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ?adminUnitOne org:hasPrimarySite ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphOne)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ?adminUnitTwo org:hasSite ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphTwo)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ?adminUnitThree org:hasPrimarySite ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphThree)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ?adminUnitFour org:hasSite ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphFour)
+          }
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        subjectType: "schema:ContactPoint"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ${subject} ext:goesInGraph ?graphOne.
+          ${subject} ext:goesInGraph ?graphTwo.
+        } WHERE {
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ?siteOne org:siteAddress ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphOne)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ?siteTwo org:siteAddress ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphTwo)
+          }
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        subjectType: "locn:Address"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ${subject} ext:goesInGraph ?graphOne.
+          ${subject} ext:goesInGraph ?graphTwo.
+        } WHERE {
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ?siteOne organisatie:bestaatUit ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphOne)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ?siteTwo organisatie:bestaatUit ${subject}.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphTwo)
+          }
+        }`
+    },
+    //Predicate triggers to solve out of order deltas
+    {
+      trigger: { // subjectType or predicateValue
+        predicateValue: "org:hasSite"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ${subject} ext:goesInGraph ?graphOne.
+          ${subject} ext:goesInGraph ?graphTwo.
+          ?site ?pSite ?oSite.
+          ?contact ?pContact ?oContact.
+          ?address ?pAddress ?oAddress.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        } WHERE {
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ${subject} a ?type.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphOne)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ${subject} a ?type.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphTwo)
+          }
+          ${subject} org:hasSite ?site.
+          ?site ?pSite ?oSite.
+          {
+            ?site org:siteAddress ?contact.
+            ?contact ?pContact ?oContact.
+            ?contact locn:address ?addressContact.
+            ?addressContact ?pAddressContact ?oAddressContact.
+          }
+          UNION
+          {
+            ?site org:siteAddress ?contact.
+            ?contact ?pContact ?oContact.
+          }
+          UNION
+          {
+            ?site organisatie:bestaatUit ?address.
+            ?address ?pAddress ?oAddress.
+          }        
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        predicateValue: "org:hasPrimarySite"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ${subject} ext:goesInGraph ?graphOne.
+          ${subject} ext:goesInGraph ?graphTwo.
+          ?site ?pSite ?oSite.
+          ?contact ?pContact ?oContact.
+          ?address ?pAddress ?oAddress.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        } WHERE {
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+              ${subject} a ?type.
+            }
+            BIND(<http://mu.semte.ch/graphs/administrative-unit> AS ?graphOne)
+          }
+          OPTIONAL {
+            GRAPH <http://mu.semte.ch/graphs/worship-service> {
+              ${subject} a ?type.
+            }
+            BIND(<http://mu.semte.ch/graphs/worship-service> AS ?graphTwo)
+          }
+          ${subject} org:hasPrimarySite ?site.
+          ?site ?pSite ?oSite.
+          {
+            ?site org:siteAddress ?contact.
+            ?contact ?pContact ?oContact.
+            ?contact locn:address ?addressContact.
+            ?addressContact ?pAddressContact ?oAddressContact.
+          }
+          UNION
+          {
+            ?site org:siteAddress ?contact.
+            ?contact ?pContact ?oContact.
+          }
+          UNION
+          {
+            ?site organisatie:bestaatUit ?address.
+            ?address ?pAddress ?oAddress.
+          }        
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        predicateValue: "org:siteAddress"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ?contact ?pContact ?oContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        } WHERE {
+          ${subject} org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+          OPTIONAL {
+            ?contact locn:address ?addressContact.
+            ?addressContact ?pAddressContact ?oAddressContact.
+          }
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        predicateValue: "locn:address"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ?addressContact ?pAddressContact ?oAddressContact.
+        } WHERE {
+          ${subject} locn:address ?addressContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        }`
+    },
+    {
+      trigger: { // subjectType or predicateValue
+        predicateValue: "organisatie:bestaatUit"
+      },
+      queryTemplate: (subject) => `
+        ${PREFIXES}
+        CONSTRUCT {
+          ?address ?pAddress ?oAddress.
+        } WHERE {
+          ${subject} organisatie:bestaatUit ?address.
+          ?address ?pAddress ?oAddress.
+        }`
+    }
+  ]
+}
+
+module.exports = {
+  contextConfig,
+  PREFIXES
+};
--- a/config/clb-consumer/contact-data/delta-sync-dispatching.js
+++ b/config/clb-consumer/contact-data/delta-sync-dispatching.js
@@ -0,0 +1,106 @@
+const {
+  deleteFromPublicGraph,
+  insertIntoPublicGraph,
+  deleteFromSpecificGraphs,
+  insertIntoSpecificGraphs
+} = require('./util');
+
+const {
+  LANDING_ZONE_GRAPH
+} = require('./config')
+
+const publicTypes = [
+  '<http://lblod.data.gift/vocabularies/organisatie/TypeVestiging>'
+]
+
+/**
+* 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) {
+  let { termObjectChangeSets, termObjectChangeSetsWithContext } = data;
+
+  // Both arrays are the same length, so we can zip them together for easier processing
+  zippedChangeSets = termObjectChangeSets.map((o, i) => ({
+    original: o,
+    withContext: termObjectChangeSetsWithContext && termObjectChangeSetsWithContext[i]
+  }));
+
+  console.log(`Received ${zippedChangeSets.length} change sets`)
+
+  for (let { original, withContext } of zippedChangeSets) {
+    
+    const {public: insertsOnPublic, target: insertsOnGraphs} = separateBetweenTargetAndPublic(original.inserts, withContext.inserts)
+    const {public: deletesOnPublic, target: deletesOnGraphs} = separateBetweenTargetAndPublic(original.deletes, withContext.deletes)
+
+    await deleteFromPublicGraph(lib, deletesOnPublic);
+    await deleteFromSpecificGraphs(lib, deletesOnGraphs);
+    await insertIntoPublicGraph(lib, insertsOnPublic);
+    await insertIntoSpecificGraphs(lib, insertsOnGraphs);
+
+  }
+}
+
+function separateBetweenTargetAndPublic(changes, context) {
+  const changesOnPublic = [];
+  const changesOnGraphs = {};
+  for(let change of changes) {
+    const subject = change.subject;
+    const contextTriples = context.filter((context) => context.subject === subject);
+    const graphTriple = contextTriples.find(
+      (context) => context.predicate === '<http://mu.semte.ch/vocabularies/ext/goesInGraph>' 
+        && context.object !== `<${LANDING_ZONE_GRAPH}>`
+    )
+    const otherContextTriples = context.filter(
+      (context) => context.predicate !== '<http://mu.semte.ch/vocabularies/ext/goesInGraph>' 
+        && context.predicate !== '<http://mu.semte.ch/vocabularies/ext/contextDataGoesInGraph>' 
+        && !(context.subject === subject && context.predicate === '<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>')
+    );
+    if(graphTriple) {
+      const graph = graphTriple.object.slice(1,-1); // We have to slice it to remove the "<" and ">"
+      if(change.object !== '') {
+        if(!changesOnGraphs[graph]) {
+          changesOnGraphs[graph] = [`${change.subject} ${change.predicate} ${change.object}.`]                                         
+        } else {
+          changesOnGraphs[graph].push(`${change.subject} ${change.predicate} ${change.object}.`)
+        }   
+      }                    
+      for(let triple of otherContextTriples) {
+        if(triple.object === '') continue;
+        if(!changesOnGraphs[graph]) {
+          changesOnGraphs[graph] = [`${triple.subject} ${triple.predicate} ${triple.object}.`]                                      
+        } else {
+          changesOnGraphs[graph].push(`${triple.subject} ${triple.predicate} ${triple.object}.`)
+        }                                        
+      }
+    }
+    const typeTriple = contextTriples.find(
+      (context) => context.subject === subject 
+        && context.predicate === '<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>'
+    );
+    if(!typeTriple) continue;
+    const type = typeTriple.object;
+    if(publicTypes.includes(type)) {
+      changesOnPublic.push(`${change.subject} ${change.predicate} ${change.object}.`)
+      for(let triple of otherContextTriples) {
+        changesOnPublic.push(`${triple.subject} ${triple.predicate} ${triple.object}.`)
+      }
+    }
+  }
+  return {public: changesOnPublic, target: changesOnGraphs}
+}
+
+module.exports = {
+  dispatch
+};
--- a/config/clb-consumer/contact-data/initial-sync-dispatching.js
+++ b/config/clb-consumer/contact-data/initial-sync-dispatching.js
@@ -0,0 +1,69 @@
+const { batchedDbUpdate, moveDataAfterInitialSync } = 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,
+  LANDING_ZONE_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 } = lib;
+  const { termObjects } = data;
+
+  // Steps:
+  // 1. Write the triples to the landing zone graph - without any mapping or filtering
+  // 2. One-time reasoning run with the full landing zone graph and write the results to the target graph (on-finish)
+
+  if (BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES) {
+    console.warn(`Service configured to skip MU_AUTH!`);
+  }
+  console.log(`Using ${endpoint} to insert triples`);
+
+  if (termObjects.length) {
+    const originalInsertTriples = termObjects.map(o => `${o.subject} ${o.predicate} ${o.object}.`);
+
+    await batchedDbUpdate(
+      muAuthSudo.updateSudo,
+      LANDING_ZONE_GRAPH,
+      originalInsertTriples,
+      { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC },
+      endpoint,
+      BATCH_SIZE,
+      MAX_DB_RETRY_ATTEMPTS,
+      SLEEP_BETWEEN_BATCHES,
+      SLEEP_TIME_AFTER_FAILED_DB_OPERATION
+    );
+  }
+}
+
+async function onFinishInitialIngest(lib,data) {
+  const { muAuthSudo } = lib;
+
+  console.log(`!! On-finish triggered !!`);
+
+  await moveDataAfterInitialSync(muAuthSudo.updateSudo, endpoint)
+}
+
+module.exports = {
+  dispatch,
+  onFinishInitialIngest
+};
--- a/config/clb-consumer/contact-data/util.js
+++ b/config/clb-consumer/contact-data/util.js
@@ -0,0 +1,337 @@
+const {
+  MAX_DB_RETRY_ATTEMPTS,
+  SLEEP_BETWEEN_BATCHES,
+  SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+  LANDING_ZONE_GRAPH,
+  BATCH_SIZE,
+} = require('./config');
+
+const prefixes = `
+PREFIX besluit: <http://data.vlaanderen.be/ns/besluit#>
+PREFIX skos: <http://www.w3.org/2004/02/skos/core#>
+PREFIX lblodgeneriek: <https://data.lblod.info/vocabularies/generiek/>
+PREFIX org: <http://www.w3.org/ns/org#>
+PREFIX code: <http://lblod.data.gift/vocabularies/organisatie/>
+PREFIX adms: <http://www.w3.org/ns/adms#>
+PREFIX generiek: <https://data.vlaanderen.be/ns/generiek#>
+PREFIX ere: <http://data.lblod.info/vocabularies/erediensten/>
+PREFIX organisatie: <https://data.vlaanderen.be/ns/organisatie#>
+PREFIX mu: <http://mu.semte.ch/vocabularies/core/>
+PREFIX euvoc: <http://publications.europa.eu/ontology/euvoc#>
+PREFIX prov: <http://www.w3.org/ns/prov#>
+PREFIX schema: <http://schema.org/>
+PREFIX locn: <http://www.w3.org/ns/locn#>
+PREFIX foaf: <http://xmlns.com/foaf/0.1/>
+PREFIX ext:<http://mu.semte.ch/vocabularies/ext/>
+`
+
+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);
+  }
+}
+
+
+async function insertIntoPublicGraph(lib, statements) {
+  console.log(`Inserting ${statements.length} statements into public graph`);
+
+  await batchedDbUpdate(
+    lib.muAuthSudo.updateSudo,
+    'http://mu.semte.ch/graphs/public',
+    statements,
+    {},
+    process.env.MU_SPARQL_ENDPOINT,
+    BATCH_SIZE,
+    MAX_DB_RETRY_ATTEMPTS,
+    SLEEP_BETWEEN_BATCHES,
+    SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+    'INSERT');
+}
+
+async function insertIntoSpecificGraphs(lib, statementsWithGraphs) {
+
+  for( let graph in statementsWithGraphs) {
+    console.log(`Inserting ${statementsWithGraphs[graph].length} statements into ${graph} graph`);
+    await batchedDbUpdate(
+      lib.muAuthSudo.updateSudo,
+      graph,
+      statementsWithGraphs[graph],
+      {},
+      process.env.MU_SPARQL_ENDPOINT,
+      BATCH_SIZE,
+      MAX_DB_RETRY_ATTEMPTS,
+      SLEEP_BETWEEN_BATCHES,
+      SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+      'INSERT');
+  }
+  
+}
+
+async function deleteFromPublicGraph(lib, statements) {
+  console.log(`Deleting ${statements.length} statements from public graph`);
+
+  await batchedDbUpdate(
+    lib.muAuthSudo.updateSudo,
+    'http://mu.semte.ch/graphs/public',
+    statements,
+    {},
+    process.env.MU_SPARQL_ENDPOINT,
+    BATCH_SIZE,
+    MAX_DB_RETRY_ATTEMPTS,
+    SLEEP_BETWEEN_BATCHES,
+    SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+    'DELETE');
+}
+
+async function deleteFromSpecificGraphs(lib, statementsWithGraphs) {
+  
+
+  for( let graph in statementsWithGraphs) {
+    console.log(`Deleting ${statementsWithGraphs[graph].length} statements from ${graph} graph`);
+    await batchedDbUpdate(
+      lib.muAuthSudo.updateSudo,
+      graph,
+      statementsWithGraphs[graph],
+      {},
+      process.env.MU_SPARQL_ENDPOINT,
+      BATCH_SIZE,
+      MAX_DB_RETRY_ATTEMPTS,
+      SLEEP_BETWEEN_BATCHES,
+      SLEEP_TIME_AFTER_FAILED_DB_OPERATION,
+      'DELETE');
+  }
+  
+}
+
+
+async function moveDataAfterInitialSync(muUpdate, endpoint) {
+  // Move primary sites admin units
+  await muUpdate(`
+    ${prefixes}
+    INSERT {
+      GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+        ?adminUnit org:hasPrimarySite ?site.
+        ?site ?pSite ?oSite.
+        ?contact ?pContact ?oContact.
+        ?address ?pAddress ?oAddress.
+        ?addressContact ?pAddressContact ?oAddressContact.
+      }
+    } WHERE {
+      GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+        ?adminUnit a ?type.
+      }
+      GRAPH <${LANDING_ZONE_GRAPH}> {
+        ?adminUnit org:hasPrimarySite ?site.
+        ?site ?pSite ?oSite.
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+          ?contact locn:address ?addressContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        }
+        UNION
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+        }
+        UNION
+        {
+          ?site organisatie:bestaatUit ?address.
+          ?address ?pAddress ?oAddress.
+        }
+      }        
+    }
+  `, undefined, endpoint)
+
+  // Move sites administrative units
+  await muUpdate(`
+    ${prefixes}
+    INSERT {
+      GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+        ?adminUnit org:hasSite ?site.
+        ?site ?pSite ?oSite.
+        ?contact ?pContact ?oContact.
+        ?address ?pAddress ?oAddress.
+        ?addressContact ?pAddressContact ?oAddressContact.
+      }
+    } WHERE {
+      GRAPH <http://mu.semte.ch/graphs/administrative-unit> {
+        ?adminUnit a ?type.
+      }
+      GRAPH <${LANDING_ZONE_GRAPH}> {
+        ?adminUnit org:hasSite ?site.
+        ?site ?pSite ?oSite.
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+          ?contact locn:address ?addressContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        }
+        UNION
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+        }
+        UNION
+        {
+          ?site organisatie:bestaatUit ?address.
+          ?address ?pAddress ?oAddress.
+        }     
+      }   
+    }
+  `, undefined, endpoint)
+
+  // Move primary sites worships
+  await muUpdate(`
+    ${prefixes}
+    INSERT {
+      GRAPH <http://mu.semte.ch/graphs/worship-service> {
+        ?adminUnit org:hasPrimarySite ?site.
+        ?site ?pSite ?oSite.
+        ?contact ?pContact ?oContact.
+        ?address ?pAddress ?oAddress.
+        ?addressContact ?pAddressContact ?oAddressContact.
+      }
+    } WHERE {
+      GRAPH <http://mu.semte.ch/graphs/worship-service> {
+        ?adminUnit a ?type.
+      }
+      GRAPH <${LANDING_ZONE_GRAPH}> {
+        ?adminUnit org:hasPrimarySite ?site.
+        ?site ?pSite ?oSite.
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+          ?contact locn:address ?addressContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        }
+        UNION
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+        }
+        UNION
+        {
+          ?site organisatie:bestaatUit ?address.
+          ?address ?pAddress ?oAddress.
+        }        
+      }
+    }
+  `, undefined, endpoint)
+
+  // Move sites worships
+  await muUpdate(`
+    ${prefixes}
+    INSERT {
+      GRAPH <http://mu.semte.ch/graphs/worship-service> {
+        ?adminUnit org:hasSite ?site.
+        ?site ?pSite ?oSite.
+        ?contact ?pContact ?oContact.
+        ?address ?pAddress ?oAddress.
+        ?addressContact ?pAddressContact ?oAddressContact.
+      }
+    } WHERE {
+      GRAPH <http://mu.semte.ch/graphs/worship-service> {
+        ?adminUnit a ?type.
+      }
+      GRAPH <${LANDING_ZONE_GRAPH}> {
+        ?adminUnit org:hasSite ?site.
+        ?site ?pSite ?oSite.
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+          ?contact locn:address ?addressContact.
+          ?addressContact ?pAddressContact ?oAddressContact.
+        }
+        UNION
+        {
+          ?site org:siteAddress ?contact.
+          ?contact ?pContact ?oContact.
+        }
+        UNION
+        {
+          ?site organisatie:bestaatUit ?address.
+          ?address ?pAddress ?oAddress.
+        }        
+      }
+    }
+  `, undefined, endpoint)
+
+  // Move sites worships
+  await muUpdate(`
+    ${prefixes}
+    INSERT {
+      GRAPH <http://mu.semte.ch/graphs/public> {
+        ?type a <http://lblod.data.gift/vocabularies/organisatie/TypeVestiging>.
+        ?type ?a ?b.
+      }
+    } WHERE {
+      GRAPH <${LANDING_ZONE_GRAPH}> {
+        ?type a <http://lblod.data.gift/vocabularies/organisatie/TypeVestiging>.
+        ?type ?a ?b.
+      }
+    }
+  `, undefined, endpoint)
+}
+
+module.exports = {
+  batchedDbUpdate,
+  moveDataAfterInitialSync,
+  insertIntoPublicGraph,
+  deleteFromPublicGraph,
+  insertIntoSpecificGraphs,
+  deleteFromSpecificGraphs,
+};
--- a/config/delta-producer/organizations-public-info/export.json
+++ b/config/delta-producer/organizations-public-info/export.json
@@ -13,6 +13,7 @@
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
        "http://www.w3.org/2004/02/skos/core#altLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://www.w3.org/ns/org#hasPrimarySite",
@@ -40,6 +41,7 @@
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
        "http://www.w3.org/2004/02/skos/core#altLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://www.w3.org/ns/org#hasPrimarySite",
@@ -72,6 +74,7 @@
        "http://www.opengis.net/ont/geosparql#sfWithin",
        "http://www.w3.org/2002/07/owl#sameAs",
        "http://www.w3.org/2004/02/skos/core#altLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/org#changedBy",
@@ -98,7 +101,9 @@
        "http://www.w3.org/2004/02/skos/core#prefLabel",
        "http://www.w3.org/ns/org#linkedTo",
        "http://data.lblod.info/vocabularies/erediensten/typeEredienst",
-        "http://www.w3.org/ns/adms#identifier"
+        "http://www.w3.org/ns/adms#identifier",
+        "http://www.w3.org/ns/regorg#legalName",
+        "http://www.w3.org/2004/02/skos/core#altLabel"
      ]
    },
    {
--- a/config/delta-producer/organizations/export.json
+++ b/config/delta-producer/organizations/export.json
@@ -127,6 +127,7 @@
      ],
      "properties": [
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
@@ -155,6 +156,7 @@
      ],
      "properties": [
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
@@ -191,6 +193,7 @@
        "http://www.w3.org/2002/07/owl#sameAs",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/org#changedBy",
        "http://www.w3.org/ns/org#classification",
@@ -214,6 +217,7 @@
      ],
      "properties": [
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
--- a/config/delta-producer/public/export.json
+++ b/config/delta-producer/public/export.json
@@ -15,6 +15,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
@@ -46,6 +47,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/regorg#orgStatus",
@@ -84,6 +86,7 @@
        "http://www.w3.org/2002/07/owl#sameAs",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/ns/adms#identifier",
        "http://www.w3.org/ns/org#changedBy",
        "http://www.w3.org/ns/org#classification",
@@ -105,6 +108,8 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
+        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/org#linkedTo",
        "http://data.lblod.info/vocabularies/erediensten/typeEredienst",
        "http://www.w3.org/ns/adms#identifier"
--- a/config/delta-producer/worship-posts/export.json
+++ b/config/delta-producer/worship-posts/export.json
@@ -22,6 +22,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://data.lblod.info/vocabularies/erediensten/typeEredienst",
@@ -42,6 +43,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://data.lblod.info/vocabularies/erediensten/typeEredienst",
@@ -61,6 +63,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://data.lblod.info/vocabularies/erediensten/typeEredienst",
@@ -121,6 +124,7 @@
      "properties": [
        "http://mu.semte.ch/vocabularies/core/uuid",
        "http://www.w3.org/2004/02/skos/core#prefLabel",
+        "http://www.w3.org/ns/regorg#legalName",
        "http://www.w3.org/2004/02/skos/core#altLabel",
        "http://www.w3.org/ns/regorg#orgStatus",
        "http://www.w3.org/ns/org#hasPrimarySite",
--- a/config/migrations/2024/20240327131000-delete-all-empty-literals.sparql
+++ b/config/migrations/2024/20240327131000-delete-all-empty-literals.sparql
@@ -0,0 +1,20 @@
+DELETE {
+  GRAPH ?g { 
+    ?s ?p ?o .
+  }
+}
+WHERE {
+  VALUES ?g {
+    <http://mu.semte.ch/graphs/administrative-unit>
+    <http://mu.semte.ch/graphs/worship-service>
+    <http://mu.semte.ch/graphs/public>
+    <http://mu.semte.ch/graphs/privacy-centric-graph>
+    <http://mu.semte.ch/graphs/worship-privacy-centric-graph>
+  }
+  graph ?g {
+    ?s ?p ?o
+
+    FILTER (isLiteral(?o))
+    FILTER regex(str(?o), "^\\s*$")
+  }
+}
--- a/config/migrations/2024/20240415085335-correct-representative-body-classification-code-type.sparql
+++ b/config/migrations/2024/20240415085335-correct-representative-body-classification-code-type.sparql
@@ -0,0 +1,15 @@
+PREFIX ext: <http://mu.semte.ch/vocabularies/ext/>
+
+DELETE {
+  GRAPH <http://mu.semte.ch/graphs/public> {
+    ?s a ext:RepresentatieveOrgaanClassificatieCode .
+  }
+} INSERT {
+  GRAPH <http://mu.semte.ch/graphs/public> {
+    ?s a ext:RepresentatiefOrgaanClassificatieCode .
+  }
+} WHERE {
+  GRAPH <http://mu.semte.ch/graphs/public> {
+    ?s a ext:RepresentatieveOrgaanClassificatieCode .
+  }
+}
--- a/config/resources/domain.json
+++ b/config/resources/domain.json
@@ -646,6 +646,10 @@
        "start-date": {
          "type": "date",
          "predicate": "ext:startDate"
+        },
+        "modified": {
+          "type": "date",
+          "predicate": "dc_terms:modified"
        }
      },
      "relationships": {
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -181,7 +181,7 @@ services:
    restart: always
    logging: *default-logging
  frontend:
-    image: lblod/frontend-organization-portal:1.23.0
+    image: lblod/frontend-organization-portal:1.23.2
    volumes:
      - ./config/frontend/add-x-frame-options-header.conf:/config/add-x-frame-options.conf
    environment:
@@ -299,7 +299,7 @@ services:
    restart: always
    logging: *default-logging
  kbo-data-sync:
-    image: lblod/sync-ovo-numbers-service:1.0.2
+    image: lblod/sync-wegwijs-organization-service:2.0.0
    links:
      - db:database
    labels:
@@ -633,6 +633,37 @@ services:
      - triplestore:virtuoso
    logging: *default-logging

+################################################################################
+  # CLB CONSUMERS
+  ################################################################################
+  clb-contact-data-consumer:
+    image: lblod/delta-consumer:0.0.18
+    environment:
+      DCR_SYNC_BASE_URL: "https://dev.contactgegevens-loket.lblod.info/"
+      DCR_SERVICE_NAME: "clb-contact-data-consumer"
+      DCR_SYNC_DATASET_SUBJECT: "http://data.lblod.info/datasets/delta-producer/dumps/ContactGegevensCacheGraphDump"
+      DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/contactData"
+      DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/contactDataDeltaFileSyncing"
+      DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/clb-contact-data-consumer"
+      DCR_DISABLE_INITIAL_SYNC: "false"
+      BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true"
+      DCR_SYNC_FILES_PATH: "/sync/contact-data/files"
+      BATCH_SIZE: 1000
+      DCR_KEEP_DELTA_FILES: "true"
+      DCR_ENABLE_DELTA_CONTEXT: "true"
+      DCR_LANDING_ZONE_DATABASE: "triplestore"
+      DCR_LANDING_ZONE_GRAPH: "http://mu.semte.ch/graphs/landing-zone/clb-contact-data"
+    volumes:
+      - ./config/clb-consumer/contact-data:/config/triples-dispatching/custom-dispatching
+      - ./data/files/consumer-files/clb-contact-data:/consumer-files/
+    restart: always
+    labels:
+      - "logging=true"
+    links:
+      - db:database
+      - triplestore:virtuoso
+    logging: *default-logging
+
  ################################################################################
  # DELTA CONSUMERS: END
  ################################################################################