From ba0eba7132a35d2ea984925d76635343d09ca021 Mon Sep 17 00:00:00 2001
From: "SPRINX0\\prochazka" <jan.prochazka@sprinx.com>
Date: Wed, 5 Mar 2025 18:09:17 +0100
Subject: [PATCH] SYNC:  mognodb - correct handle stream errors

---
 e2e-tests/.localconfig.js                     | 10 +++---
 e2e-tests/cypress/e2e/multi-sql.cy.js         |  4 +--
 .../src/backend/createBulkInsertStream.js     | 33 ++++++++++++-------
 3 files changed, 28 insertions(+), 19 deletions(-)

diff --git a/e2e-tests/.localconfig.js b/e2e-tests/.localconfig.js
index 96f2bbc13..45a700f93 100644
--- a/e2e-tests/.localconfig.js
+++ b/e2e-tests/.localconfig.js
@@ -1,8 +1,8 @@
 module.exports = {
-  // mysql: true,
-  // postgres: true,
-  // mssql: true,
-  // oracle: true,
-  // sqlite: true,
+  mysql: true,
+  postgres: true,
+  mssql: true,
+  oracle: true,
+  sqlite: true,
   mongo: true
 };
diff --git a/e2e-tests/cypress/e2e/multi-sql.cy.js b/e2e-tests/cypress/e2e/multi-sql.cy.js
index 2ed8c943e..6f6340c23 100644
--- a/e2e-tests/cypress/e2e/multi-sql.cy.js
+++ b/e2e-tests/cypress/e2e/multi-sql.cy.js
@@ -184,7 +184,7 @@ describe('Import CSV - source error', () => {
 });
 
 describe('Import CSV - target error', () => {
-  multiTest({ skipMongo: true }, (connectionName, databaseName, engine, options = {}) => {
+  multiTest({}, (connectionName, databaseName, engine, options = {}) => {
     cy.contains(connectionName).click();
     if (databaseName) cy.contains(databaseName).click();
     cy.testid('ConnectionList_container')
@@ -194,7 +194,7 @@ describe('Import CSV - target error', () => {
 
     cy.get('input[type=file]').selectFile('cypress/fixtures/customers-20.csv', { force: true });
     cy.contains('customers-20');
-    cy.testid('ImportExportConfigurator_targetName_customers-20').clear().type('"]`');
+    cy.testid('ImportExportConfigurator_targetName_customers-20').clear().type('system."]`');
     cy.testid('ImportExportTab_executeButton').click();
     cy.testid('ImportExportConfigurator_errorInfoIcon_customers-20').click();
     cy.testid('ErrorMessageModal_message').should('be.visible');
diff --git a/plugins/dbgate-plugin-mongo/src/backend/createBulkInsertStream.js b/plugins/dbgate-plugin-mongo/src/backend/createBulkInsertStream.js
index 153f3f704..11c533292 100644
--- a/plugins/dbgate-plugin-mongo/src/backend/createBulkInsertStream.js
+++ b/plugins/dbgate-plugin-mongo/src/backend/createBulkInsertStream.js
@@ -1,10 +1,9 @@
 const ObjectId = require('mongodb').ObjectId;
-const { getLogger } = global.DBGATE_PACKAGES['dbgate-tools'];
+const { getLogger, extractErrorLogData } = global.DBGATE_PACKAGES['dbgate-tools'];
 const { EJSON } = require('bson');
 
 const logger = getLogger('mongoBulkInsert');
 
-
 function createBulkInsertStream(driver, stream, dbhan, name, options) {
   const collectionName = name.pureName;
   const db = dbhan.getDatabase();
@@ -31,21 +30,31 @@ function createBulkInsertStream(driver, stream, dbhan, name, options) {
   };
 
   writable.checkStructure = async () => {
-    if (options.dropIfExists) {
-      logger.info(`Dropping collection ${collectionName}`);
-      await db.collection(collectionName).drop();
-    }
-    if (options.truncate) {
-      logger.info(`Truncating collection ${collectionName}`);
-      await db.collection(collectionName).deleteMany({});
+    try {
+      if (options.dropIfExists) {
+        logger.info(`Dropping collection ${collectionName}`);
+        await db.collection(collectionName).drop();
+      }
+      if (options.truncate) {
+        logger.info(`Truncating collection ${collectionName}`);
+        await db.collection(collectionName).deleteMany({});
+      }
+    } catch (err) {
+      logger.error(extractErrorLogData(err), 'Error during preparing mongo bulk insert collection, stopped');
+      writable.destroy(err);
     }
   };
 
   writable.send = async () => {
-    const rows = writable.buffer;
-    writable.buffer = [];
+    try {
+      const rows = writable.buffer;
+      writable.buffer = [];
 
-    await db.collection(collectionName).insertMany(rows);
+      await db.collection(collectionName).insertMany(rows);
+    } catch (err) {
+      logger.error(extractErrorLogData(err), 'Error bulk insert collection, stopped');
+      writable.destroy(err);
+    }
   };
 
   writable.sendIfFull = async () => {