nft持有查询定时任务

Makefile

@@ -32,7 +32,7 @@ api:
 .PHONY: build
 # 编译二进制可执行文件
 build:
-	mkdir -p bin/ && go build -ldflags="-s -w" -tags no_k8s -o ./bin/ ./...
+	mkdir -p bin/ && CGO_ENABLED=0 go build -ldflags="-s -w" -tags no_k8s -o ./bin/ ./...
 
 .PHONY: img
 # 构建Docker镜像

doc/sql/novatask.sql

@@ -1,3 +1,4 @@
+#DROP TABLE IF EXISTS `nh_task_progress`;
 CREATE TABLE `nh_task_progress`
 (
     `id`          int(11) NOT NULL AUTO_INCREMENT,
@@ -7,6 +8,34 @@ CREATE TABLE `nh_task_progress`
     `stage`       tinyint NOT NULL DEFAULT 0 COMMENT '任务的阶段, 0:未完成 1:待校验 2:已完成未领取 3:已领取',
     `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
     `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
-    PRIMARY KEY (`id`) USING BTREE,
+    PRIMARY KEY (`id`),
     UNIQUE KEY `uid_task_id_seq` (`uid`, `task_id`, `task_seq`)
 ) COMMENT='用户任务节点';
+
+#DROP TABLE IF EXISTS `nh_nft_holder`;
+CREATE TABLE `nh_nft_holder`
+(
+    `id`         int         NOT NULL AUTO_INCREMENT,
+    `address`    varchar(80) NOT NULL COMMENT '钱包地址',
+    `token_id`   varchar(80) NOT NULL COMMENT 'token id',
+    `balance`    int(11)     NOT NULL DEFAULT 0 COMMENT '余额',
+    `update_seq` int         NOT NULL COMMENT '更新序列号',
+    `created_at` timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    `updated_at` timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
+    PRIMARY KEY (`id`),
+    UNIQUE KEY (`address`, `token_id`),
+    INDEX (`update_seq`)
+) COMMENT ='nft 持有表';
+
+#DROP TABLE IF EXISTS `nh_nft_holder_change_log`;
+CREATE TABLE `nh_nft_holder_change_log`
+(
+    `id`         int         NOT NULL AUTO_INCREMENT,
+    `address`    varchar(80) NOT NULL COMMENT '钱包地址',
+    `token_id`   varchar(80) NOT NULL COMMENT 'token id',
+    `value`      int(11)     NOT NULL COMMENT '变化数量',
+    `balance`    int(11)     NOT NULL COMMENT '余额',
+    `created_at` timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    `updated_at` timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
+    PRIMARY KEY (`id`)
+) COMMENT ='nft 持有表变化日志';

etc/novatask.yaml

@@ -4,6 +4,16 @@ Port: 443
 CertFile: "etc/cert/saas.crt"
 KeyFile: "etc/cert/saas.key"
 
+Log:
+#  Mode: "file"
+  Encoding: "json"
+  Path: "logs"
+  Level: "debug"
+  Compress: true
+  KeepDays: 7
+  MaxBackups: 3
+  MaxSize: 100
+
 Auth: # js-sdk鉴权相关配置
   AccessSecret: "Mj2G%szYe&$MP@ytNv8JktQN1n5^cPq%" # 鉴权token密钥
 
@@ -22,4 +32,13 @@ Earn:
   GameId: "c0deda99-bb15-47a2-a3be-f1fe2983cde2"
 
 DailyPay:
-  Contract: "xxx"
+  Contract: "0xf5b1c3f66c418071b025e3fa63e0f0100240951477fa45c760de80e3232b95bf::my_counter"
+  Network: "testnet"
+
+EarnCorn:
+  Spec: "@every 5m"
+  RunOnStart : true
+
+PledgeCron:
+  Spec: "@every 30m"
+  RunOnStart : true

internal/config/config.go

