add comments

Author: leolambo

packages/bitcore-node/src/providers/chain-state/evm/api/ecsp.ts

@@ -20,18 +20,18 @@ import {
 import { unixToDate } from '../../../../utils/convert';
 import { StatsUtil } from '../../../../utils/stats';
 import MoralisAPI from '../../external/providers/moralis';
-import { ExternalApiStream, MergedStream, ParseTransform } from '../../external/streams/apiStream';
+import { ExternalApiStream, MergedStream, ParseStream } from '../../external/streams/apiStream';
 import { NodeQueryStream } from '../../external/streams/nodeStream';
 import { InternalStateProvider } from '../../internal/internal';
 import { EVMTransactionStorage } from '../models/transaction';
 import { EVMTransactionJSON } from '../types';
 import { BaseEVMStateProvider } from './csp';
+import { PopulateReceiptTransform } from './populateReceiptTransform';
 import {
   getProvider,
   isValidProviderType
 } from './provider';
 import { EVMListTransactionsStream } from './transform';
-import { PopulateReceiptTransform } from './populateReceiptTransform';
 
 
 export interface GetWeb3Response { rpc: CryptoRpc; web3: Web3; dataType: string }
@@ -268,16 +268,10 @@ export class BaseEVMExternalStateProvider extends InternalStateProvider implemen
       const txStreams: Readable[] = [];
       const ethTransactionTransform = new EVMListTransactionsStream(walletAddresses);
       const populateReceipt = new PopulateReceiptTransform();
-      const parseStrings = new ParseTransform();
+      const parseStream = new ParseStream();
       const mergedStream = new MergedStream(); // Stream to combine the output of multiple streams
       const resultStream = new MergedStream(); // Stream to write to the res object
 
-      // Transform transactions proccessed through merged stream
-      mergedStream
-      .pipe(parseStrings)
-      .pipe(populateReceipt)
-      .pipe(ethTransactionTransform)
-      .pipe(resultStream);
       // Tip height used to calculate confirmations
       args.tipHeight = tip ? tip.height : 0;
       // Defaults to pulling only the first 10 transactions per address
@@ -287,7 +281,13 @@ export class BaseEVMExternalStateProvider extends InternalStateProvider implemen
       }
       // Pipe all txStreams to the mergedStream
       ExternalApiStream.mergeStreams(txStreams, mergedStream);
-      // Ensure mergeStream resolves
+      // Transform transactions proccessed through merged stream
+      mergedStream
+      .pipe(parseStream)
+      .pipe(populateReceipt)
+      .pipe(ethTransactionTransform)
+      .pipe(resultStream);
+      // Ensure resultStream resolves
       const result = await ExternalApiStream.onStream(resultStream, req!, res!, { jsonl: true });
       if (result?.success === false) {
         logger.error('Error mid-stream (streamWalletTransactions): %o', result.error?.log || result.error);

packages/bitcore-node/src/providers/chain-state/external/streams/apiStream.ts

@@ -103,6 +103,8 @@ export class ExternalApiStream extends Readable {
             }
           }
           if (!isFirst) {
+            // Data has already been written to the stream and status 200 headers have already been sent
+            // We notify and log the error instead of throwing
             const errMsg = '{"error": "An error occurred during data stream"}';
             if (opts.jsonl) {
               res.write(`${errMsg}`);
@@ -113,27 +115,26 @@ export class ExternalApiStream extends Readable {
             res.destroy();
             return resolve({ success: false, error: err });
           } else {
+            // Rejecting here allows downstream to send status 500
             return reject(err);
           }
         }
         return;
       });
       stream.on('data', function(data) {
         if (!closed) {
-          if (opts.jsonl) {
-            if (isFirst) {
-              isFirst = false;
-            } else if (!data.endsWith('\n')) {
-              res.write('\n');
-            }
-          } else {
+          // We are assuming jsonl data appended a new line upstream
+          if (!opts.jsonl) {
             if (isFirst) {
               res.write('[\n');
-              isFirst = false;
             } else {
               res.write(',\n');
             }
           }
+          if (isFirst) {
+            // All cases need isFirst set correctly for proper error handling
+            isFirst = false;
+          }
           res.write(data);
         } else {
           stream.destroy();
@@ -175,7 +176,7 @@ export class ExternalApiStream extends Readable {
   }
 }
 
-export class ParseTransform extends Transform {
+export class ParseStream extends Transform {
   constructor() {
     super({ objectMode: true });
   }