feat: 完善小组绩效与有效端统计并修复审查问题
This commit is contained in:
@@ -161,6 +161,11 @@ func (r *accountRepo) Create(ctx context.Context, a *business.Account) error {
|
||||
if err := tx.Create(&po).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if a.HireDate != "" {
|
||||
if err := tx.Create(&businessMemberProfilePO{UserID: po.ID, HireDate: a.HireDate, Employment: "employed"}).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
}
|
||||
if err := tx.Table("sys_user_authority").Create(map[string]any{"sys_user_id": po.ID, "sys_authority_authority_id": po.AuthorityID}).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package business
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
"gorm.io/gorm"
|
||||
@@ -218,6 +219,22 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
|
||||
if len(rowMap) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
if kind == business.ChannelAnalytics {
|
||||
location := time.FixedZone("Asia/Shanghai", 8*3600)
|
||||
begin, _ := time.ParseInLocation("2006-01-02", f.Begin, location)
|
||||
end, _ := time.ParseInLocation("2006-01-02", f.End, location)
|
||||
keys := []business.AnalyticsKey{}
|
||||
for key := range rowMap {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
for start := 0; start < len(keys); start += 300 {
|
||||
var count int64
|
||||
if err := db.Table("sys_business_channel_effectiveness e").Joins("JOIN sys_business_effectiveness_state s ON s.id=1 AND s.rule_key=e.rule_key").Where("(e.source_db,e.promote_id) IN ? AND e.qualified_at>=? AND e.qualified_at<?", analyticsPairs(keys[start:min(start+300, len(keys))]), begin.Unix(), end.AddDate(0, 0, 1).Unix()).Count(&count).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
result.EffectiveCount += count
|
||||
}
|
||||
}
|
||||
if kind == business.GameAnalytics {
|
||||
available, err := businessTableAvailable(db, "tab_game")
|
||||
if err != nil {
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
)
|
||||
|
||||
type channelPO struct {
|
||||
QualifiedAt *int64 `gorm:"->;-:migration"`
|
||||
ID uint64
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
Account, Nickname, ParentName, RealName string
|
||||
@@ -21,7 +22,7 @@ type channelPO struct {
|
||||
|
||||
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, CreatedUnix: p.CreateTime, LastLoginUnix: p.LastLoginTime}
|
||||
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, CreatedUnix: p.CreateTime, LastLoginUnix: p.LastLoginTime, QualifiedAt: p.QualifiedAt}
|
||||
}
|
||||
|
||||
type channelRepo struct{ data Provider }
|
||||
@@ -95,14 +96,22 @@ func (r *channelRepo) List(ctx context.Context, scope business.UserScope, filter
|
||||
q = q.Where("p."+field.column+" = ?", *field.value)
|
||||
}
|
||||
}
|
||||
if filter.Effective != nil {
|
||||
q = joinEffectiveness(q)
|
||||
if *filter.Effective {
|
||||
q = q.Where("e.qualified_at IS NOT NULL")
|
||||
} else {
|
||||
q = q.Where("e.qualified_at IS NULL")
|
||||
}
|
||||
}
|
||||
if err := q.Count(&page.Total).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
var pos []channelPO
|
||||
o := business.NewListOptions(options...)
|
||||
keysQuery := q.Select("p.id,p.source_db").Order("p.create_time DESC,p.id DESC,p.source_db ASC").Offset(o.Offset).Limit(o.Limit)
|
||||
pageQuery := db.Table("tab_promote p").Joins("JOIN (?) page_ids ON page_ids.id=p.id AND page_ids.source_db=p.source_db", keysQuery)
|
||||
if err := pageQuery.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").Scan(&pos).Error; err != nil {
|
||||
pageQuery := joinEffectiveness(db.Table("tab_promote p").Joins("JOIN (?) page_ids ON page_ids.id=p.id AND page_ids.source_db=p.source_db", keysQuery))
|
||||
if err := pageQuery.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,p.create_time,e.qualified_at").Order("p.create_time DESC,p.id DESC,p.source_db ASC").Scan(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if len(pos) == 0 {
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type dashboardRepo struct{ data Provider }
|
||||
|
||||
func NewDashboardRepo(data Provider) business.DashboardRepo { return &dashboardRepo{data: data} }
|
||||
func (r *dashboardRepo) Overview(ctx context.Context, scope business.AnalyticsScope, sources []string, w business.DashboardWindow) (*business.DashboardData, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
out := &business.DashboardData{Current: business.DashboardMetrics{Pay: "0.00"}, Previous: business.DashboardMetrics{Pay: "0.00"}, Available: true}
|
||||
empty, available, err := workbenchAvailable(db, scope, sources, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out.Available = available
|
||||
if empty || !available {
|
||||
return out, nil
|
||||
}
|
||||
begin, end, previous := w.Begin.Format("2006-01-02"), w.End.Format("2006-01-02"), w.PreviousBegin.Format("2006-01-02")
|
||||
f := business.AnalyticsFilter{Sources: sources, Begin: previous, End: end}
|
||||
mode := 0
|
||||
if scope.IDCardDedup {
|
||||
mode = 1
|
||||
}
|
||||
var counts []struct {
|
||||
Current bool
|
||||
Metric int
|
||||
Count int64
|
||||
}
|
||||
var amounts struct{ CurrentPay, PreviousPay string }
|
||||
type activeRow struct {
|
||||
Current bool
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
PromoteID uint64
|
||||
}
|
||||
var registered, paid []activeRow
|
||||
var channels struct{ CurrentNew, PreviousNew, CurrentEffective, PreviousEffective int64 }
|
||||
group, gctx := errgroup.WithContext(ctx)
|
||||
group.SetLimit(2)
|
||||
qdb := db.WithContext(gctx)
|
||||
group.Go(func() error {
|
||||
return storageError(analyticsQuery(qdb, scope, f, "tab_sw_report_day_key", mode).Where("r.metric IN (1,2)").Select("(r.stat_date>=?) AS current,r.metric,COUNT(DISTINCT "+portraitIdentity(qdb, scope.IDCardDedup)+") AS count", begin).Group("current,r.metric").Scan(&counts).Error)
|
||||
})
|
||||
group.Go(func() error {
|
||||
return storageError(analyticsQuery(qdb, scope, f, "tab_sw_report_day", 0).Select("CAST(COALESCE(SUM(CASE WHEN r.stat_date>=? THEN r.total_pay ELSE 0 END),0) AS CHAR) AS current_pay,CAST(COALESCE(SUM(CASE WHEN r.stat_date<? THEN r.total_pay ELSE 0 END),0) AS CHAR) AS previous_pay", begin, begin).Scan(&amounts).Error)
|
||||
})
|
||||
group.Go(func() error {
|
||||
return storageError(analyticsQuery(qdb, scope, f, "tab_sw_report_day_key", mode).Where("r.metric=1 AND r.identity_key IS NOT NULL").Select("DISTINCT (r.stat_date>=?) AS current,r.source_db,r.promote_id", begin).Scan(®istered).Error)
|
||||
})
|
||||
group.Go(func() error {
|
||||
return storageError(analyticsQuery(qdb, scope, f, "tab_sw_report_day", 0).Select("(r.stat_date>=?) AS current,r.source_db,r.promote_id", begin).Group("current,r.source_db,r.promote_id").Having("SUM(r.total_pay)>0").Scan(&paid).Error)
|
||||
})
|
||||
group.Go(func() error {
|
||||
q := analyticsBasePromotes(qdb, scope, f, "p")
|
||||
q = joinEffectiveness(q)
|
||||
return storageError(q.Select("COALESCE(SUM(CASE WHEN p.create_time>=? AND p.create_time<? THEN 1 ELSE 0 END),0) AS current_new,"+
|
||||
"COALESCE(SUM(CASE WHEN p.create_time>=? AND p.create_time<? THEN 1 ELSE 0 END),0) AS previous_new,"+
|
||||
"COALESCE(SUM(CASE WHEN e.qualified_at>=? AND e.qualified_at<? THEN 1 ELSE 0 END),0) AS current_effective,"+
|
||||
"COALESCE(SUM(CASE WHEN e.qualified_at>=? AND e.qualified_at<? THEN 1 ELSE 0 END),0) AS previous_effective",
|
||||
w.Begin.Unix(), w.End.AddDate(0, 0, 1).Unix(), w.PreviousBegin.Unix(), w.Begin.Unix(), w.Begin.Unix(), w.End.AddDate(0, 0, 1).Unix(), w.PreviousBegin.Unix(), w.Begin.Unix()).Scan(&channels).Error)
|
||||
})
|
||||
if err := group.Wait(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, c := range counts {
|
||||
target := &out.Previous
|
||||
if c.Current {
|
||||
target = &out.Current
|
||||
}
|
||||
if c.Metric == 1 {
|
||||
target.Register = c.Count
|
||||
} else {
|
||||
target.Active = c.Count
|
||||
}
|
||||
}
|
||||
current, prior := map[business.ChannelKey]bool{}, map[business.ChannelKey]bool{}
|
||||
for _, a := range append(registered, paid...) {
|
||||
target := prior
|
||||
if a.Current {
|
||||
target = current
|
||||
}
|
||||
target[business.ChannelKey{SourceDB: a.SourceDB, ID: a.PromoteID}] = true
|
||||
}
|
||||
out.Current.ActiveChannels, out.Previous.ActiveChannels = int64(len(current)), int64(len(prior))
|
||||
out.Current.NewChannels, out.Previous.NewChannels = channels.CurrentNew, channels.PreviousNew
|
||||
out.Current.Effective, out.Previous.Effective = channels.CurrentEffective, channels.PreviousEffective
|
||||
out.Current.Pay, out.Previous.Pay = amounts.CurrentPay, amounts.PreviousPay
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type effectivenessRepo struct{ data Provider }
|
||||
|
||||
func NewEffectivenessRepo(data Provider) business.EffectivenessRepo {
|
||||
return &effectivenessRepo{data: data}
|
||||
}
|
||||
func (r *effectivenessRepo) State(ctx context.Context) (*business.EffectivenessState, error) {
|
||||
var po effectivenessStatePO
|
||||
if err := r.data.DB().WithContext(ctx).First(&po, "id=1").Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
s := &business.EffectivenessState{RuleKey: po.RuleKey, Revision: po.Revision}
|
||||
if po.BuiltAt != nil {
|
||||
s.BuiltAt = *po.BuiltAt
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
func (r *effectivenessRepo) Candidates(ctx context.Context, key string, duration, now int64, full bool) ([]*business.EffectivenessCandidate, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
available, err := businessTablesAvailable(db, "tab_promote", "tab_spend")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrEffectivenessUnavailable
|
||||
}
|
||||
q := db.Table("tab_promote p").Joins("LEFT JOIN sys_business_channel_effectiveness e ON e.rule_key=? AND e.source_db=p.source_db AND e.promote_id=p.id", key).Where("p.source_db IN ?", []string{"lx", "hw", "yq"})
|
||||
if !full {
|
||||
q = q.Where("e.promote_id IS NULL OR e.complete=?", false)
|
||||
}
|
||||
var pos []struct {
|
||||
ID uint64
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
CreateTime, QualifiedAt *int64
|
||||
WindowPay string
|
||||
}
|
||||
if err := q.Select("p.id,p.source_db,p.create_time,e.qualified_at,e.window_pay").Order("p.id,p.source_db").Scan(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
rows := make([]*business.EffectivenessCandidate, 0, len(pos))
|
||||
for _, po := range pos {
|
||||
rows = append(rows, &business.EffectivenessCandidate{ChannelKey: business.ChannelKey{ID: po.ID, SourceDB: po.SourceDB}, CreatedAt: po.CreateTime, PreviousAt: po.QualifiedAt, PreviousPay: po.WindowPay})
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
func (r *effectivenessRepo) Charges(ctx context.Context, candidates []*business.EffectivenessCandidate, duration, now int64, full bool, consume func(business.EffectivenessCharge) error) error {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
// Bound by now as future charges are excluded. Avoid unsigned subtraction
|
||||
// on legacy pay_time columns and overflow for very large configured windows.
|
||||
windowSeconds := min(duration, now+1)
|
||||
// The full-history scan is reserved for explicit rebuilds. Minute jobs
|
||||
// must keep their candidate restriction in SQL, even with many open windows.
|
||||
if full && len(candidates) > 3000 {
|
||||
return r.scanAllCharges(db, candidates, windowSeconds, now, consume)
|
||||
}
|
||||
for start := 0; start < len(candidates); start += 300 {
|
||||
keys := [][]any{}
|
||||
ids := []uint64{}
|
||||
for _, c := range candidates[start:min(start+300, len(candidates))] {
|
||||
if c.CreatedAt != nil && *c.CreatedAt > 0 && *c.CreatedAt <= now {
|
||||
keys = append(keys, []any{c.SourceDB, c.ID})
|
||||
ids = append(ids, c.ID)
|
||||
}
|
||||
}
|
||||
if len(keys) == 0 {
|
||||
continue
|
||||
}
|
||||
q := db.Table("tab_spend s").Joins("JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id").Where("s.promote_id IN ? AND (s.source_db,s.promote_id) IN ? AND s.pay_status=1", ids, keys).Where("s.pay_time>=p.create_time AND s.pay_time<=? AND s.pay_time<p.create_time+?", now, windowSeconds)
|
||||
rows, err := q.Select("s.source_db,s.promote_id,s.pay_time,CAST(COALESCE(s.pay_amount,0) AS CHAR)").Order("s.promote_id,s.source_db,s.pay_time,s.id").Rows()
|
||||
if err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
for rows.Next() {
|
||||
var c business.EffectivenessCharge
|
||||
if err := rows.Scan(&c.SourceDB, &c.ID, &c.PaidAt, &c.Amount); err != nil {
|
||||
rows.Close()
|
||||
return business.ErrStorage
|
||||
}
|
||||
if err := consume(c); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Large rebuilds stream the covering index once. Repeating hundreds of IN
|
||||
// batches can make the optimizer repeatedly sort large payment ranges.
|
||||
func (r *effectivenessRepo) scanAllCharges(db *gorm.DB, candidates []*business.EffectivenessCandidate, windowSeconds, now int64, consume func(business.EffectivenessCharge) error) error {
|
||||
selected := make(map[business.ChannelKey]bool, len(candidates))
|
||||
for _, c := range candidates {
|
||||
selected[c.ChannelKey] = true
|
||||
}
|
||||
table, join := "tab_spend s", "JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id"
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
join = "STRAIGHT_JOIN tab_promote p ON p.source_db=s.source_db AND p.id=s.promote_id"
|
||||
if db.Migrator().HasIndex("tab_spend", "idx_sw_spend_effective_scan") {
|
||||
table += " FORCE INDEX (idx_sw_spend_effective_scan)"
|
||||
}
|
||||
}
|
||||
q := db.Table(table).Joins(join).Where("s.pay_status=1 AND s.source_db IN ? AND p.create_time>0 AND s.pay_time>=p.create_time AND s.pay_time<=? AND s.pay_time<p.create_time+?", []string{"lx", "hw", "yq"}, now, windowSeconds)
|
||||
rows, err := q.Select("s.source_db,s.promote_id,s.pay_time,CAST(COALESCE(s.pay_amount,0) AS CHAR)").Order("s.promote_id,s.source_db,s.pay_time,s.id").Rows()
|
||||
if err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var c business.EffectivenessCharge
|
||||
if err := rows.Scan(&c.SourceDB, &c.ID, &c.PaidAt, &c.Amount); err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if selected[c.ChannelKey] {
|
||||
if err := consume(c); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return storageError(rows.Err())
|
||||
}
|
||||
func (r *effectivenessRepo) Publish(ctx context.Context, expected *business.EffectivenessState, rule business.OtherSettings, key string, rows []business.EffectivenessResult, save bool) (*business.OtherSettings, error) {
|
||||
var saved = rule
|
||||
err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var state effectivenessStatePO
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&state, "id=1").Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if state.Revision != expected.Revision {
|
||||
return business.ErrEffectivenessConflict
|
||||
}
|
||||
if !save {
|
||||
var current otherSettingsPO
|
||||
result := tx.Where(map[string]any{"key": otherSettingsKey}).Limit(1).Find(¤t)
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
active := business.DefaultOtherSettings()
|
||||
if result.RowsAffected > 0 {
|
||||
v, err := toBizOtherSettings(current)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
active = *v
|
||||
}
|
||||
if business.EffectivenessRuleKey(active) != key {
|
||||
return business.ErrEffectivenessConflict
|
||||
}
|
||||
}
|
||||
now := time.Now()
|
||||
for start := 0; start < len(rows); start += 500 {
|
||||
pos := make([]effectivenessPO, 0, min(500, len(rows)-start))
|
||||
for _, row := range rows[start:min(start+500, len(rows))] {
|
||||
pos = append(pos, effectivenessPO{RuleKey: key, SourceDB: row.SourceDB, PromoteID: row.ID, QualifiedAt: row.QualifiedAt, WindowPay: row.WindowPay, Complete: row.Complete, UpdatedAt: now})
|
||||
}
|
||||
if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "rule_key"}, {Name: "source_db"}, {Name: "promote_id"}}, DoUpdates: clause.AssignmentColumns([]string{"qualified_at", "window_pay", "complete", "updated_at"})}).Create(&pos).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if save {
|
||||
po := newOtherSettings(rule)
|
||||
var old otherSettingsPO
|
||||
result := tx.Where(map[string]any{"key": otherSettingsKey}).Limit(1).Find(&old)
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
if err := tx.Create(&po).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
if err := tx.Model(&old).Updates(map[string]any{"value": po.Value, "updated_at": now}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
saved.UpdatedAt = now
|
||||
}
|
||||
return tx.Model(&effectivenessStatePO{}).Where("id=1").Updates(map[string]any{"rule_key": key, "revision": state.Revision + 1, "built_at": now}).Error
|
||||
})
|
||||
if err != nil {
|
||||
return nil, storageError(err)
|
||||
}
|
||||
return &saved, nil
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/pkg/database/migration"
|
||||
)
|
||||
|
||||
type businessGroupPO struct {
|
||||
ID uint64 `gorm:"primaryKey"`
|
||||
Name string `gorm:"size:40;not null"`
|
||||
OwnerID uint `gorm:"index"`
|
||||
Sort int
|
||||
CreatedAt, UpdatedAt time.Time
|
||||
DeletedAt gorm.DeletedAt `gorm:"index"`
|
||||
}
|
||||
|
||||
func (businessGroupPO) TableName() string { return "sys_business_groups" }
|
||||
|
||||
type businessGroupMemberPO struct {
|
||||
UserID uint `gorm:"primaryKey;autoIncrement:false"`
|
||||
GroupID uint64 `gorm:"index;not null"`
|
||||
}
|
||||
|
||||
func (businessGroupMemberPO) TableName() string { return "sys_business_group_members" }
|
||||
|
||||
type businessMemberProfilePO struct {
|
||||
UserID uint `gorm:"primaryKey;autoIncrement:false"`
|
||||
HireDate string `gorm:"size:10"`
|
||||
Employment string `gorm:"size:16;not null;default:employed"`
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (businessMemberProfilePO) TableName() string { return "sys_business_member_profiles" }
|
||||
|
||||
type businessGroupStatePO struct {
|
||||
ID uint `gorm:"primaryKey;autoIncrement:false"`
|
||||
Revision uint64
|
||||
}
|
||||
|
||||
func (businessGroupStatePO) TableName() string { return "sys_business_group_state" }
|
||||
|
||||
type performanceGoalPO struct {
|
||||
Kind string `gorm:"primaryKey;size:8"`
|
||||
TargetID uint64 `gorm:"primaryKey;autoIncrement:false"`
|
||||
Month string `gorm:"primaryKey;size:7;index"`
|
||||
Open, Effective, Amount *int64
|
||||
UpdatedBy uint
|
||||
CreatedAt, UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (performanceGoalPO) TableName() string { return "sys_business_performance_goals" }
|
||||
|
||||
type effectivenessStatePO struct {
|
||||
ID uint `gorm:"primaryKey;autoIncrement:false"`
|
||||
RuleKey string `gorm:"size:64"`
|
||||
Revision uint64
|
||||
BuiltAt *time.Time
|
||||
}
|
||||
|
||||
func (effectivenessStatePO) TableName() string { return "sys_business_effectiveness_state" }
|
||||
|
||||
type effectivenessPO struct {
|
||||
RuleKey string `gorm:"primaryKey;size:64;index:idx_sw_effective_date,priority:1"`
|
||||
// Platform codes are identifiers. Binary storage also avoids collation
|
||||
// conflicts when joining historical SDK tables imported with another collation.
|
||||
SourceDB string `gorm:"primaryKey;type:varbinary(10)"`
|
||||
PromoteID uint64 `gorm:"primaryKey;autoIncrement:false"`
|
||||
QualifiedAt *int64 `gorm:"index:idx_sw_effective_date,priority:2"`
|
||||
WindowPay string `gorm:"type:decimal(28,2);not null;default:0"`
|
||||
Complete bool
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (effectivenessPO) TableName() string { return "sys_business_channel_effectiveness" }
|
||||
|
||||
func migrateBusinessManagement(db *gorm.DB) error {
|
||||
if err := migration.CreateMissingTables(db, &businessGroupPO{}, &businessGroupMemberPO{}, &businessMemberProfilePO{}, &businessGroupStatePO{}, &performanceGoalPO{}, &effectivenessStatePO{}, &effectivenessPO{}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&businessGroupStatePO{ID: 1}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return db.Clauses(clause.OnConflict{DoNothing: true}).Create(&effectivenessStatePO{ID: 1}).Error
|
||||
}
|
||||
|
||||
func migrateEffectivenessQueryIndex(db *gorm.DB) error {
|
||||
return migrateQueryIndexes(db, []businessQueryIndex{{"tab_spend", "idx_sw_spend_effective_scan", []string{"promote_id", "source_db", "pay_status", "pay_time", "id", "pay_amount"}}})
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math/big"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
"kra/internal/biz/system"
|
||||
)
|
||||
|
||||
func TestEffectivenessRepositoryWindowAndAtomicRulePublication(t *testing.T) {
|
||||
db, _, _, _ := prepareLedgerData(t)
|
||||
if err := db.AutoMigrate(&otherSettingsPO{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, sql := range []string{
|
||||
"INSERT INTO tab_spend(id,source_db,user_id,promote_id,pay_amount,pay_time,pay_status) VALUES(1001,'lx',1,100,5,87399,1),(1002,'lx',1,100,999,87400,1),(1003,'lx',1,100,999,900,1)",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
repo := NewEffectivenessRepo(testData{db})
|
||||
settings := NewOtherSettingsRepo(testData{db})
|
||||
uc := business.NewEffectivenessUsecase(repo, settings)
|
||||
ctx := system.NewActorContext(context.Background(), system.Actor{UserID: 1, AuthorityID: 888})
|
||||
rule := business.OtherSettings{Amount: 10, Days: 1}
|
||||
if _, err := uc.Reconfigure(ctx, rule); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var row effectivenessPO
|
||||
if err := db.Where("rule_key=? AND source_db='lx' AND promote_id=100", business.EffectivenessRuleKey(rule)).First(&row).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if row.QualifiedAt == nil || *row.QualifiedAt != 1500 || portraitTestMoney(row.WindowPay) != "15.10" {
|
||||
t.Fatalf("窗口须排除注册前、结束瞬间及失败订单: %+v", row)
|
||||
}
|
||||
var other effectivenessPO
|
||||
if err := db.Where("rule_key=? AND source_db='hw' AND promote_id=100", business.EffectivenessRuleKey(rule)).First(&other).Error; err != nil || other.QualifiedAt == nil {
|
||||
t.Fatal("同号渠道必须按平台分别判定")
|
||||
}
|
||||
if err := db.Exec("UPDATE tab_spend SET pay_status=0 WHERE source_db='lx' AND id=901").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := uc.Refresh(ctx, true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Where("rule_key=? AND source_db='lx' AND promote_id=100", business.EffectivenessRuleKey(rule)).First(&row).Error; err != nil || row.QualifiedAt == nil || *row.QualifiedAt != 1500 {
|
||||
t.Fatal("后续流水回落不能取消已达标标记")
|
||||
}
|
||||
next := business.OtherSettings{Amount: 20, Days: 1}
|
||||
if _, err := uc.Reconfigure(ctx, next); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var newRow effectivenessPO
|
||||
if err := db.Where("rule_key=? AND source_db='lx' AND promote_id=100", business.EffectivenessRuleKey(next)).First(&newRow).Error; err != nil || newRow.QualifiedAt != nil {
|
||||
t.Fatal("更改规则须按新门槛重算,而不是继承旧规则结果")
|
||||
}
|
||||
state, err := repo.State(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Callback().Create().Before("gorm:create").Register("fail_effectiveness", func(tx *gorm.DB) {
|
||||
if tx.Statement.Table == "sys_business_channel_effectiveness" {
|
||||
tx.AddError(errors.New("injected failure"))
|
||||
}
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Callback().Create().Remove("fail_effectiveness")
|
||||
if _, err := uc.Reconfigure(ctx, business.OtherSettings{Amount: 100, Days: 1}); err == nil {
|
||||
t.Fatal("注入的持久化失败必须上报")
|
||||
}
|
||||
stored, _ := settings.Find(ctx)
|
||||
after, _ := repo.State(ctx)
|
||||
if stored.Amount != 20 || after.Revision != state.Revision || after.RuleKey != state.RuleKey {
|
||||
t.Fatal("失败不能发布规则或部分结果")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLargeEffectivenessScanMatchesBatchedCharges(t *testing.T) {
|
||||
db, _, _, _ := prepareLedgerData(t)
|
||||
if err := db.Exec("WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n<3200) INSERT INTO tab_promote(id,source_db,account,promote_level,parent_id,top_promote_id,busier_id,create_time) SELECT 10000+n,'lx','bulk-'||n,1,0,0,14,1000 FROM seq").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := &effectivenessRepo{data: testData{db}}
|
||||
ctx := context.Background()
|
||||
candidates, err := r.Candidates(ctx, "test", 86400, 90000, true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(candidates) <= 3000 {
|
||||
t.Fatal("fixture must exercise streaming path")
|
||||
}
|
||||
var full, batched []business.EffectivenessCharge
|
||||
if err := r.Charges(ctx, candidates, 86400, 90000, true, func(c business.EffectivenessCharge) error { full = append(full, c); return nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for start := 0; start < len(candidates); start += 300 {
|
||||
if err := r.Charges(ctx, candidates[start:min(start+300, len(candidates))], 86400, 90000, false, func(c business.EffectivenessCharge) error { batched = append(batched, c); return nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if !reflect.DeepEqual(full, batched) {
|
||||
t.Fatalf("流式优化不能改变时间顺序、平台身份或流水: full=%v batch=%v", full, batched)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPerformanceOwnerAggregationAndDrilldownUseSameScope(t *testing.T) {
|
||||
db, _, scope, _ := prepareAnalyticsData(t)
|
||||
repo := NewPerformanceRepo(testData{db})
|
||||
location := time.FixedZone("Asia/Shanghai", 8*3600)
|
||||
begin := time.Date(2026, 11, 1, 0, 0, 0, 0, location)
|
||||
end := begin.AddDate(0, 0, 2)
|
||||
if err := db.Exec("UPDATE tab_promote SET create_time=? WHERE source_db='lx' AND id=100", begin.Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&effectivenessStatePO{}).Where("id=1").Updates(map[string]any{"rule_key": "fixture", "built_at": time.Now()}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Create(&effectivenessPO{RuleKey: "fixture", SourceDB: "lx", PromoteID: 100, QualifiedAt: func() *int64 { n := begin.AddDate(0, 0, 1).Unix(); return &n }(), WindowPay: "3000"}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
facts, err := repo.Facts(context.Background(), scope.Viewer, nil, []string{"lx"}, begin, end)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(facts.Cash) != 2 {
|
||||
t.Fatalf("同一SDK商务应按日聚合,避免按渠道返回所有明细: %+v", facts.Cash)
|
||||
}
|
||||
if len(facts.Channels) != 1 || facts.Channels[0].ID != 100 {
|
||||
t.Fatal("historical channels with no in-period opening/qualification must not be materialized")
|
||||
}
|
||||
total := new(big.Rat)
|
||||
for _, cash := range facts.Cash {
|
||||
if len(cash.OwnerIDs) != 1 || cash.OwnerIDs[0] != 9 {
|
||||
t.Fatal("聚合丢失商务归属")
|
||||
}
|
||||
n, ok := new(big.Rat).SetString(cash.Pay)
|
||||
if !ok {
|
||||
t.Fatal("invalid amount")
|
||||
}
|
||||
total.Add(total, n)
|
||||
}
|
||||
if total.FloatString(2) != "1059.60" {
|
||||
t.Fatalf("汇总金额改变: %v", total)
|
||||
}
|
||||
rows, count, err := repo.Details(context.Background(), scope.Viewer, business.PerformanceDetailSelection{OwnerIDs: []uint{9}}, []string{"lx"}, begin, end, "effective", 0, 20)
|
||||
if err != nil || count != 1 || rows[0].ID != 100 || portraitTestMoney(rows[0].PeriodPay) != "10.10" {
|
||||
t.Fatalf("下钻应准确筛选并只补本页充值: %+v %d %v", rows, count, err)
|
||||
}
|
||||
_, count, err = repo.Details(context.Background(), business.UserScope{OnlySelf: true, SelfID: 99}, business.PerformanceDetailSelection{OwnerIDs: []uint{9}}, nil, begin, end, "effective", 0, 20)
|
||||
if err != nil || count != 0 {
|
||||
t.Fatal("不能通过指定其他商务ID越权下钻")
|
||||
}
|
||||
selection := business.PerformanceDetailSelection{OwnerIDs: []uint{8, 9}, ContributorIDs: []uint{8, 9}}
|
||||
rows, count, err = repo.Details(context.Background(), scope.Viewer, selection, []string{"lx"}, begin, end, "effective", 0, 20)
|
||||
if err != nil || count != 2 || len(rows) != 2 || rows[0].ContributorID != 8 || rows[1].ContributorID != 9 || rows[0].ID != rows[1].ID || portraitTestMoney(rows[0].PeriodPay) != "10.10" || rows[0].ContributorName == "" {
|
||||
t.Fatalf("team drilldown must show both member contributions: %+v count=%d err=%v", rows, count, err)
|
||||
}
|
||||
rows, count, err = repo.Details(context.Background(), scope.Viewer, selection, []string{"lx"}, begin, end, "effective", 1, 1)
|
||||
if err != nil || count != 2 || len(rows) != 1 || rows[0].ContributorID != 9 {
|
||||
t.Fatal("contribution pagination must preserve total and stable row identity")
|
||||
}
|
||||
data, err := NewDashboardRepo(testData{db}).Overview(context.Background(), scope, []string{"lx"}, business.DashboardWindow{Begin: begin, End: end, PreviousBegin: begin.AddDate(0, 0, -3), PreviousEnd: begin.AddDate(0, 0, -1)})
|
||||
if err != nil || data.Current.Effective != 1 || data.Current.ActiveChannels != 4 || portraitTestMoney(data.Current.Pay) != "1059.60" {
|
||||
t.Fatalf("首页按区间统计与绩效口径不一致: %+v %v", data, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTeamCatalogVersionedMembershipAndIndependentGoals(t *testing.T) {
|
||||
db, _ := prepareChannels(t)
|
||||
repo := NewTeamRepo(testData{db})
|
||||
ctx := context.Background()
|
||||
initial, err := repo.Catalog(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
view, err := repo.SaveCatalog(ctx, initial.Revision, []*business.BusinessGroup{{Name: "同名组", OwnerID: 7, MemberIDs: []uint{8}}, {Name: "同名组", OwnerID: 7, Sort: 1, MemberIDs: []uint{9}}})
|
||||
if err != nil || len(view.Groups) != 2 || view.Groups[0].ID == view.Groups[1].ID {
|
||||
t.Fatal("同名组应以独立ID区分")
|
||||
}
|
||||
if _, err := repo.SaveCatalog(ctx, initial.Revision, nil); err != business.ErrTeamConflict {
|
||||
t.Fatal("过期版本不能删除成员")
|
||||
}
|
||||
goals := NewGoalRepo(testData{db})
|
||||
target := int64(30)
|
||||
if err := goals.Save(ctx, &business.PerformanceGoal{Kind: "person", TargetID: 9, Month: "2026-11", Open: &target}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := goals.Save(ctx, &business.PerformanceGoal{Kind: "group", TargetID: view.Groups[1].ID, Month: "2026-11", Open: &target}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
zero := int64(0)
|
||||
if err := goals.Save(ctx, &business.PerformanceGoal{Kind: "person", TargetID: 9, Month: "2026-11", Open: &zero}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
values, err := goals.List(ctx, "2026-11", "2026-11")
|
||||
if err != nil || len(values) != 2 {
|
||||
t.Fatal("个人和小组目标必须各自保存")
|
||||
}
|
||||
for _, value := range values {
|
||||
if value.Kind == "group" && *value.Open != 30 {
|
||||
t.Fatal("修改个人目标不应改变团队目标")
|
||||
}
|
||||
if value.Kind == "person" && (*value.Open != 0 || value.Effective != nil) {
|
||||
t.Fatal("零目标和留空必须区分")
|
||||
}
|
||||
}
|
||||
if _, err := repo.SaveCatalog(ctx, view.Revision, []*business.BusinessGroup{{ID: view.Groups[0].ID, Name: "同名组", OwnerID: 7, MemberIDs: []uint{8, 9}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var members []businessGroupMemberPO
|
||||
if err := db.Find(&members).Error; err != nil || len(members) != 2 {
|
||||
t.Fatal("移动成员不能丢失或重复")
|
||||
}
|
||||
for _, member := range members {
|
||||
if member.GroupID != view.Groups[0].ID {
|
||||
t.Fatal("成员应移动到目标小组")
|
||||
}
|
||||
}
|
||||
if err := repo.UpdateProfile(ctx, business.MemberProfile{ID: 9, RealName: "成员", Phone: "13900000009", HireDate: "2025-01-01", Employment: "departed"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var user accountPO
|
||||
if err := db.First(&user, 9).Error; err != nil || user.Enable != 0 || user.BusinessStatus != 1 {
|
||||
t.Fatal("离职应停用登录但保留账号")
|
||||
}
|
||||
var total int64
|
||||
db.Table("tab_spend").Count(&total)
|
||||
if total == 0 {
|
||||
t.Fatal("离职不能删除历史业绩")
|
||||
}
|
||||
}
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
)
|
||||
|
||||
func Migrations() []migration.Step {
|
||||
return []migration.Step{{ID: "202611040015_business_approved_game_index", Migrate: migrateApprovedGameIndexes}, {ID: "202611040014_business_report_build_indexes", Migrate: migrateReportBuildQueryIndexes}, {ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
|
||||
return []migration.Step{{ID: "202611040017_business_groups_goals_effectiveness", Migrate: migrateBusinessManagement}, {ID: "202611040018_business_effectiveness_query_index", Migrate: migrateEffectivenessQueryIndex}, {ID: "202611040015_business_approved_game_index", Migrate: migrateApprovedGameIndexes}, {ID: "202611040014_business_report_build_indexes", Migrate: migrateReportBuildQueryIndexes}, {ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
|
||||
return migration.CreateMissingTables(db, &resetCodePO{}, &resetTokenPO{})
|
||||
}}, {ID: "202611030004_business_binding_locks", Migrate: func(db *gorm.DB) error {
|
||||
return migration.CreateMissingTables(db, &bindingLockPO{})
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type performanceRepo struct{ data Provider }
|
||||
|
||||
func NewPerformanceRepo(data Provider) business.PerformanceRepo { return &performanceRepo{data: data} }
|
||||
|
||||
func performanceChannels(db *gorm.DB, scope business.UserScope, owners []uint, sources []string) *gorm.DB {
|
||||
q := db.Table("tab_promote p").Where("p.source_db IN ?", []string{"lx", "hw", "yq"})
|
||||
if len(sources) > 0 {
|
||||
q = q.Where("p.source_db IN ?", sources)
|
||||
}
|
||||
if !scope.All || owners != nil {
|
||||
bindings := channelBindingQuery(db, scope, business.ChannelFilter{Sources: sources}).Select("b.source_db,b.platform_busier_id")
|
||||
if owners != nil {
|
||||
bindings = bindings.Where("b.user_id IN ?", owners)
|
||||
}
|
||||
q = q.Where("(p.source_db,p.busier_id) IN (?)", bindings)
|
||||
}
|
||||
return q
|
||||
}
|
||||
func performancePayQuery(db *gorm.DB, scope business.UserScope, owners []uint, sources []string, begin, end time.Time) *gorm.DB {
|
||||
keys := performanceChannels(db, scope, owners, sources).Select("p.source_db,p.id")
|
||||
return db.Table("tab_sw_report_day r").Where("r.dedup_mode=0 AND r.game_id>0 AND r.stat_date BETWEEN ? AND ?", begin.Format("2006-01-02"), end.Format("2006-01-02")).Where("(r.source_db,r.promote_id) IN (?)", keys)
|
||||
}
|
||||
func joinEffectiveness(q *gorm.DB) *gorm.DB {
|
||||
return q.Joins("LEFT JOIN sys_business_effectiveness_state es ON es.id=1").Joins("LEFT JOIN sys_business_channel_effectiveness e ON e.rule_key=es.rule_key AND e.source_db=p.source_db AND e.promote_id=p.id")
|
||||
}
|
||||
|
||||
type performanceChannelPO struct {
|
||||
ContributorID uint
|
||||
ContributorName string
|
||||
ID, BusierID uint64
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
Account, Nickname string
|
||||
CreateTime, QualifiedAt *int64
|
||||
WindowPay, PeriodPay string
|
||||
}
|
||||
|
||||
func performanceChannelToBiz(po performanceChannelPO) *business.PerformanceChannel {
|
||||
return &business.PerformanceChannel{ContributorID: po.ContributorID, ContributorName: po.ContributorName, ChannelKey: business.ChannelKey{ID: po.ID, SourceDB: po.SourceDB}, Account: po.Account, Name: po.Nickname, CreatedAt: po.CreateTime, QualifiedAt: po.QualifiedAt, WindowPay: po.WindowPay, PeriodPay: po.PeriodPay, OwnerIDs: []uint{}}
|
||||
}
|
||||
|
||||
type performanceOwnerPO struct {
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
PlatformBusierID uint64
|
||||
UserID uint
|
||||
Username, NickName string
|
||||
}
|
||||
|
||||
func performanceOwners(db *gorm.DB, scope business.UserScope, sources []string) (map[business.PlatformKey][]performanceOwnerPO, error) {
|
||||
var rows []performanceOwnerPO
|
||||
if err := channelBindingQuery(db, scope, business.ChannelFilter{Sources: sources}).Select("b.source_db,b.platform_busier_id,b.user_id,u.username,u.nick_name").Order("b.user_id").Scan(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
byKey := map[business.PlatformKey][]performanceOwnerPO{}
|
||||
for _, row := range rows {
|
||||
key := business.PlatformKey{SourceDB: row.SourceDB, BusierID: row.PlatformBusierID}
|
||||
byKey[key] = append(byKey[key], row)
|
||||
}
|
||||
return byKey, nil
|
||||
}
|
||||
func (r *performanceRepo) Facts(ctx context.Context, scope business.UserScope, ownerIDs []uint, sources []string, begin, end time.Time) (*business.PerformanceFacts, error) {
|
||||
var out *business.PerformanceFacts
|
||||
err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var err error
|
||||
out, err = readPerformanceFacts(tx, scope, ownerIDs, sources, begin, end)
|
||||
return err
|
||||
})
|
||||
return out, storageError(err)
|
||||
}
|
||||
func readPerformanceFacts(db *gorm.DB, scope business.UserScope, ownerIDs []uint, sources []string, begin, end time.Time) (*business.PerformanceFacts, error) {
|
||||
out := &business.PerformanceFacts{Channels: []*business.PerformanceChannel{}, Cash: []business.PerformanceCash{}, Available: true}
|
||||
available, err := businessTablesAvailable(db, "tab_promote", "tab_sw_report_day")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out.Available = available
|
||||
if !available {
|
||||
return out, nil
|
||||
}
|
||||
owners, err := performanceOwners(db, scope, sources)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pos []performanceChannelPO
|
||||
start, finish := begin.Unix(), min(end.AddDate(0, 0, 1).Unix(), time.Now().Unix()+1)
|
||||
channelFacts := joinEffectiveness(performanceChannels(db, scope, ownerIDs, sources)).Where("(p.create_time>=? AND p.create_time<?) OR (e.qualified_at>=? AND e.qualified_at<?)", start, finish, start, finish)
|
||||
if err := channelFacts.Select("p.id,p.source_db,p.busier_id,p.account,p.nickname,p.create_time,e.qualified_at,COALESCE(e.window_pay,0) AS window_pay").Scan(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, po := range pos {
|
||||
channel := performanceChannelToBiz(po)
|
||||
for _, owner := range owners[business.PlatformKey{SourceDB: po.SourceDB, BusierID: po.BusierID}] {
|
||||
channel.OwnerIDs = append(channel.OwnerIDs, owner.UserID)
|
||||
if channel.OwnerID == 0 {
|
||||
channel.OwnerID = owner.UserID
|
||||
channel.OwnerName = owner.NickName
|
||||
if channel.OwnerName == "" {
|
||||
channel.OwnerName = owner.Username
|
||||
}
|
||||
}
|
||||
}
|
||||
out.Channels = append(out.Channels, channel)
|
||||
}
|
||||
if begin.After(end) {
|
||||
return out, nil
|
||||
}
|
||||
var cash []struct {
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
BusierID uint64
|
||||
StatDate, Pay string
|
||||
}
|
||||
// Aggregate once per SDK business/date, not per channel/date. All channels
|
||||
// under the same SDK owner have the same business-tree attribution. This
|
||||
// keeps the exact totals while avoiding hundreds of thousands of Go rows.
|
||||
visibleChannels := performanceChannels(db, scope, ownerIDs, sources).Select("p.source_db,p.id,p.busier_id")
|
||||
if err := db.Table("tab_sw_report_day r").Joins("JOIN (?) owner_p ON owner_p.source_db=r.source_db AND owner_p.id=r.promote_id", visibleChannels).Where("r.dedup_mode=0 AND r.game_id>0 AND r.stat_date BETWEEN ? AND ?", begin.Format("2006-01-02"), end.Format("2006-01-02")).Select("r.source_db,owner_p.busier_id,CAST(r.stat_date AS CHAR) AS stat_date,CAST(SUM(r.total_pay) AS CHAR) AS pay").Group("r.source_db,owner_p.busier_id,r.stat_date").Scan(&cash).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range cash {
|
||||
ids := []uint{}
|
||||
for _, owner := range owners[business.PlatformKey{SourceDB: row.SourceDB, BusierID: row.BusierID}] {
|
||||
ids = append(ids, owner.UserID)
|
||||
}
|
||||
out.Cash = append(out.Cash, business.PerformanceCash{ChannelKey: business.ChannelKey{SourceDB: row.SourceDB, ID: row.BusierID}, OwnerIDs: ids, Date: row.StatDate, Pay: row.Pay})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (r *performanceRepo) Details(ctx context.Context, scope business.UserScope, selection business.PerformanceDetailSelection, sources []string, begin, end time.Time, metric string, offset, limit int) ([]*business.PerformanceChannel, int64, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
owners := selection.OwnerIDs
|
||||
pay := performancePayQuery(db, scope, owners, sources, begin, end).Select("r.source_db,r.promote_id,SUM(r.total_pay) AS pay").Group("r.source_db,r.promote_id")
|
||||
needsPay := metric == "pay" || metric == "all"
|
||||
q := joinEffectiveness(performanceChannels(db, scope, owners, sources))
|
||||
contributorSelect, contributorOrder := "", ""
|
||||
if len(selection.ContributorIDs) > 0 {
|
||||
// Keep one row per member/platform owner, even if bindings overlap.
|
||||
prefix := "(COALESCE(NULLIF(TRIM(c.id_path),''),'0') || ',' || c.id)"
|
||||
pattern := "(" + prefix + " || ',%')"
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
prefix = "CONCAT(COALESCE(NULLIF(TRIM(c.id_path),''),'0'),',',c.id)"
|
||||
pattern = "CONCAT(" + prefix + ",',%')"
|
||||
}
|
||||
contributors := channelBindingQuery(db, scope, business.ChannelFilter{Sources: sources}).
|
||||
Joins("JOIN sys_users c ON c.deleted_at IS NULL AND (c.id=u.id OR c.id=u.parent_id OR u.id_path="+prefix+" OR u.id_path LIKE "+pattern+")").
|
||||
Where("c.id IN ? AND u.deleted_at IS NULL", selection.ContributorIDs).
|
||||
Select("DISTINCT b.source_db,b.platform_busier_id,c.id AS contributor_id,COALESCE(NULLIF(c.nick_name,''),c.username) AS contributor_name")
|
||||
if owners != nil {
|
||||
contributors = contributors.Where("b.user_id IN ?", owners)
|
||||
}
|
||||
q = q.Joins("JOIN (?) contribution ON contribution.source_db=p.source_db AND contribution.platform_busier_id=p.busier_id", contributors)
|
||||
contributorSelect = ",contribution.contributor_id,contribution.contributor_name"
|
||||
contributorOrder = ",contribution.contributor_id"
|
||||
}
|
||||
if needsPay {
|
||||
q = q.Joins("LEFT JOIN (?) pp ON pp.source_db=p.source_db AND pp.promote_id=p.id", pay)
|
||||
}
|
||||
start, finish := begin.Unix(), min(end.AddDate(0, 0, 1).Unix(), time.Now().Unix()+1)
|
||||
switch metric {
|
||||
case "open":
|
||||
q = q.Where("p.create_time>=? AND p.create_time<?", start, finish)
|
||||
case "effective":
|
||||
q = q.Where("e.qualified_at>=? AND e.qualified_at<?", start, finish)
|
||||
case "pay":
|
||||
q = q.Where("COALESCE(pp.pay,0)<>0")
|
||||
default:
|
||||
q = q.Where("(p.create_time>=? AND p.create_time<?) OR (e.qualified_at>=? AND e.qualified_at<?) OR COALESCE(pp.pay,0)<>0", start, finish, start, finish)
|
||||
}
|
||||
var total int64
|
||||
if err := q.Count(&total).Error; err != nil {
|
||||
return nil, 0, business.ErrStorage
|
||||
}
|
||||
var pos []performanceChannelPO
|
||||
paySelect := "'0.00' AS period_pay"
|
||||
if needsPay {
|
||||
paySelect = "CAST(COALESCE(pp.pay,0) AS CHAR) AS period_pay"
|
||||
}
|
||||
if err := q.Select("p.id,p.source_db,p.busier_id,p.account,p.nickname,p.create_time,e.qualified_at,CAST(COALESCE(e.window_pay,0) AS CHAR) AS window_pay," + paySelect + contributorSelect).Order("p.create_time DESC,p.source_db,p.id" + contributorOrder).Offset(offset).Limit(limit).Scan(&pos).Error; err != nil {
|
||||
return nil, 0, business.ErrStorage
|
||||
}
|
||||
if !needsPay && len(pos) > 0 {
|
||||
keys := [][]any{}
|
||||
for _, po := range pos {
|
||||
keys = append(keys, []any{po.SourceDB, po.ID})
|
||||
}
|
||||
var amounts []struct {
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
PromoteID uint64
|
||||
Pay string
|
||||
}
|
||||
if err := db.Table("tab_sw_report_day r").Where("r.dedup_mode=0 AND r.game_id>0 AND r.stat_date BETWEEN ? AND ? AND (r.source_db,r.promote_id) IN ?", begin.Format("2006-01-02"), end.Format("2006-01-02"), keys).Select("r.source_db,r.promote_id,CAST(SUM(r.total_pay) AS CHAR) AS pay").Group("r.source_db,r.promote_id").Scan(&amounts).Error; err != nil {
|
||||
return nil, 0, business.ErrStorage
|
||||
}
|
||||
byKey := map[business.ChannelKey]string{}
|
||||
for _, a := range amounts {
|
||||
byKey[business.ChannelKey{SourceDB: a.SourceDB, ID: a.PromoteID}] = a.Pay
|
||||
}
|
||||
for i := range pos {
|
||||
if value, ok := byKey[business.ChannelKey{SourceDB: pos[i].SourceDB, ID: pos[i].ID}]; ok {
|
||||
pos[i].PeriodPay = value
|
||||
}
|
||||
}
|
||||
}
|
||||
bindings, err := performanceOwners(db, scope, sources)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
rows := make([]*business.PerformanceChannel, 0, len(pos))
|
||||
for _, po := range pos {
|
||||
row := performanceChannelToBiz(po)
|
||||
for _, owner := range bindings[business.PlatformKey{SourceDB: po.SourceDB, BusierID: po.BusierID}] {
|
||||
row.OwnerIDs = append(row.OwnerIDs, owner.UserID)
|
||||
if row.OwnerID == 0 {
|
||||
row.OwnerID = owner.UserID
|
||||
row.OwnerName = owner.NickName
|
||||
if row.OwnerName == "" {
|
||||
row.OwnerName = owner.Username
|
||||
}
|
||||
}
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
return rows, total, nil
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type goalRepo struct{ data Provider }
|
||||
|
||||
func NewGoalRepo(data Provider) business.GoalRepo { return &goalRepo{data: data} }
|
||||
func (r *goalRepo) List(ctx context.Context, begin, end string) ([]*business.PerformanceGoal, error) {
|
||||
var rows []performanceGoalPO
|
||||
if err := r.data.DB().WithContext(ctx).Where("month BETWEEN ? AND ?", begin, end).Order("month DESC,kind,target_id").Find(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
out := make([]*business.PerformanceGoal, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
out = append(out, &business.PerformanceGoal{Kind: r.Kind, TargetID: r.TargetID, Month: r.Month, Open: r.Open, Effective: r.Effective, Amount: r.Amount, UpdatedBy: r.UpdatedBy, UpdatedAt: r.UpdatedAt})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (r *goalRepo) Save(ctx context.Context, g *business.PerformanceGoal) error {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
if g.Open == nil && g.Effective == nil && g.Amount == nil {
|
||||
return storageError(db.Where("kind=? AND target_id=? AND month=?", g.Kind, g.TargetID, g.Month).Delete(&performanceGoalPO{}).Error)
|
||||
}
|
||||
po := performanceGoalPO{Kind: g.Kind, TargetID: g.TargetID, Month: g.Month, Open: g.Open, Effective: g.Effective, Amount: g.Amount, UpdatedBy: g.UpdatedBy, UpdatedAt: time.Now()}
|
||||
return storageError(db.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "kind"}, {Name: "target_id"}, {Name: "month"}}, DoUpdates: clause.AssignmentColumns([]string{"open", "effective", "amount", "updated_by", "updated_at"})}).Create(&po).Error)
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type teamRepo struct{ data Provider }
|
||||
|
||||
func NewTeamRepo(data Provider) business.TeamRepo { return &teamRepo{data: data} }
|
||||
|
||||
func (r *teamRepo) Members(ctx context.Context, scope business.UserScope) ([]*business.BusinessMember, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
var rows []struct {
|
||||
Account accountPO `gorm:"embedded"`
|
||||
HireDate, Employment string
|
||||
GroupID uint64
|
||||
}
|
||||
q := applyUserScope(db.Table("sys_users u").Where("u.deleted_at IS NULL AND u.level BETWEEN 1 AND 3"), scope, "u.")
|
||||
if err := q.Select("u.*,p.username AS parent_user_name,mp.hire_date,mp.employment,gm.group_id").Joins("LEFT JOIN sys_users p ON p.id=u.parent_id AND p.deleted_at IS NULL").Joins("LEFT JOIN sys_business_member_profiles mp ON mp.user_id=u.id").Joins("LEFT JOIN sys_business_group_members gm ON gm.user_id=u.id").Order("u.id").Scan(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
ids := make([]uint, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
ids = append(ids, row.Account.ID)
|
||||
}
|
||||
counts, err := (&accountRepo{data: r.data}).DirectCounts(ctx, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]*business.BusinessMember, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
a := toBizAccount(row.Account)
|
||||
a.Password = ""
|
||||
a.PromoteCount = counts[a.ID].PromoteCount
|
||||
a.BindCount = counts[a.ID].BindCount
|
||||
employment := row.Employment
|
||||
if employment == "" {
|
||||
employment = "employed"
|
||||
if a.Status != 0 {
|
||||
employment = "departed"
|
||||
}
|
||||
}
|
||||
a.HireDate, a.Employment = row.HireDate, employment
|
||||
out = append(out, &business.BusinessMember{Account: *a, GroupID: row.GroupID})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func readTeamCatalog(db *gorm.DB) (*business.TeamCatalog, error) {
|
||||
var state businessGroupStatePO
|
||||
if err := db.First(&state, "id=1").Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
var groups []businessGroupPO
|
||||
if err := db.Order("sort,id").Find(&groups).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
var members []businessGroupMemberPO
|
||||
if err := db.Order("user_id").Find(&members).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
byGroup := map[uint64][]uint{}
|
||||
for _, m := range members {
|
||||
byGroup[m.GroupID] = append(byGroup[m.GroupID], m.UserID)
|
||||
}
|
||||
out := &business.TeamCatalog{Revision: state.Revision, Groups: []*business.BusinessGroup{}}
|
||||
for _, g := range groups {
|
||||
out.Groups = append(out.Groups, &business.BusinessGroup{ID: g.ID, Name: g.Name, OwnerID: g.OwnerID, Sort: g.Sort, MemberIDs: append([]uint{}, byGroup[g.ID]...)})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (r *teamRepo) Catalog(ctx context.Context) (*business.TeamCatalog, error) {
|
||||
var value *business.TeamCatalog
|
||||
err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { var err error; value, err = readTeamCatalog(tx); return err })
|
||||
return value, storageError(err)
|
||||
}
|
||||
func (r *teamRepo) SaveCatalog(ctx context.Context, revision uint64, groups []*business.BusinessGroup) (*business.TeamCatalog, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
err := db.Transaction(func(tx *gorm.DB) error {
|
||||
updated := tx.Model(&businessGroupStatePO{}).Where("id=1 AND revision=?", revision).Update("revision", gorm.Expr("revision+1"))
|
||||
if updated.Error != nil {
|
||||
return updated.Error
|
||||
}
|
||||
if updated.RowsAffected != 1 {
|
||||
return business.ErrTeamConflict
|
||||
}
|
||||
var current []businessGroupPO
|
||||
if err := tx.Find(¤t).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
byID := map[uint64]businessGroupPO{}
|
||||
for _, g := range current {
|
||||
byID[g.ID] = g
|
||||
}
|
||||
ids := []uint64{}
|
||||
members := []businessGroupMemberPO{}
|
||||
for _, g := range groups {
|
||||
po := businessGroupPO{ID: g.ID, Name: g.Name, OwnerID: g.OwnerID, Sort: g.Sort}
|
||||
if g.ID == 0 {
|
||||
if err := tx.Create(&po).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
old, exists := byID[g.ID]
|
||||
if !exists {
|
||||
return business.ErrTeamConflict
|
||||
}
|
||||
if old.Name != g.Name || old.Sort != g.Sort {
|
||||
if err := tx.Model(&businessGroupPO{}).Where("id=?", g.ID).Updates(map[string]any{"name": g.Name, "sort": g.Sort, "updated_at": time.Now()}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
ids = append(ids, po.ID)
|
||||
for _, id := range g.MemberIDs {
|
||||
members = append(members, businessGroupMemberPO{UserID: id, GroupID: po.ID})
|
||||
}
|
||||
}
|
||||
deleted := tx.Model(&businessGroupPO{})
|
||||
if len(ids) > 0 {
|
||||
deleted = deleted.Where("id NOT IN ?", ids)
|
||||
} else {
|
||||
deleted = deleted.Where("1=1")
|
||||
}
|
||||
if err := deleted.Delete(&businessGroupPO{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Where("1=1").Delete(&businessGroupMemberPO{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(members) > 0 {
|
||||
if err := tx.CreateInBatches(&members, 300).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, storageError(err)
|
||||
}
|
||||
return r.Catalog(ctx)
|
||||
}
|
||||
func (r *teamRepo) UpdateProfile(ctx context.Context, p business.MemberProfile) error {
|
||||
return storageError(r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var account accountPO
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&account, p.ID).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
domain := toBizAccount(account)
|
||||
domain.Phone = p.Phone
|
||||
if err := checkAccountUnique(tx, domain); err != nil {
|
||||
return err
|
||||
}
|
||||
status, enable := 0, 1
|
||||
if p.Employment == "departed" {
|
||||
status, enable = 1, 0
|
||||
}
|
||||
values := map[string]any{"real_name": p.RealName, "phone": p.Phone, "business_status": status, "enable": enable, "updated_at": time.Now()}
|
||||
if p.Password != "" {
|
||||
values["password"] = p.Password
|
||||
values["password_updated_at"] = time.Now()
|
||||
}
|
||||
if err := tx.Model(&accountPO{}).Where("id=?", p.ID).Updates(values).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "user_id"}}, DoUpdates: clause.AssignmentColumns([]string{"hire_date", "employment", "updated_at"})}).Create(&businessMemberProfilePO{UserID: p.ID, HireDate: p.HireDate, Employment: p.Employment, UpdatedAt: time.Now()}).Error
|
||||
}))
|
||||
}
|
||||
@@ -30,6 +30,11 @@ var ProviderSet = wire.NewSet(
|
||||
wire.Bind(new(databusiness.PlatformProvider), new(*Data)),
|
||||
databusiness.NewUserPreferenceRepo,
|
||||
databusiness.NewOtherSettingsRepo,
|
||||
databusiness.NewTeamRepo,
|
||||
databusiness.NewGoalRepo,
|
||||
databusiness.NewEffectivenessRepo,
|
||||
databusiness.NewPerformanceRepo,
|
||||
databusiness.NewDashboardRepo,
|
||||
databusiness.NewAccountRepo,
|
||||
databusiness.NewBindingRepo,
|
||||
databusiness.NewPlatformAccountRepo,
|
||||
|
||||
Reference in New Issue
Block a user