@@ -20,6 +20,13 @@ type Config struct {
 	}
 	Earn       earn.Config
 	DailyPay   DailyPay
+	EarnCorn   Cron `json:",optional"`
+	PledgeCron Cron `json:",optional"`
+}
+
+type Cron struct {
+	Spec       string
+	RunOnStart bool `json:",optional"`
 }
 
 type DailyPay struct {

internal/job/earn/earn.go

@@ -17,15 +17,18 @@ type Earn struct {
 
 func NewEarn(ctx context.Context, svcCtx *svc.ServiceContext) *Earn {
 	e := &Earn{ctx: ctx, svcCtx: svcCtx}
+	if e.svcCtx.Config.EarnCorn.RunOnStart {
 		e.Run()
+	}
 	return e
 }
 
 func (e *Earn) Spec() string {
-	return "@every 5m"
+	return e.svcCtx.Config.EarnCorn.Spec
 }
 
 func (e *Earn) Run() {
+	logx.Debugw("run earn cron task")
 	c, err := e.svcCtx.ConfigModel.FindOneByName(e.ctx, consts.EarnAllianceInviterId)
 	if err != nil {
 		logx.Errorw("find earn alliance inviter id failed", logx.Field("err", err))
@@ -41,6 +44,7 @@ func (e *Earn) pushUserInfo(shareId uint) {
 		logx.Errorw("find require push user failed", logx.Field("err", err), logx.Field("share_id", shareId))
 		return
 	}
+	logx.Debugw("find require push user", logx.Field("count", len(us)))
 	for _, u := range us {
 		ui, err := e.svcCtx.UserModel.FindOne(e.ctx, u.InvitedUid)
 		if err != nil {
@@ -53,15 +57,15 @@ func (e *Earn) pushUserInfo(shareId uint) {
 			twitterId = ut.TwitterId
 		}
 
+		err = e.svcCtx.PromoteBindModel.UpdatePushUser(e.ctx, u.Id)
+		if err != nil {
+			logx.Errorw("update push user failed", logx.Field("err", err), logx.Field("uid", u.InvitedUid), logx.Field("twitter_id", twitterId), logx.Field("email", ui.Email))
+		} else {
 			e.svcCtx.Earn.SetIdentifiers(cast.ToString(ui.Id), &ea.Identifiers{
 				Email:     ea.IdentifierFrom(ui.Email),
 				TwitterId: ea.IdentifierFrom(twitterId),
 			})
-		err = e.svcCtx.PromoteBindModel.UpdatePushUser(e.ctx, u.Id)
+			logx.Infow("push user info success", logx.Field("uid", u.InvitedUid), logx.Field("twitter_id", twitterId), logx.Field("email", ui.Email))
-		if err != nil {
-			logx.Errorw("update push user failed", logx.Field("err", err), logx.Field("uid", u.InvitedUid))
-		} else {
-			logx.Infow("push user info success", logx.Field("uid", u.InvitedUid))
 		}
 	}
 }
@@ -72,12 +76,13 @@ func (e *Earn) pushUserBind(shareId uint) {
 		logx.Errorw("find require push bind role user failed", logx.Field("err", err), logx.Field("share_id", shareId))
 		return
 	}
+	logx.Debugw("find require push bind role user", logx.Field("count", len(us)))
 	for _, u := range us {
-		e.svcCtx.Earn.Track(cast.ToString(u.Uid), "BIND_ROLE", nil, nil)
 		err = e.svcCtx.PromoteBindModel.UpdatePushRole(e.ctx, u.Id)
 		if err != nil {
 			logx.Errorw("update push user failed", logx.Field("err", err), logx.Field("uid", u.Uid))
 		} else {
+			e.svcCtx.Earn.Track(cast.ToString(u.Uid), "BIND_ROLE", nil, nil)
 			logx.Infow("push user info success", logx.Field("uid", u.Uid))
 		}
 	}

internal/job/job.go

@@ -31,7 +31,12 @@ func NewJob(svcCtx *svc.ServiceContext) *Job {
 	c := cron.New()
 	for _, cr := range cronList {
 		if cs, ok := cr.(Spec); ok {
-			_, err = c.AddJob(cs.Spec(), cr)
+			spec := cs.Spec()
+			if spec == "" {
+				logx.Errorw("cron job spec is empty")
+				continue
+			}
+			_, err = c.AddJob(spec, cr)
 			logx.Must(err)
 			continue
 		}

internal/job/pledge/nft.go

@@ -0,0 +1,31 @@
+package pledge
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+type OwnerList struct {
+	Owners []struct {
+		OwnerAddress  string `json:"ownerAddress"`
+		TokenBalances []struct {
+			TokenID string `json:"tokenId"`
+			Balance string `json:"balance"`
+		} `json:"tokenBalances"`
+	} `json:"owners"`
+	PageKey string `json:"pageKey"`
+}
+
+func GetOwnerList() (*OwnerList, error) {
+	contractAddress := "0x89B8D549feA2eBd2aA0b375ce0DCaBba79e7e636"
+	url := fmt.Sprintf("https://eth-mainnet.g.alchemy.com/nft/v3/alcht_1a183fAsPqF9upfTfp3AC1l0iedGLo/getOwnersForContract?contractAddress=%s&withTokenBalances=true", contractAddress)
+	resp, err := http.Get(url)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	result := new(OwnerList)
+	err = json.NewDecoder(resp.Body).Decode(result)
+	return result, err
+}

internal/job/pledge/pledge.go

@@ -2,7 +2,12 @@ package pledge
 
 import (
 	"context"
+	"errors"
+	"github.com/spf13/cast"
+	"github.com/zeromicro/go-zero/core/logx"
+	"nova_task/internal/model"
 	"nova_task/internal/svc"
+	"time"
 )
 
 type Pledge struct {
@@ -11,16 +16,100 @@ type Pledge struct {
 }
 
 func NewPledge(ctx context.Context, svcCtx *svc.ServiceContext) *Pledge {
-	return &Pledge{
+	pg := &Pledge{
 		ctx:    ctx,
 		svcCtx: svcCtx,
 	}
+	if svcCtx.Config.PledgeCron.RunOnStart {
+		pg.Run()
+	}
+	return pg
 }
 
 func (p *Pledge) Spec() string {
-	return "@every 30m"
+	return p.svcCtx.Config.PledgeCron.Spec
 }
 
 func (p *Pledge) Run() {
-
+	logx.Debugw("run pledge cron task")
+	ols, err := GetOwnerList()
+	if err != nil {
+		logx.Errorw("get owner list error", logx.Field("error", err))
+		return
+	}
+	updateSeq := int(time.Now().Unix())
+	for _, o := range ols.Owners {
+		for _, tk := range o.TokenBalances {
+			balance := cast.ToInt(tk.Balance)
+			logx.Debugw("owner token", logx.Field("address", o.OwnerAddress), logx.Field("token", tk.TokenID), logx.Field("balance", tk.Balance))
+			nft, err := p.svcCtx.NftHolderModel.FindOneByAddressTokenId(p.ctx, o.OwnerAddress, tk.TokenID)
+			if err != nil {
+				if errors.Is(err, model.ErrNotFound) {
+					nft = &model.NhNftHolder{
+						Address:   o.OwnerAddress,
+						TokenId:   tk.TokenID,
+						Balance:   balance,
+						UpdateSeq: updateSeq,
+					}
+				} else {
+					logx.Errorw("find nft holder error", logx.Field("error", err), logx.Field("address", o.OwnerAddress), logx.Field("token", tk.TokenID))
+					continue
+				}
+			}
+
+			var value int
+			if nft.Id == 0 {
+				// 新增
+				_, err = p.svcCtx.NftHolderModel.Insert(p.ctx, nft)
+				value = balance
+			} else {
+				// 持有数量变化
+				value = balance - nft.Balance
+				if value != 0 {
+					nft.Balance = balance
+					nft.UpdateSeq = updateSeq
+					err = p.svcCtx.NftHolderModel.Update(p.ctx, nft)
+				}
+			}
+
+			if err != nil {
+				logx.Errorw("insert or update nft holder error", logx.Field("error", err), logx.Field("address", o.OwnerAddress), logx.Field("token", tk.TokenID), logx.Field("balance", balance))
+			}
+
+			// 余额有变化，记录变化日志
+			if value != 0 {
+				_, err = p.svcCtx.NftHolderChangeLogModel.Insert(p.ctx, &model.NhNftHolderChangeLog{
+					Address: o.OwnerAddress,
+					TokenId: tk.TokenID,
+					Value:   balance,
+					Balance: balance,
+				})
+				if err != nil {
+					logx.Errorw("insert nft holder change log error", logx.Field("error", err), logx.Field("address", o.OwnerAddress), logx.Field("token", tk.TokenID))
+				}
+			}
+		}
+	}
+
+	// 删除已经不持有的地址，且添加变化日志
+	nfts, err := p.svcCtx.NftHolderModel.FindOtherUpdateSeq(p.ctx, updateSeq)
+	if err != nil {
+		logx.Errorw("find other update seq error", logx.Field("error", err))
+		return
+	}
+	for _, nft := range nfts {
+		_, err = p.svcCtx.NftHolderChangeLogModel.Insert(p.ctx, &model.NhNftHolderChangeLog{
+			Address: nft.Address,
+			TokenId: nft.TokenId,
+			Value:   -nft.Balance,
+			Balance: 0,
+		})
+		if err != nil {
+			logx.Errorw("delete nft holder error", logx.Field("error", err), logx.Field("address", nft.Address), logx.Field("token", nft.TokenId))
+		}
+	}
+	err = p.svcCtx.NftHolderModel.DeleteOtherUpdateSeq(p.ctx, updateSeq)
+	if err != nil {
+		logx.Errorw("delete other update seq error", logx.Field("error", err), logx.Field("update_seq", updateSeq))
+	}
 }

internal/logic/task/verify_task_result_logic.go

@@ -104,7 +104,7 @@ func (l *VerifyTaskResultLogic) checkoutTranscation(uid int, txHash string) bool
 		l.Errorw("get transaction owner address error", logx.Field("err", err))
 		return false
 	}
-	targetUid, err := l.svcCtx.WalletModel.FindWalletByAddress(l.ctx, address, aptos.StrPadAptosAddress(address))
+	targetUid, err := l.svcCtx.WalletModel.FindWalletByAddress(l.ctx, address)
 	if err != nil {
 		l.Errorw("find wallet by address error", logx.Field("err", err), logx.Field("address", address), logx.Field("target_uid", targetUid))
 		return false

internal/model/nh_nft_holder_change_log_model.go

@@ -0,0 +1,29 @@
+package model
+
+import "github.com/zeromicro/go-zero/core/stores/sqlx"
+
+var _ NhNftHolderChangeLogModel = (*customNhNftHolderChangeLogModel)(nil)
+
+type (
+	// NhNftHolderChangeLogModel is an interface to be customized, add more methods here,
+	// and implement the added methods in customNhNftHolderChangeLogModel.
+	NhNftHolderChangeLogModel interface {
+		nhNftHolderChangeLogModel
+		withSession(session sqlx.Session) NhNftHolderChangeLogModel
+	}
+
+	customNhNftHolderChangeLogModel struct {
+		*defaultNhNftHolderChangeLogModel
+	}
+)
+
+// NewNhNftHolderChangeLogModel returns a model for the database table.
+func NewNhNftHolderChangeLogModel(conn sqlx.SqlConn) NhNftHolderChangeLogModel {
+	return &customNhNftHolderChangeLogModel{
+		defaultNhNftHolderChangeLogModel: newNhNftHolderChangeLogModel(conn),
+	}
+}
+
+func (m *customNhNftHolderChangeLogModel) withSession(session sqlx.Session) NhNftHolderChangeLogModel {
+	return NewNhNftHolderChangeLogModel(sqlx.NewSqlConnFromSession(session))
+}

internal/model/nh_nft_holder_change_log_model_gen.go

@@ -0,0 +1,91 @@
+// Code generated by goctl. DO NOT EDIT.
+// versions:
+//  goctl version: 1.7.3
+
+package model
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/zeromicro/go-zero/core/stores/builder"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+var (
+	nhNftHolderChangeLogFieldNames          = builder.RawFieldNames(&NhNftHolderChangeLog{})
+	nhNftHolderChangeLogRows                = strings.Join(nhNftHolderChangeLogFieldNames, ",")
+	nhNftHolderChangeLogRowsExpectAutoSet   = strings.Join(stringx.Remove(nhNftHolderChangeLogFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), ",")
+	nhNftHolderChangeLogRowsWithPlaceHolder = strings.Join(stringx.Remove(nhNftHolderChangeLogFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), "=?,") + "=?"
+)
+
+type (
+	nhNftHolderChangeLogModel interface {
+		Insert(ctx context.Context, data *NhNftHolderChangeLog) (sql.Result, error)
+		FindOne(ctx context.Context, id int) (*NhNftHolderChangeLog, error)
+		Update(ctx context.Context, data *NhNftHolderChangeLog) error
+		Delete(ctx context.Context, id int) error
+	}
+
+	defaultNhNftHolderChangeLogModel struct {
+		conn  sqlx.SqlConn
+		table string
+	}
+
+	NhNftHolderChangeLog struct {
+		Id        int       `db:"id"`
+		Address   string    `db:"address"`    // 钱包地址
+		TokenId   string    `db:"token_id"`   // token id
+		Value     int       `db:"value"`      // 变化数量
+		Balance   int       `db:"balance"`    // 余额
+		CreatedAt time.Time `db:"created_at"` // 创建时间
+		UpdatedAt time.Time `db:"updated_at"` // 修改时间
+	}
+)
+
+func newNhNftHolderChangeLogModel(conn sqlx.SqlConn) *defaultNhNftHolderChangeLogModel {
+	return &defaultNhNftHolderChangeLogModel{
+		conn:  conn,
+		table: "`nh_nft_holder_change_log`",
+	}
+}
+
+func (m *defaultNhNftHolderChangeLogModel) Delete(ctx context.Context, id int) error {
+	query := fmt.Sprintf("delete from %s where `id` = ?", m.table)
+	_, err := m.conn.ExecCtx(ctx, query, id)
+	return err
+}
+
+func (m *defaultNhNftHolderChangeLogModel) FindOne(ctx context.Context, id int) (*NhNftHolderChangeLog, error) {
+	query := fmt.Sprintf("select %s from %s where `id` = ? limit 1", nhNftHolderChangeLogRows, m.table)
+	var resp NhNftHolderChangeLog
+	err := m.conn.QueryRowCtx(ctx, &resp, query, id)
+	switch err {
+	case nil:
+		return &resp, nil
+	case sqlx.ErrNotFound:
+		return nil, ErrNotFound
+	default:
+		return nil, err
+	}
+}
+
+func (m *defaultNhNftHolderChangeLogModel) Insert(ctx context.Context, data *NhNftHolderChangeLog) (sql.Result, error) {
+	query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?)", m.table, nhNftHolderChangeLogRowsExpectAutoSet)
+	ret, err := m.conn.ExecCtx(ctx, query, data.Address, data.TokenId, data.Value, data.Balance)
+	return ret, err
+}
+
+func (m *defaultNhNftHolderChangeLogModel) Update(ctx context.Context, data *NhNftHolderChangeLog) error {
+	query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, nhNftHolderChangeLogRowsWithPlaceHolder)
+	_, err := m.conn.ExecCtx(ctx, query, data.Address, data.TokenId, data.Value, data.Balance, data.Id)
+	return err
+}
+
+func (m *defaultNhNftHolderChangeLogModel) tableName() string {
+	return m.table
+}

internal/model/nh_nft_holder_model.go

@@ -0,0 +1,52 @@
+package model
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+)
+
+var _ NhNftHolderModel = (*customNhNftHolderModel)(nil)
+
+type (
+	// NhNftHolderModel is an interface to be customized, add more methods here,
+	// and implement the added methods in customNhNftHolderModel.
+	NhNftHolderModel interface {
+		nhNftHolderModel
+		withSession(session sqlx.Session) NhNftHolderModel
+		FindOtherUpdateSeq(ctx context.Context, updateSeq int) ([]*NhNftHolder, error)
+		DeleteOtherUpdateSeq(ctx context.Context, updateSeq int) error
+	}
+
+	customNhNftHolderModel struct {
+		*defaultNhNftHolderModel
+	}
+)
+
+func (m *customNhNftHolderModel) DeleteOtherUpdateSeq(ctx context.Context, updateSeq int) error {
+	delSql := fmt.Sprintf("delete from %s where `update_seq` != ?", m.table)
+	_, err := m.conn.ExecCtx(ctx, delSql, updateSeq)
+	return err
+}
+
+func (m *customNhNftHolderModel) FindOtherUpdateSeq(ctx context.Context, updateSeq int) ([]*NhNftHolder, error) {
+	query := fmt.Sprintf("select %s from %s where `update_seq` != ?", nhNftHolderRows, m.table)
+	var resp []*NhNftHolder
+	err := m.conn.QueryRowsCtx(ctx, &resp, query, updateSeq)
+	if err != nil && !errors.Is(err, sqlx.ErrNotFound) {
+		return nil, err
+	}
+	return resp, nil
+}
+
+// NewNhNftHolderModel returns a model for the database table.
+func NewNhNftHolderModel(conn sqlx.SqlConn) NhNftHolderModel {
+	return &customNhNftHolderModel{
+		defaultNhNftHolderModel: newNhNftHolderModel(conn),
+	}
+}
+
+func (m *customNhNftHolderModel) withSession(session sqlx.Session) NhNftHolderModel {
+	return NewNhNftHolderModel(sqlx.NewSqlConnFromSession(session))
+}

internal/model/nh_nft_holder_model_gen.go

@@ -0,0 +1,106 @@
+// Code generated by goctl. DO NOT EDIT.
+// versions:
+//  goctl version: 1.7.3
+
+package model
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/zeromicro/go-zero/core/stores/builder"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+var (
+	nhNftHolderFieldNames          = builder.RawFieldNames(&NhNftHolder{})
+	nhNftHolderRows                = strings.Join(nhNftHolderFieldNames, ",")
+	nhNftHolderRowsExpectAutoSet   = strings.Join(stringx.Remove(nhNftHolderFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), ",")
+	nhNftHolderRowsWithPlaceHolder = strings.Join(stringx.Remove(nhNftHolderFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), "=?,") + "=?"
+)
+
+type (
+	nhNftHolderModel interface {
+		Insert(ctx context.Context, data *NhNftHolder) (sql.Result, error)
+		FindOne(ctx context.Context, id int) (*NhNftHolder, error)
+		FindOneByAddressTokenId(ctx context.Context, address string, tokenId string) (*NhNftHolder, error)
+		Update(ctx context.Context, data *NhNftHolder) error
+		Delete(ctx context.Context, id int) error
+	}
+
+	defaultNhNftHolderModel struct {
+		conn  sqlx.SqlConn
+		table string
+	}
+
+	NhNftHolder struct {
+		Id        int       `db:"id"`
+		Address   string    `db:"address"`    // 钱包地址
+		TokenId   string    `db:"token_id"`   // token id
+		Balance   int       `db:"balance"`    // 余额
+		UpdateSeq int       `db:"update_seq"` // 更新序列号
+		CreatedAt time.Time `db:"created_at"` // 创建时间
+		UpdatedAt time.Time `db:"updated_at"` // 修改时间
+	}
+)
+
+func newNhNftHolderModel(conn sqlx.SqlConn) *defaultNhNftHolderModel {
+	return &defaultNhNftHolderModel{
+		conn:  conn,
+		table: "`nh_nft_holder`",
+	}
+}
+
+func (m *defaultNhNftHolderModel) Delete(ctx context.Context, id int) error {
+	query := fmt.Sprintf("delete from %s where `id` = ?", m.table)
+	_, err := m.conn.ExecCtx(ctx, query, id)
+	return err
+}
+
+func (m *defaultNhNftHolderModel) FindOne(ctx context.Context, id int) (*NhNftHolder, error) {
+	query := fmt.Sprintf("select %s from %s where `id` = ? limit 1", nhNftHolderRows, m.table)
+	var resp NhNftHolder
+	err := m.conn.QueryRowCtx(ctx, &resp, query, id)
+	switch err {
+	case nil:
+		return &resp, nil
+	case sqlx.ErrNotFound:
+		return nil, ErrNotFound
+	default:
+		return nil, err
+	}
+}
+
+func (m *defaultNhNftHolderModel) FindOneByAddressTokenId(ctx context.Context, address string, tokenId string) (*NhNftHolder, error) {
+	var resp NhNftHolder
+	query := fmt.Sprintf("select %s from %s where `address` = ? and `token_id` = ? limit 1", nhNftHolderRows, m.table)
+	err := m.conn.QueryRowCtx(ctx, &resp, query, address, tokenId)
+	switch err {
+	case nil:
+		return &resp, nil
+	case sqlx.ErrNotFound:
+		return nil, ErrNotFound
+	default:
+		return nil, err
+	}
+}
+
+func (m *defaultNhNftHolderModel) Insert(ctx context.Context, data *NhNftHolder) (sql.Result, error) {
+	query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?)", m.table, nhNftHolderRowsExpectAutoSet)
+	ret, err := m.conn.ExecCtx(ctx, query, data.Address, data.TokenId, data.Balance, data.UpdateSeq)
+	return ret, err
+}
+
+func (m *defaultNhNftHolderModel) Update(ctx context.Context, newData *NhNftHolder) error {
+	query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, nhNftHolderRowsWithPlaceHolder)
+	_, err := m.conn.ExecCtx(ctx, query, newData.Address, newData.TokenId, newData.Balance, newData.UpdateSeq, newData.Id)
+	return err
+}
+
+func (m *defaultNhNftHolderModel) tableName() string {
+	return m.table
+}

internal/model/nh_promote_bind_model.go

@@ -53,16 +53,36 @@ func (m *customNhPromoteBindModel) FindRequirePushUser(ctx context.Context, shar
 }
 
 func (m *customNhPromoteBindModel) UpdatePushUser(ctx context.Context, id uint) error {
-	update := fmt.Sprintf("update %s set `is_push_user` = 1 where `id` = ?", m.table)
+	update := fmt.Sprintf("update %s set `is_push_user` = 1 where `id` = ? and `is_push_user` = 0", m.table)
-	_, err := m.conn.ExecCtx(ctx, update, id)
+	result, err := m.conn.ExecCtx(ctx, update, id)
+	if err != nil {
 		return err
 	}
+	row, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if row == 0 {
+		return ErrNoRowUpdate
+	}
+	return nil
+}
 
 func (m *customNhPromoteBindModel) UpdatePushRole(ctx context.Context, id uint) error {
-	update := fmt.Sprintf("update %s set `is_push_role` = 1 where `id` = ?", m.table)
+	update := fmt.Sprintf("update %s set `is_push_role` = 1 where `id` = ? and `is_push_role` = 0", m.table)
-	_, err := m.conn.ExecCtx(ctx, update, id)
+	result, err := m.conn.ExecCtx(ctx, update, id)
+	if err != nil {
 		return err
 	}
+	row, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if row == 0 {
+		return ErrNoRowUpdate
+	}
+	return nil
+}
 
 func (m *customNhPromoteBindModel) UserInviteCount(ctx context.Context, uid uint) (int64, error) {
 	key := cast.ToString(uid)

internal/model/nh_wallet_model.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/zeromicro/go-zero/core/stores/sqlx"
+	"nova_task/internal/pkg/aptos"
 )
 
 var _ NhWalletModel = (*customNhWalletModel)(nil)
@@ -14,7 +15,7 @@ type (
 	NhWalletModel interface {
 		nhWalletModel
 		withSession(session sqlx.Session) NhWalletModel
-		FindWalletByAddress(ctx context.Context, address ...string) (uint, error)
+		FindWalletByAddress(ctx context.Context, address string) (uint, error)
 	}
 
 	customNhWalletModel struct {
@@ -22,10 +23,10 @@ type (
 	}
 )
 
-func (m *customNhWalletModel) FindWalletByAddress(ctx context.Context, address ...string) (uint, error) {
+func (m *customNhWalletModel) FindWalletByAddress(ctx context.Context, address string) (uint, error) {
-	query := fmt.Sprintf("select `uid` from %s where `address` in ? limit 1", m.table)
+	query := fmt.Sprintf("select `uid` from %s where `address` = ? or `address` = ? limit 1", m.table)
 	var uid uint
-	err := m.conn.QueryRowCtx(ctx, &uid, query, address)
+	err := m.conn.QueryRowCtx(ctx, &uid, query, address, aptos.StrPadAptosAddress(address))
 	return uid, err
 }
 

internal/model/vars.go

@@ -1,5 +1,9 @@
 package model
 
-import "github.com/zeromicro/go-zero/core/stores/sqlx"
+import (
+	"errors"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+)
 
 var ErrNotFound = sqlx.ErrNotFound
+var ErrNoRowUpdate = errors.New("no row was updated")

internal/pkg/aptos/aptos.go

@@ -35,7 +35,6 @@ func GetTransactionOwnerAddress(contract, txHash, aptosNet string) (string, erro
 	// Validate contract address
 	function := payload["function"].(string)
 	parts := strings.Split(function, "::")
-	fmt.Println(parts[0] + "::" + parts[1])
 	if len(parts) < 2 || parts[0]+"::"+parts[1] != contract {
 		return "", errors.New("invalid contract address")
 	}

internal/svc/service_context.go

@@ -22,6 +22,8 @@ type ServiceContext struct {
 	UserModel               model.NhUserModel
 	WalletModel             model.NhWalletModel
 	ConfigModel             model.NhSystemConfigModel
+	NftHolderModel          model.NhNftHolderModel
+	NftHolderChangeLogModel model.NhNftHolderChangeLogModel
 
 	Earn   *ea.Client
 	DBConn sqlx.SqlConn
@@ -43,6 +45,8 @@ func NewServiceContext(c config.Config) *ServiceContext {
 		TouristBindModel:        model.NewNhTouristBindModel(dbConn),
 		WalletModel:             model.NewNhWalletModel(dbConn),
 		ConfigModel:             model.NewNhSystemConfigModel(dbConn, c.Cache),
+		NftHolderModel:          model.NewNhNftHolderModel(dbConn),
+		NftHolderChangeLogModel: model.NewNhNftHolderChangeLogModel(dbConn),
 
 		Earn:   c.Earn.BuildEarnClient(),
 		DBConn: dbConn,