kra-new/internal/data/business/channel_write.go

package business

import (
	"context"
	"errors"
	"strconv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"kra/internal/biz/business"
)

type platformChannelPO struct {
	ID                         uint64 `gorm:"primaryKey"`
	Account, Password          string
	Status                     int
	BusierID                   uint64
	CreateTime                 int64
	RegisterType, PromoteLevel int
	ParentID, TopPromoteID     uint64
	GameIDs2                   string `gorm:"column:game_ids2"`
}

func (platformChannelPO) TableName() string { return "tab_promote" }

type channelSettlementPO struct {
	ID                                    uint64 `gorm:"primaryKey"`
	PromoteID                             uint64
	PromoteAccount                        string
	TimeType, DayPeriod                   int
	CreateTime, UpdateTime, NextCountTime int64
}

func (channelSettlementPO) TableName() string { return "tab_promote_settlement_time" }

func newPlatformChannel(input *business.ChannelWrite) platformChannelPO {
	return platformChannelPO{Account: input.Account, Password: input.Password, Status: input.Status, BusierID: input.BusierID, CreateTime: input.CreatedUnix, RegisterType: input.RegisterType, PromoteLevel: input.Level, ParentID: input.ParentID, TopPromoteID: input.TopPromoteID, GameIDs2: input.GameIDs2}
}
func newChannelSettlement(input *business.ChannelWrite, promoteID uint64) channelSettlementPO {
	return channelSettlementPO{PromoteID: promoteID, PromoteAccount: input.Account, TimeType: input.SettlementType, DayPeriod: input.SettlementDays, CreateTime: input.CreatedUnix, UpdateTime: input.CreatedUnix, NextCountTime: input.NextCountUnix}
}

type channelWriteRepo struct{ data PlatformProvider }

func NewChannelWriteRepo(data PlatformProvider) business.ChannelWriteRepo {
	return &channelWriteRepo{data: data}
}

func (r *channelWriteRepo) Create(ctx context.Context, input *business.ChannelWrite) (uint64, error) {
	db, err := r.data.Database(input.SourceDB)
	if err != nil || db == nil {
		return 0, business.ErrPlatformUnavailable
	}
	var id uint64
	err = db.WithContext(ctx).Connection(func(conn *gorm.DB) error {
		if conn.Dialector.Name() == "mysql" {
			// 命名锁覆盖账号尚不存在时的竞争；固定借用连接，提交事务后才释放锁。
			var lockName string
			if err := conn.Raw("SELECT CONCAT('sw-channel:', MD5(CONCAT(DATABASE(), ':', LOWER(?))))", input.Account).Scan(&lockName).Error; err != nil {
				return err
			}
			var locked int
			if err := conn.Raw("SELECT GET_LOCK(?, 10)", lockName).Scan(&locked).Error; err != nil {
				return err
			}
			if locked != 1 {
				return business.ErrChannelBusy
			}
			defer func() {
				cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
				defer cancel()
				conn.WithContext(cleanup).Exec("DO RELEASE_LOCK(?)", lockName)
			}()
		}
		// GET_LOCK 的标量 Scan 会修改固定连接上的 Statement.Model。
		// 清空查询状态后再开事务，仍复用持锁连接，避免后续结构体 Take 读取到标量模型。
		return conn.Session(&gorm.Session{NewDB: true}).Transaction(func(tx *gorm.DB) error {
			var count int64
			if err := tx.Model(&platformChannelPO{}).Where("account = ?", input.Account).Count(&count).Error; err != nil {
				return err
			}
			if count > 0 {
				return business.ErrChannelExists
			}
			var owner struct {
				ID         uint64
				PromoteIDs *string
			}
			if err := tx.Table("tab_promote_business").Clauses(clause.Locking{Strength: "UPDATE"}).Select("id,promote_ids").Where("id = ?", input.BusierID).Take(&owner).Error; err != nil {
				if errors.Is(err, gorm.ErrRecordNotFound) {
					return business.ErrChannelPlatformOwner
				}
				return err
			}
			po := newPlatformChannel(input)
			if err := tx.Create(&po).Error; err != nil {
				var duplicate *mysql.MySQLError
				if errors.Is(err, gorm.ErrDuplicatedKey) || (errors.As(err, &duplicate) && duplicate.Number == 1062) {
					return business.ErrChannelExists
				}
				return err
			}
			settlement := newChannelSettlement(input, po.ID)
			if err := tx.Create(&settlement).Error; err != nil {
				return err
			}
			ids := strconv.FormatUint(po.ID, 10)
			if owner.PromoteIDs != nil && *owner.PromoteIDs != "" {
				ids = *owner.PromoteIDs + "," + ids
			}
			result := tx.Table("tab_promote_business").Where("id = ?", input.BusierID).Update("promote_ids", ids)
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected != 1 {
				return business.ErrChannelPlatformOwner
			}
			id = po.ID
			return nil
		})
	})
	if err != nil {
		if strings.Contains(strings.ToLower(err.Error()), "unique constraint failed: tab_promote.account") {
			return 0, business.ErrChannelExists
		}
		return 0, storageError(err)
	}
	return id, nil
}