feat: add bi-sync module (#1617)

* feat: add sync module to bi-sync mysql

* feat: fix the delay problem

* feat: fix go mod

* feat: fix the varchar(100) parse error

* fix: fix go.mod space inconsistency

* fix: fix go.mod space inconsistency

* fix: use sql builder instead of concatenation

* fix: remove serverId

* fix: fix file is not `gofumpt`-ed (gofumpt) error

* feat: add mysql bi-sync

* feat: fix some data inconsistency problems

* feat: add function atuo get server uuid

* fix: encapsulate the struct to optimize the code

captcha/aliyun.go

@@ -86,8 +86,8 @@ func (captcha *AliyunCaptchaProvider) VerifyCaptcha(token, clientSecret string)
 	}
 
 	type captchaResponse struct {
-		Code    string `json:"Code"`
-		Message string `json:"Message"`
+		Code int    `json:"Code"`
+		Msg  string `json:"Msg"`
 	}
 	captchaResp := &captchaResponse{}
 
@@ -96,8 +96,8 @@ func (captcha *AliyunCaptchaProvider) VerifyCaptcha(token, clientSecret string)
 		return false, err
 	}
 
-	if captchaResp.Code != "100" {
-		return false, errors.New(captchaResp.Message)
+	if captchaResp.Code != 100 {
+		return false, errors.New(captchaResp.Msg)
 	}
 
 	return true, nil

sync/bi_sync_mysql.go

@@ -16,6 +16,8 @@ package sync
 
 import (
 	"fmt"
+	"strings"
+	"sync"
 
 	"github.com/go-mysql-org/go-mysql/canal"
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -24,60 +26,59 @@ import (
 	"github.com/xorm-io/xorm"
 )
 
-var (
-	dataSourceName1 string
-	dataSourceName2 string
-	engin1          *xorm.Engine
-	engin2          *xorm.Engine
-)
-
-func InitConfig() *canal.Config {
-	// init dataSource
-	dataSourceName1 = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", username1, password1, host1, port1, database1)
-	dataSourceName2 = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", username2, password2, host2, port2, database2)
-
-	// create engine
-	engin1, _ = CreateEngine(dataSourceName1)
-	engin2, _ = CreateEngine(dataSourceName2)
-	log.Info("init engine success…")
-
-	// config canal
-	cfg := canal.NewDefaultConfig()
-	cfg.Addr = fmt.Sprintf("%s:%d", host1, port1)
-	cfg.Password = password1
-	cfg.User = username1
-	// We only care table in database1
-	cfg.Dump.TableDB = database1
-	// cfg.Dump.Tables = []string{"user"}
-	log.Info("config canal success…")
-	return cfg
+type MyEventHandler struct {
+	dataSourceName string
+	engine         *xorm.Engine
+	serverId       uint32
+	serverUUID     string
+	GTID           string
+	canal.DummyEventHandler
 }
 
-func StartBinlogSync() error {
-	// init config
-	config := InitConfig()
-
-	c, err := canal.NewCanal(config)
-	pos, err := c.GetMasterPos()
+func StartCanal(cfg *canal.Config, username string, password string, host string, port int, database string) error {
+	c, err := canal.NewCanal(cfg)
 	if err != nil {
 		return err
 	}
 
+	GTIDSet, err := c.GetMasterGTIDSet()
+	if err != nil {
+		return err
+	}
+
+	eventHandler := GetMyEventHandler(username, password, host, port, database)
 	// Register a handler to handle RowsEvent
-	c.SetEventHandler(&MyEventHandler{})
-
-	// Start canal
-	c.RunFrom(pos)
+	c.SetEventHandler(&eventHandler)
 
+	// Start replication
+	err = c.StartFromGTID(GTIDSet)
+	if err != nil {
+		return err
+	}
 	return nil
 }
 
-type MyEventHandler struct {
-	canal.DummyEventHandler
+func StartBinlogSync() error {
+	var wg sync.WaitGroup
+	// init config
+	cfg1 := GetCanalConfig(username1, password1, host1, port1, database1)
+	cfg2 := GetCanalConfig(username2, password2, host2, port2, database2)
+
+	// start canal1 replication
+	go StartCanal(cfg1, username2, password2, host2, port2, database2)
+	wg.Add(1)
+
+	// start canal2 replication
+	go StartCanal(cfg2, username1, password1, host1, port1, database1)
+	wg.Add(1)
+
+	wg.Wait()
+	return nil
 }
 
-func OnTableChanged(header *replication.EventHeader, schema string, table string) error {
-	log.Info("table changed event")
+func (h *MyEventHandler) OnGTID(header *replication.EventHeader, gtid mysql.GTIDSet) error {
+	log.Info("OnGTID: ", gtid.String())
+	h.GTID = gtid.String()
 	return nil
 }
 
@@ -87,6 +88,13 @@ func (h *MyEventHandler) onDDL(header *replication.EventHeader, nextPos mysql.Po
 }
 
 func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
+	log.Info("serverId: ", e.Header.ServerID)
+	if strings.Contains(h.GTID, h.serverUUID) {
+		return nil
+	}
+
+	// Set the next gtid of the target library to the gtid of the current target library to avoid loopbacks
+	h.engine.Exec(fmt.Sprintf("SET GTID_NEXT= '%s'", h.GTID))
 	length := len(e.Table.Columns)
 	columnNames := make([]string, length)
 	oldColumnValue := make([]interface{}, length)
@@ -101,9 +109,12 @@ func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
 			isChar[i] = true
 		}
 	}
+	// get pk column name
+	pkColumnNames := GetPKColumnNames(columnNames, e.Table.PKColumns)
 
 	switch e.Action {
 	case canal.UpdateAction:
+		h.engine.Exec("BEGIN")
 		for i, row := range e.Rows {
 			for j, item := range row {
 				if i%2 == 0 {
@@ -114,27 +125,34 @@ func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
 					}
 				} else {
 					if isChar[j] == true {
-						newColumnValue[j] = fmt.Sprintf("%s", item)
+						if item == nil {
+							newColumnValue[j] = nil
+						} else {
+							newColumnValue[j] = fmt.Sprintf("%s", item)
+						}
 					} else {
 						newColumnValue[j] = fmt.Sprintf("%d", item)
 					}
 				}
 			}
-
 			if i%2 == 1 {
-				updateSql, args, err := GetUpdateSql(e.Table.Schema, e.Table.Name, columnNames, newColumnValue, oldColumnValue)
+				pkColumnValue := GetPKColumnValues(oldColumnValue, e.Table.PKColumns)
+				updateSql, args, err := GetUpdateSql(e.Table.Schema, e.Table.Name, columnNames, newColumnValue, pkColumnNames, pkColumnValue)
 				if err != nil {
 					return err
 				}
 
-				res, err := engin2.DB().Exec(updateSql, args...)
+				res, err := h.engine.DB().Exec(updateSql, args...)
 				if err != nil {
 					return err
 				}
 				log.Info(updateSql, args, res)
 			}
 		}
+		h.engine.Exec("COMMIT")
+		h.engine.Exec("SET GTID_NEXT='automatic'")
 	case canal.DeleteAction:
+		h.engine.Exec("BEGIN")
 		for _, row := range e.Rows {
 			for j, item := range row {
 				if isChar[j] == true {
@@ -144,22 +162,30 @@ func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
 				}
 			}
 
-			deleteSql, args, err := GetDeleteSql(e.Table.Schema, e.Table.Name, columnNames, oldColumnValue)
+			pkColumnValue := GetPKColumnValues(oldColumnValue, e.Table.PKColumns)
+			deleteSql, args, err := GetDeleteSql(e.Table.Schema, e.Table.Name, pkColumnNames, pkColumnValue)
 			if err != nil {
 				return err
 			}
 
-			res, err := engin2.DB().Exec(deleteSql, args...)
+			res, err := h.engine.DB().Exec(deleteSql, args...)
 			if err != nil {
 				return err
 			}
 			log.Info(deleteSql, args, res)
 		}
+		h.engine.Exec("COMMIT")
+		h.engine.Exec("SET GTID_NEXT='automatic'")
 	case canal.InsertAction:
+		h.engine.Exec("BEGIN")
 		for _, row := range e.Rows {
 			for j, item := range row {
 				if isChar[j] == true {
-					newColumnValue[j] = fmt.Sprintf("%s", item)
+					if item == nil {
+						newColumnValue[j] = nil
+					} else {
+						newColumnValue[j] = fmt.Sprintf("%s", item)
+					}
 				} else {
 					newColumnValue[j] = fmt.Sprintf("%d", item)
 				}
@@ -170,12 +196,14 @@ func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
 				return err
 			}
 
-			res, err := engin2.DB().Exec(insertSql, args...)
+			res, err := h.engine.DB().Exec(insertSql, args...)
 			if err != nil {
 				return err
 			}
 			log.Info(insertSql, args, res)
 		}
+		h.engine.Exec("COMMIT")
+		h.engine.Exec("SET GTID_NEXT='automatic'")
 	default:
 		log.Infof("%v", e.String())
 	}

sync/utils.go

@@ -15,20 +15,24 @@
 package sync
 
 import (
+	"fmt"
 	"log"
+	"strconv"
+
+	"github.com/go-mysql-org/go-mysql/canal"
 
 	"github.com/Masterminds/squirrel"
 	"github.com/xorm-io/xorm"
 )
 
-func GetUpdateSql(schemaName string, tableName string, columnNames []string, newColumnVal []interface{}, oldColumnVal []interface{}) (string, []interface{}, error) {
+func GetUpdateSql(schemaName string, tableName string, columnNames []string, newColumnVal []interface{}, pkColumnNames []string, pkColumnValue []interface{}) (string, []interface{}, error) {
 	updateSql := squirrel.Update(schemaName + "." + tableName)
 	for i, columnName := range columnNames {
 		updateSql = updateSql.Set(columnName, newColumnVal[i])
 	}
 
-	for i, columnName := range columnNames {
-		updateSql = updateSql.Where(squirrel.Eq{columnName: oldColumnVal[i]})
+	for i, pkColumnName := range pkColumnNames {
+		updateSql = updateSql.Where(squirrel.Eq{pkColumnName: pkColumnValue[i]})
 	}
 
 	sql, args, err := updateSql.ToSql()
@@ -45,11 +49,11 @@ func GetInsertSql(schemaName string, tableName string, columnNames []string, col
 	return insertSql.ToSql()
 }
 
-func GetDeleteSql(schemaName string, tableName string, columnNames []string, columnValue []interface{}) (string, []interface{}, error) {
+func GetDeleteSql(schemaName string, tableName string, pkColumnNames []string, pkColumnValue []interface{}) (string, []interface{}, error) {
 	deleteSql := squirrel.Delete(schemaName + "." + tableName)
 
-	for i, columnName := range columnNames {
-		deleteSql = deleteSql.Where(squirrel.Eq{columnName: columnValue[i]})
+	for i, columnName := range pkColumnNames {
+		deleteSql = deleteSql.Where(squirrel.Eq{columnName: pkColumnValue[i]})
 	}
 
 	return deleteSql.ToSql()
@@ -70,3 +74,57 @@ func CreateEngine(dataSourceName string) (*xorm.Engine, error) {
 	log.Println("mysql connection success……")
 	return engine, nil
 }
+
+func GetServerId(engin *xorm.Engine) (uint32, error) {
+	res, err := engin.QueryInterface("SELECT @@server_id")
+	if err != nil {
+		return 0, err
+	}
+	serverId, _ := strconv.ParseUint(fmt.Sprintf("%s", res[0]["@@server_id"]), 10, 32)
+	return uint32(serverId), nil
+}
+
+func GetServerUUID(engin *xorm.Engine) (string, error) {
+	res, err := engin.QueryString("show variables like 'server_uuid'")
+	if err != nil {
+		return "", err
+	}
+	serverUUID := fmt.Sprintf("%s", res[0]["Value"])
+	return serverUUID, err
+}
+
+func GetPKColumnNames(columnNames []string, PKColumns []int) []string {
+	pkColumnNames := make([]string, len(PKColumns))
+	for i, index := range PKColumns {
+		pkColumnNames[i] = columnNames[index]
+	}
+	return pkColumnNames
+}
+
+func GetPKColumnValues(columnValues []interface{}, PKColumns []int) []interface{} {
+	pkColumnNames := make([]interface{}, len(PKColumns))
+	for i, index := range PKColumns {
+		pkColumnNames[i] = columnValues[index]
+	}
+	return pkColumnNames
+}
+
+func GetCanalConfig(username string, password string, host string, port int, database string) *canal.Config {
+	// config canal
+	cfg := canal.NewDefaultConfig()
+	cfg.Addr = fmt.Sprintf("%s:%d", host, port)
+	cfg.Password = password
+	cfg.User = username
+	// We only care table in database1
+	cfg.Dump.TableDB = database
+	return cfg
+}
+
+func GetMyEventHandler(username string, password string, host string, port int, database string) MyEventHandler {
+	var eventHandler MyEventHandler
+	eventHandler.dataSourceName = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", username, password, host, port, database)
+	eventHandler.engine, _ = CreateEngine(eventHandler.dataSourceName)
+	eventHandler.serverId, _ = GetServerId(eventHandler.engine)
+	eventHandler.serverUUID, _ = GetServerUUID(eventHandler.engine)
+	return eventHandler
+}

web/src/ProviderEditPage.js

@@ -24,7 +24,7 @@ import * as ProviderEditTestSms from "./TestSmsWidget";
 import copy from "copy-to-clipboard";
 import {CaptchaPreview} from "./common/CaptchaPreview";
 import * as OrganizationBackend from "./backend/OrganizationBackend";
-import {PhoneNumberInput} from "./common/PhoneNumberInput";
+import {CountryCodeSelect} from "./common/CountryCodeSelect";
 
 const {Option} = Select;
 const {TextArea} = Input;
@@ -667,7 +667,7 @@ class ProviderEditPage extends React.Component {
                 </Col>
                 <Col span={4} >
                   <Input.Group compact>
-                    <PhoneNumberInput
+                    <CountryCodeSelect
                       style={{width: "30%"}}
                       value={this.state.provider.content}
                       onChange={(value) => {

web/src/UserEditPage.js

@@ -29,7 +29,7 @@ import SelectRegionBox from "./SelectRegionBox";
 import WebAuthnCredentialTable from "./WebauthnCredentialTable";
 import ManagedAccountTable from "./ManagedAccountTable";
 import PropertyTable from "./propertyTable";
-import {PhoneNumberInput} from "./common/PhoneNumberInput";
+import {CountryCodeSelect} from "./common/CountryCodeSelect";
 
 const {Option} = Select;
 
@@ -312,7 +312,7 @@ class UserEditPage extends React.Component {
           </Col>
           <Col style={{paddingRight: "20px"}} span={11} >
             <Input.Group compact style={{width: "280Px"}}>
-              <PhoneNumberInput
+              <CountryCodeSelect
                 style={{width: "30%"}}
                 // disabled={!Setting.isLocalAdminUser(this.props.account) ? true : disabled}
                 value={this.state.user.countryCode}

web/src/auth/SignupPage.js

@@ -26,7 +26,7 @@ import SelectRegionBox from "../SelectRegionBox";
 import CustomGithubCorner from "../CustomGithubCorner";
 import SelectLanguageBox from "../SelectLanguageBox";
 import {withRouter} from "react-router-dom";
-import {PhoneNumberInput} from "../common/PhoneNumberInput";
+import {CountryCodeSelect} from "../common/CountryCodeSelect";
 
 const formItemLayout = {
   labelCol: {
@@ -208,7 +208,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="username"
-          key="username"
           label={i18next.t("signup:Username")}
           rules={[
             {
@@ -227,7 +226,6 @@ class SignupPage extends React.Component {
           <React.Fragment>
             <Form.Item
               name="firstName"
-              key="firstName"
               label={i18next.t("general:First name")}
               rules={[
                 {
@@ -241,7 +239,6 @@ class SignupPage extends React.Component {
             </Form.Item>
             <Form.Item
               name="lastName"
-              key="lastName"
               label={i18next.t("general:Last name")}
               rules={[
                 {
@@ -260,7 +257,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="name"
-          key="name"
           label={(signupItem.rule === "Real name" || signupItem.rule === "First, last") ? i18next.t("general:Real name") : i18next.t("general:Display name")}
           rules={[
             {
@@ -277,7 +273,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="affiliation"
-          key="affiliation"
           label={i18next.t("user:Affiliation")}
           rules={[
             {
@@ -294,7 +289,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="idCard"
-          key="idCard"
           label={i18next.t("user:ID card")}
           rules={[
             {
@@ -316,7 +310,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="country_region"
-          key="region"
           label={i18next.t("user:Country/Region")}
           rules={[
             {
@@ -333,7 +326,6 @@ class SignupPage extends React.Component {
         <React.Fragment>
           <Form.Item
             name="email"
-            key="email"
             label={i18next.t("general:Email")}
             rules={[
               {
@@ -359,7 +351,6 @@ class SignupPage extends React.Component {
             signupItem.rule !== "No verification" &&
             <Form.Item
               name="emailCode"
-              key="emailCode"
               label={i18next.t("code:Email code")}
               rules={[{
                 required: required,
@@ -383,7 +374,6 @@ class SignupPage extends React.Component {
             <Input.Group compact>
               <Form.Item
                 name="countryCode"
-                key="countryCode"
                 noStyle
                 rules={[
                   {
@@ -392,14 +382,13 @@ class SignupPage extends React.Component {
                   },
                 ]}
               >
-                <PhoneNumberInput
+                <CountryCodeSelect
                   style={{width: "35%"}}
                   countryCodes={this.getApplicationObj().organizationObj.countryCodes}
                 />
               </Form.Item>
               <Form.Item
                 name="phone"
-                key="phone"
                 dependencies={["countryCode"]}
                 noStyle
                 rules={[
@@ -429,7 +418,6 @@ class SignupPage extends React.Component {
           </Form.Item>
           <Form.Item
             name="phoneCode"
-            key="phoneCode"
             label={i18next.t("code:Phone code")}
             rules={[
               {
@@ -443,7 +431,7 @@ class SignupPage extends React.Component {
               method={"signup"}
               onButtonClickArgs={[this.state.phone, "phone", Setting.getApplicationName(application)]}
               application={application}
-              countryCode={this.state.countryCode}
+              countryCode={this.form.current?.getFieldValue("countryCode")}
             />
           </Form.Item>
         </React.Fragment>
@@ -452,7 +440,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="password"
-          key="password"
           label={i18next.t("general:Password")}
           rules={[
             {
@@ -470,7 +457,6 @@ class SignupPage extends React.Component {
       return (
         <Form.Item
           name="confirm"
-          key="confirm"
           label={i18next.t("signup:Confirm")}
           dependencies={["password"]}
           hasFeedback

web/src/common/CaptchaModal.js

@@ -144,7 +144,7 @@ export const CaptchaModal = (props) => {
 
     return [
       <Button key="cancel" onClick={handleCancel}>{i18next.t("user:Cancel")}</Button>,
-      <Button key="ok" disabled={isOkDisabled} onClick={handleOk}>{i18next.t("user:OK")}</Button>,
+      <Button key="ok" disabled={isOkDisabled} type="primary" onClick={handleOk}>{i18next.t("user:OK")}</Button>,
     ];
   };
 

web/src/common/CountryCodeSelect.js

@@ -16,7 +16,7 @@ import {Select} from "antd";
 import * as Setting from "../Setting";
 import React from "react";
 
-export const PhoneNumberInput = (props) => {
+export const CountryCodeSelect = (props) => {
   const {onChange, style, disabled, value} = props;
   const countryCodes = props.countryCodes ?? [];