Updates processors to convert go maps to json bits for non json columns (#3473)

Author: alishakawaguchi

worker/pkg/benthos/sql/processor_neosync_mysql.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/doug-martin/goqu/v9"
+	"github.com/nucleuscloud/neosync/internal/gotypeutil"
 	neosynctypes "github.com/nucleuscloud/neosync/internal/neosync-types"
 	neosync_benthos "github.com/nucleuscloud/neosync/worker/pkg/benthos"
 	"github.com/redpanda-data/benthos/v4/public/service"
@@ -154,8 +155,7 @@ func getMysqlValue(
 		return value, nil
 	}
 
-	switch datatype {
-	case "json":
+	if datatype == "json" {
 		if v, ok := value.([]byte); ok {
 			validJson, err := getValidJson(v)
 			if err != nil {
@@ -171,9 +171,17 @@ func getMysqlValue(
 			return nil, fmt.Errorf("unable to marshal mysql json to bits: %w", err)
 		}
 		return bits, nil
-	default:
-		return value, nil
 	}
+
+	if gotypeutil.IsMap(value) {
+		bits, err := json.Marshal(value)
+		if err != nil {
+			return nil, fmt.Errorf("unable to marshal go map to json bits: %w", err)
+		}
+		return bits, nil
+	}
+
+	return value, nil
 }
 
 func getMysqlNeosyncValue(root any) (value any, isNeosyncValue bool, err error) {

worker/pkg/benthos/sql/processor_neosync_pgx.go

@@ -191,9 +191,15 @@ func getPgxValue(
 			return string(byteSlice), nil
 		}
 		return value, nil
-	default:
-		return value, nil
 	}
+	if gotypeutil.IsMap(value) {
+		bits, err := json.Marshal(value)
+		if err != nil {
+			return nil, fmt.Errorf("unable to marshal go map to json bits: %w", err)
+		}
+		return bits, nil
+	}
+	return value, nil
 }
 
 func getPgxNeosyncValue(root any) (value any, isNeosyncValue bool, err error) {

worker/pkg/benthos/utils.go

@@ -47,6 +47,7 @@ func IsCriticalError(errMsg string) bool {
 		"ON CONFLICT does not support deferrable unique constraints",
 		"ON CONFLICT",
 		"SQLSTATE", // any sqlstate error should result in ending
+		"goqu_encode_error",
 		"doesn't have a default value",
 		"column does not allow nulls",
 	}

worker/pkg/query-builder/query-builder.go

@@ -2,6 +2,7 @@ package querybuilder
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/doug-martin/goqu/v9"
 	mgmtv1alpha1 "github.com/nucleuscloud/neosync/backend/gen/go/protos/mgmt/v1alpha1"
@@ -115,6 +116,10 @@ func BuildInsertQuery(
 
 	query, args, err := insert.ToSQL()
 	if err != nil {
+		// check if it's a goqu encoding error and sanitize it
+		if strings.Contains(err.Error(), "goqu_encode_error") {
+			return "", nil, fmt.Errorf("goqu_encode_error: Unable to encode value")
+		}
 		return "", nil, err
 	}
 	return query, args, nil
@@ -147,6 +152,10 @@ func BuildUpdateQuery(
 
 	query, _, err := update.ToSQL()
 	if err != nil {
+		// check if it's a goqu encoding error and sanitize it
+		if strings.Contains(err.Error(), "goqu_encode_error") {
+			return "", fmt.Errorf("goqu_encode_error: Unable to encode value")
+		}
 		return "", err
 	}
 	return query, nil