feat: 迁移渠道列表新增和游戏累计分布

2026-11-03 21:32:51 +08:00
parent 5e58c6bcf3
commit 075d94a51e
39 changed files with 2216 additions and 12 deletions
@@ -0,0 +1,199 @@
+package business
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+type channelPO struct {
+	ID                                             uint64
+	SourceDB                                       string `gorm:"column:source_db"`
+	Account, Nickname, ParentName, RealName        string
+	PromoteLevel, RegisterType, Status, LockStatus *int
+	ParentID, TopPromoteID, BusierID               uint64
+	CreateTime                                     *int64
+}
+
+func (channelPO) TableName() string { return "tab_promote" }
+func toBizChannel(p channelPO) *business.Channel {
+	return &business.Channel{ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.ID}, Account: p.Account, Nickname: p.Nickname, ParentName: p.ParentName, RealName: p.RealName, Level: p.PromoteLevel, RegisterType: p.RegisterType, Status: p.Status, LockStatus: p.LockStatus, ParentID: p.ParentID, TopPromoteID: p.TopPromoteID, BusierID: p.BusierID}
+}
+
+type channelRepo struct{ data Provider }
+
+func NewChannelRepo(data Provider) business.ChannelRepo { return &channelRepo{data: data} }
+
+func channelBindingQuery(db *gorm.DB, scope business.UserScope, filter business.ChannelFilter) *gorm.DB {
+	q := db.Table("sys_user_bind_platform b").Joins("LEFT JOIN sys_users u ON u.id=b.user_id").Where("b.status = ?", "0")
+	if !scope.All {
+		q = applyUserScope(q.Where("u.deleted_at IS NULL"), scope, "u.")
+	}
+	if filter.BusierUserID != nil {
+		q = q.Where("b.user_id = ?", *filter.BusierUserID)
+	}
+	if len(filter.Sources) > 0 {
+		q = q.Where("b.source_db IN ?", filter.Sources)
+	}
+	if keyword := strings.ToLower(strings.TrimSpace(filter.BusierAccount)); keyword != "" {
+		q = q.Where("(INSTR(LOWER(b.platform_busier_account),?) > 0 OR INSTR(LOWER(u.username),?) > 0)", keyword, keyword)
+	}
+	return q
+}
+func (r *channelRepo) Bindings(ctx context.Context, scope business.UserScope, source string) ([]*business.Binding, error) {
+	var pos []bindingPO
+	if err := channelBindingQuery(r.data.DB().WithContext(ctx), scope, business.ChannelFilter{Sources: []string{source}}).Select("b.*").Order("b.user_id,b.source_db,b.platform_busier_id").Scan(&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 *channelRepo) List(ctx context.Context, scope business.UserScope, filter business.ChannelFilter, options ...business.ListOption) (*business.ChannelPage, error) {
+	db := r.data.DB().WithContext(ctx)
+	page := &business.ChannelPage{Rows: []*business.Channel{}, Available: true}
+	bindings := channelBindingQuery(db, scope, filter)
+	constrain := !scope.All || filter.BusierUserID != nil || filter.BusierAccount != ""
+	if constrain {
+		var count int64
+		if err := bindings.Count(&count).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		if count == 0 {
+			return page, nil
+		}
+	}
+	available, err := businessTableAvailable(db, "tab_promote")
+	if err != nil {
+		return nil, err
+	}
+	if !available {
+		page.Available = false
+		return page, nil
+	}
+	q := db.Table("tab_promote p")
+	if constrain {
+		q = q.Where("EXISTS (?)", bindings.Select("1").Where("b.source_db=p.source_db AND b.platform_busier_id=p.busier_id"))
+	}
+	if len(filter.Sources) > 0 {
+		q = q.Where("p.source_db IN ?", filter.Sources)
+	}
+	if filter.Keyword != "" {
+		q = q.Where("p.account LIKE ?", "%"+filter.Keyword+"%")
+	}
+	for _, field := range []struct {
+		column string
+		value  *int
+	}{{"promote_level", filter.Level}, {"register_type", filter.RegisterType}, {"status", filter.Status}, {"lock_status", filter.LockStatus}} {
+		if field.value != nil {
+			q = q.Where("p."+field.column+" = ?", *field.value)
+		}
+	}
+	if err := q.Count(&page.Total).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	var pos []channelPO
+	o := business.NewListOptions(options...)
+	if err := q.Select("p.id,p.source_db,p.account,p.nickname,p.parent_name,p.real_name,p.promote_level,p.register_type,p.status,p.lock_status,p.parent_id,p.top_promote_id,p.busier_id").Order("p.create_time DESC,p.id DESC,p.source_db ASC").Offset(o.Offset).Limit(o.Limit).Scan(&pos).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	if len(pos) == 0 {
+		return page, nil
+	}
+	pairs := make([][]any, 0, len(pos))
+	for _, po := range pos {
+		pairs = append(pairs, []any{po.SourceDB, po.BusierID})
+	}
+	var bound []struct {
+		SourceDB                        string `gorm:"column:source_db"`
+		PlatformBusierID                uint64
+		UserID                          uint
+		PlatformBusierAccount, UserName string
+	}
+	if err := channelBindingQuery(db, scope, filter).Select("b.source_db,b.platform_busier_id,b.user_id,b.platform_busier_account,u.username AS user_name").Where("(b.source_db,b.platform_busier_id) IN ?", pairs).Order("b.user_id").Scan(&bound).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	byKey := map[business.PlatformKey]int{}
+	for i, b := range bound {
+		key := business.PlatformKey{SourceDB: b.SourceDB, BusierID: b.PlatformBusierID}
+		if _, ok := byKey[key]; !ok {
+			byKey[key] = i
+		}
+	}
+	for _, po := range pos {
+		c := toBizChannel(po)
+		if i, ok := byKey[business.PlatformKey{SourceDB: po.SourceDB, BusierID: po.BusierID}]; ok {
+			b := bound[i]
+			c.BusierUserID = &b.UserID
+			c.BindingAccount = b.PlatformBusierAccount
+			c.BindingUserName = b.UserName
+		}
+		page.Rows = append(page.Rows, c)
+	}
+	return page, nil
+}
+func (r *channelRepo) Find(ctx context.Context, scope business.UserScope, key business.ChannelKey) (*business.Channel, error) {
+	db := r.data.DB().WithContext(ctx)
+	available, err := businessTableAvailable(db, "tab_promote")
+	if err != nil {
+		return nil, err
+	}
+	if !available {
+		return nil, business.ErrChannelUnavailable
+	}
+	var po channelPO
+	if err := db.Where("source_db = ? AND id = ?", key.SourceDB, key.ID).Take(&po).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, business.ErrChannelNotFound
+		}
+		return nil, business.ErrStorage
+	}
+	if !scope.All {
+		var count int64
+		if err := channelBindingQuery(db, scope, business.ChannelFilter{Sources: []string{key.SourceDB}}).Where("b.platform_busier_id = ?", po.BusierID).Count(&count).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		if count == 0 {
+			return nil, business.ErrChannelScope
+		}
+	}
+	return toBizChannel(po), nil
+}
+func channelPairs(keys []business.ChannelKey) [][]any {
+	pairs := make([][]any, 0, len(keys))
+	for _, key := range keys {
+		pairs = append(pairs, []any{key.SourceDB, key.ID})
+	}
+	return pairs
+}
+func (r *channelRepo) StatChannels(ctx context.Context, roots []*business.Channel) ([]*business.Channel, error) {
+	keys := make([]business.ChannelKey, 0, len(roots))
+	for _, root := range roots {
+		if root.Level != nil && *root.Level < 3 {
+			keys = append(keys, root.ChannelKey)
+		}
+	}
+	rows := []*business.Channel{}
+	seen := map[business.ChannelKey]bool{}
+	for start := 0; start < len(keys); start += 200 {
+		pairs := channelPairs(keys[start:min(start+200, len(keys))])
+		var pos []channelPO
+		err := r.data.DB().WithContext(ctx).Table("tab_promote").Select("id,source_db,promote_level,parent_id,top_promote_id").Where("(source_db,parent_id) IN ? OR (source_db,top_promote_id) IN ?", pairs, pairs).Scan(&pos).Error
+		if err != nil {
+			return nil, business.ErrStorage
+		}
+		for _, po := range pos {
+			c := toBizChannel(po)
+			if !seen[c.ChannelKey] {
+				rows = append(rows, c)
+				seen[c.ChannelKey] = true
+			}
+		}
+	}
+	return rows, nil
+}
@@ -0,0 +1,179 @@
+package business
+
+import (
+	"context"
+
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+func (r *channelRepo) Metrics(ctx context.Context, roots, statKeys []business.ChannelKey, dedup bool) (*business.ChannelMetricData, error) {
+	db := r.data.DB().WithContext(ctx)
+	metrics := &business.ChannelMetricData{Games: map[business.ChannelKey]int64{}, Players: map[business.ChannelKey]int64{}, Pay: map[business.ChannelKey]string{}}
+	var err error
+	metrics.GamesAvailable, err = businessTableAvailable(db, "tab_promote_apply")
+	if err != nil {
+		return nil, err
+	}
+	metrics.PlayersAvailable, err = businessTableAvailable(db, "tab_user")
+	if err != nil {
+		return nil, err
+	}
+	metrics.PayAvailable, err = businessTableAvailable(db, "tab_spend")
+	if err != nil {
+		return nil, err
+	}
+	type countRow struct {
+		SourceDB  string `gorm:"column:source_db"`
+		PromoteID uint64
+		Count     int64
+	}
+	if metrics.GamesAvailable {
+		for start := 0; start < len(roots); start += 300 {
+			var rows []countRow
+			if err := db.Table("tab_promote_apply").Select("source_db,promote_id,COUNT(DISTINCT game_id) AS count").Where("status = 1 AND (source_db,promote_id) IN ?", channelPairs(roots[start:min(start+300, len(roots))])).Group("source_db,promote_id").Scan(&rows).Error; err != nil {
+				return nil, business.ErrStorage
+			}
+			for _, row := range rows {
+				metrics.Games[business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}] = row.Count
+			}
+		}
+	}
+	// 每批按平台/渠道一起查询，避免逐个渠道访问数据库，也避免参数数量无限增长。
+	for start := 0; start < len(statKeys); start += 300 {
+		pairs := channelPairs(statKeys[start:min(start+300, len(statKeys))])
+		if metrics.PlayersAvailable {
+			q := db.Table("tab_user").Where("(puid = 0 OR puid IS NULL) AND (source_db,promote_id) IN ?", pairs)
+			if dedup {
+				var players []struct {
+					SourceDB          string `gorm:"column:source_db"`
+					PromoteID, UserID uint64
+					IDCard            string `gorm:"column:idcard"`
+				}
+				if err := q.Select("source_db,promote_id,id AS user_id,idcard").Scan(&players).Error; err != nil {
+					return nil, business.ErrStorage
+				}
+				for _, p := range players {
+					metrics.PlayerCards = append(metrics.PlayerCards, business.ChannelPlayer{ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}, UserID: p.UserID, IDCard: p.IDCard})
+				}
+			} else {
+				var rows []countRow
+				if err := q.Select("source_db,promote_id,COUNT(*) AS count").Group("source_db,promote_id").Scan(&rows).Error; err != nil {
+					return nil, business.ErrStorage
+				}
+				for _, row := range rows {
+					metrics.Players[business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}] = row.Count
+				}
+			}
+		}
+		if metrics.PayAvailable {
+			var rows []struct {
+				SourceDB  string `gorm:"column:source_db"`
+				PromoteID uint64
+				Amount    string
+			}
+			if err := db.Table("tab_spend").Select("source_db,promote_id,COALESCE(SUM(pay_amount),0) AS amount").Where("pay_status = 1 AND (source_db,promote_id) IN ?", pairs).Group("source_db,promote_id").Scan(&rows).Error; err != nil {
+				return nil, business.ErrStorage
+			}
+			for _, row := range rows {
+				metrics.Pay[business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}] = row.Amount
+			}
+		}
+	}
+	return metrics, nil
+}
+
+func channelGameTree(db *gorm.DB, scope business.ChannelGameScope) *gorm.DB {
+	root := scope.Root
+	q := db.Table("tab_promote p").Where("p.source_db = ?", root.SourceDB)
+	if !scope.IncludeSub || root.Level == nil || *root.Level >= 3 {
+		return q.Where("p.id = ?", root.ID)
+	}
+	if *root.Level == 2 {
+		return q.Where("p.id = ? OR (p.parent_id = ? AND p.promote_level = 3)", root.ID, root.ID)
+	}
+	return q.Where("p.id = ? OR p.parent_id = ? OR p.top_promote_id = ?", root.ID, root.ID, root.ID)
+}
+func (r *channelRepo) Games(ctx context.Context, scope business.ChannelGameScope) ([]*business.ChannelGame, error) {
+	db := r.data.DB().WithContext(ctx)
+	available, err := businessTableAvailable(db, "tab_promote_apply")
+	if err != nil {
+		return nil, err
+	}
+	if !available {
+		return nil, business.ErrChannelReportsUnavailable
+	}
+	ids := []uint64{}
+	if err := db.Table("tab_promote_apply").Where("source_db = ? AND promote_id = ? AND status = 1 AND game_id IS NOT NULL", scope.Root.SourceDB, scope.Root.ID).Distinct("game_id").Order("game_id").Pluck("game_id", &ids).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	result := []*business.ChannelGame{}
+	if len(ids) == 0 {
+		return result, nil
+	}
+	for _, table := range []string{"tab_game", "tab_sw_report_day", "tab_sw_report_day_key"} {
+		available, err := businessTableAvailable(db, table)
+		if err != nil {
+			return nil, err
+		}
+		if !available {
+			return nil, business.ErrChannelReportsUnavailable
+		}
+	}
+	mode := 0
+	if scope.IDCardDedup {
+		mode = 1
+	}
+	query := func(table string, dedup int) *gorm.DB {
+		return db.Table(table+" r").Where("r.source_db = ? AND r.game_id > 0 AND r.dedup_mode = ?", scope.Root.SourceDB, dedup).Where("EXISTS (?)", channelGameTree(db, scope).Select("1").Where("p.id=r.promote_id AND p.source_db=r.source_db"))
+	}
+	// 累计起点按当前人数口径的日汇总最早日期决定，与原画像查询一致。
+	var minimum struct{ FirstDate *string }
+	if err := query("tab_sw_report_day", mode).Select("CAST(MIN(r.stat_date) AS CHAR) AS first_date").Scan(&minimum).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	begin := scope.EndDate
+	if minimum.FirstDate != nil && *minimum.FirstDate != "" {
+		begin = *minimum.FirstDate
+	}
+	byID := map[uint64]*business.ChannelGame{}
+	for _, id := range ids {
+		item := &business.ChannelGame{ID: id, PayAmount: "0.00"}
+		byID[id] = item
+		result = append(result, item)
+	}
+	for start := 0; start < len(ids); start += 300 {
+		batch := ids[start:min(start+300, len(ids))]
+		var names []struct {
+			ID       uint64
+			GameName string
+		}
+		if err := db.Table("tab_game").Select("id,game_name").Where("source_db = ? AND id IN ?", scope.Root.SourceDB, batch).Scan(&names).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		for _, n := range names {
+			byID[n.ID].Name = n.GameName
+		}
+		var counts []struct {
+			GameID uint64
+			Count  int64
+		}
+		if err := query("tab_sw_report_day_key", mode).Select("r.game_id,COUNT(DISTINCT r.identity_key) AS count").Where("r.metric = 1 AND r.stat_date >= ? AND r.stat_date <= ? AND r.game_id IN ?", begin, scope.EndDate, batch).Group("r.game_id").Scan(&counts).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		for _, c := range counts {
+			byID[c.GameID].RegisterCount = c.Count
+		}
+		var amounts []struct {
+			GameID uint64
+			Amount string
+		}
+		if err := query("tab_sw_report_day", 0).Select("r.game_id,COALESCE(SUM(r.total_pay),0) AS amount").Where("r.stat_date >= ? AND r.stat_date <= ? AND r.game_id IN ?", begin, scope.EndDate, batch).Group("r.game_id").Scan(&amounts).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		for _, a := range amounts {
+			byID[a.GameID].PayAmount = a.Amount
+		}
+	}
+	return result, nil
+}
@@ -0,0 +1,165 @@
+package business
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+func prepareChannels(t *testing.T) (*gorm.DB, business.ChannelRepo) {
+	t.Helper()
+	db := openTestDB(t)
+	if err := db.AutoMigrate(&accountPO{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, sql := range []string{
+		"INSERT INTO sys_users(id,username,level,parent_id,id_path) VALUES(1,'root',0,0,'0'),(7,'parent',1,0,'0'),(8,'branch',2,7,'0,7'),(9,'leaf',3,8,'0,7,8'),(10,'sibling',3,7,'0,7'),(99,'outsider',1,0,'0')",
+		"INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,platform_busier_account,status) VALUES(9,'lx',14,'lx-business','0'),(8,'hw',14,'hw-business','0'),(10,'lx',16,'sibling-business','0'),(99,'yq',14,'other-business','0'),(9,'yq',14,'pending','3')",
+		"CREATE TABLE tab_promote(id INTEGER,source_db TEXT,account TEXT,nickname TEXT,parent_name TEXT,real_name TEXT,promote_level INTEGER,register_type INTEGER DEFAULT 0,status INTEGER DEFAULT 1,lock_status INTEGER DEFAULT 0,parent_id INTEGER,top_promote_id INTEGER,busier_id INTEGER,create_time INTEGER DEFAULT 1000,PRIMARY KEY(source_db,id))",
+		"INSERT INTO tab_promote(id,source_db,account,promote_level,parent_id,top_promote_id,busier_id) VALUES(100,'lx','root-channel',1,0,0,14),(101,'lx','child-channel',2,100,100,14),(102,'lx','grandchild',3,101,100,14),(103,'lx','missing-top',3,101,0,14),(104,'lx','other-branch',2,999,999,16),(100,'hw','same-id',1,0,0,14),(100,'yq','outside',1,0,0,14)",
+		"CREATE TABLE tab_promote_apply(source_db TEXT,promote_id INTEGER,game_id INTEGER,status INTEGER)",
+		"INSERT INTO tab_promote_apply VALUES('lx',100,1,1),('lx',100,1,1),('lx',100,2,1),('lx',100,3,0),('lx',101,3,1),('hw',100,99,1)",
+		"CREATE TABLE tab_user(source_db TEXT,id INTEGER,promote_id INTEGER,puid INTEGER,idcard TEXT,PRIMARY KEY(source_db,id))",
+		"INSERT INTO tab_user VALUES('lx',1,100,0,' abcX '),('lx',2,101,NULL,'abcx'),('lx',3,102,0,''),('lx',4,101,2,'subaccount'),('lx',5,103,0,'thirdX'),('hw',1,100,0,'different')",
+		"CREATE TABLE tab_spend(source_db TEXT,promote_id INTEGER,pay_status INTEGER,pay_amount DECIMAL(20,2))",
+		"INSERT INTO tab_spend VALUES('lx',100,1,10.10),('lx',101,1,20.20),('lx',102,1,30.30),('lx',103,1,40.40),('hw',100,1,99.90),('lx',100,0,999)",
+		"CREATE TABLE tab_game(source_db TEXT,id INTEGER,game_name TEXT)",
+		"INSERT INTO tab_game VALUES('lx',1,'游戏甲'),('lx',2,'游戏乙'),('hw',1,'同号其他游戏')",
+		"CREATE TABLE tab_sw_report_day(source_db TEXT,promote_id INTEGER,game_id INTEGER,dedup_mode INTEGER,stat_date DATE,total_pay DECIMAL(20,2))",
+		"INSERT INTO tab_sw_report_day VALUES('lx',100,1,0,'2026-10-01',1.10),('lx',100,1,0,'2026-11-01',10.10),('lx',101,1,0,'2026-11-02',20.20),('lx',102,2,0,'2026-11-02',30.30),('lx',103,1,0,'2026-11-02',999),('lx',100,1,0,'2026-11-04',999),('lx',100,1,1,'2026-11-01',9999),('hw',100,1,0,'2026-11-01',999)",
+		"CREATE TABLE tab_sw_report_day_key(source_db TEXT,promote_id INTEGER,game_id INTEGER,dedup_mode INTEGER,stat_date DATE,metric INTEGER,identity_key TEXT)",
+		"INSERT INTO tab_sw_report_day_key VALUES('lx',100,1,1,'2026-11-01',1,'A'),('lx',101,1,1,'2026-11-02',1,'A'),('lx',102,2,1,'2026-11-02',1,'B'),('lx',102,2,1,'2026-11-02',2,'C'),('lx',103,1,1,'2026-11-02',1,'N'),('lx',100,1,1,'2026-11-04',1,'F'),('hw',100,1,1,'2026-11-01',1,'Z'),('lx',100,1,0,'2026-10-01',1,'old'),('lx',100,1,0,'2026-11-01',1,'u1'),('lx',101,1,0,'2026-11-02',1,'u2')",
+	} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	return db, NewChannelRepo(testData{db})
+}
+
+func channelTestScope() business.UserScope {
+	return business.UserScope{SelfID: 8, IncludeSelf: true, DescendantPath: "0,7,8"}
+}
+func TestChannelListMatchesBindingsAndLiteralBusinessKeyword(t *testing.T) {
+	db, repo := prepareChannels(t)
+	ctx := context.Background()
+	scope := channelTestScope()
+	page, err := repo.List(ctx, scope, business.ChannelFilter{})
+	if err != nil || page.Total != 5 {
+		t.Fatalf("本人及下级绑定范围错误: %+v %v", page, err)
+	}
+	user := uint(9)
+	page, err = repo.List(ctx, scope, business.ChannelFilter{BusierUserID: &user})
+	if err != nil || page.Total != 4 {
+		t.Fatalf("指定商务应只取本人绑定: %+v %v", page, err)
+	}
+	for _, row := range page.Rows {
+		if row.SourceDB != "lx" || row.BusierUserID == nil || *row.BusierUserID != 9 || row.BusinessDisplayName() != "lx-business（leaf）" {
+			t.Fatalf("归属展示错误: %+v", row)
+		}
+	}
+	for _, keyword := range []string{"LEAF", "lx-business"} {
+		page, err = repo.List(ctx, scope, business.ChannelFilter{BusierAccount: keyword})
+		if err != nil || page.Total != 4 {
+			t.Fatal("商务账号或 SW 用户名匹配失败")
+		}
+	}
+	page, err = repo.List(ctx, scope, business.ChannelFilter{BusierAccount: "%"})
+	if err != nil || page.Total != 0 {
+		t.Fatal("商务关键词被当成 SQL 通配符")
+	}
+	page, err = repo.List(ctx, business.UserScope{All: true}, business.ChannelFilter{})
+	if err != nil || page.Total != 7 {
+		t.Fatal("超管默认不限绑定")
+	}
+	page, err = repo.List(ctx, business.UserScope{All: true}, business.ChannelFilter{BusierAccount: "leaf"})
+	if err != nil || page.Total != 4 {
+		t.Fatal("超管输入商务账号后没有收窄")
+	}
+	if _, err := repo.Find(ctx, scope, business.ChannelKey{SourceDB: "yq", ID: 100}); err != business.ErrChannelScope {
+		t.Fatal("同号其他平台未拒绝")
+	}
+	if err := db.Exec("DROP TABLE tab_promote").Error; err != nil {
+		t.Fatal(err)
+	}
+	page, err = repo.List(ctx, scope, business.ChannelFilter{})
+	if err != nil || page.Available {
+		t.Fatal("缺渠道表未标记不可用")
+	}
+	page, err = repo.List(ctx, scope, business.ChannelFilter{BusierAccount: "not-found"})
+	if err != nil || !page.Available || page.Total != 0 {
+		t.Fatal("没有绑定是确定空集")
+	}
+}
+func TestChannelMetricStorageSeparatesSourcesAndMissingTables(t *testing.T) {
+	db, repo := prepareChannels(t)
+	ctx := context.Background()
+	keys := []business.ChannelKey{{SourceDB: "lx", ID: 100}, {SourceDB: "lx", ID: 101}, {SourceDB: "lx", ID: 102}}
+	metrics, err := repo.Metrics(ctx, keys[:1], keys, true)
+	if err != nil || metrics.Games[keys[0]] != 2 || len(metrics.PlayerCards) != 3 || metrics.Pay[keys[0]] != "10.1" {
+		t.Fatalf("原始指标查询范围错误: %+v %v", metrics, err)
+	}
+	var card string
+	for _, player := range metrics.PlayerCards {
+		if player.UserID == 1 {
+			card = player.IDCard
+		}
+	}
+	if card != " abcX " {
+		t.Fatal("idcard 存储字段未映射，导致身份证去重人数错误")
+	}
+	if err := db.Exec("DROP TABLE tab_user").Error; err != nil {
+		t.Fatal(err)
+	}
+	metrics, err = repo.Metrics(ctx, keys[:1], keys, false)
+	if err != nil || metrics.PlayersAvailable || !metrics.GamesAvailable || !metrics.PayAvailable {
+		t.Fatal("一个指标缺表不应掩盖其余指标")
+	}
+}
+func TestChannelGameDistributionUsesReportsAndCurrentDedupStart(t *testing.T) {
+	db, repo := prepareChannels(t)
+	ctx := context.Background()
+	root, err := repo.Find(ctx, channelTestScope(), business.ChannelKey{SourceDB: "lx", ID: 100})
+	if err != nil {
+		t.Fatal(err)
+	}
+	scope := business.ChannelGameScope{Root: root, IncludeSub: true, IDCardDedup: true, EndDate: "2026-11-03"}
+	money := func(raw string) string {
+		n, ok := new(big.Rat).SetString(raw)
+		if !ok {
+			return "invalid"
+		}
+		return n.FloatString(2)
+	}
+	games, err := repo.Games(ctx, scope)
+	if err != nil || len(games) != 2 || games[0].ID != 1 || games[0].Name != "游戏甲" || games[0].RegisterCount != 1 || money(games[0].PayAmount) != "30.30" || games[1].RegisterCount != 1 || money(games[1].PayAmount) != "30.30" {
+		t.Fatalf("报表日期、去重口径、直接推广范围或渠道树错误: %+v %v", games, err)
+	}
+	scope.IDCardDedup = false
+	games, err = repo.Games(ctx, scope)
+	if err != nil || games[0].RegisterCount != 3 || money(games[0].PayAmount) != "31.40" {
+		t.Fatal("不去重口径没有使用自己的累计起点")
+	}
+	scope.IDCardDedup = true
+	scope.IncludeSub = false
+	games, err = repo.Games(ctx, scope)
+	if err != nil || money(games[0].PayAmount) != "10.10" || money(games[1].PayAmount) != "0.00" {
+		t.Fatal("不包含下级仍汇总了子渠道")
+	}
+	if err := db.Exec("DELETE FROM tab_sw_report_day WHERE dedup_mode=1").Error; err != nil {
+		t.Fatal(err)
+	}
+	games, err = repo.Games(ctx, scope)
+	if err != nil || games[0].RegisterCount != 0 || money(games[0].PayAmount) != "0.00" {
+		t.Fatal("无当前口径日汇总时必须沿用旧接口的当天起点")
+	}
+	if err := db.Exec("DROP TABLE tab_sw_report_day_key").Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := repo.Games(ctx, scope); err != business.ErrChannelReportsUnavailable {
+		t.Fatal("缺报表表不能回退原始流水冒充同口径")
+	}
+}
@@ -0,0 +1,130 @@
+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)
+			}()
+		}
+		return conn.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
+}
@@ -0,0 +1,87 @@
+package business
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path/filepath"
+	"testing"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+type channelDatabases struct{ platform *gorm.DB }
+
+func (d channelDatabases) Database(name string) (*gorm.DB, error) {
+	if name == "lx" {
+		return d.platform, nil
+	}
+	return nil, errors.New("数据源未配置")
+}
+func TestChannelWriteIsAtomicInNamedPlatformDatabase(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "platform.db")), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pool, _ := db.DB()
+	t.Cleanup(func() { _ = pool.Close() })
+	if err := db.AutoMigrate(&platformChannelPO{}, &channelSettlementPO{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, sql := range []string{"CREATE UNIQUE INDEX channel_account ON tab_promote(account)", "CREATE TABLE tab_promote_business(id INTEGER PRIMARY KEY,promote_ids TEXT)", "INSERT INTO tab_promote_business VALUES(14,'5,6')"} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	repo := NewChannelWriteRepo(channelDatabases{db})
+	input := &business.ChannelWrite{ChannelCreate: business.ChannelCreate{SourceDB: "lx", Account: "中文渠道", Password: "encoded-password", BusierID: 14, RegisterType: 1}, Status: -3, Level: 1, SettlementDays: 7, CreatedUnix: 1000, NextCountUnix: 605800}
+	id, err := repo.Create(context.Background(), input)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var channel platformChannelPO
+	if err := db.First(&channel, id).Error; err != nil {
+		t.Fatal(err)
+	}
+	if channel.Status != -3 || channel.PromoteLevel != 1 || channel.ParentID != 0 || channel.TopPromoteID != 0 || channel.Password != "encoded-password" || channel.CreateTime != 1000 || channel.RegisterType != 1 {
+		t.Fatalf("源平台渠道默认字段错误: %+v", channel)
+	}
+	var settlement channelSettlementPO
+	if err := db.Where("promote_id = ?", id).First(&settlement).Error; err != nil {
+		t.Fatal(err)
+	}
+	if settlement.TimeType != 0 || settlement.DayPeriod != 7 || settlement.NextCountTime != 605800 || settlement.PromoteAccount != "中文渠道" {
+		t.Fatalf("结算周期字段错误: %+v", settlement)
+	}
+	var related string
+	db.Table("tab_promote_business").Select("promote_ids").Where("id=14").Scan(&related)
+	if related != fmt.Sprintf("5,6,%d", id) {
+		t.Fatal("追加关系覆盖了已有渠道")
+	}
+	if _, err := repo.Create(context.Background(), input); err != business.ErrChannelExists {
+		t.Fatal("重复账号没有拒绝")
+	}
+	if err := db.Exec("DROP TABLE tab_promote_settlement_time").Error; err != nil {
+		t.Fatal(err)
+	}
+	input.Account = "rollback"
+	if _, err := repo.Create(context.Background(), input); err == nil {
+		t.Fatal("缺结算表应回滚")
+	}
+	var count int64
+	db.Model(&platformChannelPO{}).Count(&count)
+	if count != 1 {
+		t.Fatal("创建失败留下了渠道")
+	}
+	var after string
+	db.Table("tab_promote_business").Select("promote_ids").Where("id=14").Scan(&after)
+	if after != related {
+		t.Fatal("创建失败仍改动关联")
+	}
+	input.SourceDB = "hw"
+	if _, err := repo.Create(context.Background(), input); err != business.ErrPlatformUnavailable {
+		t.Fatal("未配置平台不应回退到其他数据源")
+	}
+}
@@ -37,6 +37,8 @@ var ProviderSet = wire.NewSet(
 	databusiness.NewAlertRuleRepo,
 	databusiness.NewAlertRecordRepo,
 	databusiness.NewActivationRepo,
+	databusiness.NewChannelRepo,
+	databusiness.NewChannelWriteRepo,
 	NewIntegrationRuntime,
 	wire.Bind(new(datasystem.Provider), new(*Data)),
 	wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),