From 5e01721f756c5f61918e3b372514445604c36f07 Mon Sep 17 00:00:00 2001
From: Juan Ferrer Toribio <juan@verdnatura.es>
Date: Wed, 8 May 2024 14:18:45 +0200
Subject: [PATCH] fix: refs #4409 Columns changed detection fix

---
 config/producer.yml |  1 +
 mycdc.js            | 28 ++++++++++++++++++----------
 package.json        |  2 +-
 3 files changed, 20 insertions(+), 11 deletions(-)

diff --git a/config/producer.yml b/config/producer.yml
index 2e2c4b7..848e8a1 100644
--- a/config/producer.yml
+++ b/config/producer.yml
@@ -5,6 +5,7 @@ deleteNonEmpty: false
 amqp: amqp://user:password@localhost:5672
 pingInterval: 60
 flushInterval: 10
+serverId: 1
 db:
   host: localhost
   port: 3306
diff --git a/mycdc.js b/mycdc.js
index 9529672..64e93ed 100644
--- a/mycdc.js
+++ b/mycdc.js
@@ -60,7 +60,7 @@ module.exports = class MyCDC {
             tableInfo = {
               queues: new Map(),
               events: new Map(),
-              columns: false,
+              columnSet: false,
               fk: 'id'
             };
             tableMap.set(tableName, tableInfo);
@@ -79,13 +79,13 @@ module.exports = class MyCDC {
 
           const columns = table.columns;
           if (columns) {
-            if (tableInfo.columns === false)
-              tableInfo.columns = new Set();
-            if (tableInfo.columns !== true)
+            if (tableInfo.columnSet === false)
+              tableInfo.columnSet = new Set();
+            if (tableInfo.columnSet !== true)
               for (const column of columns)
-                tableInfo.columns.add(column);
+                tableInfo.columnSet.add(column);
           } else
-            tableInfo.columns = true;
+            tableInfo.columnSet = true;
 
           if (table.id)
             tableInfo.id = table.id;
@@ -96,8 +96,15 @@ module.exports = class MyCDC {
     }
 
     const includeSchema = {};
-    for (const [schemaName, tableMap] of this.schemaMap)
+
+    for (const [schemaName, tableMap] of this.schemaMap) {
       includeSchema[schemaName] = Array.from(tableMap.keys());
+
+      for (const [tableName, tableInfo] of tableMap) {
+        if (tableInfo.columnSet !== true)
+          tableInfo.columns = Array.from(tableInfo.columnSet.keys());
+      }
+    }
   
     this.opts = {
       includeEvents: [
@@ -107,7 +114,8 @@ module.exports = class MyCDC {
         'updaterows',
         'deleterows'
       ],
-      includeSchema
+      includeSchema,
+      serverId: conf.serverId
     };
 
     if (conf.testMode)
@@ -353,9 +361,9 @@ module.exports = class MyCDC {
     if (isUpdate) {
       rows = [];
       cols = new Set();
-      let columns = tableInfo.columns === true
+      const columns = tableInfo.columnSet === true
         ? Object.keys(evt.rows[0].after)
-        : tableInfo.columns.keys();
+        : tableInfo.columns;
 
       for (const row of evt.rows) {
         let nColsChanged = 0;
diff --git a/package.json b/package.json
index fa89ccd..42eda9c 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "mycdc",
-  "version": "0.0.13",
+  "version": "0.0.14",
   "author": "Verdnatura Levante SL",
   "description": "Asynchronous DB calculations reading the binary log",
   "license": "GPL-3.0",