yvan/kra-new
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: align business reports and record task migration progress

Browse Source
This commit is contained in:
Yvan
2026-11-06 00:47:06 +08:00
parent 1bc2082718
commit b1f14bd525
32 changed files with 501 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/biz/business/alert_governance.go
+6 -2
View File
@@ -53,11 +53,15 @@ func AlertLossAt(now time.Time) AlertLossWindow {
return AlertLossWindow{LoginBefore: now.AddDate(0, 0, -15).Unix(), PayBegin: now.AddDate(0, 0, -30).Unix(), PayEnd: now.Unix(), MinPay: "500"}
}
func (w AlertLossWindow) IsLost(lastLogin, created int64, pay string) (bool, error) {
func (w AlertLossWindow) LoginExpired(lastLogin, created int64) bool {
if lastLogin <= 0 {
lastLogin = created
}
if lastLogin > 0 && lastLogin < w.LoginBefore {
return lastLogin > 0 && lastLogin < w.LoginBefore
}
func (w AlertLossWindow) IsLost(lastLogin, created int64, pay string) (bool, error) {
if w.LoginExpired(lastLogin, created) {
return true, nil
}
amount, err := portraitNumber(pay)
internal/biz/business/alert_trigger.go
+5
View File
@@ -3,6 +3,7 @@ package business
import (
"context"
"fmt"
"kra/pkg/task"
"log/slog"
"math/big"
"sort"
@@ -152,6 +153,10 @@ func (uc *AlertTriggerUsecase) Detect(ctx context.Context, now time.Time) (*Aler
if input == nil {
return result, ErrStorage
}
task.Logf(ctx, "预警检测：规则=%d；候选一级渠道=%d；商务账号=%d", len(input.Rules), len(input.Channels), len(input.Accounts))
defer func() {
task.Logf(ctx, "预警检测结果：已检测规则=%d；跳过规则=%d；写入预警=%d", result.Rules, result.Skipped, result.Written)
}()
zone := time.FixedZone("Asia/Shanghai", 8*3600)
now = now.In(zone)
today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, zone)
internal/biz/business/analytics.go
+28 -3
View File
@@ -27,6 +27,7 @@ type AnalyticsKey struct {
ID int64
}
type AnalyticsFilter struct {
HasRegister, HasPay *bool
Sources, GameNames []string
PromoteName string
PromoteType *int
@@ -47,6 +48,22 @@ type AnalyticsRow struct {
Pay, PayRate, ARPU, ARPPU string
}
func (f AnalyticsFilter) MatchesMetrics(row *AnalyticsRow) (bool, error) {
if f.HasRegister != nil && (row.Counts.Register > 0) != *f.HasRegister {
return false, nil
}
if f.HasPay != nil {
amount, err := portraitNumber(row.Pay)
if err != nil {
return false, err
}
if (amount.Sign() > 0) != *f.HasPay {
return false, nil
}
}
return true, nil
}
func (r *AnalyticsRow) BusinessName() string {
if value := strings.TrimSpace(r.BindingAccount); value != "" {
return value
@@ -135,12 +152,20 @@ func (uc *AnalyticsUsecase) list(ctx context.Context, kind AnalyticsKind, filter
if kind == GameAnalytics {
filter.PromoteName = ""
filter.PromoteType = nil
filter.HasRegister, filter.HasPay = nil, nil
}
filter.StartDate, filter.EndDate = reportDateDefaults(filter.StartDate, filter.EndDate, 30, uc.now())
if filter.EndDate.Before(*filter.StartDate) {
if kind == GameAnalytics {
filter.StartDate, filter.EndDate = reportDateDefaults(filter.StartDate, filter.EndDate, 1, uc.now())
} else if filter.EndDate == nil {
_, filter.EndDate = reportDateDefaults(nil, nil, 1, uc.now())
}
if filter.StartDate != nil && filter.EndDate.Before(*filter.StartDate) {
return nil, ErrAnalyticsDateOrder
}
filter.Begin = filter.StartDate.Format("2006-01-02")
filter.Begin = ""
if filter.StartDate != nil {
filter.Begin = filter.StartDate.Format("2006-01-02")
}
filter.End = filter.EndDate.Format("2006-01-02")
out := &AnalyticsPage{Rows: []*AnalyticsRow{}, Begin: filter.Begin, End: filter.End, Available: true}
scope := AnalyticsScope{Viewer: viewer, IncludeSub: pref.IncludeSubPromote, IDCardDedup: pref.IDCardDedup}
internal/biz/business/analytics_test.go
+4 -4
View File
@@ -63,8 +63,8 @@ func TestGameAnalyticsSummaryUsesAllRowsAndCorrectDenominators(t *testing.T) {
if len(page.Rows) != 1 || page.Total != 2 || page.Rows[0].ID != 2 || page.Summary.Counts.Register != 5 || page.Summary.Pay != "50.00" || page.Summary.PayRate != "30.00" || page.Summary.ARPU != "5.00" || page.Summary.ARPPU != "16.67" {
t.Fatalf("游戏汇总、分页或分母错误: %+v", page)
}
if page.Begin != "2026-10-06" || page.End != "2026-11-04" {
t.Fatal("后台缺省统计日期应为含今天的近30天")
if page.Begin != "2026-11-04" || page.End != "2026-11-04" {
t.Fatal("游戏数据缺省统计日期应为当天")
}
}
func TestChannelAnalyticsSummaryDependsOnViewer(t *testing.T) {
@@ -90,8 +90,8 @@ func TestChannelAnalyticsAllPagesSummaryAndDefaultDateRange(t *testing.T) {
if err != nil || page.Total != 2 || page.Summary.Counts.Register != 5 || page.Summary.Counts.Active != 10 || page.Summary.Counts.Pay != 3 || page.Summary.Counts.NewPay != 2 || page.Summary.Pay != "50.00" || page.Summary.PayRate != "30.00" || page.EffectiveCount != 3 {
t.Fatalf("翻页不能改变完整合计: page=%d %+v %v", pageNum, page, err)
}
if page.Begin != "2026-10-06" || page.End != "2026-11-04" || repo.filter.StartDate == nil {
t.Fatalf("渠道统计缺省应为含今天的近30天: %+v", repo.filter)
if page.Begin != "" || page.End != "2026-11-04" || repo.filter.StartDate != nil {
t.Fatalf("渠道统计开始留空不应截断历史数据: %+v", repo.filter)
}
}
for _, sample := range []struct {
internal/biz/business/effectiveness.go
+15 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"crypto/sha256"
"fmt"
"kra/pkg/task"
"math"
"math/big"
"strconv"
@@ -98,6 +99,7 @@ func (uc *EffectivenessUsecase) rebuild(ctx context.Context, rule OtherSettings,
if err != nil {
return nil, err
}
task.Logf(ctx, "有效端计算：全量=%t；注册后 %d 天内满 %d 元；待计算渠道=%d", full, rule.Days, rule.Amount, len(candidates))
type accumulator struct {
candidate *EffectivenessCandidate
sum *big.Rat
@@ -120,7 +122,9 @@ func (uc *EffectivenessUsecase) rebuild(ctx context.Context, rule OtherSettings,
}
values[c.ChannelKey] = a
}
charges := 0
err = uc.repo.Charges(ctx, candidates, duration, now, full, func(charge EffectivenessCharge) error {
charges++
a := values[charge.ChannelKey]
if a == nil {
return nil
@@ -144,5 +148,15 @@ func (uc *EffectivenessUsecase) rebuild(ctx context.Context, rule OtherSettings,
a := values[c.ChannelKey]
results = append(results, EffectivenessResult{ChannelKey: c.ChannelKey, QualifiedAt: a.at, WindowPay: portraitDecimal(a.sum, 2), Complete: a.complete})
}
return uc.repo.Publish(ctx, state, rule, key, results, save)
updated, err := uc.repo.Publish(ctx, state, rule, key, results, save)
if err == nil {
qualified := 0
for _, result := range results {
if result.QualifiedAt != nil {
qualified++
}
}
task.Logf(ctx, "有效端结果已提交；扫描流水=%d；更新渠道=%d；本批达标渠道=%d", charges, len(results), qualified)
}
return updated, err
}
internal/biz/business/management_test.go
+12 -7
View File
@@ -172,7 +172,7 @@ func TestPerformanceRateSortKeepsUndefinedRatesLast(t *testing.T) {
}
}
func TestPerformanceMatrixTotalsKeepVisibleScopeWhenRowsAreFiltered(t *testing.T) {
func TestPerformanceMatrixTotalsDeduplicateDisplayedMembers(t *testing.T) {
_, uc, _, teams, _ := managementFixture()
teams.catalog.Groups[0].MemberIDs = []uint{8, 9, 10}
ctx := accountContext(7, 901)
@@ -181,22 +181,27 @@ func TestPerformanceMatrixTotalsKeepVisibleScopeWhenRowsAreFiltered(t *testing.T
if err != nil {
t.Fatal(err)
}
if len(page.Rows) != 1 || page.MatrixTotal != page.Total || page.MatrixTotal.Open != 2 || page.Groups[0].UniqueTotal.Open != 1 {
t.Fatalf("matrix total must retain the full visible scope, with a filtered group subtotal: %+v", page)
if len(page.Rows) != 1 || page.MatrixTotal.Open != 1 || page.Groups[0].UniqueTotal.Open != 1 {
t.Fatalf("matrix and group subtotal must reflect the displayed member scope: %+v", page)
}
if page.Total.Open != 2 || page.Groups[0].Total.Open != 3 {
t.Fatal("scope cards and team contributions must retain their own scope")
}
f.Name, f.GroupID = "", func() *uint64 { id := uint64(999); return &id }()
page, err = uc.Query(ctx, f)
if err != nil || len(page.Rows) != 0 || page.MatrixTotal != page.Total || page.Total.Open != 2 {
if err != nil || len(page.Rows) != 0 || page.MatrixTotal.Open != 0 || page.MatrixTotal.Effective != 0 || page.MatrixTotal.Pay != "0.00" || page.Total.Open != 2 {
t.Fatalf("empty displayed rows must not erase visible-scope totals: %+v %v", page, err)
}
for period, total := range page.Trend {
if page.MatrixTrend[period] != total {
t.Fatalf("period total changed with filtering: %s", period)
for period, total := range page.MatrixTrend {
if total.Open != 0 || total.Effective != 0 || total.Pay != "0.00" {
t.Fatalf("empty matrix retained period values: %s", period)
}
}
f.GroupID = nil
page, err = uc.Query(ctx, f)
if err != nil || page.MatrixTotal != page.Total || page.MatrixTotal.Open != 2 {
t.Fatalf("overlapping ancestors must contribute only once: %+v %v", page, err)
}
}
func TestBusinessManagementIncludesCurrentGoalWithoutCurrentPerformance(t *testing.T) {
internal/biz/business/performance.go
+11 -2
View File
@@ -173,6 +173,7 @@ func (uc *PerformanceUsecase) query(ctx context.Context, f PerformanceFilter, ow
return out, nil
}
all := newPerformanceBuckets(periods)
matrix := newPerformanceBuckets(periods)
needle := strings.ToLower(strings.TrimSpace(f.Name))
visible := func(m *BusinessMember) bool {
return (f.Level == nil || m.Level == *f.Level) && (needle == "" || strings.Contains(strings.ToLower(memberName(m)+" "+m.Username+" "+m.RealName), needle))
@@ -259,6 +260,14 @@ func (uc *PerformanceUsecase) query(ctx context.Context, f PerformanceFilter, ow
return
}
all.add(key, metric, pay)
// The matrix is the union of the displayed members' descendant scopes;
// overlapping parent/child rows must not count the same channel twice.
for _, id := range a.members {
if member := byID[id]; member != nil && matrixVisible(member) {
matrix.add(key, metric, pay)
break
}
}
for _, id := range a.members {
memberBuckets[id].add(key, metric, pay)
}
@@ -361,9 +370,9 @@ func (uc *PerformanceUsecase) query(ctx context.Context, f PerformanceFilter, ow
}
}
out.Total = all.total.metrics()
out.MatrixTotal = out.Total
out.MatrixTotal = matrix.total.metrics()
out.MatrixRate = performanceRate(out.MatrixTotal)
out.MatrixTrend = all.values()
out.MatrixTrend = matrix.values()
out.Prior = prior.metrics()
out.Rate = performanceRate(out.Total)
out.Trend = all.values()
internal/biz/business/report_build.go
+24 -1
View File
@@ -2,6 +2,7 @@ package business
import (
"context"
"kra/pkg/task"
"math/big"
"sort"
"strconv"
@@ -103,13 +104,35 @@ func (uc *ReportBuildUsecase) Run(ctx context.Context, kind ReportBuildKind, per
if finish.Before(begin) {
return ErrAnalyticsDateOrder
}
task.Logf(ctx, "报表类型=%s；日期区间=%s 至 %s", kind, begin.Format("2006-01-02"), finish.Format("2006-01-02"))
completed, totalRows, totalKeys := 0, 0, 0
defer func() {
task.Logf(ctx, "已提交日期=%d；汇总行=%d；身份键=%d", completed, totalRows, totalKeys)
}()
for day := begin; !day.After(finish); day = day.AddDate(0, 0, 1) {
if err := ctx.Err(); err != nil {
return err
}
if err := uc.repo.Rebuild(ctx, kind, day, func(input *ReportInput) ([]*ReportBuiltRow, error) { return compileReport(kind, input) }); err != nil {
started := time.Now()
registers, actives, payments, rowCount, keyCount := 0, 0, 0, 0, 0
if err := uc.repo.Rebuild(ctx, kind, day, func(input *ReportInput) ([]*ReportBuiltRow, error) {
registers, actives, payments = len(input.Registers), len(input.Actives), len(input.Payments)
rows, err := compileReport(kind, input)
rowCount = len(rows)
for _, row := range rows {
for _, keys := range row.Keys {
keyCount += len(keys)
}
}
return rows, err
}); err != nil {
task.Logf(ctx, "日期=%s 未提交：%v", day.Format("2006-01-02"), err)
return err
}
completed++
totalRows += rowCount
totalKeys += keyCount
task.Logf(ctx, "日期=%s 已提交；原始注册=%d，活跃=%d，充值=%d；汇总行=%d，身份键=%d；耗时=%d ms", day.Format("2006-01-02"), registers, actives, payments, rowCount, keyCount, time.Since(started).Milliseconds())
if day.Before(finish) {
if err := uc.pause(ctx); err != nil {
return err
internal/biz/business/report_build_test.go
+25
View File
@@ -2,6 +2,9 @@ package business
import (
"context"
"errors"
"kra/pkg/task"
"strings"
"testing"
"time"
)
@@ -77,6 +80,28 @@ func TestReportBuildIdentityRulesAndAnonymousLogins(t *testing.T) {
type reportBuildFake struct{ days []string }
type failingReportCommit struct{}
func (failingReportCommit) Rebuild(_ context.Context, _ ReportBuildKind, _ time.Time, compile func(*ReportInput) ([]*ReportBuiltRow, error)) error {
_, err := compile(&ReportInput{Registers: []ReportEvent{{SourceDB: "lx", UserID: reportPtr(1), PromoteID: reportPtr(1), GameID: reportPtr(1)}}})
if err != nil {
return err
}
return errors.New("commit failed")
}
func TestReportBuildDetailsDoNotClaimRolledBackRows(t *testing.T) {
ctx, output := task.WithOutput(context.Background())
uc := NewReportBuildUsecase(failingReportCommit{})
if err := uc.Run(ctx, DailyReport, "backfill", "2026-11-03", "2026-11-03"); err == nil {
t.Fatal("commit error was lost")
}
text := output.String()
if !strings.Contains(text, "2026-11-03 未提交") || !strings.Contains(text, "已提交日期=0；汇总行=0；身份键=0") {
t.Fatalf("rollback was reported as success: %s", text)
}
}
func (r *reportBuildFake) Rebuild(_ context.Context, _ ReportBuildKind, day time.Time, _ func(*ReportInput) ([]*ReportBuiltRow, error)) error {
r.days = append(r.days, day.Format("2006-01-02"))
return nil
internal/biz/business/report_dates_test.go
+8 -5
View File
@@ -5,7 +5,7 @@ import (
"time"
)
func TestAnalyticsDefaultDatesBoundListAndExportQueries(t *testing.T) {
func TestAnalyticsDateRulesAgreeForListAndExport(t *testing.T) {
date := func(value string) *time.Time {
parsed, err := time.ParseInLocation("2006-01-02", value, portraitLocation)
if err != nil {
@@ -18,13 +18,16 @@ func TestAnalyticsDefaultDatesBoundListAndExportQueries(t *testing.T) {
name, start, end, wantStart, wantEnd string
now time.Time
}{
{"empty", "", "", "2026-10-06", "2026-11-04", time.Date(2026, 11, 4, 9, 0, 0, 0, portraitLocation)},
{"China day across UTC midnight", "", "", "2025-12-03", "2026-01-01", time.Date(2025, 12, 31, 17, 0, 0, 0, time.UTC)},
{"only end across leap day", "", "2024-03-01", "2024-02-01", "2024-03-01", time.Now()},
{"empty", "", "", "", "2026-11-04", time.Date(2026, 11, 4, 9, 0, 0, 0, portraitLocation)},
{"China day across UTC midnight", "", "", "", "2026-01-01", time.Date(2025, 12, 31, 17, 0, 0, 0, time.UTC)},
{"only end across leap day", "", "2024-03-01", "", "2024-03-01", time.Now()},
{"only start", "2026-01-01", "", "2026-01-01", "2026-11-04", time.Date(2026, 11, 4, 9, 0, 0, 0, portraitLocation)},
{"explicit historical year", "2023-01-01", "2023-12-31", "2023-01-01", "2023-12-31", time.Now()},
} {
t.Run(string(kind)+"/"+sample.name, func(t *testing.T) {
if kind == GameAnalytics && sample.wantStart == "" {
sample.wantStart = sample.wantEnd
}
uc, repo := analyticsFixture()
uc.now = func() time.Time { return sample.now }
filter := AnalyticsFilter{}
@@ -45,7 +48,7 @@ func TestAnalyticsDefaultDatesBoundListAndExportQueries(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if page.Begin != sample.wantStart || page.End != sample.wantEnd || repo.filter.Begin != page.Begin || repo.filter.End != page.End || repo.filter.StartDate == nil || repo.filter.EndDate == nil {
if page.Begin != sample.wantStart || page.End != sample.wantEnd || repo.filter.Begin != page.Begin || repo.filter.End != page.End || (repo.filter.StartDate == nil) != (sample.wantStart == "") || repo.filter.EndDate == nil {
t.Fatalf("export=%v expected %s..%s, got %+v", export, sample.wantStart, sample.wantEnd, repo.filter)
}
}
internal/data/business/alert_governance.go
+12 -2
View File
@@ -262,12 +262,22 @@ func loadAlertLoss(db *gorm.DB, keys []business.ChannelKey, window business.Aler
if err := db.Table("tab_promote").Select("id,last_login_time,create_time").Where("source_db=? AND id IN ?", source, batch).Scan(&channels).Error; err != nil {
return nil, business.ErrStorage
}
// Login expiry already establishes loss; only the remaining channels
// need the potentially expensive payment scan on a cold database.
paymentIDs := make([]uint64, 0, len(channels))
for _, channel := range channels {
if !window.LoginExpired(channel.LastLoginTime, channel.CreateTime) {
paymentIDs = append(paymentIDs, channel.ID)
}
}
var amounts []struct {
PromoteID uint64
RecentAmount string
}
if err := db.Table("tab_spend").Select("promote_id,COALESCE(SUM(pay_amount),0) AS recent_amount").Where("source_db=? AND promote_id IN ? AND pay_status=1 AND pay_time>=? AND pay_time<?", source, batch, window.PayBegin, window.PayEnd).Group("promote_id").Scan(&amounts).Error; err != nil {
return nil, business.ErrStorage
if len(paymentIDs) > 0 {
if err := db.Table("tab_spend").Select("promote_id,COALESCE(SUM(pay_amount),0) AS recent_amount").Where("source_db=? AND promote_id IN ? AND pay_status=1 AND pay_time>=? AND pay_time<?", source, paymentIDs, window.PayBegin, window.PayEnd).Group("promote_id").Scan(&amounts).Error; err != nil {
return nil, business.ErrStorage
}
}
payments := map[uint64]string{}
for _, amount := range amounts {
internal/data/business/alert_v2_test.go
+30
View File
@@ -2,6 +2,7 @@ package business
import (
"context"
"sync/atomic"
"testing"
"time"
@@ -9,6 +10,35 @@ import (
"kra/internal/biz/business"
)
func TestAlertLossSkipsPaymentReadsForExpiredLogins(t *testing.T) {
db := openTestDB(t)
now := prepareAlertV2(t, db)
if err := db.Exec("UPDATE tab_promote SET last_login_time=0,create_time=1").Error; err != nil {
t.Fatal(err)
}
var paymentReads atomic.Int32
if err := db.Callback().Row().Before("gorm:row").Register("count_loss_payment_reads", func(tx *gorm.DB) {
if tx.Statement.Table == "tab_spend" {
paymentReads.Add(1)
}
}); err != nil {
t.Fatal(err)
}
keys := []business.ChannelKey{{SourceDB: "lx", ID: 1}, {SourceDB: "hw", ID: 1}}
window := business.AlertLossAt(now)
states, err := loadAlertLoss(db, keys, window)
if err != nil || paymentReads.Load() != 0 || len(states) != 2 || !*states[keys[0]] || !*states[keys[1]] {
t.Fatalf("expired logins should establish loss without payment reads: %+v %v reads=%d", states, err, paymentReads.Load())
}
if err := db.Exec("UPDATE tab_promote SET last_login_time=? WHERE source_db='hw'", now.Unix()).Error; err != nil {
t.Fatal(err)
}
states, err = loadAlertLoss(db, keys, window)
if err != nil || paymentReads.Load() != 1 || !*states[keys[0]] || *states[keys[1]] {
t.Fatalf("active login must still check payments within its own platform: %+v %v reads=%d", states, err, paymentReads.Load())
}
}
func prepareAlertV2(t *testing.T, db *gorm.DB) time.Time {
t.Helper()
now := time.Date(2026, 11, 5, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
internal/data/business/analytics.go
+31 -1
View File
@@ -189,7 +189,8 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
}
return nil
})
if withGlobal {
metricFilter := kind == business.ChannelAnalytics && (f.HasRegister != nil || f.HasPay != nil)
if withGlobal && !metricFilter {
group.Go(func() error {
q := analyticsQuery(queryDB, scope, f, "tab_sw_report_day_key", mode)
if err := q.Select("r.metric,COUNT(DISTINCT " + portraitIdentity(queryDB, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&global).Error; err != nil {
@@ -226,6 +227,35 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
return nil, err
}
}
if metricFilter {
for key, row := range rowMap {
match, err := f.MatchesMetrics(row)
if err != nil {
return nil, err
}
if !match {
delete(rowMap, key)
}
}
if withGlobal {
keys := make([]business.AnalyticsKey, 0, len(rowMap))
for key := range rowMap {
keys = append(keys, key)
}
// Deduplicate identities across all matched channels, before pagination.
// Summing per-channel counts would double-count shared players.
if len(keys) > 0 {
q, id := analyticsRoot(db, kind, scope, f, "tab_sw_report_day_key", mode, false)
if err := q.Where("(r.source_db,"+id+") IN ?", analyticsPairs(keys)).Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&global).Error; err != nil {
return nil, business.ErrStorage
}
q, id = analyticsRoot(db, kind, scope, f, "tab_sw_report_day", 0, true)
if err := q.Where("(r.source_db,"+id+") IN ?", analyticsPairs(keys)).Select("COALESCE(SUM(r.total_pay),0)").Scan(&globalPay).Error; err != nil {
return nil, business.ErrStorage
}
}
}
}
if withGlobal {
totals := toPortraitCounts(global)
result.GlobalCounts, result.GlobalPay = &totals, globalPay
internal/data/business/analytics_test.go
+26
View File
@@ -61,6 +61,32 @@ func TestAnalyticsChannelTreeAndAdminOfficialRows(t *testing.T) {
}
}
func TestAnalyticsMetricFiltersPreserveGlobalDedupAndEffectiveCount(t *testing.T) {
db, repo, scope, filter := prepareAnalyticsData(t)
scope.Viewer = business.UserScope{All: true}
yes, no := true, false
filter.HasRegister, filter.HasPay = &yes, &yes
// The same real identity occurs under different root channels/platforms.
// All rows match, so the filtered total must keep the original global count.
data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || len(data.Rows) != 5 || data.GlobalCounts.Register != 4 || portraitTestMoney(data.GlobalPay) != "2070.60" {
t.Fatalf("filtered summary lost global deduplication: %+v %v", data, err)
}
if err := db.Exec("UPDATE tab_sw_report_day SET total_pay=0 WHERE source_db='lx' AND promote_id IN (100,101,102)").Error; err != nil {
t.Fatal(err)
}
filter.HasPay = &no
data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || len(data.Rows) != 1 || data.Rows[0].ID != 100 || portraitTestMoney(data.GlobalPay) != "0.00" || data.GlobalCounts.Register != 2 {
t.Fatalf("no-payment filter must keep the whole root scope and matching totals: %+v %v", data, err)
}
filter.HasRegister = &no
data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || len(data.Rows) != 0 || data.GlobalCounts.Register != 0 || data.EffectiveCount != 0 {
t.Fatalf("combined filters must use AND and return empty totals: %+v %v", data, err)
}
}
func TestAnalyticsRowsOnlyComeFromSelectedPeriod(t *testing.T) {
db, repo, scope, filter := prepareAnalyticsData(t)
for _, sql := range []string{
internal/data/business/approved_games.go
+1 -7
View File
@@ -6,13 +6,7 @@ import (
)
func approvedGamesAvailable(db *gorm.DB) (bool, error) {
for _, table := range []string{"tab_game", "tab_promote", "tab_promote_apply"} {
available, err := businessTableAvailable(db, table)
if err != nil || !available {
return available, err
}
}
return true, nil
return businessTablesAvailable(db, "tab_game", "tab_promote", "tab_promote_apply")
}
// 游戏可选范围来自可见渠道的有效申请，不受统计时是否包含下级的开关影响。
internal/service/business/analytics.go
+1
View File
@@ -26,6 +26,7 @@ func analyticsRequest(req *dto.AnalyticsQuery) (business.AnalyticsFilter, error)
return business.AnalyticsFilter{}, err
}
f := business.AnalyticsFilter{Sources: sources, GameNames: splitQueryValues(req.GameNames), PromoteName: strings.TrimSpace(req.PromoteName), PromoteType: req.PromoteType, SortField: strings.TrimSpace(req.SortField), Ascending: strings.EqualFold(strings.TrimSpace(req.SortOrder), "asc")}
f.HasRegister, f.HasPay = req.HasRegister, req.HasPay
begin, end := strings.TrimSpace(req.BeginTime), strings.TrimSpace(req.EndTime)
location := time.FixedZone("Asia/Shanghai", 8*3600)
if begin != "" {
internal/service/dto/analytics.go
+2
View File
@@ -8,6 +8,8 @@ type AnalyticsQuery struct {
GameNames []string `form:"gameName"`
PromoteType *int `form:"promoteType"`
PromoteName string `form:"promoteName"`
HasRegister *bool `form:"hasRegister"`
HasPay *bool `form:"hasPay"`
BeginTime string `form:"beginTime"`
EndTime string `form:"endTime"`
SortField string `form:"sortField"`
internal/worker/task_executor.go
+9 -1
View File
@@ -216,7 +216,11 @@ func (e *TaskExecutor) runMethod(ctx context.Context, task *taskbiz.TimedTask) e
return err
}
for attempt := 0; attempt <= options.MaxRetryTimes; attempt++ {
platformtask.Logf(ctx, "开始第 %d 次执行；单次超时 %s，最多重试 %d 次", attempt+1, options.Timeout, options.MaxRetryTimes)
err = e.runMethodAttempt(ctx, task, method, options.Timeout)
if err != nil {
platformtask.Logf(ctx, "第 %d 次执行失败：%s", attempt+1, truncateTaskText(err.Error()))
}
if ctxErr := ctx.Err(); ctxErr != nil {
if errors.Is(ctxErr, context.DeadlineExceeded) {
return errTaskTimeout
@@ -278,6 +282,7 @@ func (e *TaskExecutor) Run(ctx context.Context, task *taskbiz.TimedTask, trigger
ctx = context.Background()
}
started := time.Now()
ctx, output := platformtask.WithOutput(ctx)
log = &taskbiz.TimedTaskLog{TriggerType: trigger, StartedAt: started, Status: "success"}
if task != nil {
log.TaskID = task.ID
@@ -288,9 +293,12 @@ func (e *TaskExecutor) Run(ctx context.Context, task *taskbiz.TimedTask, trigger
log.Status = "fail"
log.ErrorMsg = truncateTaskText(fmt.Sprint(recovered))
}
log.Output = truncateTaskText(log.Output)
log.FinishedAt = time.Now()
log.DurationMS = log.FinishedAt.Sub(started).Milliseconds()
if task != nil && task.ExecutorType == taskbiz.TaskExecutorMethod {
log.Output = fmt.Sprintf("方法：%s；触发方式：%s\n%s\n执行结束：%s；耗时 %d ms", task.MethodName, trigger, output.String(), log.Status, log.DurationMS)
}
log.Output = truncateTaskText(log.Output)
logCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
defer cancel()
if task != nil {
internal/worker/task_executor_test.go
+46
View File
@@ -24,6 +24,52 @@ func TestPrivateIP(t *testing.T) {
}
}
func TestMethodExecutionPersistsProgressAndRetryOutcome(t *testing.T) {
registry := platformtask.NewRegistry()
attempts := 0
registry.Register(platformtask.Method{Name: "report", Run: func(ctx context.Context, _ json.RawMessage) error {
attempts++
if attempts == 1 {
return errors.New("temporary storage failure")
}
platformtask.Logf(ctx, "日期=2026-11-05；已提交 12 行")
return nil
}})
repo := &workerTaskRepo{logs: make(chan *taskbiz.TimedTaskLog, 1)}
executor := NewTaskExecutorWithRegistry(taskbiz.NewTaskUsecaseWithRegistry(repo, registry), registry)
log := executor.Run(context.Background(), &taskbiz.TimedTask{ID: 1, Name: "日报", ExecutorType: taskbiz.TaskExecutorMethod, MethodName: "report", Params: []byte(`{"_execution":{"maxRetryTimes":1,"retryIntervalSeconds":0}}`)}, "manual")
stored := <-repo.logs
if log.Status != "success" || stored.Output != log.Output {
t.Fatalf("result was not persisted: %+v", log)
}
for _, detail := range []string{"方法：report", "第 1 次执行失败", "第 2 次执行", "已提交 12 行", "执行结束：success"} {
if !strings.Contains(log.Output, detail) {
t.Fatalf("missing %q in %s", detail, log.Output)
}
}
}
func TestMethodTimeoutKeepsPartialProgressWithoutClaimingCompletion(t *testing.T) {
registry := platformtask.NewRegistry()
release := make(chan struct{})
finished := make(chan struct{})
registry.Register(platformtask.Method{Name: "slow", Run: func(ctx context.Context, _ json.RawMessage) error {
defer close(finished)
platformtask.Logf(ctx, "已处理第一批")
<-release
platformtask.Logf(ctx, "迟到的结果")
return nil
}})
ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond)
defer cancel()
log := (&TaskExecutor{methods: registry}).Run(ctx, &taskbiz.TimedTask{ID: 1, ExecutorType: taskbiz.TaskExecutorMethod, MethodName: "slow"}, "manual")
close(release)
<-finished
if log.Status != "timeout" || !strings.Contains(log.Output, "已处理第一批") || strings.Contains(log.Output, "迟到的结果") {
t.Fatalf("timeout did not preserve the execution snapshot: %+v", log)
}
}
func TestRunMethodReportsParentDeadlineAsTimeout(t *testing.T) {
const methodName = "worker-test-parent-deadline"
methods := platformtask.NewRegistry()