From 98113c46fd9e1a1386f84f18cdcf660491952198 Mon Sep 17 00:00:00 2001
From: Uwe Klinger <uwe.klinger@sap.com>
Date: Mon, 3 May 2021 16:37:16 +0200
Subject: [PATCH] Own transaction when raising an event

---
 README.md               |  6 ++++++
 suppliers/requests.http |  3 ++-
 suppliers/srv/mashup.js | 40 +++++++++++++++++++++++++++-------------
 3 files changed, 35 insertions(+), 14 deletions(-)

diff --git a/README.md b/README.md
index 16ee56f1..9c94df18 100644
--- a/README.md
+++ b/README.md
@@ -84,3 +84,9 @@ In case you've a question, find a bug, or otherwise need support, use our [commu
 ## License
 
 Copyright (c) 2021 SAP SE or an SAP affiliate company. All rights reserved. This file is licensed under the Apache Software License, version 2.0 except as noted otherwise in the [LICENSE](LICENSE.txt) file.
+
+## TODOs
+
+1. Move `Suppliers` out of `API_BUSINESS_PARTNER` - remove additional lines in monkey patch --> Uwe
+2. Fix problem with `await S4bupa.read (Suppliers).where('ID in',IDs)` --> Johannes
+3.
\ No newline at end of file
diff --git a/suppliers/requests.http b/suppliers/requests.http
index 2572f89d..8877829c 100644
--- a/suppliers/requests.http
+++ b/suppliers/requests.http
@@ -1,8 +1,9 @@
 
 @server = http://localhost:4004
+@bpServer = http://localhost:59847
 @authAlice = Authorization: Basic alice:
 
-PUT {{server}}/api-business-partner/A_BusinessPartner('ACME')
+PATCH {{server}}/api-business-partner/A_BusinessPartner('ACME')
 Content-Type: application/json
 
 {
diff --git a/suppliers/srv/mashup.js b/suppliers/srv/mashup.js
index 98ac7773..2fdb1ac6 100644
--- a/suppliers/srv/mashup.js
+++ b/suppliers/srv/mashup.js
@@ -26,26 +26,26 @@ module.exports = async()=>{ // called by server.js
     admin.on (['CREATE','UPDATE'], 'Books', async ({data:{supplier_ID: supplierId}}, next) => {
       // Using Promise.all(...) to parallelize local write, i.e. next(), and replication
 
-      /*
-      // ERROR: Reference integrity is violated for association "supplier"
-      if (supplierId) return Promise.all ([ next(), async()=>{
-        let replicated = await db.exists (Suppliers, supplierId)
-        if (!replicated) await replicate (supplierId, 'initial')
-      }])
-      else return next() //> don't forget to pass down the interceptor stack
-      */
+      const replicateIfNotExists = async()=>{
+        let replicated = await db.exists (Suppliers, supplierId);
+        if (!replicated) await replicate (supplierId, 'initial');
+      };
 
-      let replicated = await db.exists (Suppliers, supplierId);
-      if (!replicated) await replicate (supplierId, 'initial');
-      return next();
+      if (supplierId) return Promise.all ([ next(), replicateIfNotExists() ])
+      else return next() //> don't forget to pass down the interceptor stack
     })
 
   })
 
   // Subscribe to changes in the S4 origin of Suppliers data
+  // REVISIT: cds context is still from the UPDAT method when running in same programm, but should
+  // be a separate
   S4bupa.on ('BusinessPartners/Changed', async msg => { //> would be great if we had batch events from S/4
-    let replicas = await SELECT('ID').from (Suppliers) .where ('ID in', msg.businessPartners)
-    return replicate (replicas.map(each => each.ID))
+    await new Promise( resolve => setTimeout( resolve, 1000 ));
+    const tx = cds.db.tx(msg);
+    let replicas = await tx.run(SELECT('ID').from (Suppliers) .where ('ID in', msg.data.businessPartners));
+    await replicateTx(replicas.map(each => each.ID), undefined, tx, msg);
+    await tx.commit();
   })
 
   /**
@@ -57,6 +57,7 @@ module.exports = async()=>{ // called by server.js
     if (!Array.isArray(IDs)) IDs = [ IDs ]
     // TODO: Doesn't work when running in same process with mocked API_BUSINESS_PARTNER
 
+    // TODO: Issue
     let suppliers = await S4bupa.read (Suppliers).where(...([[]].concat(IDs).reduce( (where, id, index ) => { where.push(`${index>1 ? "OR ":""}ID = `, id); return where })));
     //let suppliers = await S4bupa.read (Suppliers).where('ID in',IDs)
     if (_initial) return db.insert (suppliers) .into (Suppliers) //> using bulk insert
@@ -65,4 +66,17 @@ module.exports = async()=>{ // called by server.js
     ))
   }
 
+  async function replicateTx (IDs,_initial, tx, msg) {
+    if (!Array.isArray(IDs)) IDs = [ IDs ]
+    // TODO: Doesn't work when running in same process with mocked API_BUSINESS_PARTNER
+
+    // TODO: Issue
+    let suppliers = await S4bupa.tx(msg).read (Suppliers).where(...([[]].concat(IDs).reduce( (where, id, index ) => { where.push(`${index>1 ? "OR ":""}ID = `, id); return where })));
+    //let suppliers = await S4bupa.read (Suppliers).where('ID in',IDs)
+    if (_initial) return tx.insert (suppliers) .into (Suppliers) //> using bulk insert
+    else return Promise.all(suppliers.map ( //> parallelizing updates
+      each => tx.update (Suppliers,each.ID) .with (each)
+    ))
+  }
+
 }