feat: 迁移游戏商务绑定审核流程和列表

internal/data/business/account.go

@@ -153,7 +153,7 @@ func checkAccountUnique(tx *gorm.DB, a *business.Account) error {
 }
 
 func (r *accountRepo) Create(ctx context.Context, a *business.Account) error {
-	return r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+	err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		if err := checkAccountUnique(tx, a); err != nil {
 			return err
 		}
@@ -171,10 +171,11 @@ func (r *accountRepo) Create(ctx context.Context, a *business.Account) error {
 		a.ID, a.CreatedAt, a.UpdatedAt = po.ID, po.CreatedAt, po.UpdatedAt
 		return nil
 	})
+	return storageError(err)
 }
 
 func (r *accountRepo) Update(ctx context.Context, a *business.Account) error {
-	return r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+	err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		if err := checkAccountUnique(tx, a); err != nil {
 			return err
 		}
@@ -194,6 +195,7 @@ func (r *accountRepo) Update(ctx context.Context, a *business.Account) error {
 		}
 		return nil
 	})
+	return storageError(err)
 }
 
 func (r *accountRepo) AuthorityForLevel(ctx context.Context, level int) (uint, error) {
@@ -250,17 +252,10 @@ func (r *accountRepo) DirectCounts(ctx context.Context, ids []uint) (map[uint]bu
 		result[b.UserID] = business.AccountCounts{PromoteCount: &zero, BindCount: b.Total}
 	}
 	// 系统数据演练库尚未包含渠道表时，基础资料与已通过绑定数仍可使用。
-	tables, err := db.Migrator().GetTables()
+	available, err := businessTableAvailable(db, "tab_promote")
 	if err != nil {
 		return nil, business.ErrStorage
 	}
-	available := false
-	for _, table := range tables {
-		if table == "tab_promote" {
-			available = true
-			break
-		}
-	}
 	if !available {
 		for _, b := range binds {
 			value := result[b.UserID]

internal/data/business/binding_repo.go

@@ -0,0 +1,161 @@
+package business
+
+import (
+	"context"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"kra/internal/biz/business"
+)
+
+type bindingLockPO struct {
+	SourceDB         string `gorm:"column:source_db;primaryKey;size:50"`
+	PlatformBusierID uint64 `gorm:"primaryKey;autoIncrement:false"`
+}
+
+func (bindingLockPO) TableName() string { return "sys_business_binding_locks" }
+
+type bindingRepo struct{ data Provider }
+
+func NewBindingRepo(data Provider) business.BindingRepo { return &bindingRepo{data: data} }
+
+type bindingTxKey struct{}
+
+func (r *bindingRepo) db(ctx context.Context) *gorm.DB {
+	if tx, ok := ctx.Value(bindingTxKey{}).(*gorm.DB); ok {
+		return tx.WithContext(ctx)
+	}
+	return r.data.DB().WithContext(ctx)
+}
+func (r *bindingRepo) WithPlatformLock(ctx context.Context, key business.PlatformKey, fn func(context.Context) error) error {
+	err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		// 单独的键锁覆盖“尚无绑定记录”的并发场景，不修改平台数据库。
+		lock := bindingLockPO{SourceDB: key.SourceDB, PlatformBusierID: key.BusierID}
+		if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&lock).Error; err != nil {
+			return business.ErrStorage
+		}
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("source_db = ? AND platform_busier_id = ?", key.SourceDB, key.BusierID).First(&lock).Error; err != nil {
+			return business.ErrStorage
+		}
+		return fn(context.WithValue(ctx, bindingTxKey{}, tx))
+	})
+	return storageError(err)
+}
+
+func toBizBinding(po bindingPO) *business.Binding {
+	return &business.Binding{UserID: po.UserID, PlatformKey: business.PlatformKey{SourceDB: po.SourceDB, BusierID: po.PlatformBusierID}, Account: po.PlatformBusierAccount,
+		Status: po.Status, RejectReason: po.RejectReason, CreatedAt: po.CreateTime, ReviewTime: po.ReviewTime, ReviewAccount: po.ReviewAccount}
+}
+func newBinding(b *business.Binding) bindingPO {
+	return bindingPO{UserID: b.UserID, SourceDB: b.SourceDB, PlatformBusierID: b.BusierID, PlatformBusierAccount: b.Account, Status: b.Status,
+		RejectReason: b.RejectReason, CreateTime: b.CreatedAt, ReviewTime: b.ReviewTime, ReviewAccount: b.ReviewAccount}
+}
+func (r *bindingRepo) ForPlatform(ctx context.Context, key business.PlatformKey) ([]*business.Binding, error) {
+	var pos []bindingPO
+	if err := r.db(ctx).Where("source_db = ? AND platform_busier_id = ?", key.SourceDB, key.BusierID).Find(&pos).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	rows := make([]*business.Binding, 0, len(pos))
+	for _, po := range pos {
+		rows = append(rows, toBizBinding(po))
+	}
+	return rows, nil
+}
+func (r *bindingRepo) Save(ctx context.Context, b *business.Binding) error {
+	po := newBinding(b)
+	err := r.db(ctx).Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "user_id"}, {Name: "source_db"}, {Name: "platform_busier_id"}},
+		DoUpdates: clause.AssignmentColumns([]string{"platform_busier_account", "status", "reject_reason", "create_time", "review_time", "review_account"})}).Create(&po).Error
+	if err != nil {
+		return business.ErrStorage
+	}
+	return nil
+}
+func (r *bindingRepo) Delete(ctx context.Context, userID uint, key business.PlatformKey) error {
+	result := r.db(ctx).Where("user_id = ? AND source_db = ? AND platform_busier_id = ?", userID, key.SourceDB, key.BusierID).Delete(&bindingPO{})
+	if result.Error != nil {
+		return business.ErrStorage
+	}
+	if result.RowsAffected == 0 {
+		return business.ErrBindingNotFound
+	}
+	return nil
+}
+
+func (r *bindingRepo) List(ctx context.Context, scope business.UserScope, q business.BindingFilter, options ...business.ListOption) ([]*business.Binding, int64, error) {
+	db := r.db(ctx).Table("sys_user_bind_platform b").Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL")
+	db = applyUserScope(db, scope, "u.")
+	if q.UserID != 0 {
+		db = db.Where("b.user_id = ?", q.UserID)
+	}
+	if len(q.SourceDBs) > 0 {
+		db = db.Where("b.source_db IN ?", q.SourceDBs)
+	}
+	if q.Account != "" {
+		db = db.Where("b.platform_busier_account LIKE ?", "%"+q.Account+"%")
+	}
+	if q.Status != "" {
+		db = db.Where("b.status = ?", q.Status)
+	}
+	var total int64
+	if err := db.Count(&total).Error; err != nil {
+		return nil, 0, business.ErrStorage
+	}
+	o := business.NewListOptions(options...)
+	var pos []struct {
+		Binding  bindingPO `gorm:"embedded"`
+		UserName string
+	}
+	if err := db.Select("b.*,u.username AS user_name").Order("b.create_time DESC").Order("b.source_db,b.platform_busier_id,b.user_id").Offset(o.Offset).Limit(o.Limit).Scan(&pos).Error; err != nil {
+		return nil, 0, business.ErrStorage
+	}
+	rows := make([]*business.Binding, 0, len(pos))
+	keys := make(map[business.PlatformKey]bool)
+	for _, po := range pos {
+		b := toBizBinding(po.Binding)
+		zero := int64(0)
+		b.PromoteCount = &zero
+		b.UserName = po.UserName
+		rows = append(rows, b)
+		if b.Status == "0" {
+			keys[b.PlatformKey] = true
+		}
+	}
+	if len(keys) == 0 {
+		return rows, total, nil
+	}
+	available, err := businessTableAvailable(r.db(ctx), "tab_promote")
+	if err != nil {
+		return nil, 0, err
+	}
+	if !available {
+		for _, b := range rows {
+			if b.Status == "0" {
+				b.PromoteCount = nil
+			}
+		}
+		return rows, total, nil
+	}
+	query := r.db(ctx).Table("tab_promote").Where("promote_level = ?", 1)
+	pairs := r.db(ctx).Session(&gorm.Session{NewDB: true})
+	for key := range keys {
+		pairs = pairs.Or("(source_db = ? AND busier_id = ?)", key.SourceDB, key.BusierID)
+	}
+	var counts []struct {
+		SourceDB string `gorm:"column:source_db"`
+		BusierID uint64
+		Total    int64
+	}
+	if err := query.Where(pairs).Select("source_db,busier_id,COUNT(*) AS total").Group("source_db,busier_id").Scan(&counts).Error; err != nil {
+		return nil, 0, business.ErrStorage
+	}
+	byKey := make(map[business.PlatformKey]int64, len(counts))
+	for _, c := range counts {
+		byKey[business.PlatformKey{SourceDB: c.SourceDB, BusierID: c.BusierID}] = c.Total
+	}
+	for _, b := range rows {
+		if b.Status == "0" {
+			count := byKey[b.PlatformKey]
+			b.PromoteCount = &count
+		}
+	}
+	return rows, total, nil
+}

