Merge pull request #486 from pryv/feature/stream-eventdeletions

Streaming eventDeletion on streams.delete

Author: perki

components/api-server/src/Result.js

@@ -7,7 +7,8 @@
 const commonMeta = require('./methods/helpers/setCommonMeta');
 const MultiStream = require('multistream');
 const DrainStream = require('./methods/streams/DrainStream');
-const ArrayStream = require('./methods/streams/ArrayStream');
+const ArraySerializationStream = require('./methods/streams/ArraySerializationStream');
+const SingleObjectSerializationStream = require('./methods/streams/SingleObjectSerializationStream');
 const async = require('async');
 
 const { Transform } = require('stream');
@@ -131,9 +132,9 @@ class Result {
    * @param {Readable} stream
    * @returns {void}
    */
-  addStream (arrayName, stream) {
+  addStream (arrayName, stream, isArray = true) {
     this._private.isStreamResult = true;
-    this._private.streamsArray.push({ name: arrayName, stream });
+    this._private.streamsArray.push({ name: arrayName, stream, isArray });
   }
 
   // Returns true if the Result holds any streams, false otherwise.
@@ -191,20 +192,14 @@ class Result {
     const streamsArray = this._private.streamsArray;
     if (!this._private.isStreamResult) { throw new Error('AF: not a stream result.'); }
     if (streamsArray.length < 1) { throw new Error('streams array empty'); }
-    // Are we handling a single stream?
-    if (streamsArray.length === 1) {
-      const first = streamsArray[0];
-      return first.stream
-        .pipe(new ArrayStream(first.name, true))
-        .pipe(new ResultStream(this._private.tracing, this._private.tracingId))
-        .pipe(res);
-    }
-    // assert: streamsArray.length > 1
+
     const streams = [];
     for (let i = 0; i < streamsArray.length; i++) {
       const s = streamsArray[i];
-      streams.push(s.stream.pipe(new ArrayStream(s.name, i === 0)));
+      const serializedStream = s.stream.pipe(s.isArray ? new ArraySerializationStream(s.name) : new SingleObjectSerializationStream(s.name));
+      streams.push(serializedStream);
     }
+
     return new MultiStream(streams)
       .pipe(new ResultStream(this._private.tracing, this._private.tracingId))
       .pipe(res);
@@ -245,7 +240,7 @@ class Result {
     const streamsArray = _private.streamsArray;
     const resultObj = {};
     async.forEachOfSeries(streamsArray, (elementDef, i, done) => {
-      const drain = new DrainStream({ limit: _private.arrayLimit }, (err, list) => {
+      const drain = new DrainStream({ limit: _private.arrayLimit, isArray: elementDef.isArray }, (err, list) => {
         if (err) {
           return done(err);
         }
@@ -275,15 +270,15 @@ class Result {
 /** @extends Transform */
 class ResultStream extends Transform {
   isStart;
-
   tracing;
-
   tracingId;
+  debugString;
   constructor (tracing, parentTracingId) {
-    super({ objectMode: true });
+    super({ writableObjectMode: true });
     this.isStart = true;
     this.tracing = tracing;
     this.tracingId = this.tracing.startSpan('resultStream', {}, parentTracingId);
+    this.debugString = '';
   }
 
   /**
@@ -300,13 +295,18 @@ class ResultStream extends Transform {
     callback();
   }
 
+  // uncomment to debug
+  // push (data) { this.debugString += data; super.push(data); }
+
   /**
    * @returns {void}
    */
   _flush (callback) {
-    const thing = ', "meta": ' + JSON.stringify(commonMeta.setCommonMeta({}).meta);
+    const thing = ' "meta": ' + JSON.stringify(commonMeta.setCommonMeta({}).meta);
     this.push(thing + '}');
     this.tracing.finishSpan('resultStream');
+
+    if (this.debugString !== '') { console.log('***** RESULT DATA **********\n' + this.debugString + '\n*********************'); }
     callback();
   }
 }

components/api-server/src/methods/streams.js

@@ -22,6 +22,7 @@ const logger = getLogger('methods:streams');
 const { getMall, storeDataUtils } = require('mall');
 const { changePrefixIdForStreams, replaceWithNewPrefix } = require('./helpers/backwardCompatibility');
 const { pubsub } = require('messages');
+const Readable = require('stream').Readable;
 /**
  * Event streams API methods implementation.
  *
@@ -345,6 +346,7 @@ module.exports = async function (api) {
       childrenDepth: -1,
       storeId
     });
+
     const streamToDelete = streamToDeleteSingleArray[0]; // no need to check existence: done before in verifyStreamExistenceAndPermissions
     const streamAndDescendantIds = treeUtils.collectPluckFromRootItem(streamToDelete, 'id');
     // keep stream and children to delete in next step
@@ -359,13 +361,25 @@ module.exports = async function (api) {
     const events = await mall.events.getWithParamsByStore(context.user.id, {
       [storeId]: { streams: [{ any: cleanDescendantIds }], limit: 1 }
     });
+
     const hasLinkedEvents = !!events.length;
     if (hasLinkedEvents) {
       // has linked events -----------------
       if (params.mergeEventsWithParent === null) {
         return next(errors.invalidParametersFormat('There are events referring to the deleted items ' +
                     'and the `mergeEventsWithParent` parameter is missing.'));
       }
+    }
+
+    // --- all tests are passed
+    // --- create result streams and start send result as they come
+    const updatedEventsStream = new ItemsStream();
+    result.addStream('updatedEvents', updatedEventsStream);
+    const singleItemDeletedStream = new ItemsStream();
+    result.addStream('streamDeletion', singleItemDeletedStream, false);
+    next(); // <== call next here to avoid await blocking
+
+    if (hasLinkedEvents) {
       if (params.mergeEventsWithParent) {
         // -- Case 1 -- Merge events with parent
         // add parent stream Id if needed and remove deleted stream ids
@@ -374,6 +388,9 @@ module.exports = async function (api) {
         await mall.events.updateMany(context.user.id, query, {
           addStreams: [parentId],
           removeStreams: streamAndDescendantIds
+        }, function (event) {
+          if (event == null) return;
+          updatedEventsStream.add({ action: 'mergedToParent', id: event.id });
         });
       } else {
         // case  mergeEventsWithParent = false
@@ -382,14 +399,17 @@ module.exports = async function (api) {
           const remaningStreamsIds = _.difference(event.streamIds, streamAndDescendantIds);
           if (remaningStreamsIds.length === 0) { // no more streams deleted event
             await mall.events.delete(context.user.id, event);
+            updatedEventsStream.add({ action: 'deleted', id: event.id });
           } else { // update event without these streams
             event.streamIds = remaningStreamsIds;
             await mall.events.update(context.user.id, event);
+            updatedEventsStream.add({ action: 'updatedStreamIds', id: event.id });
           }
         }
       }
       pubsub.notifications.emit(context.user.username, pubsub.USERNAME_BASED_EVENTS_CHANGED);
     }
+    updatedEventsStream.add(null); // close stream
     // finally delete stream
     for (const streamIdToDelete of context.streamToDeleteAndDescendantIds) {
       try {
@@ -398,8 +418,25 @@ module.exports = async function (api) {
         logger.error('Failed deleted some streams', err);
       }
     }
-    result.streamDeletion = { id: params.id };
+    singleItemDeletedStream.push({ id: params.id });
+    singleItemDeletedStream.push(null); // close stream
     pubsub.notifications.emit(context.user.username, pubsub.USERNAME_BASED_STREAMS_CHANGED);
-    next();
   }
 };
+
+class ItemsStream extends Readable {
+  buffer;
+  constructor () {
+    super({ objectMode: true });
+    this.buffer = [];
+  }
+
+  add (item) { this.push(item); }
+
+  _read () {
+    let push = true;
+    while (this.buffer.length > 0 && push) {
+      push = this.push(this.buffer.shift());
+    }
+  }
+}

components/api-server/src/methods/streams/ArraySerializationStream.js

@@ -0,0 +1,56 @@
+/**
+ * @license
+ * Copyright (C) 2012–2023 Pryv S.A. https://pryv.com - All Rights Reserved
+ * Unauthorized copying of this file, via any medium is strictly prohibited
+ * Proprietary and confidential
+ */
+const Transform = require('stream').Transform;
+
+// serialize every n objects
+const OBJECT_BUFFER_SIZE = 100;
+// event if OBJECT_BUFFER_SIZE is not reach, serialize if MAX_WAIT_MS is reached
+const MAX_WAIT_MS = 100;
+
+/**
+ * Stream that encapsulates the items it receives in a stringified array.
+ *
+ * @param arrayName {String} array name that will prefix the array
+ * @constructor
+ */
+module.exports = class ArraySerializationStream extends Transform {
+  constructor (arrayName) {
+    super({ writableObjectMode: true });
+    this.isStart = true;
+    this.prefix = '"' + arrayName + '":';
+    this.size = OBJECT_BUFFER_SIZE;
+    this.stack = [];
+    this.lastSerialization = Date.now();
+  }
+
+  _transform (item, encoding, callback) {
+    this.stack.push(item);
+
+    if (this.stack.length >= this.size || (Date.now() - this.lastSerialization) > MAX_WAIT_MS) {
+      if (this.isStart) {
+        this.isStart = false;
+        this.push((this.prefix + JSON.stringify(this.stack)).slice(0, -1));
+      } else {
+        this.push(',' + (JSON.stringify(this.stack)).slice(1, -1));
+      }
+      this.lastSerialization = Date.now();
+      this.stack = [];
+    }
+    callback();
+  }
+
+  _flush = function (callback) {
+    if (this.isStart) {
+      this.push(this.prefix + JSON.stringify(this.stack));
+    } else {
+      const joiningComma = this.stack.length > 0 ? ',' : '';
+      this.push(joiningComma + (JSON.stringify(this.stack)).slice(1));
+    }
+    this.push(',');
+    callback();
+  };
+};

components/api-server/src/methods/streams/ArrayStream.js

@@ -34,7 +34,13 @@ function DrainStream (params, callback) {
 
   if (callback) {
     this.on('finish', function () {
-      callback(null, this.array);
+      if (params.isArray) {
+        return callback(null, this.array);
+      }
+      if (this.array.length !== 1) {
+        return callback(new Error('Expected to find 1 item in array got: ' + JSON.stringify(this.array)));
+      }
+      callback(null, this.array[0]);
     });
   }
 

components/api-server/src/methods/streams/DrainStream.js

@@ -0,0 +1,30 @@
+/**
+ * @license
+ * Copyright (C) 2012–2023 Pryv S.A. https://pryv.com - All Rights Reserved
+ * Unauthorized copying of this file, via any medium is strictly prohibited
+ * Proprietary and confidential
+ */
+const Transform = require('stream').Transform;
+
+/**
+ * Stream that serialize the first object it receives.
+ *
+ * @param objectName {String} array name that will prefix the array
+ * @constructor
+ */
+module.exports = class SingleObjectSerializationStream extends Transform {
+  name;
+  constructor (objectName) {
+    super({ writableObjectMode: true });
+    this.name = objectName;
+  }
+
+  _transform = function (item, encoding, callback) {
+    this.push('"' + this.name + '": ' + JSON.stringify(item) + ', ');
+    callback();
+  };
+
+  _flush = function (callback) {
+    callback();
+  };
+};

components/api-server/src/methods/streams/SingleObjectSerializationStream.js

@@ -18,6 +18,15 @@ const string = helpers.string;
 const number = helpers.number;
 const boolean = helpers.boolean;
 
+const updatedEvent = helpers.object({
+  id: helpers.string(),
+  action: helpers.string()
+}, {
+  id: 'updatedEvent',
+  required: ['id', 'action'],
+  additionalProperties: false
+});
+
 module.exports = {
   get: {
     params: object({
@@ -83,7 +92,7 @@ module.exports = {
           required: ['stream'],
           additionalProperties: false
         }),
-        object({ streamDeletion: itemDeletion }, {
+        object({ streamDeletion: itemDeletion, updatedEvents: array(updatedEvent) }, {
           required: ['streamDeletion'],
           additionalProperties: false
         })