From 078e4085588bcf4a77fd73723bb62c452dad3a63 Mon Sep 17 00:00:00 2001 From: DengBiao <2319963317@qq.com> Date: Thu, 8 Feb 2024 14:25:35 +0800 Subject: [PATCH] =?UTF-8?q?add=20=E6=88=90=E9=95=BF=E5=80=BC=E7=9B=91?= =?UTF-8?q?=E5=90=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...r_public_platoon_double_network_setting.go | 121 +++++++++++++++++ ...platoon_double_network_user_coin_record.go | 121 +++++++++++++++++ ...r_public_platoon_double_network_setting.go | 18 +++ ...platoon_double_network_user_coin_record.go | 12 ++ .../canal_user_virtual_coin_flow_consume.go | 127 ++++++++++++++++++ consume/init.go | 2 + consume/md/consume_key.go | 19 +-- consume/md/md_canal_user_virtual_coin_flow.go | 38 ++++++ 8 files changed, 449 insertions(+), 9 deletions(-) create mode 100644 app/db/db_user_public_platoon_double_network_setting.go create mode 100644 app/db/db_user_public_platoon_double_network_user_coin_record.go create mode 100644 app/db/model/user_public_platoon_double_network_setting.go create mode 100644 app/db/model/user_public_platoon_double_network_user_coin_record.go create mode 100644 consume/canal_user_virtual_coin_flow_consume.go create mode 100644 consume/md/md_canal_user_virtual_coin_flow.go diff --git a/app/db/db_user_public_platoon_double_network_setting.go b/app/db/db_user_public_platoon_double_network_setting.go new file mode 100644 index 0000000..48b0101 --- /dev/null +++ b/app/db/db_user_public_platoon_double_network_setting.go @@ -0,0 +1,121 @@ +package db + +import ( + "applet/app/db/model" + "applet/app/utils" + "applet/app/utils/logx" + "errors" + "fmt" + "reflect" + "xorm.io/xorm" +) + +// BatchSelectUserPublicPlatoonDoubleNetworkSettings 批量查询数据 TODO::和下面的方法重复了，建议采用下面的 `UserPublicPlatoonDoubleNetworkSettingFindByParams` 方法 +func BatchSelectUserPublicPlatoonDoubleNetworkSettings(Db *xorm.Engine, params map[string]interface{}) (*[]model.UserPublicPlatoonDoubleNetworkSetting, error) { + var UserPublicPlatoonDoubleNetworkSettingData []model.UserPublicPlatoonDoubleNetworkSetting + if err := Db.In(utils.AnyToString(params["key"]), params["value"]). + Find(&UserPublicPlatoonDoubleNetworkSettingData); err != nil { + return nil, logx.Warn(err) + } + return &UserPublicPlatoonDoubleNetworkSettingData, nil +} + +// UserPublicPlatoonDoubleNetworkSettingInsert 插入单条数据 +func UserPublicPlatoonDoubleNetworkSettingInsert(Db *xorm.Engine, UserPublicPlatoonDoubleNetworkSetting *model.UserPublicPlatoonDoubleNetworkSetting) (int, error) { + _, err := Db.InsertOne(UserPublicPlatoonDoubleNetworkSetting) + if err != nil { + return 0, err + } + return UserPublicPlatoonDoubleNetworkSetting.Id, nil +} + +// BatchAddUserPublicPlatoonDoubleNetworkSettings 批量新增数据 +func BatchAddUserPublicPlatoonDoubleNetworkSettings(Db *xorm.Engine, UserPublicPlatoonDoubleNetworkSettingData []*model.UserPublicPlatoonDoubleNetworkSetting) (int64, error) { + affected, err := Db.Insert(UserPublicPlatoonDoubleNetworkSettingData) + if err != nil { + return 0, err + } + return affected, nil +} + +func GetUserPublicPlatoonDoubleNetworkSettingCount(Db *xorm.Engine) int { + var UserPublicPlatoonDoubleNetworkSetting model.UserPublicPlatoonDoubleNetworkSetting + session := Db.Where("") + count, err := session.Count(&UserPublicPlatoonDoubleNetworkSetting) + if err != nil { + return 0 + } + return int(count) +} + +// UserPublicPlatoonDoubleNetworkSettingDelete 删除记录 +func UserPublicPlatoonDoubleNetworkSettingDelete(Db *xorm.Engine, id interface{}) (int64, error) { + if reflect.TypeOf(id).Kind() == reflect.Slice { + return Db.In("id", id).Delete(model.UserPublicPlatoonDoubleNetworkSetting{}) + } else { + return Db.Where("id = ?", id).Delete(model.UserPublicPlatoonDoubleNetworkSetting{}) + } +} + +// UserPublicPlatoonDoubleNetworkSettingUpdate 更新记录 +func UserPublicPlatoonDoubleNetworkSettingUpdate(Db *xorm.Engine, id interface{}, UserPublicPlatoonDoubleNetworkSetting *model.UserPublicPlatoonDoubleNetworkSetting, forceColums ...string) (int64, error) { + var ( + affected int64 + err error + ) + if forceColums != nil { + affected, err = Db.Where("id=?", id).Cols(forceColums...).Update(UserPublicPlatoonDoubleNetworkSetting) + } else { + affected, err = Db.Where("id=?", id).Update(UserPublicPlatoonDoubleNetworkSetting) + } + if err != nil { + return 0, err + } + return affected, nil +} + +// UserPublicPlatoonDoubleNetworkSettingGetOneByParams 通过传入的参数查询数据(单条) +func UserPublicPlatoonDoubleNetworkSettingGetOneByParams(Db *xorm.Engine, params map[string]interface{}) (*model.UserPublicPlatoonDoubleNetworkSetting, error) { + var m model.UserPublicPlatoonDoubleNetworkSetting + var query = fmt.Sprintf("%s =?", params["key"]) + has, err := Db.Where(query, params["value"]).Get(&m) + if err != nil { + return nil, logx.Error(err) + } + if has == false { + return nil, nil + } + return &m, nil +} + +// UserPublicPlatoonDoubleNetworkSettingFindByParams 通过传入的参数查询数据(多条) +func UserPublicPlatoonDoubleNetworkSettingFindByParams(Db *xorm.Engine, params map[string]interface{}) (*[]model.UserPublicPlatoonDoubleNetworkSetting, error) { + var m []model.UserPublicPlatoonDoubleNetworkSetting + if params["value"] == nil { + return nil, errors.New("参数有误") + } + if params["key"] == nil { + //查询全部数据 + err := Db.Find(&m) + if err != nil { + return nil, logx.Error(err) + } + return &m, nil + } else { + if reflect.TypeOf(params["value"]).Kind() == reflect.Slice { + //指定In查询 + if err := Db.In(utils.AnyToString(params["key"]), params["value"]).Find(&m); err != nil { + return nil, logx.Warn(err) + } + return &m, nil + } else { + var query = fmt.Sprintf("%s =?", params["key"]) + err := Db.Where(query, params["value"]).Find(&m) + if err != nil { + return nil, logx.Error(err) + } + return &m, nil + } + + } +} diff --git a/app/db/db_user_public_platoon_double_network_user_coin_record.go b/app/db/db_user_public_platoon_double_network_user_coin_record.go new file mode 100644 index 0000000..f6dc1e5 --- /dev/null +++ b/app/db/db_user_public_platoon_double_network_user_coin_record.go @@ -0,0 +1,121 @@ +package db + +import ( + "applet/app/db/model" + "applet/app/utils" + "applet/app/utils/logx" + "errors" + "fmt" + "reflect" + "xorm.io/xorm" +) + +// BatchSelectUserPublicPlatoonDoubleNetworkUserCoinRecords 批量查询数据 TODO::和下面的方法重复了，建议采用下面的 `UserPublicPlatoonDoubleNetworkUserCoinRecordFindByParams` 方法 +func BatchSelectUserPublicPlatoonDoubleNetworkUserCoinRecords(Db *xorm.Engine, params map[string]interface{}) (*[]model.UserPublicPlatoonDoubleNetworkUserCoinRecord, error) { + var UserPublicPlatoonDoubleNetworkUserCoinRecordData []model.UserPublicPlatoonDoubleNetworkUserCoinRecord + if err := Db.In(utils.AnyToString(params["key"]), params["value"]). + Find(&UserPublicPlatoonDoubleNetworkUserCoinRecordData); err != nil { + return nil, logx.Warn(err) + } + return &UserPublicPlatoonDoubleNetworkUserCoinRecordData, nil +} + +// UserPublicPlatoonDoubleNetworkUserCoinRecordInsert 插入单条数据 +func UserPublicPlatoonDoubleNetworkUserCoinRecordInsert(Db *xorm.Engine, UserPublicPlatoonDoubleNetworkUserCoinRecord *model.UserPublicPlatoonDoubleNetworkUserCoinRecord) (int, error) { + _, err := Db.InsertOne(UserPublicPlatoonDoubleNetworkUserCoinRecord) + if err != nil { + return 0, err + } + return UserPublicPlatoonDoubleNetworkUserCoinRecord.Id, nil +} + +// BatchAddUserPublicPlatoonDoubleNetworkUserCoinRecords 批量新增数据 +func BatchAddUserPublicPlatoonDoubleNetworkUserCoinRecords(Db *xorm.Engine, UserPublicPlatoonDoubleNetworkUserCoinRecordData []*model.UserPublicPlatoonDoubleNetworkUserCoinRecord) (int64, error) { + affected, err := Db.Insert(UserPublicPlatoonDoubleNetworkUserCoinRecordData) + if err != nil { + return 0, err + } + return affected, nil +} + +func GetUserPublicPlatoonDoubleNetworkUserCoinRecordCount(Db *xorm.Engine) int { + var UserPublicPlatoonDoubleNetworkUserCoinRecord model.UserPublicPlatoonDoubleNetworkUserCoinRecord + session := Db.Where("") + count, err := session.Count(&UserPublicPlatoonDoubleNetworkUserCoinRecord) + if err != nil { + return 0 + } + return int(count) +} + +// UserPublicPlatoonDoubleNetworkUserCoinRecordDelete 删除记录 +func UserPublicPlatoonDoubleNetworkUserCoinRecordDelete(Db *xorm.Engine, id interface{}) (int64, error) { + if reflect.TypeOf(id).Kind() == reflect.Slice { + return Db.In("id", id).Delete(model.UserPublicPlatoonDoubleNetworkUserCoinRecord{}) + } else { + return Db.Where("id = ?", id).Delete(model.UserPublicPlatoonDoubleNetworkUserCoinRecord{}) + } +} + +// UserPublicPlatoonDoubleNetworkUserCoinRecordUpdate 更新记录 +func UserPublicPlatoonDoubleNetworkUserCoinRecordUpdate(Db *xorm.Engine, id interface{}, UserPublicPlatoonDoubleNetworkUserCoinRecord *model.UserPublicPlatoonDoubleNetworkUserCoinRecord, forceColums ...string) (int64, error) { + var ( + affected int64 + err error + ) + if forceColums != nil { + affected, err = Db.Where("id=?", id).Cols(forceColums...).Update(UserPublicPlatoonDoubleNetworkUserCoinRecord) + } else { + affected, err = Db.Where("id=?", id).Update(UserPublicPlatoonDoubleNetworkUserCoinRecord) + } + if err != nil { + return 0, err + } + return affected, nil +} + +// UserPublicPlatoonDoubleNetworkUserCoinRecordGetOneByParams 通过传入的参数查询数据(单条) +func UserPublicPlatoonDoubleNetworkUserCoinRecordGetOneByParams(Db *xorm.Engine, params map[string]interface{}) (*model.UserPublicPlatoonDoubleNetworkUserCoinRecord, error) { + var m model.UserPublicPlatoonDoubleNetworkUserCoinRecord + var query = fmt.Sprintf("%s =?", params["key"]) + has, err := Db.Where(query, params["value"]).Get(&m) + if err != nil { + return nil, logx.Error(err) + } + if has == false { + return nil, nil + } + return &m, nil +} + +// UserPublicPlatoonDoubleNetworkUserCoinRecordFindByParams 通过传入的参数查询数据(多条) +func UserPublicPlatoonDoubleNetworkUserCoinRecordFindByParams(Db *xorm.Engine, params map[string]interface{}) (*[]model.UserPublicPlatoonDoubleNetworkUserCoinRecord, error) { + var m []model.UserPublicPlatoonDoubleNetworkUserCoinRecord + if params["value"] == nil { + return nil, errors.New("参数有误") + } + if params["key"] == nil { + //查询全部数据 + err := Db.Find(&m) + if err != nil { + return nil, logx.Error(err) + } + return &m, nil + } else { + if reflect.TypeOf(params["value"]).Kind() == reflect.Slice { + //指定In查询 + if err := Db.In(utils.AnyToString(params["key"]), params["value"]).Find(&m); err != nil { + return nil, logx.Warn(err) + } + return &m, nil + } else { + var query = fmt.Sprintf("%s =?", params["key"]) + err := Db.Where(query, params["value"]).Find(&m) + if err != nil { + return nil, logx.Error(err) + } + return &m, nil + } + + } +} diff --git a/app/db/model/user_public_platoon_double_network_setting.go b/app/db/model/user_public_platoon_double_network_setting.go new file mode 100644 index 0000000..af3c862 --- /dev/null +++ b/app/db/model/user_public_platoon_double_network_setting.go @@ -0,0 +1,18 @@ +package model + +import ( + "time" +) + +type UserPublicPlatoonDoubleNetworkSetting struct { + Id int `json:"id" xorm:"not null pk autoincr INT(11)"` + IsOpen int `json:"is_open" xorm:"not null default 0 comment('是否开启(0:关闭 1:开启)') TINYINT(1)"` + SeveralTimes int `json:"several_times" xorm:"not null default 0 comment('几乘') TINYINT(3)"` + SeveralRows int `json:"several_rows" xorm:"not null default 0 comment('几排') TINYINT(3)"` + OriginatorUid int `json:"originator_uid" xorm:"not null default -1 comment('创始人uid') INT(11)"` + SettlementDate string `json:"settlement_date" xorm:"not null default '0000-00-00 00:00:00' comment('结算日期') CHAR(50)"` + CoinId int `json:"coin_id" xorm:"not null default 0 comment('虚拟币id(作用于成长值)') INT(11)"` + Ext string `json:"ext" xorm:"comment('拓展字段(json存储)') TEXT"` + CreateAt time.Time `json:"create_at" xorm:"not null default 'CURRENT_TIMESTAMP' DATETIME"` + UpdateAt time.Time `json:"update_at" xorm:"not null default 'CURRENT_TIMESTAMP' DATETIME"` +} diff --git a/app/db/model/user_public_platoon_double_network_user_coin_record.go b/app/db/model/user_public_platoon_double_network_user_coin_record.go new file mode 100644 index 0000000..a2ec6ba --- /dev/null +++ b/app/db/model/user_public_platoon_double_network_user_coin_record.go @@ -0,0 +1,12 @@ +package model + +type UserPublicPlatoonDoubleNetworkUserCoinRecord struct { + Id int `json:"id" xorm:"not null pk autoincr INT(11)"` + Uid int `json:"uid" xorm:"not null default 0 comment('uid') INT(11)"` + RecommendUid int `json:"recommend_uid" xorm:"not null default 0 comment('推荐人uid') INT(11)"` + LastAmount string `json:"last_amount" xorm:"not null default 0.0000 comment('上次金额') DECIMAL(10,4)"` + Amount string `json:"amount" xorm:"not null default 0.0000 comment('当前金额') DECIMAL(10,4)"` + CoinId int `json:"coin_id" xorm:"not null default 0 comment('虚拟币id(作用于成长值)') INT(11)"` + CreateAt string `json:"create_at" xorm:"not null default 'CURRENT_TIMESTAMP' DATETIME"` + UpdateAt string `json:"update_at" xorm:"not null default 'CURRENT_TIMESTAMP' DATETIME"` +} diff --git a/consume/canal_user_virtual_coin_flow_consume.go b/consume/canal_user_virtual_coin_flow_consume.go new file mode 100644 index 0000000..5591a7a --- /dev/null +++ b/consume/canal_user_virtual_coin_flow_consume.go @@ -0,0 +1,127 @@ +package consume + +import ( + "applet/app/db" + "applet/app/db/model" + "applet/app/utils" + "applet/app/utils/logx" + "applet/consume/md" + "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" + "encoding/json" + "errors" + "fmt" + "github.com/shopspring/decimal" + "github.com/streadway/amqp" + "strings" + "time" +) + +func CanalUserVirtualCoinFlowConsume(queue md.MqQueue) { + fmt.Println(">>>>>>>>>>>>CanalUserVirtualCoinFlowConsume>>>>>>>>>>>>") + ch, err := rabbit.Cfg.Pool.GetChannel() + if err != nil { + logx.Error(err) + return + } + defer ch.Release() + //1、将自己绑定到交换机上 + ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) + //2、取出数据进行消费 + ch.Qos(1) + delivery := ch.Consume(queue.Name, false) + + var res amqp.Delivery + var ok bool + for { + res, ok = <-delivery + if ok == true { + //fmt.Println(string(res.Body)) + fmt.Println(">>>>>>>>>>>>>>>>CanalUserVirtualCoinFlowConsume<<<<<<<<<<<<<<<<<<<<<<<<<") + err = handleCanalUserVirtualCoinFlow(res.Body) + if err != nil { + fmt.Println("handleCanalUserVirtualCoinFlow_ERR:::::", err.Error()) + } + //_ = res.Reject(false) + err = res.Ack(true) + fmt.Println("err ::: ", err) + } else { + panic(errors.New("error getting message")) + } + } + fmt.Println("get msg done") +} + +func handleCanalUserVirtualCoinFlow(msg []byte) error { + //1、解析canal采集至mq中queue的数据结构体 + var canalMsg *md.CanalUserVirtualCoinFlowOrderMessage[md.CanalUserVirtualCoinFlowOrder] + err := json.Unmarshal(msg, &canalMsg) + if err != nil { + return err + } + + masterId := strings.Split(canalMsg.Database, "_")[1] + if masterId != "32053480" { + return nil + } + engine := db.DBs[masterId] + now := time.Now() + + //2、查找 one_circles_green_energy_basic_setting 基础设置 + userPublicPlatoonDoubleNetworkSetting, err := db.UserPublicPlatoonDoubleNetworkSettingGetOneByParams(engine, map[string]interface{}{ + "key": "is_open", + "value": 1, + }) + if err != nil { + return err + } + if userPublicPlatoonDoubleNetworkSetting == nil { + return errors.New("公排双网未开启") + } + if canalMsg.Type == md.CanalMsgInsertSqlType { + if canalMsg.Data[0].CoinId == utils.IntToStr(userPublicPlatoonDoubleNetworkSetting.CoinId) { + //3、查找 user_public_platoon_double_network_user_coin_record + userPublicPlatoonDoubleNetworkUserCoinRecord, err1 := db.UserPublicPlatoonDoubleNetworkUserCoinRecordGetOneByParams(engine, map[string]interface{}{ + "key": "uid", + "value": canalMsg.Data[0].Uid, + }) + if err1 != nil { + return err1 + } + if userPublicPlatoonDoubleNetworkUserCoinRecord == nil { + userProfile, err2 := db.UserProfileFindByIDSess(engine.NewSession(), canalMsg.Data[0].Uid) + if err2 != nil { + return err2 + } + //新增记录 + _, err3 := db.UserPublicPlatoonDoubleNetworkUserCoinRecordInsert(engine, &model.UserPublicPlatoonDoubleNetworkUserCoinRecord{ + Uid: utils.StrToInt(canalMsg.Data[0].Uid), + LastAmount: canalMsg.Data[0].AfterAmout, + Amount: canalMsg.Data[0].AfterAmout, + RecommendUid: userProfile.ParentUid, + CoinId: utils.StrToInt(canalMsg.Data[0].CoinId), + CreateAt: now.Format("2006-01-02 15:04:05"), + UpdateAt: now.Format("2006-01-02 15:04:05"), + }) + if err3 != nil { + return err3 + } + } else { + //更新记录 + afterAmount, _ := decimal.NewFromString(canalMsg.Data[0].AfterAmout) + amount, _ := decimal.NewFromString(userPublicPlatoonDoubleNetworkUserCoinRecord.Amount) + lastAmount, _ := decimal.NewFromString(userPublicPlatoonDoubleNetworkUserCoinRecord.LastAmount) + if canalMsg.Data[0].Direction == "1" || canalMsg.Data[0].Direction == "2" { + //收入 && 支出 + userPublicPlatoonDoubleNetworkUserCoinRecord.Amount = amount.Add(afterAmount.Sub(lastAmount)).String() + } + userPublicPlatoonDoubleNetworkUserCoinRecord.LastAmount = canalMsg.Data[0].AfterAmout + _, err2 := db.UserPublicPlatoonDoubleNetworkUserCoinRecordUpdate(engine, userPublicPlatoonDoubleNetworkUserCoinRecord.Id, userPublicPlatoonDoubleNetworkUserCoinRecord, "amount", "last_amount") + if err2 != nil { + return err2 + } + } + } + } + + return nil +} diff --git a/consume/init.go b/consume/init.go index cbdb6ab..3e88fe5 100644 --- a/consume/init.go +++ b/consume/init.go @@ -62,6 +62,8 @@ func initConsumes() { jobs[consumeMd.ZhiosExpressOrderFail] = ZhiosExpressOrderFail jobs[consumeMd.ZhiosWithdrawReward] = ZhiosWithdrawReward + jobs[consumeMd.CanalUserVirtualCcoinFlowFunName] = CanalUserVirtualCoinFlowConsume + ////////////////////////////////////// V1 ///////////////////////////////////////////////////// //jobs[consumeMd.CloudIssuanceMsgCallBackFunName] = CloudIssuanceMsgCallBackConsume diff --git a/consume/md/consume_key.go b/consume/md/consume_key.go index b850b5c..8a1be54 100644 --- a/consume/md/consume_key.go +++ b/consume/md/consume_key.go @@ -380,15 +380,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ BindKey: "", ConsumeFunName: "ZhiosIntegralProxyRecharge", }, - //{ - // ExchangeName: "zhios.order_buckle.exchange", - // Name: "zhios_order_buckle_dev", - // Type: DirectQueueType, - // IsPersistent: false, - // RoutKey: "order_buckle_dev", - // BindKey: "", - // ConsumeFunName: "ZhiosOrderBuckle", - //}, + { + ExchangeName: "canal.topic", + Name: "canal_user_virtual_coin_flow", + Type: TopicQueueType, + IsPersistent: false, + RoutKey: "canal_user_virtual_coin_flow", + BindKey: "", + ConsumeFunName: "CanalUserVirtualCcoinFlowConsume", + }, } const ( @@ -433,4 +433,5 @@ const ( ZhiosAcquisitionConditionDevFunName = "ZhiosAcquisitionConditionDev" SupplyCloudChainFenxiaoNewChangeFunName = "SupplyCloudChainFenxiaoNewChangeConsume" MallAddSupplyGoodsFunName = "MallAddSupplyGoodsConsume" + CanalUserVirtualCcoinFlowFunName = "CanalUserVirtualCcoinFlowConsume" ) diff --git a/consume/md/md_canal_user_virtual_coin_flow.go b/consume/md/md_canal_user_virtual_coin_flow.go new file mode 100644 index 0000000..f89b5bc --- /dev/null +++ b/consume/md/md_canal_user_virtual_coin_flow.go @@ -0,0 +1,38 @@ +package md + +type CanalUserVirtualCoinFlowOrder struct { + Id string `json:"id"` + Uid string `json:"uid"` + CoinId string `json:"coin_id"` + Direction string `json:"direction"` + Title string `json:"title"` + OrdId string `json:"ord_id"` + Amout string `json:"amout"` + BeforeAmout string `json:"before_amout"` + AfterAmout string `json:"after_amout"` + SysFee string `json:"sys_fee"` + CreateTime string `json:"create_time"` + CoinIdTo string `json:"coin_id_to"` + TransferType string `json:"transfer_type"` + IsRevoke string `json:"is_revoke"` + TransferId string `json:"transfer_id"` + TransferMoney string `json:"transfer_money"` + ToUid string `json:"to_uid"` + State string `json:"state"` + ThirdPartId string `json:"third_part_id"` + Date string `json:"date"` + Price string `json:"price"` +} + +type CanalUserVirtualCoinFlowOrderMessage[T any] struct { + Data []T `json:"data"` + Database string `json:"database"` + ES int64 `json:"es"` + ID int64 `json:"id"` + IsDdl bool `json:"isDdl"` + Old []T `json:"old"` + PkNames []string `json:"pkNames"` + Table string `json:"table"` + TS int64 `json:"ts"` + Type string `json:"type"` +}