internal/data/business/binding_test.go

@@ -0,0 +1,76 @@
+package business
+
+import (
+	"context"
+	"kra/internal/biz/business"
+	"testing"
+	"time"
+)
+
+func TestBindingRepositoryClearsAuditAndUsesCompositeIdentity(t *testing.T) {
+	db := prepareAccounts(t)
+	r := NewBindingRepo(testData{db})
+	ctx := context.Background()
+	reason, review := "旧驳回", "审核员"
+	now := time.Now()
+	key := business.PlatformKey{SourceDB: "lx", BusierID: 7}
+	if err := r.Save(ctx, &business.Binding{UserID: 3, PlatformKey: key, Status: "2", RejectReason: &reason, ReviewTime: &now, ReviewAccount: &review}); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Save(ctx, &business.Binding{UserID: 3, PlatformKey: business.PlatformKey{SourceDB: "hw", BusierID: 7}, Status: "0"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.WithPlatformLock(ctx, key, func(txctx context.Context) error {
+		return r.Save(txctx, &business.Binding{UserID: 3, PlatformKey: key, Status: "3", CreatedAt: now})
+	}); err != nil {
+		t.Fatal(err)
+	}
+	rows, err := r.ForPlatform(ctx, key)
+	if err != nil || len(rows) != 1 {
+		t.Fatalf("查询失败: %+v %v", rows, err)
+	}
+	if rows[0].Status != "3" || rows[0].RejectReason != nil || rows[0].ReviewTime != nil || rows[0].ReviewAccount != nil {
+		t.Fatal("重提未清除旧审核信息")
+	}
+	if err := r.Delete(ctx, 4, key); err != business.ErrBindingNotFound {
+		t.Fatal("不能删除其他用户绑定")
+	}
+	if err := r.Delete(ctx, 3, key); err != nil {
+		t.Fatal(err)
+	}
+	other, err := r.ForPlatform(ctx, business.PlatformKey{SourceDB: "hw", BusierID: 7})
+	if err != nil || len(other) != 1 {
+		t.Fatal("解绑影响了同 ID 的其他平台")
+	}
+}
+
+func TestBindingListLoadsDTOFieldsAndSurvivesMissingChannels(t *testing.T) {
+	db := prepareAccounts(t)
+	if err := db.Create(&accountPO{ID: 3, Username: "child", Level: 3, IDPath: "0,1,2"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	r := NewBindingRepo(testData{db})
+	ctx := context.Background()
+	for _, source := range []string{"lx", "hw"} {
+		if err := r.Save(ctx, &business.Binding{UserID: 3, PlatformKey: business.PlatformKey{SourceDB: source, BusierID: 7}, Account: "sdk", Status: "0", CreatedAt: time.Now()}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := db.Exec("INSERT INTO tab_promote VALUES(1,'lx',7,1,1),(2,'lx',7,2,1)").Error; err != nil {
+		t.Fatal(err)
+	}
+	rows, total, err := r.List(ctx, business.UserScope{OnlySelf: true, SelfID: 3}, business.BindingFilter{SourceDBs: []string{"lx"}})
+	if err != nil || total != 1 || len(rows) != 1 {
+		t.Fatalf("绑定列表错误: %+v %d %v", rows, total, err)
+	}
+	if rows[0].UserID != 3 || rows[0].UserName != "child" || rows[0].Account != "sdk" || rows[0].PromoteCount == nil || *rows[0].PromoteCount != 1 {
+		t.Fatalf("绑定字段或渠道统计错误: %+v", rows[0])
+	}
+	if err := db.Exec("DROP TABLE tab_promote").Error; err != nil {
+		t.Fatal(err)
+	}
+	rows, total, err = r.List(ctx, business.UserScope{OnlySelf: true, SelfID: 3}, business.BindingFilter{})
+	if err != nil || total != 2 || len(rows) != 2 || rows[0].PromoteCount != nil {
+		t.Fatalf("缺渠道表不应阻断绑定资料: %+v %d %v", rows, total, err)
+	}
+}

internal/data/business/migrations.go

@@ -6,7 +6,9 @@ import (
 )
 
 func Migrations() []migration.Step {
-	return []migration.Step{{ID: "202611030003_business_bindings", Migrate: func(db *gorm.DB) error {
+	return []migration.Step{{ID: "202611030004_business_binding_locks", Migrate: func(db *gorm.DB) error {
+		return migration.CreateMissingTables(db, &bindingLockPO{})
+	}}, {ID: "202611030003_business_bindings", Migrate: func(db *gorm.DB) error {
 		return migration.CreateMissingTables(db, &bindingPO{})
 	}}, {ID: "202611030001_business_preference", Migrate: func(db *gorm.DB) error {
 		// 旧表沿用；已迁移数据库不重建或重置用户偏好。

internal/data/business/platform.go

@@ -0,0 +1,45 @@
+package business
+
+import (
+	"context"
+	"errors"
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+type PlatformProvider interface {
+	Database(string) (*gorm.DB, error)
+}
+type platformAccountRepo struct{ data PlatformProvider }
+
+func NewPlatformAccountRepo(data PlatformProvider) business.PlatformAccountRepo {
+	return &platformAccountRepo{data: data}
+}
+
+type platformAccountPO struct {
+	ID                uint64
+	Account, Password string
+	Status            int
+}
+
+func (platformAccountPO) TableName() string { return "tab_promote_business" }
+
+func (r *platformAccountRepo) FindByAccount(ctx context.Context, source, account string) (*business.PlatformAccount, error) {
+	source, err := business.NormalizePlatform(source)
+	if err != nil {
+		return nil, err
+	}
+	db, err := r.data.Database(source)
+	if err != nil || db == nil {
+		return nil, business.ErrPlatformUnavailable
+	}
+	var po platformAccountPO
+	err = db.WithContext(ctx).Where("account = ?", account).First(&po).Error
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, business.ErrPlatformUnavailable
+	}
+	return &business.PlatformAccount{ID: po.ID, Account: po.Account, Password: po.Password, Status: po.Status}, nil
+}

internal/data/business/schema.go

@@ -0,0 +1,32 @@
+package business
+
+import (
+	"errors"
+	kratoserrors "github.com/go-kratos/kratos/v3/errors"
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+func storageError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var typed *kratoserrors.Error
+	if errors.As(err, &typed) {
+		return err
+	}
+	return business.ErrStorage
+}
+
+func businessTableAvailable(db *gorm.DB, name string) (bool, error) {
+	tables, err := db.Migrator().GetTables()
+	if err != nil {
+		return false, business.ErrStorage
+	}
+	for _, table := range tables {
+		if table == name {
+			return true, nil
+		}
+	}
+	return false, nil
+}

internal/data/data.go

@@ -27,8 +27,11 @@ import (
 var ProviderSet = wire.NewSet(
 	NewData,
 	wire.Bind(new(databusiness.Provider), new(*Data)),
+	wire.Bind(new(databusiness.PlatformProvider), new(*Data)),
 	databusiness.NewUserPreferenceRepo,
 	databusiness.NewAccountRepo,
+	databusiness.NewBindingRepo,
+	databusiness.NewPlatformAccountRepo,
 	NewIntegrationRuntime,
 	wire.Bind(new(datasystem.Provider), new(*Data)),
 	wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),

internal/data/migrations_test.go

@@ -34,8 +34,8 @@ func TestMigrateAllRunsModuleSchemasWithoutBootstrapSeed(t *testing.T) {
 	if err = db.Table(migration.TableName).Count(&versions).Error; err != nil {
 		t.Fatal(err)
 	}
-	if versions != 17 {
-		t.Fatalf("migration versions = %d, want 17", versions)
+	if versions != 18 {
+		t.Fatalf("migration versions = %d, want 18", versions)
 	}
 	var communicationRows []dataintegration.ConfigPO
 	if err = db.Where("kind IN ?", []string{"mq", "websocket"}).Order("kind, provider").Find(&communicationRows).Error; err != nil {