Port CSV Generation into master

package.json

@@ -81,6 +81,7 @@
     "mongodb-memory-server": "^8.4.1",
     "mongoose": "^6.2.7",
     "node-cron": "^3.0.1",
+    "objects-to-csv": "^1.3.6",
     "picomatch": "^2.3.1",
     "toml": "^3.0.0",
     "winston": "^3.6.0",

src/replication/DealReplicationService.ts

@@ -1,3 +1,4 @@
+/* eslint @typescript-eslint/no-var-requires: "off" */
 import bodyParser from 'body-parser';
 import config from 'config';
 import express, { Express, Request, Response } from 'express';
@@ -10,6 +11,9 @@ import GetReplicationDetailsResponse from './GetReplicationDetailsResponse';
 import { GetReplicationsResponse } from './GetReplicationsResponse';
 import UpdateReplicationRequest from './UpdateReplicationRequest';
 import { ObjectId } from 'mongodb';
+import fs from 'fs-extra';
+import path from 'path';
+const ObjectsToCsv: any = require('objects-to-csv');// no TS support
 
 export default class DealReplicationService extends BaseService {
 
@@ -21,6 +25,7 @@ export default class DealReplicationService extends BaseService {
       this.handleUpdateReplicationRequest = this.handleUpdateReplicationRequest.bind(this);
       this.handleListReplicationRequests = this.handleListReplicationRequests.bind(this);
       this.handleGetReplicationRequest = this.handleGetReplicationRequest.bind(this);
+      this.handlePrintCSVReplicationRequest = this.handlePrintCSVReplicationRequest.bind(this);
       this.startCleanupHealthCheck = this.startCleanupHealthCheck.bind(this);
       if (!this.enabled) {
         this.logger.warn('Deal Replication Service is not enabled. Exit now...');
@@ -37,6 +42,7 @@ export default class DealReplicationService extends BaseService {
       this.app.post('/replication/:id', this.handleUpdateReplicationRequest);
       this.app.get('/replications', this.handleListReplicationRequests);
       this.app.get('/replication/:id', this.handleGetReplicationRequest);
+      this.app.get('/replication/:id/csv', this.handlePrintCSVReplicationRequest);
     }
 
     public start (): void {
@@ -247,6 +253,45 @@ export default class DealReplicationService extends BaseService {
       response.end(JSON.stringify({ id: replicationRequest.id }));
     }
 
+    private async handlePrintCSVReplicationRequest (request: Request, response: Response) {
+      const id = request.params['id'];
+      const replicationRequest = await Datastore.ReplicationRequestModel.findById(id);
+      if (replicationRequest) {
+        const deals = await Datastore.DealStateModel.find({
+          replicationRequestId: id
+        });
+        this.logger.info(`Found ${deals.length} deals from replication request ${id}`);
+        if (deals.length > 0) {
+          const csvRow = [];
+          for (let i = 0; i < deals.length; i++) {
+            const deal = deals[i];
+            csvRow.push({
+              miner_id: deal.provider,
+              deal_cid: deal.dealCid,
+              filename: `${deal.pieceCid}.car`,
+              piece_cid: deal.pieceCid,
+              start_epoch: deal.startEpoch,
+              full_url: `${replicationRequest.urlPrefix}/${deal.pieceCid}.car`
+            });
+          }
+          const csv = new ObjectsToCsv(csvRow);
+          const configDir = config.util.getEnv('NODE_CONFIG_DIR');
+          await fs.mkdirp(path.join(configDir, 'csv'));
+          if (await fs.pathExists(path.join(configDir, 'csv'))) {
+            const filename = `${configDir}/csv/${deals[0].provider}_${id}.csv`;
+            await csv.toDisk(filename);
+            response.end(`CSV saved to ${filename}`);
+          } else {
+            this.logger.error('CSV dir could not be created');
+            response.end(`CSV dir could not be created`);
+          }
+        }
+      } else {
+        console.error(`Could not find replication request ${id}`);
+      }
+      response.end();
+    }
+
     private async cleanupHealthCheck (): Promise<void> {
       this.logger.debug(`Cleaning up health check table`);
       // Find all active workerId

src/singularity.ts

@@ -521,6 +521,20 @@ replication.command('resume').description('Resume a paused deal replication requ
     }
     CliUtil.renderResponse(response.data, options.json);
   });
+
+  replication.command('csv').description('Write a deal replication result as csv.')
+  .argument('<id>', 'Existing ID of deal replication request.')
+  .action(async (id) => {
+    let response!: AxiosResponse;
+    try {
+      const url: string = config.get('connection.deal_replication_service');
+      response = await axios.get(`${url}/replication/${id}/csv`);
+    } catch (error) {
+      CliUtil.renderErrorAndExit(error);
+    }
+    CliUtil.renderResponse(response.data, false);
+  });
+
 program.showSuggestionAfterError();
 program.showHelpAfterError('(add --help for additional information)');
 program.parse();