diff --git a/cmd/analytics_http_test.go b/cmd/analytics_http_test.go index 76dbaf0..c741ab3 100644 --- a/cmd/analytics_http_test.go +++ b/cmd/analytics_http_test.go @@ -172,6 +172,14 @@ func TestChannelAndGameAnalyticsHTTP(t *testing.T) { if data["total"] != float64(1) || data["totalPayAmount"] != "30.00" { t.Fatal("模糊命中子渠道后应统计可见的同树数据") } + data = call(7, "channelData", "sourceDb=lx&hasRegister=false&hasPay=true&pageSize=1", 0) + if data["total"] != float64(0) || data["totalPayAmount"] != "0.00" { + t.Fatalf("指标过滤未传递或合计仍包含筛选外数据: %+v", data) + } + data = call(7, "channelData", "sourceDb=lx&hasRegister=true&hasPay=true&pageSize=1&pageNum=2", 0) + if data["total"] != float64(2) || len(data["rows"].([]any)) != 1 || data["totalPayAmount"] != "60.00" { + t.Fatalf("筛选后的全量合计不能随翻页变化: %+v", data) + } if err := db.Exec("INSERT INTO sys_user_preference(user_id,include_sub_promote,idcard_dedup) VALUES(7,1,0)").Error; err != nil { t.Fatal(err) } diff --git a/deploy/docker/README.md b/deploy/docker/README.md index c7112ff..d9d3516 100644 --- a/deploy/docker/README.md +++ b/deploy/docker/README.md @@ -37,6 +37,8 @@ bash /www/wwwroot/sw2.fuhua91.com/releases//deploy.sh 脚本先构建和检查镜像，再停止当前受管容器，保留一份停止的 `-previous` 容器。健康检查失败时恢复前一组容器。配置、上传文件和数据库不随镜像替换；索引迁移由后端启动按版本执行。 +首次启动持续等待建表与索引迁移完成，以后端健康检查通过为准，不设置总等待时限。容器退出或重启时立即回退；迁移日志记录每一步的名称、结果和耗时。切换数据库前需备份运行配置；镜像回退不会自动还原数据库连接配置。 + 首次接入域名时，备份宝塔原站点配置，再安装 `bt-nginx-sw2.fuhua91.com.conf` 至 `/www/server/panel/vhost/nginx/`，执行 `/www/server/nginx/sbin/nginx -t` 通过后重载。模板保留证书验证目录，当前监听 HTTP；后续由宝塔为 sw2 单独配置匹配域名的证书。 验证首页、静态资源、`/api/health`、登录状态、商务接口和任务启停/执行日志。回退版本可执行 `bash /www/wwwroot/sw2.fuhua91.com/deploy.sh <旧commit>`；数据库迁移及业务数据不由镜像回退脚本撤销。 diff --git a/deploy/docker/build-package.ps1 b/deploy/docker/build-package.ps1 index b6288b6..adfff11 100644 --- a/deploy/docker/build-package.ps1 +++ b/deploy/docker/build-package.ps1 @@ -44,7 +44,7 @@ try { Get-ChildItem -LiteralPath (Join-Path $repoRoot 'web\dist') | ForEach-Object { Copy-Item -LiteralPath $_.FullName -Destination $frontendDist -Recurse -Force } - foreach ($name in @('deploy.sh', 'bt-nginx-sw2.fuhua91.com.conf')) { + foreach ($name in @('deploy.sh', 'wait-backend.sh', 'bt-nginx-sw2.fuhua91.com.conf')) { Copy-Item -LiteralPath (Join-Path $PSScriptRoot $name) -Destination (Join-Path $stage $name) } & tar -czf $archive -C $stage . diff --git a/deploy/docker/deploy.sh b/deploy/docker/deploy.sh index 38c351c..b439e04 100644 --- a/deploy/docker/deploy.sh +++ b/deploy/docker/deploy.sh @@ -47,6 +47,7 @@ if [[ "$MODE" == '--build-only' ]]; then exit 0; fi test -s "$BASE_DIR/config/config.yaml" test -f "$BASE_DIR/runtime.env" +test -f "$RELEASE/wait-backend.sh" install -d -o 1000 -g 1000 -m 0750 "$BASE_DIR/config" "$BASE_DIR/log" "$BASE_DIR/uploads" chown 1000:1000 "$BASE_DIR/config/config.yaml" chmod 0600 "$BASE_DIR/config/config.yaml" "$BASE_DIR/runtime.env" @@ -98,12 +99,7 @@ docker run -d --name sw2-backend --label com.sw2.application=sw2 \ -v "$BASE_DIR/config:/data/conf" -v "$BASE_DIR/log:/app/logs" -v "$BASE_DIR/uploads:/app/uploads" \ "$BACKEND_IMAGE" -ready=false -for attempt in $(seq 1 45); do - if curl -fsS --max-time 3 http://127.0.0.1:9602/health >/dev/null 2>&1; then ready=true; break; fi - sleep 2 -done -[[ "$ready" == true ]] || { echo 'Backend health check failed' >&2; false; } +bash "$RELEASE/wait-backend.sh" created+=(sw2-frontend) docker run -d --name sw2-frontend --label com.sw2.application=sw2 \ diff --git a/deploy/docker/wait-backend.sh b/deploy/docker/wait-backend.sh new file mode 100644 index 0000000..9825a9b --- /dev/null +++ b/deploy/docker/wait-backend.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +# The HTTP listener starts after migrations. Large indexes have no meaningful +# fixed startup deadline; a stopped/restarted process is an actual failure. +started=$SECONDS +echo 'Waiting for backend startup and database migrations to finish...' +while true; do + state=$(docker inspect --format '{{.State.Running}} {{.RestartCount}}' sw2-backend) || exit 1 + if [[ "$state" != 'true 0' ]]; then + echo "Backend exited or restarted during initialization ($state); check the migration logs." >&2 + exit 1 + fi + if curl -fsS --max-time 3 http://127.0.0.1:9602/health >/dev/null 2>&1; then + printf 'Backend ready after %s seconds.\n' "$((SECONDS - started))" + exit 0 + fi + sleep 2 +done diff --git a/deploy/docker/wait-backend.test.sh b/deploy/docker/wait-backend.test.sh new file mode 100644 index 0000000..a6b5eb1 --- /dev/null +++ b/deploy/docker/wait-backend.test.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +set -euo pipefail +SCRIPT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)/wait-backend.sh +TEST_TMP=$(cd -- "${TMPDIR:-/tmp}" && pwd -P) +TEST_ROOT=$(mktemp -d "$TEST_TMP/sw-backend-readiness.XXXXXX") +cleanup() { + [[ "$TEST_ROOT" == "$TEST_TMP/sw-backend-readiness."* ]] || exit 1 + rm -rf -- "$TEST_ROOT" +} +trap cleanup EXIT +mkdir "$TEST_ROOT/bin" +cat > "$TEST_ROOT/bin/docker" <<'DOCKER' +#!/usr/bin/env bash +set -euo pipefail +[[ "$1" == inspect && "${!#}" == sw2-backend ]] +case "$TEST_MODE" in + exited) echo 'false 0' ;; + restarted) echo 'true 1' ;; + inspect-failed) exit 1 ;; + *) echo 'true 0' ;; +esac +DOCKER +cat > "$TEST_ROOT/bin/curl" <<'CURL' +#!/usr/bin/env bash +set -euo pipefail +count=$(cat "$TEST_ROOT/polls") +count=$((count + 1)) +echo "$count" > "$TEST_ROOT/polls" +[[ "$TEST_MODE" == ready || "$count" -gt 60 ]] +CURL +printf '#!/usr/bin/env bash\nexit 0\n' > "$TEST_ROOT/bin/sleep" +chmod +x "$TEST_ROOT/bin/"* +export TEST_ROOT PATH="$TEST_ROOT/bin:$PATH" +for scenario in ready long-migration exited restarted inspect-failed; do + export TEST_MODE="$scenario" + echo 0 > "$TEST_ROOT/polls" + result=0 + bash "$SCRIPT" > "$TEST_ROOT/output" 2>&1 || result=$? + case "$scenario" in + ready) [[ "$result" == 0 && "$(cat "$TEST_ROOT/polls")" == 1 ]] ;; + long-migration) [[ "$result" == 0 && "$(cat "$TEST_ROOT/polls")" == 61 ]] ;; + *) [[ "$result" != 0 && "$(cat "$TEST_ROOT/polls")" == 0 ]] ;; + esac + printf 'passed: %s\n' "$scenario" +done diff --git a/internal/biz/business/alert_governance.go b/internal/biz/business/alert_governance.go index 3ecbf02..9592db3 100644 --- a/internal/biz/business/alert_governance.go +++ b/internal/biz/business/alert_governance.go @@ -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) diff --git a/internal/biz/business/alert_trigger.go b/internal/biz/business/alert_trigger.go index ddf3986..e74f6af 100644 --- a/internal/biz/business/alert_trigger.go +++ b/internal/biz/business/alert_trigger.go @@ -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) diff --git a/internal/biz/business/analytics.go b/internal/biz/business/analytics.go index 7f989b3..ca4c6ce 100644 --- a/internal/biz/business/analytics.go +++ b/internal/biz/business/analytics.go @@ -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} diff --git a/internal/biz/business/analytics_test.go b/internal/biz/business/analytics_test.go index 8bd6b67..7e2c32d 100644 --- a/internal/biz/business/analytics_test.go +++ b/internal/biz/business/analytics_test.go @@ -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 { diff --git a/internal/biz/business/effectiveness.go b/internal/biz/business/effectiveness.go index 235b395..9236c65 100644 --- a/internal/biz/business/effectiveness.go +++ b/internal/biz/business/effectiveness.go @@ -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 } diff --git a/internal/biz/business/management_test.go b/internal/biz/business/management_test.go index c504607..edcccc6 100644 --- a/internal/biz/business/management_test.go +++ b/internal/biz/business/management_test.go @@ -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) { diff --git a/internal/biz/business/performance.go b/internal/biz/business/performance.go index f9e7c32..43df244 100644 --- a/internal/biz/business/performance.go +++ b/internal/biz/business/performance.go @@ -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() diff --git a/internal/biz/business/report_build.go b/internal/biz/business/report_build.go index a2c346b..ae07a09 100644 --- a/internal/biz/business/report_build.go +++ b/internal/biz/business/report_build.go @@ -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 diff --git a/internal/biz/business/report_build_test.go b/internal/biz/business/report_build_test.go index be8a8b1..6a68e9a 100644 --- a/internal/biz/business/report_build_test.go +++ b/internal/biz/business/report_build_test.go @@ -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 diff --git a/internal/biz/business/report_dates_test.go b/internal/biz/business/report_dates_test.go index 40db353..4e5c131 100644 --- a/internal/biz/business/report_dates_test.go +++ b/internal/biz/business/report_dates_test.go @@ -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) } } diff --git a/internal/data/business/alert_governance.go b/internal/data/business/alert_governance.go index 58506f2..05ebd81 100644 --- a/internal/data/business/alert_governance.go +++ b/internal/data/business/alert_governance.go @@ -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 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 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 diff --git a/internal/data/business/analytics_test.go b/internal/data/business/analytics_test.go index a3365de..a3703bb 100644 --- a/internal/data/business/analytics_test.go +++ b/internal/data/business/analytics_test.go @@ -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{ diff --git a/internal/data/business/approved_games.go b/internal/data/business/approved_games.go index f8cbb5e..b428fc7 100644 --- a/internal/data/business/approved_games.go +++ b/internal/data/business/approved_games.go @@ -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") } // 游戏可选范围来自可见渠道的有效申请，不受统计时是否包含下级的开关影响。 diff --git a/internal/service/business/analytics.go b/internal/service/business/analytics.go index 52e479b..d4d6889 100644 --- a/internal/service/business/analytics.go +++ b/internal/service/business/analytics.go @@ -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 != "" { diff --git a/internal/service/dto/analytics.go b/internal/service/dto/analytics.go index 2b3a3b0..96183cd 100644 --- a/internal/service/dto/analytics.go +++ b/internal/service/dto/analytics.go @@ -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"` diff --git a/internal/worker/task_executor.go b/internal/worker/task_executor.go index 066ef4f..cea3564 100644 --- a/internal/worker/task_executor.go +++ b/internal/worker/task_executor.go @@ -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 { diff --git a/internal/worker/task_executor_test.go b/internal/worker/task_executor_test.go index 69e5751..e991196 100644 --- a/internal/worker/task_executor_test.go +++ b/internal/worker/task_executor_test.go @@ -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() diff --git a/pkg/database/migration/runner.go b/pkg/database/migration/runner.go index 0bed148..4729a7f 100644 --- a/pkg/database/migration/runner.go +++ b/pkg/database/migration/runner.go @@ -5,7 +5,9 @@ package migration import ( "fmt" + "log/slog" "strings" + "time" "gorm.io/gorm" ) @@ -23,10 +25,20 @@ type record struct { func (record) TableName() string { return TableName } -func Run(db *gorm.DB, steps []Step) error { +func Run(db *gorm.DB, steps []Step) (runErr error) { if db == nil { return fmt.Errorf("database is nil") } + started := time.Now() + completed := 0 + slog.Info("database migrations started", "steps", len(steps)) + defer func() { + status := "completed" + if runErr != nil { + status = "failed" + } + slog.Info("database migrations finished", "status", status, "applied", completed, "elapsed_ms", time.Since(started).Milliseconds()) + }() db = db.Session(&gorm.Session{NewDB: true}) if err := repairMigrationTable(db); err != nil { return err @@ -66,12 +78,17 @@ func Run(db *gorm.DB, steps []Step) error { if step.Migrate == nil { return fmt.Errorf("migration %q has no function", step.ID) } + stepStarted := time.Now() + slog.Info("database migration started", "migration", step.ID) if err := step.Migrate(db.Session(&gorm.Session{NewDB: true})); err != nil { + slog.Error("database migration failed", "migration", step.ID, "elapsed_ms", time.Since(stepStarted).Milliseconds(), "error", err) return fmt.Errorf("apply migration %q: %w", step.ID, err) } if err := db.Session(&gorm.Session{NewDB: true}).Create(&record{ID: step.ID}).Error; err != nil { return fmt.Errorf("record migration %q: %w", step.ID, err) } + completed++ + slog.Info("database migration completed", "migration", step.ID, "elapsed_ms", time.Since(stepStarted).Milliseconds()) } return nil } diff --git a/pkg/task/output.go b/pkg/task/output.go new file mode 100644 index 0000000..0a10618 --- /dev/null +++ b/pkg/task/output.go @@ -0,0 +1,63 @@ +package task + +import ( + "context" + "fmt" + "strings" + "sync" + "unicode/utf8" +) + +type outputKey struct{} + +// Output collects bounded execution details without changing the method contract. +// A timed-out method may still write, so snapshots and writes must be synchronized. +type Output struct { + mu sync.Mutex + text string + truncated bool +} + +func WithOutput(ctx context.Context) (context.Context, *Output) { + output := &Output{} + return context.WithValue(ctx, outputKey{}, output), output +} + +// Logf records operational counts and progress. Do not include credentials or +// whole request/response bodies. Outside a task execution it is a no-op. +func Logf(ctx context.Context, format string, args ...any) { + if ctx == nil { + return + } + output, _ := ctx.Value(outputKey{}).(*Output) + if output == nil { + return + } + line := fmt.Sprintf(format, args...) + output.mu.Lock() + defer output.mu.Unlock() + const limit = 3000 // Reserve space for the executor's final status. + if output.truncated { + return + } + remaining := limit - len(output.text) + if len(line)+1 > remaining { + cut := max(0, remaining-1) + for cut > 0 && !utf8.RuneStart(line[cut]) { + cut-- + } + line = line[:cut] + output.truncated = true + } + output.text += line + "\n" +} + +func (output *Output) String() string { + output.mu.Lock() + defer output.mu.Unlock() + text := strings.TrimSpace(output.text) + if output.truncated { + text += "\n...(过程详情已截断)" + } + return text +} diff --git a/pkg/task/output_test.go b/pkg/task/output_test.go new file mode 100644 index 0000000..aa77248 --- /dev/null +++ b/pkg/task/output_test.go @@ -0,0 +1,35 @@ +package task + +import ( + "context" + "strings" + "sync" + "testing" + "unicode/utf8" +) + +func TestOutputIsBoundedConcurrentAndExecutionLocal(t *testing.T) { + ctx, output := WithOutput(context.Background()) + other, separate := WithOutput(context.Background()) + Logf(other, "独立任务") + var workers sync.WaitGroup + for i := 0; i < 20; i++ { + workers.Add(1) + go func() { + defer workers.Done() + for j := 0; j < 100; j++ { + Logf(ctx, "%s", strings.Repeat("渠道", 20)) + _ = output.String() + } + }() + } + workers.Wait() + text := output.String() + if !utf8.ValidString(text) || len(text) > 3100 || !strings.Contains(text, "已截断") { + t.Fatalf("invalid bounded output: bytes=%d", len(text)) + } + if separate.String() != "独立任务" { + t.Fatal("execution details leaked between tasks") + } + Logf(context.Background(), "outside an execution") +} diff --git a/web/src/view/sw/BusinessPerformance.vue b/web/src/view/sw/BusinessPerformance.vue index 6f6c1e2..492d6ce 100644 --- a/web/src/view/sw/BusinessPerformance.vue +++ b/web/src/view/sw/BusinessPerformance.vue @@ -23,7 +23,7 @@ {{ rateText(effectiveRate(row.total)) }} 暂无符合条件的商务 - 可见范围合计{{ displayMetric(metric.key, data?.matrixTotal?.[metric.key] ?? 0) }}{{ data?.matrixRate == null ? '—' : `${data.matrixRate}%` }} + 当前筛选合计{{ displayMetric(metric.key, data?.matrixTotal?.[metric.key] ?? 0) }}{{ data?.matrixRate == null ? '—' : `${data.matrixRate}%` }}