Adding lock for eliminating any race conditions between publish and r…

…eplicate

db/change_log.go

@@ -130,6 +130,9 @@ func (conn *SqliteStreamDB) tableCDCScriptFor(tableName string) (string, error)
 }
 
 func (conn *SqliteStreamDB) consumeReplicationEvent(event *ChangeLogEvent) error {
+	conn.dbLock.Lock()
+	defer conn.dbLock.Unlock()
+
 	return conn.WithTx(func(tnx *goqu.TxDatabase) error {
 		primaryKeyMap := conn.getPrimaryKeyMap(event)
 		if primaryKeyMap == nil {
@@ -245,6 +248,9 @@ func (conn *SqliteStreamDB) publishChangeLog() {
 	}
 	defer conn.publishLock.Unlock()
 
+	conn.dbLock.Lock()
+	defer conn.dbLock.Unlock()
+
 	changes, err := conn.getGlobalChanges(ScanLimit)
 	if err != nil {
 		log.Error().Err(err).Msg("Unable to scan global changes")

db/sqlite.go

@@ -25,6 +25,7 @@ type SqliteStreamDB struct {
 	rawConnection *sqlite3.SQLiteConn
 	watcher       *fsnotify.Watcher
 	publishLock   *sync.Mutex
+	dbLock        *sync.Mutex
 
 	dbPath            string
 	prefix            string
@@ -68,6 +69,7 @@ func OpenStreamDB(path string) (*SqliteStreamDB, error) {
 		watcher:           nil,
 		dbPath:            path,
 		prefix:            MarmotPrefix,
+		dbLock:            &sync.Mutex{},
 		publishLock:       &sync.Mutex{},
 		watchTablesSchema: map[string][]*ColumnInfo{},
 	}