From 0eab438c31cadc22e8f721ab99b15222fdbc018a Mon Sep 17 00:00:00 2001
From: Romuald Lemesle <53513584+RomuDeuxfois@users.noreply.github.com>
Date: Thu, 19 Oct 2023 16:47:10 +0200
Subject: [PATCH] [backend] Improv error handling in import csv connector
 (#4655)

---
 .../importCsv/importCsv-connector.ts          | 56 +++++++++++--------
 1 file changed, 32 insertions(+), 24 deletions(-)

diff --git a/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts b/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts
index 8691656b66e5..24cfa19b2c1d 100644
--- a/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts
+++ b/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts
@@ -52,36 +52,44 @@ const initImportCsvConnector = () => {
       await updateReceivedTime(context, applicantUser, workId, 'Connector ready to process the operation');
       if (stream) {
         const chunks: Uint8Array[] = [];
+        let hasError: boolean = false;
         stream.on('data', async (chunk) => {
           chunks.push(chunk.toString('utf8'));
-        }).on('error', (err) => {
-          throw err;
+        }).on('error', async (error) => {
+          hasError = true;
+          const errorData = {
+            error: error.message,
+            source: fileId,
+          };
+          await reportExpectation(context, applicantUser, workId, errorData);
         })
           .on('end', async () => {
-            const string = chunks.join('');
-            const bundle = await bundleProcess(context, applicantUser, Buffer.from(string), csvMapper, entity);
-            await updateExpectationsNumber(context, applicantUser, workId, 1);
+            if (!hasError) {
+              const string = chunks.join('');
+              const bundle = await bundleProcess(context, applicantUser, Buffer.from(string), csvMapper, entity);
+              await updateExpectationsNumber(context, applicantUser, workId, 1);
 
-            const validateBeforeImport = connectorConfig.config.validate_before_import;
-            if (validateBeforeImport) {
-              const contentStream = Readable.from([JSON.stringify(bundle, null, '  ')]);
-              const file = {
-                createReadStream: () => contentStream,
-                filename: `${workId}.json`,
-                mimetype: 'application/json',
-              };
-              await upload(context, applicantUser, 'import/pending', file, { entity });
+              const validateBeforeImport = connectorConfig.config.validate_before_import;
+              if (validateBeforeImport) {
+                const contentStream = Readable.from([JSON.stringify(bundle, null, '  ')]);
+                const file = {
+                  createReadStream: () => contentStream,
+                  filename: `${workId}.json`,
+                  mimetype: 'application/json',
+                };
+                await upload(context, applicantUser, 'import/pending', file, { entity });
 
-              await reportExpectation(context, applicantUser, workId);
-            } else {
-              const content = Buffer.from(JSON.stringify(bundle), 'utf-8').toString('base64');
-              await pushToSync({
-                type: 'bundle',
-                update: true,
-                applicant_id: applicantId ?? OPENCTI_SYSTEM_UUID,
-                work_id: workId,
-                content
-              });
+                await reportExpectation(context, applicantUser, workId);
+              } else {
+                const content = Buffer.from(JSON.stringify(bundle), 'utf-8').toString('base64');
+                await pushToSync({
+                  type: 'bundle',
+                  update: true,
+                  applicant_id: applicantId ?? OPENCTI_SYSTEM_UUID,
+                  work_id: workId,
+                  content
+                });
+              }
             }
           });
       }