feat: Extract chronicleEventLog from narrateEventLog, changes flags

Also wraps _receiveTruth sites into createReceiveTruth/Batch
Renames/moves _decorateRetrieveMediaContent into _downstreamOps

Author: iridiankin

packages/inspire/InspireGateway.js

@@ -342,7 +342,7 @@ export default class InspireGateway extends LogEventGenerator {
       this.warnEvent(`Narrated revelation with ${prologues.length} prologues`,
           "\n\tprologue partitions:",
               `'${prologues.map(({ partitionURI }) => String(partitionURI)).join("', '")}'`);
-      const ret = await Promise.all(prologues.map(this._connectAndNarratePrologue));
+      const ret = await Promise.all(prologues.map(this._connectChronicleAndNarratePrologue));
       this.warnEvent(`Acquired active connections for all revelation prologue partitions:`,
           "\n\tconnections:", ...dumpObject(ret));
       return ret;
@@ -391,21 +391,20 @@ export default class InspireGateway extends LogEventGenerator {
     }
   }
 
-  _connectAndNarratePrologue = async ({ partitionURI, info }: any) => {
+  _connectChronicleAndNarratePrologue = async ({ partitionURI, info }: any) => {
     if ((await info.commandId) >= 0) {
       throw new Error("Command queues in revelation are not supported yet");
     }
     // Acquire connection without remote narration to determine the current last authorized event
     // so that we can narrate any content in the prologue before any remote activity.
     const connection = await this.falseProphet.acquirePartitionConnection(partitionURI, {
-      dontRemoteNarrate: true,
+      narrateRemote: false, subscribeRemote: false,
     });
-    const eventId = await info.eventId;
-    const lastEventId = connection.getLastAuthorizedEventId();
-    if ((typeof eventId === "undefined") || (eventId <= lastEventId)) {
-      const remoteNarration = connection.narrateEventLog();
-      if (!(lastEventId >= 0)) await remoteNarration;
-    } else {
+    const lastPrologueEventId = await info.eventId;
+    const lastChronicledEventId = connection.getLastAuthorizedEventId() || 0;
+    const shouldChroniclePrologue = (lastPrologueEventId !== undefined)
+        && (lastPrologueEventId > lastChronicledEventId);
+    if (shouldChroniclePrologue) {
       // If no event logs are replayed, we don't need to precache the blobs either, so we delay
       // loading them up to this point.
       await (this.blobInfos || (this.blobInfos = this._getBlobInfos()));
@@ -416,9 +415,8 @@ export default class InspireGateway extends LogEventGenerator {
         throw new Error("commandQueue revelation not implemented yet");
       }
       const latestMediaInfos = await logs.latestMediaInfos;
-      await connection.narrateEventLog({
-        eventLog,
-        firstEventId: lastEventId + 1,
+      await connection.chronicleEventLog(eventLog, {
+        firstEventId: lastChronicledEventId + 1,
         retrieveMediaContent (mediaId: VRef, mediaInfo: Object) {
           if (!latestMediaInfos[mediaId.rawId()] ||
               (mediaInfo.blobId !== latestMediaInfos[mediaId.rawId()].mediaInfo.blobId)) {
@@ -433,6 +431,9 @@ export default class InspireGateway extends LogEventGenerator {
         }
       });
     }
+    // Initiate remote narration.
+    const remoteNarration = connection.narrateEventLog();
+    if (!shouldChroniclePrologue && !(lastChronicledEventId >= 0)) await remoteNarration;
     return connection;
   }
 

packages/prophet/Oracle/OraclePartitionConnection.js

@@ -1,15 +1,16 @@
 // @flow
 
-import type Command from "~/raem/command";
+import type Command, { UniversalEvent } from "~/raem/command";
 import { VRef } from "~/raem/ValaaReference";
 
 import PartitionConnection from "~/prophet/api/PartitionConnection";
-import type { NarrateOptions, MediaInfo, RetrieveMediaContent } from "~/prophet/api/Prophet";
+import type { ChronicleOptions, NarrateOptions, MediaInfo, RetrieveMediaContent }
+    from "~/prophet/api/Prophet";
 
 import { dumpObject, thenChainEagerly } from "~/tools";
 
-import { _connect, _narrateEventLog } from "./_connectionOps";
-import { _onConfirmTruth } from "./_downstreamOps";
+import { _connect, _chronicleEventLog, _narrateEventLog } from "./_connectionOps";
+import { _createReceiveTruthBatch, _receiveTruthOf } from "./_downstreamOps";
 import { _readMediaContent, _getMediaURL, _prepareBlob } from "./_mediaOps";
 
 /**
@@ -29,8 +30,7 @@ import { _readMediaContent, _getMediaURL, _prepareBlob } from "./_mediaOps";
 export default class OraclePartitionConnection extends PartitionConnection {
   _lastAuthorizedEventId: number;
   _downstreamTruthQueue: Object[];
-  _authorityRetrieveMediaContent: ?RetrieveMediaContent;
-  _narrationRetrieveMediaContent: ?RetrieveMediaContent;
+  _retrieveMediaContentFromAuthority: ?RetrieveMediaContent;
 
   constructor (options: Object) {
     super(options);
@@ -44,7 +44,7 @@ export default class OraclePartitionConnection extends PartitionConnection {
   }
 
   getRetrieveMediaContent () {
-    return this._narrationRetrieveMediaContent || this._authorityRetrieveMediaContent;
+    return this._retrieveMediaContentFromAuthority;
   }
 
   isConnected (): boolean {
@@ -85,30 +85,47 @@ export default class OraclePartitionConnection extends PartitionConnection {
 
   async narrateEventLog (options: NarrateOptions = {}): Promise<any> {
     const ret = {};
-    const retrievals = {};
     try {
-      return await _narrateEventLog(this, options, ret, retrievals);
+      return await _narrateEventLog(this, options, ret);
     } catch (error) {
       throw this.wrapErrorEvent(error, "narrateEventLog()",
           "\n\toptions:", ...dumpObject(options),
           "\n\tcurrent ret:", ...dumpObject(ret));
-    } finally {
-      if (options.retrieveMediaContent) {
-        delete this._narrationRetrieveMediaContent;
-      }
     }
   }
 
-  async _receiveTruth (originName: string, truthEvent: UniversalEvent): Promise<Object> {
+  async chronicleEventLog (eventLog: UniversalEvent[], options: ChronicleOptions = {}):
+      Promise<any> {
+    const ret = {};
+    try {
+      return await _chronicleEventLog(this, eventLog, options, ret);
+    } catch (error) {
+      throw this.wrapErrorEvent(error, "chronicleEventLog()",
+          "\n\toptions:", ...dumpObject(options),
+          "\n\tcurrent ret:", ...dumpObject(ret),
+      );
+    }
+  }
+
+  createReceiveTruth (originName: string) {
+    return this._receiveTruthOf.bind(this, { name: originName });
+  }
+
+  createReceiveTruthBatch (batchName: string,
+      retrieveMediaContent: Function = this.getRetrieveMediaContent()) {
+    return _createReceiveTruthBatch(this, batchName, retrieveMediaContent);
+  }
+
+  async _receiveTruthOf (group: Object, truthEvent: UniversalEvent): Promise<Object> {
     const partitionData = truthEvent.partitions && truthEvent.partitions[this.partitionRawId()];
     try {
       if (!partitionData) {
-        throw new Error(`truthEvent is missing partition info for ${this.debugId()}`);
+        throw new Error(`truthEvent of '${group.name}' has no partition ${this.debugId()} info`);
       }
-      return _onConfirmTruth(this, originName, authorizedEvent, partitionData);
+      return _receiveTruthOf(this, group, partitionData.eventId, truthEvent);
     } catch (error) {
-      throw this.wrapErrorEvent(error, `_receiveTruth('${originName}')`,
-          "\n\toriginName:", originName,
+      throw this.wrapErrorEvent(error, `_receiveTruthOf('${group.name}')`,
+          "\n\tgroup:", ...dumpObject(group),
           "\n\teventId:", partitionData && partitionData.eventId,
           "\n\ttruthEvent:", ...dumpObject(truthEvent),
           "\n\tthis:", ...dumpObject(this));
@@ -119,8 +136,7 @@ export default class OraclePartitionConnection extends PartitionConnection {
     return this.getScribeConnection().claimCommandEvent(command, this.getRetrieveMediaContent());
   }
 
-  _preAuthorizeCommand = (preAuthorizedEvent: Object) =>
-      this._receiveTruth("preAuthorizer", preAuthorizedEvent)
+  _preAuthorizeCommand = this.createReceiveTruth("preAuthorizer")
 
   // Coming from downstream: tries scribe first, otherwise forwards the request to authority.
   // In latter case forwards the result received from authority to Scribe for caching.