From 9caf5b1abcdfa3bd3bb9122e8addfda8587b8131 Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Sat, 7 Nov 2026 00:09:19 +0800 Subject: [PATCH] feat: integrate OA activation workflows and business migration --- cmd/activation_http_test.go | 38 ++- cmd/alert_record_http_test.go | 2 +- cmd/business_management_http_test.go | 2 +- cmd/wire_gen.go | 12 +- deploy/migrations/cleanup-legacy.sql | 2 + docs/BUSINESS_OA.md | 44 +++ internal/biz/biz.go | 2 +- internal/biz/business/activation.go | 41 ++- internal/biz/business/activation_pool.go | 117 +++++++ internal/biz/business/activation_pool_test.go | 56 ++++ internal/biz/business/activation_test.go | 6 +- internal/biz/business/dashboard.go | 7 +- internal/config/business_oa_test.go | 34 ++ internal/config/clone.go | 3 + internal/config/runtime.go | 3 + internal/config/types.go | 13 +- internal/data/business/account.go | 3 +- internal/data/business/activation.go | 104 ++++--- .../business/activation_integration_test.go | 159 ++++++++++ internal/data/business/activation_metrics.go | 192 ++++++++++++ internal/data/business/activation_pool.go | 292 ++++++++++++++++++ internal/data/business/activation_test.go | 47 ++- internal/data/business/alert_governance.go | 64 ++-- internal/data/business/alert_review_test.go | 4 +- internal/data/business/alert_v2_test.go | 13 +- internal/data/business/dashboard.go | 12 +- internal/data/business/dashboard_events.go | 68 ++++ internal/data/business/management_test.go | 2 +- internal/data/business/migrations.go | 7 +- internal/data/data.go | 1 + internal/integration/oa/activation.go | 73 +++++ internal/integration/oa/activation_test.go | 51 +++ internal/integration/provider.go | 2 + .../modules/business/activation_surface.go | 41 +++ .../business/activation_surface_test.go | 43 +++ internal/modules/business/definition.go | 3 +- internal/modules/catalog_test.go | 8 +- internal/routecatalog/catalog.go | 4 + internal/server/handler/activation_pool.go | 54 ++++ internal/server/router/routes.go | 4 + internal/service/business/activation.go | 18 +- internal/service/business/activation_pool.go | 112 +++++++ internal/service/dto/activation.go | 29 +- internal/service/dto/activation_pool.go | 58 ++++ web/src/api/activation.js | 5 + web/src/view/sw/ChannelActivationList.vue | 31 +- .../sw/components/ActivationResources.vue | 97 ++++++ 47 files changed, 1831 insertions(+), 152 deletions(-) create mode 100644 docs/BUSINESS_OA.md create mode 100644 internal/biz/business/activation_pool.go create mode 100644 internal/biz/business/activation_pool_test.go create mode 100644 internal/config/business_oa_test.go create mode 100644 internal/data/business/activation_integration_test.go create mode 100644 internal/data/business/activation_metrics.go create mode 100644 internal/data/business/activation_pool.go create mode 100644 internal/data/business/dashboard_events.go create mode 100644 internal/integration/oa/activation.go create mode 100644 internal/integration/oa/activation_test.go create mode 100644 internal/modules/business/activation_surface.go create mode 100644 internal/modules/business/activation_surface_test.go create mode 100644 internal/server/handler/activation_pool.go create mode 100644 internal/service/business/activation_pool.go create mode 100644 internal/service/dto/activation_pool.go create mode 100644 web/src/view/sw/components/ActivationResources.vue diff --git a/cmd/activation_http_test.go b/cmd/activation_http_test.go index 0da0f7a..c45e176 100644 --- a/cmd/activation_http_test.go +++ b/cmd/activation_http_test.go @@ -20,9 +20,15 @@ import ( businessservice "kra/internal/service/business" ) -type activationHTTPDatabases struct{ oa *gorm.DB } +type activationHTTPDatabases struct { + oa *gorm.DB + sdk *gorm.DB +} func (p *activationHTTPDatabases) Database(name string) (*gorm.DB, error) { + if name == "lx" && p.sdk != nil { + return p.sdk, nil + } if name == "oa" && p.oa != nil { return p.oa, nil } @@ -66,9 +72,27 @@ func TestActivationHTTPNamedDatabaseAndSpend(t *testing.T) { } } provider := businessHTTPData{db} - databases := &activationHTTPDatabases{oa: oa} - uc := businessbiz.NewActivationUsecase(businessdata.NewActivationRepo(provider, databases), businessbiz.NewAccountUsecase(businessdata.NewAccountRepo(provider))) - h := handler.NewActivation(businessservice.NewActivationService(uc)) + sdk := open("sdk.db") + pool, _ := sdk.DB() + pool.SetMaxOpenConns(1) + var files []struct{ Name, File string } + db.Raw("PRAGMA database_list").Scan(&files) + for _, file := range files { + if file.Name == "main" { + if err := sdk.Exec("ATTACH DATABASE ? AS sample", file.File).Error; err != nil { + t.Fatal(err) + } + } + } + for _, table := range []string{"tab_promote", "tab_spend"} { + if err := sdk.Exec("CREATE TABLE " + table + " AS SELECT * FROM sample." + table + " WHERE source_db='lx'").Error; err != nil { + t.Fatal(err) + } + } + sdk.Exec("DETACH DATABASE sample") + databases := &activationHTTPDatabases{oa: oa, sdk: sdk} + uc := businessbiz.NewActivationUsecase(businessdata.NewActivationRepo(provider, databases), businessbiz.NewAccountUsecase(businessdata.NewAccountRepo(provider)), businessbiz.NewUserPreferenceUsecase(businessdata.NewUserPreferenceRepo(provider))) + h := handler.NewActivation(businessservice.NewActivationService(uc, nil)) engine := gin.New() engine.Use(func(c *gin.Context) { id, _ := strconv.ParseUint(c.GetHeader("X-Test-User"), 10, 64) @@ -113,14 +137,14 @@ func TestActivationHTTPNamedDatabaseAndSpend(t *testing.T) { } const detail = "/sw/promoteActivePool/bindSpendDetail?id=9007199254740993" data := call(8, detail, 0) - if data["totalAmount"] != "30.30" || len(data["list"].([]any)) != 2 || data["endAt"] != "1970-01-01 00:33:20" { - t.Fatalf("流水金额、树范围或达标优先截止错误: %+v", data) + if data["totalAmount"] != "1029.30" || len(data["list"].([]any)) != 2 || data["endAt"] != "1970-01-01 01:00:00" { + t.Fatalf("流水金额、树范围或实际解绑截止错误: %+v", data) } call(7, detail, 7) call(8, "/sw/promoteActivePool/bindSpendDetail?id=2", 7) call(8, "/sw/promoteActivePool/bindSpendDetail?id=3", 7) call(1, "/sw/promoteActivePool/bindSpendDetail?id=0", 7) - if err := db.Exec("DROP TABLE tab_spend").Error; err != nil { + if err := sdk.Exec("DROP TABLE tab_spend").Error; err != nil { t.Fatal(err) } call(8, detail, 7) diff --git a/cmd/alert_record_http_test.go b/cmd/alert_record_http_test.go index 0780fdc..6380a34 100644 --- a/cmd/alert_record_http_test.go +++ b/cmd/alert_record_http_test.go @@ -61,7 +61,7 @@ func TestAlertRecordHTTPCompleteWorkflowAndVisibility(t *testing.T) { c.Next() }) router.RegisterAlertRecord(engine.Group(""), handler.NewAlertRecord(service)) - router.RegisterAlertGovernance(engine.Group(""), handler.NewAlertGovernance(businessservice.NewAlertGovernanceService(businessbiz.NewAlertGovernanceUsecase(businessdata.NewAlertGovernanceRepo(provider), accounts)))) + router.RegisterAlertGovernance(engine.Group(""), handler.NewAlertGovernance(businessservice.NewAlertGovernanceService(businessbiz.NewAlertGovernanceUsecase(businessdata.NewAlertGovernanceRepo(provider, &activationHTTPDatabases{}), accounts)))) server := httptest.NewServer(engine) defer server.Close() call := func(user int, method, path, body string, expected int) map[string]any { diff --git a/cmd/business_management_http_test.go b/cmd/business_management_http_test.go index 7ebc998..8ffe3be 100644 --- a/cmd/business_management_http_test.go +++ b/cmd/business_management_http_test.go @@ -71,7 +71,7 @@ func TestBusinessManagementHTTPGroupsGoalsAndEffectivenessAgree(t *testing.T) { goalRepo := data.NewGoalRepo(provider) performance := biz.NewPerformanceUsecase(data.NewPerformanceRepo(provider), teams, effective, goalRepo) goals := biz.NewGoalUsecase(goalRepo, performance) - dashboard := biz.NewDashboardUsecase(data.NewDashboardRepo(provider), biz.NewWorkbenchUsecase(data.NewWorkbenchRepo(provider), accounts, preferences), effective) + dashboard := biz.NewDashboardUsecase(data.NewDashboardRepo(provider, &activationHTTPDatabases{}), biz.NewWorkbenchUsecase(data.NewWorkbenchRepo(provider), accounts, preferences), effective) engine := gin.New() engine.Use(func(c *gin.Context) { id, _ := strconv.ParseUint(c.GetHeader("X-Test-User"), 10, 64) diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index e40ef90..f9ead62 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -26,6 +26,7 @@ import ( "kra/internal/integration/email" "kra/internal/integration/gameplatform" "kra/internal/integration/mq" + "kra/internal/integration/oa" "kra/internal/integration/payment" "kra/internal/integration/sms" "kra/internal/integration/storage" @@ -82,7 +83,7 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg auditRecorderUsecase := system2.NewAuditRecorderUsecase(auditRecordRepo) auditRecorder := system3.NewAuditRecorder(auditRecorderUsecase) registry := taskRegistry() - alertGovernanceRepo := business.NewAlertGovernanceRepo(dataData) + alertGovernanceRepo := business.NewAlertGovernanceRepo(dataData, dataData) accountRepo := business.NewAccountRepo(dataData) accountUsecase := business2.NewAccountUsecase(accountRepo) alertGovernanceUsecase := business2.NewAlertGovernanceUsecase(alertGovernanceRepo, accountUsecase) @@ -97,7 +98,7 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg goalRepo := business.NewGoalRepo(dataData) performanceUsecase := business2.NewPerformanceUsecase(performanceRepo, teamUsecase, effectivenessUsecase, goalRepo) goalUsecase := business2.NewGoalUsecase(goalRepo, performanceUsecase) - dashboardRepo := business.NewDashboardRepo(dataData) + dashboardRepo := business.NewDashboardRepo(dataData, dataData) workbenchRepo := business.NewWorkbenchRepo(dataData) userPreferenceRepo := business.NewUserPreferenceRepo(dataData) userPreferenceUsecase := business2.NewUserPreferenceUsecase(userPreferenceRepo) @@ -141,8 +142,11 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg channelService := business3.NewChannelService(channelUsecase) channel := handler.NewChannel(channelService) activationRepo := business.NewActivationRepo(dataData, dataData) - activationUsecase := business2.NewActivationUsecase(activationRepo, accountUsecase) - activationService := business3.NewActivationService(activationUsecase) + activationUsecase := business2.NewActivationUsecase(activationRepo, accountUsecase, userPreferenceUsecase) + activationPoolRepo := business.NewActivationPoolRepo(dataData, dataData) + activationGateway := oa.NewActivationGateway(store) + activationPoolUsecase := business2.NewActivationPoolUsecase(activationPoolRepo, activationGateway, accountUsecase) + activationService := business3.NewActivationService(activationUsecase, activationPoolUsecase) activation := handler.NewActivation(activationService) alertRecordRepo := business.NewAlertRecordRepo(dataData) alertRecordUsecase := business2.NewAlertRecordUsecase(alertRecordRepo, accountUsecase) diff --git a/deploy/migrations/cleanup-legacy.sql b/deploy/migrations/cleanup-legacy.sql index a43f43f..f0a409c 100644 --- a/deploy/migrations/cleanup-legacy.sql +++ b/deploy/migrations/cleanup-legacy.sql @@ -264,6 +264,8 @@ cleanup_run: BEGIN ('202611060022_business_promote_id_capacity'), ('202611060023_business_binding_status_index'), ('202611060024_business_role_identity'), +('202611060024_business_oa_remark'), +('202611060025_business_activation_surface'), ('202611060025_automatic_upgrade_state') ; diff --git a/docs/BUSINESS_OA.md b/docs/BUSINESS_OA.md new file mode 100644 index 0000000..5458204 --- /dev/null +++ b/docs/BUSINESS_OA.md @@ -0,0 +1,44 @@ +# 商务与 OA 对接 + +商务应用负责展示和发起申请，OA 负责商务账号管理、SDK 绑定审核、渠道激活审核与渠道操作。OA 使用 `oa/server`，不依赖尚未迁移业务的 `oa-new`。 + +## 部署顺序与连接 + +1. 先升级新商务，完成自动迁移和原商务数据迁移。新商务用户表为 `sys_users`，角色表为 `sys_authorities`，用户角色关系为 `sys_user_authority`。本次新增用户备注列及激活接口权限迁移。 +2. OA 的 `db-list` 中保留连接别名 `shangwu`，把该连接的数据库指向**迁移后的新商务主库**。`lx`、`hw`、`yq` 仍指向对应 SDK 数据库。OA 启动会检查 `shangwu.sys_users`，避免误连旧库。 +3. 新商务 `data.database_list` 的 `oa` 连接指向 OA 主库；`lx`、`hw`、`yq` 指向对应 SDK 数据库。新商务只读取这些外部表，不创建它们。 +4. 两边设置同一服务间密钥。OA 配置使用 `service-credentials.business-integration-key`；新商务可使用环境变量 `KRA_BUSINESS_OA_KEY`，或配置文件中的 `business_oa.key`。密钥只保存在部署环境，不提交版本库。 +5. 新商务设置 `KRA_BUSINESS_OA_URL`（或 `business_oa.url`），值为 OA 服务地址，包含 OA 配置的 `router-prefix`，末尾无需添加接口路径。例如 OA 前缀为 `/api` 时，基地址以 `/api` 结尾。 +6. 升级 OA。启动迁移新增 OA 激活申请字段，并在已配置的三个 SDK 库创建 `tab_promote_business_event`。部署账号需要对应的建表／改表权限。应用不会把旧库用户字段自动改造成新商务结构。 + +## 接口与行为 + +| 功能 | 新商务接口 | OA 行为 | +| --- | --- | --- | +| 激活列表 | `GET /sw/promoteActivePool/bindList` | 读取 OA 的绑定记录，按当前用户已通过的 SDK 绑定限定范围 | +| 激活流水 | `GET /sw/promoteActivePool/bindSpendDetail` | 绑定起算至实际解绑；达标不中止累计；跨平台关联渠道来自申请时的快照，流水和注册读取 OA 使用的 SDK 库 | +| 可申请渠道 | `GET /sw/promoteActivePool/pool` | 仅可认领渠道，状态来自 SDK 库；不返回联系方式、注册详情、充值金额或充值详情 | +| 本人商务选项 | `GET /sw/promoteActivePool/options` | 仅本人审核通过的 SDK 商务；申请平台必须匹配 | +| 申请 | `POST /sw/promoteActivePool/apply` | 后端转发到 `POST /internal/business/activation/apply`，再次校验申请人及其绑定权限 | +| 流失记录 | `GET /sw/promoteActivePool/churn` | 按 SDK 渠道换绑事件展示离开原商务的一侧；从池返回不新增回流记录 | +| 工作台 | 原工作台概览接口 | 锁定／解除合作按事件发生时间、发生时归属商务统计渠道数，解锁不抹去历史 | + +服务间申请接口以 HMAC-SHA256 校验方法、完整路径、时间戳和请求体，允许前后五分钟时钟偏差。`X-Business-Timestamp` 为 Unix 秒，`X-Business-Signature` 为十六进制签名。签名串为 `METHOD\nPATH\nTIMESTAMP\nBODY`。业务 ID 使用十进制 JSON 字符串。密钥未配置时拒绝服务间申请。 + +OA 商务创建／编辑兼容旧请求的 `roleId` 字段，但忽略该值，按 `sys_authorities.business_level` 唯一解析角色。新增账号使用新商务自增 ID、UUID、密码更新时间和默认偏好；锁定同时设置 `business_status` 与 `enable`。调整等级会校验整棵子树并同步角色，不允许把已有 SDK 绑定的账号变成非三级商务。 + +OA SDK 绑定审核继续使用原接口。审批遇到重复 SDK 绑定时返回已有账号提示，调用方确认后需传 `confirmDuplicate: true`。通过、驳回、解绑与新商务申请共用平台商务键锁。 + +OA 激活审核接口新增 `bindAt`（`YYYY-MM-DD`，省略时取审核当天）、`reason`（驳回必填）及 `restoreCooperation`（明确确认恢复合作时传 `true`）。审核通过只进入观察期，不改变渠道商务归属；累计流水达标后才变更主渠道及下级的商务归属，并从池中移除关联渠道。`ownershipPending` 表示达标后的跨库归属同步尚待完成；定时扫描会重试，同步成功前关联渠道仍保持认领锁定。观察期内主动解绑或到期未达标，只解除认领，不改渠道归属；已达标记录保持原 OA 规则，不能通过观察期解绑接口退回。 + +绑定扫描只在观察期内判定达标，迟到的观察期外流水不能使过期申请达标；达标后的展示统计继续累计。外部换绑会结束原激活关系，新商务读取流水时也会依据换绑事件截断，防止读到新归属期间数据。 + +累计注册统计仅包含绑定期间新增的主账号；身份证去重开启时跨平台按非空实名身份证合并，缺实名不计入。流水不受身份证去重影响。 + +免预警名单自动移出依据 OA 实际入池记录，避免本地流水不足却被 OA 豁免的渠道被误移出。预警列表的实时流失风险指标仍按其原型定义计算，历史预警快照不更改。 + +## 历史与校验 + +渠道事件从本版本部署后开始记录。SDK 事件表保存首次启用时间；工作台查询或同比区间早于该时间时显示历史不完整，而不把缺失记录当成真实零值。现有 SDK 表仅保留当前状态，不能可靠还原过去每次锁定／解除合作与人工换绑，因此不以当前状态伪造历史事件。既有激活记录继续可读；没有关联快照的历史申请按原平台渠道统计。 + +联调应覆盖：新建商务登录、锁定／解锁、SDK 绑定冲突确认、申请→审核→观察期→达标→归属同步、驳回重提、达标后继续累计、到期解绑、跨库暂时不可用后的同步重试，以及不同商务／平台的访问隔离。单元测试使用临时 SQLite 数据库，真实部署还需按上述连接配置验证实际 OA 和 SDK 数据源。 diff --git a/internal/biz/biz.go b/internal/biz/biz.go index ef33cd5..acba49a 100644 --- a/internal/biz/biz.go +++ b/internal/biz/biz.go @@ -11,4 +11,4 @@ import ( ) // ProviderSet is biz providers. -var ProviderSet = wire.NewSet(system.ProviderSet, payment.ProviderSet, integration.ProviderSet, task.ProviderSet, business.NewUserPreferenceUsecase, business.NewOtherSettingsUsecase, business.NewTeamUsecase, business.NewGoalUsecase, business.NewEffectivenessUsecase, business.NewPerformanceUsecase, business.NewDashboardUsecase, business.NewAccountUsecase, business.NewBindingUsecase, business.NewPasswordResetUsecase, business.NewBusinessPromoteUsecase, business.NewAlertGovernanceUsecase, business.NewAlertRuleUsecase, business.NewAlertRecordUsecase, business.NewActivationUsecase, business.NewChannelUsecase, business.NewBusinessPortraitUsecase, business.NewChannelPortraitUsecase, business.NewLedgerUsecase, business.NewAnalyticsUsecase, business.NewServerAnalysisUsecase, business.NewWorkbenchUsecase, business.NewRankingUsecase, business.NewReportBuildUsecase, business.NewAlertTriggerUsecase) +var ProviderSet = wire.NewSet(system.ProviderSet, payment.ProviderSet, integration.ProviderSet, task.ProviderSet, business.NewUserPreferenceUsecase, business.NewOtherSettingsUsecase, business.NewTeamUsecase, business.NewGoalUsecase, business.NewEffectivenessUsecase, business.NewPerformanceUsecase, business.NewDashboardUsecase, business.NewAccountUsecase, business.NewBindingUsecase, business.NewPasswordResetUsecase, business.NewBusinessPromoteUsecase, business.NewAlertGovernanceUsecase, business.NewAlertRuleUsecase, business.NewAlertRecordUsecase, business.NewActivationUsecase, business.NewActivationPoolUsecase, business.NewChannelUsecase, business.NewBusinessPortraitUsecase, business.NewChannelPortraitUsecase, business.NewLedgerUsecase, business.NewAnalyticsUsecase, business.NewServerAnalysisUsecase, business.NewWorkbenchUsecase, business.NewRankingUsecase, business.NewReportBuildUsecase, business.NewAlertTriggerUsecase) diff --git a/internal/biz/business/activation.go b/internal/biz/business/activation.go index bbdd0c5..a1215c7 100644 --- a/internal/biz/business/activation.go +++ b/internal/biz/business/activation.go @@ -19,17 +19,26 @@ var ( ) type Activation struct { + BusierID uint64 + CreatedAt *time.Time + RegisteredPlayers *int64 + Channels []ChannelKey + MetricsAvailable bool ID uint64 PromoteID *uint64 PlatformName, PromoteAccount, BusierAccount, ApplicantOA, Status, AccumAmount string BindAt, UnbindAt, ReachAt *time.Time } -func (a *Activation) EndAt() *time.Time { - if a.Status == "reached" && a.ReachAt != nil { - return a.ReachAt +// 达标仍处于绑定关系内，流水和注册统计持续到实际解绑。 +func (a *Activation) EndAt() *time.Time { return a.UnbindAt } + +// 换绑即结束原关系，即使 OA 定时扫描还没执行，也不能读取新归属期间的流水。 +func (a *Activation) EndOnTransfer(at time.Time) { + if a.BindAt != nil && !at.Before(*a.BindAt) && (a.UnbindAt == nil || at.Before(*a.UnbindAt)) { + a.UnbindAt = &at + a.Status = "unbound" } - return a.UnbindAt } type ActivationAccount struct { @@ -41,6 +50,7 @@ type ActivationScope struct { Accounts []ActivationAccount } type ActivationFilter struct { + IDCardDedup bool PlatformNames, Statuses []string PromoteAccount, BusierAccount string } @@ -50,12 +60,13 @@ type ActivationPage struct { Available bool } type ActivationSpendQuery struct { + Channels []ChannelKey PlatformName string PromoteID uint64 Start int64 End *int64 } -type ActivationSpendItem struct{ GameName, Amount string } +type ActivationSpendItem struct{ PlatformName, GameName, Amount string } type ActivationSpendDetail struct { Activation *Activation Items []*ActivationSpendItem @@ -70,12 +81,13 @@ type ActivationRepo interface { } type ActivationUsecase struct { - repo ActivationRepo - accounts *AccountUsecase + repo ActivationRepo + accounts *AccountUsecase + preferences *UserPreferenceUsecase } -func NewActivationUsecase(repo ActivationRepo, accounts *AccountUsecase) *ActivationUsecase { - return &ActivationUsecase{repo: repo, accounts: accounts} +func NewActivationUsecase(repo ActivationRepo, accounts *AccountUsecase, preferences *UserPreferenceUsecase) *ActivationUsecase { + return &ActivationUsecase{repo: repo, accounts: accounts, preferences: preferences} } func (uc *ActivationUsecase) scope(ctx context.Context) (ActivationScope, error) { @@ -126,6 +138,11 @@ func (uc *ActivationUsecase) List(ctx context.Context, filter ActivationFilter, for i, status := range filter.Statuses { filter.Statuses[i] = NormalizeActivationStatus(status) } + pref, err := uc.preferences.Current(ctx) + if err != nil { + return nil, err + } + filter.IDCardDedup = pref.IDCardDedup return uc.repo.List(ctx, scope, filter, options...) } @@ -160,7 +177,7 @@ func (uc *ActivationUsecase) SpendDetail(ctx context.Context, id uint64) (*Activ if strings.TrimSpace(activation.PlatformName) == "" || activation.PromoteID == nil || *activation.PromoteID == 0 { return detail, nil } - query := ActivationSpendQuery{PlatformName: activation.PlatformName, PromoteID: *activation.PromoteID, Start: activation.BindAt.Unix()} + query := ActivationSpendQuery{Channels: activation.Channels, PlatformName: activation.PlatformName, PromoteID: *activation.PromoteID, Start: activation.BindAt.Unix()} if end := activation.EndAt(); end != nil { n := end.Unix() query.End = &n @@ -185,13 +202,13 @@ func (uc *ActivationUsecase) SpendDetail(ctx context.Context, id uint64) (*Activ func NormalizeActivationStatus(status string) string { status = strings.TrimSpace(status) - if code, ok := map[string]string{"待审核": "pending", "进行中": "in_progress", "已达标": "reached", "已驳回": "rejected", "已解绑": "unbound"}[status]; ok { + if code, ok := map[string]string{"待审核": "pending", "审核中": "pending", "进行中": "in_progress", "已达标": "reached", "已驳回": "rejected", "已解绑": "unbound"}[status]; ok { return code } return status } func ActivationStatusLabel(status string) string { - if label, ok := map[string]string{"pending": "待审核", "in_progress": "进行中", "reached": "已达标", "rejected": "已驳回", "unbound": "已解绑"}[status]; ok { + if label, ok := map[string]string{"pending": "审核中", "in_progress": "进行中", "reached": "已达标", "rejected": "已驳回", "unbound": "已解绑"}[status]; ok { return label } return status diff --git a/internal/biz/business/activation_pool.go b/internal/biz/business/activation_pool.go new file mode 100644 index 0000000..16ee58d --- /dev/null +++ b/internal/biz/business/activation_pool.go @@ -0,0 +1,117 @@ +package business + +import ( + "context" + "strings" + "time" +) + +type ActivationPoolFilter struct { + Sources []string + Name, ChannelType, Status, Target string + Begin, End *time.Time +} +type ActivationPoolChannel struct { + ID, PromoteID uint64 + SourceDB, Name, ChannelType, Status string + NoLoginDays *int64 + EnteredAt time.Time + Rejected bool + RejectReason string + RejectedAt *time.Time +} +type ActivationPoolPage struct { + Rows []*ActivationPoolChannel + Total int64 + Available bool +} +type ActivationBindingOption struct { + SourceDB, Account string + ID uint64 + Level int +} +type ActivationApplication struct { + UserID uint + PoolID, BusierID uint64 + SourceDB string +} +type ActivationChurn struct { + ID, PromoteID uint64 + SourceDB, Name, From, To, Reason, Operator string + At time.Time + ToPool bool +} +type ActivationChurnPage struct { + Rows []*ActivationChurn + Total int64 + Available bool +} +type ActivationPoolRepo interface { + Pool(context.Context, ActivationPoolFilter, ...ListOption) (*ActivationPoolPage, error) + PoolChannel(context.Context, uint64) (*ActivationPoolChannel, error) + BindingOptions(context.Context, uint) ([]*ActivationBindingOption, error) + Churn(context.Context, ActivationPoolFilter, ...ListOption) (*ActivationChurnPage, error) +} +type ActivationGateway interface { + Apply(context.Context, ActivationApplication) error +} +type ActivationPoolUsecase struct { + repo ActivationPoolRepo + gateway ActivationGateway + accounts *AccountUsecase +} + +func NewActivationPoolUsecase(repo ActivationPoolRepo, gateway ActivationGateway, accounts *AccountUsecase) *ActivationPoolUsecase { + return &ActivationPoolUsecase{repo: repo, gateway: gateway, accounts: accounts} +} +func (uc *ActivationPoolUsecase) Pool(ctx context.Context, f ActivationPoolFilter, o ...ListOption) (*ActivationPoolPage, error) { + if _, _, err := uc.accounts.actor(ctx); err != nil { + return nil, err + } + return uc.repo.Pool(ctx, f, o...) +} +func (uc *ActivationPoolUsecase) Options(ctx context.Context) ([]*ActivationBindingOption, error) { + a, _, err := uc.accounts.actor(ctx) + if err != nil { + return nil, err + } + return uc.repo.BindingOptions(ctx, a.UserID) +} +func (uc *ActivationPoolUsecase) Apply(ctx context.Context, poolID, busierID uint64, source string) error { + a, _, err := uc.accounts.actor(ctx) + if err != nil { + return err + } + source, err = NormalizePlatform(source) + if err != nil || poolID == 0 || busierID == 0 { + return ErrActivationInput + } + options, err := uc.repo.BindingOptions(ctx, a.UserID) + if err != nil { + return err + } + owned := false + for _, o := range options { + if o.SourceDB == source && o.ID == busierID { + owned = true + break + } + } + if !owned { + return ErrActivationScope + } + channel, err := uc.repo.PoolChannel(ctx, poolID) + if err != nil { + return err + } + if !strings.EqualFold(channel.SourceDB, source) { + return ErrActivationInput + } + return uc.gateway.Apply(ctx, ActivationApplication{UserID: a.UserID, PoolID: poolID, BusierID: busierID, SourceDB: source}) +} +func (uc *ActivationPoolUsecase) Churn(ctx context.Context, f ActivationPoolFilter, o ...ListOption) (*ActivationChurnPage, error) { + if _, _, err := uc.accounts.actor(ctx); err != nil { + return nil, err + } + return uc.repo.Churn(ctx, f, o...) +} diff --git a/internal/biz/business/activation_pool_test.go b/internal/biz/business/activation_pool_test.go new file mode 100644 index 0000000..860d39e --- /dev/null +++ b/internal/biz/business/activation_pool_test.go @@ -0,0 +1,56 @@ +package business + +import ( + "context" + "testing" +) + +type activationPoolFake struct { + ActivationPoolRepo + options []*ActivationBindingOption + requested uint +} + +func (r *activationPoolFake) BindingOptions(_ context.Context, id uint) ([]*ActivationBindingOption, error) { + r.requested = id + return r.options, nil +} +func (r *activationPoolFake) PoolChannel(context.Context, uint64) (*ActivationPoolChannel, error) { + return &ActivationPoolChannel{ID: 10, SourceDB: "lx"}, nil +} + +type activationGatewayFake struct { + calls int + request ActivationApplication +} + +func (g *activationGatewayFake) Apply(_ context.Context, a ActivationApplication) error { + g.calls++ + g.request = a + return nil +} + +func TestActivationApplicationUsesAuthenticatedIdentityAndOwnedBinding(t *testing.T) { + accountUC, _ := activationFixture() + repo := &activationPoolFake{options: []*ActivationBindingOption{{ID: 9, SourceDB: "lx", Account: "own"}}} + gateway := &activationGatewayFake{} + uc := NewActivationPoolUsecase(repo, gateway, accountUC.accounts) + ctx := accountContext(8, 902) + if err := uc.Apply(ctx, 10, 9, "lx"); err != nil { + t.Fatal(err) + } + if gateway.request.UserID != 8 || repo.requested != 8 || gateway.request.BusierID != 9 { + t.Fatal("applicant came from request rather than session") + } + for _, a := range []ActivationApplication{{PoolID: 10, BusierID: 99, SourceDB: "lx"}, {PoolID: 10, BusierID: 9, SourceDB: "hw"}, {PoolID: 0, BusierID: 9, SourceDB: "lx"}} { + if err := uc.Apply(ctx, a.PoolID, a.BusierID, a.SourceDB); err == nil { + t.Fatal("unowned application accepted") + } + } + if err := uc.Apply(context.Background(), 10, 9, "lx"); err == nil { + t.Fatal("anonymous request accepted") + } + if gateway.calls != 1 { + t.Fatal("invalid requests reached OA") + } +} diff --git a/internal/biz/business/activation_test.go b/internal/biz/business/activation_test.go index 6290bb5..3601b35 100644 --- a/internal/biz/business/activation_test.go +++ b/internal/biz/business/activation_test.go @@ -36,7 +36,7 @@ func (r *activationRepoFake) Spend(_ context.Context, q ActivationSpendQuery) ([ func activationFixture() (*ActivationUsecase, *activationRepoFake) { a := &accountRepoFake{accounts: map[uint]*Account{1: {ID: 1, Level: 0}, 7: {ID: 7, Level: 1}, 8: {ID: 8, Level: 2, IDPath: "0,7"}}} r := &activationRepoFake{accounts: []ActivationAccount{{"lx", " Own "}, {"LX", "own"}, {"hw", "Other"}, {"lx", ""}, {"unknown", "Own"}}} - return NewActivationUsecase(r, NewAccountUsecase(a)), r + return NewActivationUsecase(r, NewAccountUsecase(a), NewUserPreferenceUsecase(&preferenceRepoFake{})), r } func TestActivationScopeIncludesOwnAccountsOnly(t *testing.T) { uc, r := activationFixture() @@ -62,8 +62,8 @@ func TestActivationSpendScopeCutoffAndExactMoney(t *testing.T) { r.record = &Activation{ID: 10, PromoteID: &id, PlatformName: "lx", BusierAccount: "OWN", BindAt: &start, ReachAt: &reach, UnbindAt: &unbind, Status: "reached"} r.items = []*ActivationSpendItem{{Amount: "9007199254740993.01"}, {Amount: "0.09"}} detail, err := uc.SpendDetail(accountContext(8, 902), 10) - if err != nil || detail.TotalAmount != "9007199254740993.10" || r.query.Start != start.Unix() || r.query.End == nil || *r.query.End != reach.Unix() { - t.Fatalf("达标截止或精确金额错误: %+v %+v %v", detail, r.query, err) + if err != nil || detail.TotalAmount != "9007199254740993.10" || r.query.Start != start.Unix() || r.query.End == nil || *r.query.End != unbind.Unix() { + t.Fatalf("绑定截止或精确金额错误: %+v %+v %v", detail, r.query, err) } r.record.Status = "unbound" if _, err := uc.SpendDetail(accountContext(8, 902), 10); err != nil || *r.query.End != unbind.Unix() { diff --git a/internal/biz/business/dashboard.go b/internal/biz/business/dashboard.go index eee587a..dcfe397 100644 --- a/internal/biz/business/dashboard.go +++ b/internal/biz/business/dashboard.go @@ -13,6 +13,7 @@ type DashboardMetrics struct { Pay string } type DashboardData struct { + EventsIncomplete bool Current, Previous DashboardMetrics Available bool EventsAvailable bool @@ -84,8 +85,7 @@ func (uc *DashboardUsecase) Overview(ctx context.Context, sources []string, peri } r.Cards[card.key] = value } - // The imported SDK channel table has only current state; historical event - // counts must stay unavailable until a timestamped event source is connected. + // OA 在渠道更新事务中记录事件；按事件发生时间统计，不读取当前锁定状态。 for _, card := range []struct { key string current, previous int64 @@ -93,6 +93,9 @@ func (uc *DashboardUsecase) Overview(ctx context.Context, sources []string, peri label := "较上一同期" if !data.EventsAvailable { label = "事件数据未接入" + if data.EventsIncomplete { + label = "所选区间事件历史不完整" + } } value, err := workbenchCard(num(card.current), num(card.previous), label, data.Available && data.EventsAvailable) if err != nil { diff --git a/internal/config/business_oa_test.go b/internal/config/business_oa_test.go new file mode 100644 index 0000000..512ff58 --- /dev/null +++ b/internal/config/business_oa_test.go @@ -0,0 +1,34 @@ +package config + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestBusinessOAEnvironmentAndRuntimeMerge(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte("server:\n http:\n addr: ':8000'\n"), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("KRA_BUSINESS_OA_URL", "http://oa.test/api") + t.Setenv("KRA_BUSINESS_OA_KEY", "test-only-key") + value, err := Load(path) + if err != nil { + t.Fatal(err) + } + if value.BusinessOA == nil || value.BusinessOA.URL != "http://oa.test/api" || value.BusinessOA.Key != "test-only-key" { + t.Fatal("environment-only OA configuration not loaded") + } + merged := MergeRuntimeConfig(value, &Config{Admin: &Admin{}}) + merged.BusinessOA.Key = "changed" + if value.BusinessOA.Key != "test-only-key" { + t.Fatal("configuration snapshot shared a pointer") + } + serialized, err := json.Marshal(value) + if err != nil || strings.Contains(string(serialized), "test-only-key") { + t.Fatal("service credential leaked through public configuration") + } +} diff --git a/internal/config/clone.go b/internal/config/clone.go index 29ef351..72fcb12 100644 --- a/internal/config/clone.go +++ b/internal/config/clone.go @@ -25,6 +25,9 @@ func MergeRuntimeConfig(current, next *Config) *Config { } return merged } + if merged.BusinessOA == nil { + merged.BusinessOA = clonePtr(current.BusinessOA) + } if merged.Server == nil { merged.Server = clonePtr(current.Server) } diff --git a/internal/config/runtime.go b/internal/config/runtime.go index ed92f74..9dd2305 100644 --- a/internal/config/runtime.go +++ b/internal/config/runtime.go @@ -332,6 +332,8 @@ func newViper(path string) *viper.Viper { v.SetEnvPrefix("KRA") v.SetEnvKeyReplacer(strings.NewReplacer(".", "_", "-", "_")) v.AutomaticEnv() + _ = v.BindEnv("business_oa.url") + _ = v.BindEnv("business_oa.key") return v } @@ -385,6 +387,7 @@ func cloneConfig(value *Config) *Config { return nil } copied := *value + copied.BusinessOA = clonePtr(value.BusinessOA) copied.Server = clonePtr(value.Server) if copied.Server != nil { copied.Server.HTTP = clonePtr(value.Server.HTTP) diff --git a/internal/config/types.go b/internal/config/types.go index 2dc43b9..03ec6be 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -6,9 +6,10 @@ import "time" // It intentionally uses ordinary Go types so configuration is independent of // protobuf generation and can be consumed by every internal module. type Config struct { - Server *Server `mapstructure:"server" yaml:"server" json:"server"` - Data *Data `mapstructure:"data" yaml:"data" json:"data"` - Admin *Admin `mapstructure:"admin" yaml:"admin" json:"admin"` + BusinessOA *BusinessOA `mapstructure:"business_oa" yaml:"business_oa" json:"-"` + Server *Server `mapstructure:"server" yaml:"server" json:"server"` + Data *Data `mapstructure:"data" yaml:"data" json:"data"` + Admin *Admin `mapstructure:"admin" yaml:"admin" json:"admin"` } type Server struct { @@ -224,3 +225,9 @@ type ObjectStore struct { ForcePathStyle bool `mapstructure:"force_path_style" yaml:"force_path_style" json:"force_path_style"` AccountID string `mapstructure:"account_id" yaml:"account_id" json:"account_id"` } + +// BusinessOA.URL 包含 OA 的路由前缀；Key 与 OA 服务间密钥一致。 +type BusinessOA struct { + URL string `mapstructure:"url" yaml:"url" json:"url"` + Key string `mapstructure:"key" yaml:"key" json:"-"` +} diff --git a/internal/data/business/account.go b/internal/data/business/account.go index 74094dd..1b803da 100644 --- a/internal/data/business/account.go +++ b/internal/data/business/account.go @@ -12,7 +12,8 @@ import ( // accountPO 是 sys_users 的商务投影，表结构仍由 system 模块维护。 type accountPO struct { - ID uint `gorm:"primaryKey"` + Remark string `gorm:"size:500"` + ID uint `gorm:"primaryKey"` CreatedAt, UpdatedAt time.Time DeletedAt gorm.DeletedAt UUID, Username, Password, NickName, RealName, Phone, Email string diff --git a/internal/data/business/activation.go b/internal/data/business/activation.go index bfbb42d..f06fd81 100644 --- a/internal/data/business/activation.go +++ b/internal/data/business/activation.go @@ -2,6 +2,7 @@ package business import ( "context" + "encoding/json" "errors" "strings" "time" @@ -12,6 +13,9 @@ import ( // OA 表只读，不由新商务的系统迁移创建或改写。 type activationPO struct { + BusierID uint64 + CreatedAt *time.Time + RelatedChannels string ID uint64 `gorm:"primaryKey"` PromoteID *uint64 PlatformName, PromoteAccount, BusierAccount, ApplicantOA, Status string @@ -25,7 +29,20 @@ func toBizActivation(po activationPO) *business.Activation { if po.AccumAmount != nil { amount = *po.AccumAmount } - return &business.Activation{ID: po.ID, PromoteID: po.PromoteID, PlatformName: po.PlatformName, PromoteAccount: po.PromoteAccount, BusierAccount: po.BusierAccount, ApplicantOA: po.ApplicantOA, Status: po.Status, AccumAmount: amount, BindAt: po.BindAt, UnbindAt: po.UnbindAt, ReachAt: po.ReachAt} + row := &business.Activation{BusierID: po.BusierID, CreatedAt: po.CreatedAt, ID: po.ID, PromoteID: po.PromoteID, PlatformName: po.PlatformName, PromoteAccount: po.PromoteAccount, BusierAccount: po.BusierAccount, ApplicantOA: po.ApplicantOA, Status: po.Status, AccumAmount: amount, BindAt: po.BindAt, UnbindAt: po.UnbindAt, ReachAt: po.ReachAt} + var channels []struct { + PlatformName string `json:"platformName"` + PromoteID uint64 `json:"promoteId"` + } + if po.RelatedChannels != "" { + if json.Unmarshal([]byte(po.RelatedChannels), &channels) != nil { + return nil + } + } + for _, ch := range channels { + row.Channels = append(row.Channels, business.ChannelKey{SourceDB: ch.PlatformName, ID: ch.PromoteID}) + } + return row } type activationRepo struct { @@ -99,7 +116,41 @@ func (r *activationRepo) List(ctx context.Context, scope business.ActivationScop return nil, business.ErrActivationSource } for _, po := range pos { - page.Rows = append(page.Rows, toBizActivation(po)) + row := toBizActivation(po) + if row == nil { + return nil, business.ErrActivationSource + } + if row.BindAt != nil { + if err := r.applyTransferEnd(ctx, row); err != nil { + return nil, err + } + query := activationMetricQuery(row) + available, err := r.activationMetricsAvailable(ctx, query) + if err != nil { + return nil, err + } + if available { + items, err := r.Spend(ctx, query) + if err != nil { + return nil, err + } + amounts := make([]string, 0, len(items)) + for _, item := range items { + amounts = append(amounts, item.Amount) + } + total, err := sumActivationMoney(amounts) + if err != nil { + return nil, err + } + row.AccumAmount = total + row.MetricsAvailable = true + } + row.RegisteredPlayers, err = r.registered(ctx, query, filter.IDCardDedup) + if err != nil { + return nil, err + } + } + page.Rows = append(page.Rows, row) } return page, nil } @@ -116,49 +167,12 @@ func (r *activationRepo) Find(ctx context.Context, id uint64) (*business.Activat } return nil, business.ErrActivationSource } - return toBizActivation(po), nil -} - -func (r *activationRepo) Spend(ctx context.Context, query business.ActivationSpendQuery) ([]*business.ActivationSpendItem, error) { - db := r.data.DB().WithContext(ctx) - available, err := businessTableAvailable(db, "tab_promote") - if err != nil { + row := toBizActivation(po) + if row == nil { + return nil, business.ErrActivationSource + } + if err := r.applyTransferEnd(ctx, row); err != nil { return nil, err } - if !available { - return nil, business.ErrActivationMetrics - } - tree := func() *gorm.DB { - return db.Table("tab_promote p").Where("p.source_db = ? AND (p.id = ? OR p.top_promote_id = ? OR p.parent_id = ?)", query.PlatformName, query.PromoteID, query.PromoteID, query.PromoteID) - } - var count int64 - if err := tree().Count(&count).Error; err != nil { - return nil, business.ErrStorage - } - items := []*business.ActivationSpendItem{} - if count == 0 { - return items, nil - } - available, err = businessTableAvailable(db, "tab_spend") - if err != nil { - return nil, err - } - if !available { - return nil, business.ErrActivationMetrics - } - q := db.Table("tab_spend s").Where("s.source_db = ? AND s.pay_status = ? AND s.pay_time >= ?", query.PlatformName, 1, query.Start).Where("EXISTS (?)", tree().Select("1").Where("p.id=s.promote_id")) - if query.End != nil { - q = q.Where("s.pay_time <= ?", *query.End) - } - var rows []struct { - GameID uint64 - GameName, Amount string - } - if err := q.Select("s.game_id,COALESCE(MAX(s.game_name),'') AS game_name,COALESCE(SUM(s.pay_amount),0) AS amount").Group("s.game_id").Order("amount DESC,s.game_id").Scan(&rows).Error; err != nil { - return nil, business.ErrStorage - } - for _, row := range rows { - items = append(items, &business.ActivationSpendItem{GameName: row.GameName, Amount: row.Amount}) - } - return items, nil + return row, nil } diff --git a/internal/data/business/activation_integration_test.go b/internal/data/business/activation_integration_test.go new file mode 100644 index 0000000..9445676 --- /dev/null +++ b/internal/data/business/activation_integration_test.go @@ -0,0 +1,159 @@ +package business + +import ( + "context" + "errors" + "gorm.io/gorm" + "kra/internal/biz/business" + "testing" + "time" +) + +type activationTestSources map[string]*gorm.DB + +func (s activationTestSources) Database(name string) (*gorm.DB, error) { + db := s[name] + if db == nil { + return nil, errors.New("missing source") + } + return db, nil +} + +func TestActivationCrossPlatformMetricsAndIdentityDedup(t *testing.T) { + db, oa, repo := prepareActivations(t) + for _, sql := range []string{ + "CREATE TABLE tab_user(id INTEGER,source_db TEXT,promote_id INTEGER,puid INTEGER,register_time TEXT,idcard TEXT)", + "INSERT INTO tab_user VALUES(1,'lx',100,0,'1000','same'),(2,'lx',101,0,'1500','same'),(1,'hw',100,0,'1500','same'),(3,'lx',102,0,'1600',''),(4,'lx',103,0,'1600','outside'),(5,'lx',100,1,'1600','child-account'),(6,'lx',100,0,'999','before'),(7,'lx',100,0,'2001','after')", + } { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + if err := oa.Model(&activationPO{}).Where("id=1").Update("related_channels", `[{"platformName":"lx","promoteId":100},{"platformName":"hw","promoteId":100},{"platformName":"lx","promoteId":101}]`).Error; err != nil { + t.Fatal(err) + } + repo.(*activationRepo).databases = activationSDKFixtures(t, db, oa) + page, err := repo.List(context.Background(), business.ActivationScope{All: true}, business.ActivationFilter{Statuses: []string{"reached"}, IDCardDedup: true}) + if err != nil { + t.Fatal(err) + } + row := page.Rows[0] + if row.RegisteredPlayers == nil || *row.RegisteredPlayers != 1 || row.AccumAmount != "1032.60" { + t.Fatalf("cross platform dedup/tree overlap: %+v", row) + } + page, err = repo.List(context.Background(), business.ActivationScope{All: true}, business.ActivationFilter{Statuses: []string{"reached"}}) + if err != nil || *page.Rows[0].RegisteredPlayers != 4 { + t.Fatalf("account count: %+v %v", page, err) + } + if err := oa.Model(&activationPO{}).Where("id=1").Update("unbind_at", nil).Error; err != nil { + t.Fatal(err) + } + page, err = repo.List(context.Background(), business.ActivationScope{All: true}, business.ActivationFilter{Statuses: []string{"reached"}}) + if err != nil || page.Rows[0].AccumAmount != "1732.60" || *page.Rows[0].RegisteredPlayers != 5 { + t.Fatalf("reached should keep accumulating: %+v %v", page, err) + } +} + +func TestActivationPoolFiltersAndRedactedProjection(t *testing.T) { + db, oa, _ := prepareActivations(t) + if err := oa.AutoMigrate(&activationPoolPO{}); err != nil { + t.Fatal(err) + } + for _, sql := range []string{"ALTER TABLE tab_promote ADD COLUMN lock_status INTEGER", "ALTER TABLE tab_promote ADD COLUMN last_login_time INTEGER", "ALTER TABLE tab_promote ADD COLUMN create_time INTEGER", "UPDATE tab_promote SET status=1,lock_status=0,create_time=1000"} { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + for _, sql := range []string{"ALTER TABLE sys_promote_active_pool_bind ADD COLUMN pool_record_id INTEGER", "ALTER TABLE sys_promote_active_pool_bind ADD COLUMN reject_reason TEXT", "ALTER TABLE sys_promote_active_pool_bind ADD COLUMN review_at DATETIME"} { + if err := oa.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + now := time.Now() + rows := []activationPoolPO{{ID: 1, PlatformName: "lx", PromoteID: 100, PromoteAccount: "Open", RegisterType: 1, CreatedAt: now}, {ID: 2, PlatformName: "hw", PromoteID: 100, PromoteAccount: "Claimed", ClaimFlag: 1, CreatedAt: now}, {ID: 3, PlatformName: "lx", PromoteID: 103, PromoteAccount: "Removed", DeletedAt: &now, CreatedAt: now}} + if err := oa.Create(&rows).Error; err != nil { + t.Fatal(err) + } + r := NewActivationPoolRepo(testData{db}, activationSDKFixtures(t, db, oa)) + page, err := r.Pool(context.Background(), business.ActivationPoolFilter{Name: "Open", ChannelType: "公会", Status: "正常"}, business.ListLimit(10)) + if err != nil { + t.Fatal(err) + } + if page.Total != 1 || page.Rows[0].Name != "Open" || page.Rows[0].Status != "正常" { + t.Fatalf("pool filters: %+v", page) + } + if _, err := r.PoolChannel(context.Background(), 2); err != business.ErrActivationNotFound { + t.Fatal("claimed pool may not be applied") + } +} + +func TestDashboardEventsUseTimeAndHistoricalOwner(t *testing.T) { + db := prepareAccounts(t) + sdk := openTestDB(t) + if err := sdk.AutoMigrate(&channelEventPO{}); err != nil { + t.Fatal(err) + } + if err := db.Exec("INSERT INTO sys_users(id,username,level,id_path) VALUES(7,'owner',3,'0'); INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,status) VALUES(7,'lx',10,'0')").Error; err != nil { + t.Fatal(err) + } + begin := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC) + w := business.DashboardWindow{Begin: begin, End: begin, PreviousBegin: begin.AddDate(0, 0, -1), PreviousEnd: begin.AddDate(0, 0, -1)} + events := []channelEventPO{{ID: 1, PromoteID: 1, PromoteLevel: 1, Kind: "locked", OldBusierID: 10, OccurredAt: begin}, {ID: 2, PromoteID: 1, PromoteLevel: 1, Kind: "locked", OldBusierID: 10, OccurredAt: begin.Add(time.Hour)}, {ID: 3, PromoteID: 2, PromoteLevel: 2, Kind: "terminated", OldBusierID: 10, OccurredAt: begin}, {ID: 4, PromoteID: 3, PromoteLevel: 1, Kind: "locked", OldBusierID: 20, OccurredAt: begin}, {ID: 5, PromoteID: 1, PromoteLevel: 1, Kind: "locked", OldBusierID: 10, OccurredAt: begin.Add(-time.Hour)}, {ID: 6, PromoteID: 1, PromoteLevel: 1, Kind: "terminated", OldBusierID: 10, OccurredAt: begin.AddDate(0, 0, 1)}} + if err := sdk.Create(&events).Error; err != nil { + t.Fatal(err) + } + r := &dashboardRepo{data: testData{db}, databases: activationTestSources{"lx": sdk}} + out := &business.DashboardData{} + if err := r.events(context.Background(), business.AnalyticsScope{Viewer: business.UserScope{OnlySelf: true, SelfID: 7}, IncludeSub: true}, []string{"lx"}, w, out); err != nil { + t.Fatal(err) + } + if !out.EventsAvailable || out.Current.LockedChannels != 1 || out.Previous.LockedChannels != 1 || out.Current.TerminatedChannels != 1 { + t.Fatalf("events: %+v", out) + } + if err := sdk.Create(&channelEventPO{Kind: "coverage", OccurredAt: begin}).Error; err != nil { + t.Fatal(err) + } + incomplete := &business.DashboardData{} + if err := r.events(context.Background(), business.AnalyticsScope{Viewer: business.UserScope{All: true}}, []string{"lx"}, w, incomplete); err != nil { + t.Fatal(err) + } + if incomplete.EventsAvailable || !incomplete.EventsIncomplete { + t.Fatal("missing historical coverage was reported as real zero") + } +} + +func TestActivationStopsAtExternalTransferBeforeOAScan(t *testing.T) { + db, oa, _ := prepareActivations(t) + sources := activationSDKFixtures(t, db, oa) + sdk := sources["lx"] + if err := sdk.AutoMigrate(&channelEventPO{}); err != nil { + t.Fatal(err) + } + end := time.Unix(1600, 0).UTC() + if err := oa.Model(&activationPO{}).Where("id=1").Updates(map[string]any{"reach_at": time.Unix(1200, 0).UTC(), "unbind_at": nil}).Error; err != nil { + t.Fatal(err) + } + if err := sdk.Create(&channelEventPO{Kind: "transfer", PromoteID: 100, OldBusierAccount: "Own", OccurredAt: end}).Error; err != nil { + t.Fatal(err) + } + repo := NewActivationRepo(testData{db}, sources) + row, err := repo.Find(context.Background(), 1) + if err != nil { + t.Fatal(err) + } + if row.EndAt() == nil || !row.EndAt().Equal(end) { + t.Fatal("old owner can read past transfer") + } + items, err := repo.Spend(context.Background(), activationMetricQuery(row)) + if err != nil { + t.Fatal(err) + } + values := []string{} + for _, item := range items { + values = append(values, item.Amount) + } + sum, err := sumActivationMoney(values) + if err != nil || sum != "13.40" { + t.Fatalf("post-transfer spend leaked: %s %v", sum, err) + } +} diff --git a/internal/data/business/activation_metrics.go b/internal/data/business/activation_metrics.go new file mode 100644 index 0000000..c215925 --- /dev/null +++ b/internal/data/business/activation_metrics.go @@ -0,0 +1,192 @@ +package business + +import ( + "context" + "gorm.io/gorm" + "kra/internal/biz/business" + "math/big" + "sort" + "time" +) + +func (r *activationRepo) applyTransferEnd(ctx context.Context, row *business.Activation) error { + if row.BindAt == nil || row.PromoteID == nil || (row.Status != "reached" && row.ReachAt == nil) { + return nil + } + db, err := r.databases.Database(row.PlatformName) + if err != nil || db == nil { + return nil + } + db = db.WithContext(ctx) + ok, err := businessTableAvailable(db, "tab_promote_business_event") + if err != nil { + return err + } + if !ok { + return nil + } + begin := *row.BindAt + if row.ReachAt != nil && row.ReachAt.After(begin) { + begin = *row.ReachAt + } + q := db.Table("tab_promote_business_event").Where("kind='transfer' AND promote_id=? AND occurred_at>=?", *row.PromoteID, begin) + if row.BusierID > 0 { + q = q.Where("old_busier_id=?", row.BusierID) + } else { + q = q.Where("LOWER(old_busier_account)=LOWER(?)", row.BusierAccount) + } + var rows []struct{ OccurredAt time.Time } + if err := q.Select("occurred_at").Order("occurred_at,id").Limit(1).Find(&rows).Error; err != nil { + return business.ErrActivationSource + } + if len(rows) > 0 { + row.EndOnTransfer(rows[0].OccurredAt) + } + return nil +} + +func activationMetricQuery(row *business.Activation) business.ActivationSpendQuery { + q := business.ActivationSpendQuery{PlatformName: row.PlatformName, Channels: row.Channels, Start: row.BindAt.Unix()} + if row.PromoteID != nil { + q.PromoteID = *row.PromoteID + } + if end := row.EndAt(); end != nil { + n := end.Unix() + q.End = &n + } + return q +} +func sumActivationMoney(values []string) (string, error) { + total := new(big.Rat) + for _, v := range values { + n, ok := new(big.Rat).SetString(v) + if !ok { + return "", business.ErrStorage + } + total.Add(total, n) + } + return total.FloatString(2), nil +} + +// 激活数据与 OA 使用同一 SDK 数据源，避免商务报表副本延迟/测试副本不同导致金额不一致。 +func activationRoots(q business.ActivationSpendQuery) map[string][]uint64 { + roots := map[string][]uint64{} + keys := q.Channels + if len(keys) == 0 { + keys = []business.ChannelKey{{SourceDB: q.PlatformName, ID: q.PromoteID}} + } + for _, key := range keys { + if key.ID > 0 { + roots[key.SourceDB] = append(roots[key.SourceDB], key.ID) + } + } + return roots +} +func (r *activationRepo) activationMetricsAvailable(ctx context.Context, q business.ActivationSpendQuery) (bool, error) { + for source := range activationRoots(q) { + db, err := r.databases.Database(source) + if err != nil || db == nil { + return false, nil + } + ok, err := businessTablesAvailable(db.WithContext(ctx), "tab_promote", "tab_spend") + if err != nil { + return false, err + } + if !ok { + return false, nil + } + } + return true, nil +} +func nativeActivationTree(db *gorm.DB, roots []uint64) *gorm.DB { + return db.Table("tab_promote p").Where("p.id IN ? OR p.top_promote_id IN ? OR p.parent_id IN ?", roots, roots, roots) +} +func (r *activationRepo) Spend(ctx context.Context, q business.ActivationSpendQuery) ([]*business.ActivationSpendItem, error) { + out := []*business.ActivationSpendItem{} + for source, roots := range activationRoots(q) { + db, err := r.databases.Database(source) + if err != nil || db == nil { + return nil, business.ErrActivationMetrics + } + db = db.WithContext(ctx) + tree := nativeActivationTree(db, roots) + var n int64 + if err := tree.Count(&n).Error; err != nil { + return nil, business.ErrActivationMetrics + } + if n == 0 { + continue + } + query := db.Table("tab_spend s").Where("s.pay_status=1 AND s.pay_time>=?", q.Start).Where("EXISTS (?)", tree.Select("1").Where("p.id=s.promote_id")) + if q.End != nil { + query = query.Where("s.pay_time<=?", *q.End) + } + var rows []struct { + GameID uint64 + GameName, Amount string + } + if err := query.Select("s.game_id,COALESCE(MAX(s.game_name),'') AS game_name,COALESCE(SUM(s.pay_amount),0) AS amount").Group("s.game_id").Order("s.game_id").Scan(&rows).Error; err != nil { + return nil, business.ErrActivationMetrics + } + for _, row := range rows { + out = append(out, &business.ActivationSpendItem{PlatformName: source, GameName: row.GameName, Amount: row.Amount}) + } + } + sort.SliceStable(out, func(i, j int) bool { + a, okA := new(big.Rat).SetString(out[i].Amount) + b, okB := new(big.Rat).SetString(out[j].Amount) + if okA && okB && a.Cmp(b) != 0 { + return a.Cmp(b) > 0 + } + if out[i].PlatformName != out[j].PlatformName { + return out[i].PlatformName < out[j].PlatformName + } + return out[i].GameName < out[j].GameName + }) + return out, nil +} +func (r *activationRepo) registered(ctx context.Context, q business.ActivationSpendQuery, dedup bool) (*int64, error) { + total := int64(0) + identities := map[string]bool{} + for source, roots := range activationRoots(q) { + db, err := r.databases.Database(source) + if err != nil || db == nil { + return nil, nil + } + db = db.WithContext(ctx) + ok, err := businessTablesAvailable(db, "tab_user", "tab_promote") + if err != nil { + return nil, err + } + if !ok { + return nil, nil + } + cast := "CAST(r.register_time AS INTEGER)" + if db.Dialector.Name() == "mysql" { + cast = "CAST(r.register_time AS UNSIGNED)" + } + query := db.Table("tab_user r").Where("r.puid=0 AND "+cast+">=?", q.Start).Where("EXISTS (?)", nativeActivationTree(db, roots).Select("1").Where("p.id=r.promote_id")) + if q.End != nil { + query = query.Where(cast+"<=?", *q.End) + } + if dedup { + var cards []string + if err := query.Where("TRIM(COALESCE(r.idcard,''))<>''").Distinct("TRIM(r.idcard)").Pluck("TRIM(r.idcard)", &cards).Error; err != nil { + return nil, business.ErrActivationMetrics + } + for _, card := range cards { + identities[card] = true + } + } else { + var count int64 + if err := query.Distinct("r.id").Count(&count).Error; err != nil { + return nil, business.ErrActivationMetrics + } + total += count + } + } + if dedup { + total = int64(len(identities)) + } + return &total, nil +} diff --git a/internal/data/business/activation_pool.go b/internal/data/business/activation_pool.go new file mode 100644 index 0000000..2ab2930 --- /dev/null +++ b/internal/data/business/activation_pool.go @@ -0,0 +1,292 @@ +package business + +import ( + "context" + "errors" + "sort" + "strings" + "time" + + "gorm.io/gorm" + "kra/internal/biz/business" +) + +type activationPoolPO struct { + ID, PromoteID uint64 + PlatformName, PromoteAccount string + RegisterType int + ClaimFlag int + CreatedAt time.Time + DeletedAt *time.Time +} + +func (activationPoolPO) TableName() string { return "sys_platform_promote_active_pool" } + +type activationPoolRepo struct { + data Provider + databases PlatformProvider +} + +func NewActivationPoolRepo(data Provider, databases PlatformProvider) business.ActivationPoolRepo { + return &activationPoolRepo{data: data, databases: databases} +} +func (r *activationPoolRepo) oa(ctx context.Context) (*gorm.DB, error) { + db, err := r.databases.Database("oa") + if err != nil || db == nil { + return nil, business.ErrActivationSource + } + db = db.WithContext(ctx) + ok, err := businessTableAvailable(db, "sys_platform_promote_active_pool") + if err != nil || !ok { + return nil, business.ErrActivationSource + } + return db, nil +} +func toBizPool(p activationPoolPO) *business.ActivationPoolChannel { + kind := "个人" + if p.RegisterType == 1 { + kind = "公会" + } + return &business.ActivationPoolChannel{ID: p.ID, PromoteID: p.PromoteID, SourceDB: p.PlatformName, Name: p.PromoteAccount, ChannelType: kind, EnteredAt: p.CreatedAt} +} +func (r *activationPoolRepo) PoolChannel(ctx context.Context, id uint64) (*business.ActivationPoolChannel, error) { + db, err := r.oa(ctx) + if err != nil { + return nil, err + } + var p activationPoolPO + if err := db.Where("id=? AND claim_flag=0 AND deleted_at IS NULL", id).Take(&p).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, business.ErrActivationNotFound + } + return nil, business.ErrActivationSource + } + return toBizPool(p), nil +} +func (r *activationPoolRepo) BindingOptions(ctx context.Context, id uint) ([]*business.ActivationBindingOption, error) { + var rows []struct { + SourceDB, Account string + ID uint64 + Level int + } + err := r.data.DB().WithContext(ctx).Table("sys_user_bind_platform b").Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL").Where("b.user_id=? AND b.status='0' AND u.business_status=0 AND u.enable=1", id).Select("b.source_db,b.platform_busier_id AS id,b.platform_busier_account AS account,u.level").Order("b.source_db,b.platform_busier_id").Scan(&rows).Error + if err != nil { + return nil, business.ErrStorage + } + out := make([]*business.ActivationBindingOption, 0, len(rows)) + for _, p := range rows { + db, err := r.databases.Database(p.SourceDB) + if err != nil || db == nil { + return nil, business.ErrPlatformUnavailable + } + var sdk struct{ Level int } + if err := db.WithContext(ctx).Table("tab_promote_business").Where("id=?", p.ID).Select("level").Take(&sdk).Error; err != nil { + return nil, business.ErrPlatformUnavailable + } + p.Level = sdk.Level + out = append(out, &business.ActivationBindingOption{SourceDB: p.SourceDB, Account: p.Account, ID: p.ID, Level: p.Level}) + } + return out, nil +} +func (r *activationPoolRepo) Pool(ctx context.Context, f business.ActivationPoolFilter, opts ...business.ListOption) (*business.ActivationPoolPage, error) { + out := &business.ActivationPoolPage{Rows: []*business.ActivationPoolChannel{}, Available: true} + db, err := r.oa(ctx) + if err != nil { + out.Available = false + return out, nil + } + q := db.Model(&activationPoolPO{}).Where("deleted_at IS NULL AND claim_flag=0") + if len(f.Sources) > 0 { + q = q.Where("platform_name IN ?", f.Sources) + } + if f.Name != "" { + q = q.Where("promote_account LIKE ?", "%"+f.Name+"%") + } + if f.ChannelType == "公会" { + q = q.Where("register_type=1") + } else if f.ChannelType == "个人" { + q = q.Where("register_type<>1") + } + var pools []activationPoolPO + if err := q.Order("created_at DESC,id DESC").Find(&pools).Error; err != nil { + return nil, business.ErrActivationSource + } + if len(pools) == 0 { + return out, nil + } + bySource := map[string][]uint64{} + ids := make([]uint64, 0, len(pools)) + for _, p := range pools { + bySource[p.PlatformName] = append(bySource[p.PlatformName], p.PromoteID) + ids = append(ids, p.ID) + } + byKey := map[business.ChannelKey]channelPO{} + for source, channelIDs := range bySource { + sdk, err := r.databases.Database(source) + if err != nil || sdk == nil { + out.Available = false + return out, nil + } + sdk = sdk.WithContext(ctx) + for offset := 0; offset < len(channelIDs); offset += 500 { + var channels []channelPO + if err := sdk.Table("tab_promote").Select("id,status,lock_status,last_login_time,create_time").Where("id IN ?", channelIDs[offset:min(offset+500, len(channelIDs))]).Scan(&channels).Error; err != nil { + return nil, business.ErrActivationSource + } + for _, ch := range channels { + ch.SourceDB = source + byKey[business.ChannelKey{SourceDB: source, ID: ch.ID}] = ch + } + } + } + var rejected []struct { + PoolRecordID uint64 + RejectReason string + ReviewAt *time.Time + } + if err := db.Table("sys_promote_active_pool_bind b").Select("b.pool_record_id,b.reject_reason,b.review_at").Where("b.pool_record_id IN ? AND b.deleted_at IS NULL AND b.status='rejected'", ids).Where("NOT EXISTS (?)", db.Table("sys_promote_active_pool_bind newer").Select("1").Where("newer.pool_record_id=b.pool_record_id AND newer.deleted_at IS NULL AND newer.id>b.id")).Scan(&rejected).Error; err != nil { + return nil, business.ErrActivationSource + } + rejections := map[uint64]int{} + for i, p := range rejected { + rejections[p.PoolRecordID] = i + } + rows := []*business.ActivationPoolChannel{} + now := time.Now().Unix() + for _, p := range pools { + row := toBizPool(p) + ch, found := byKey[business.ChannelKey{SourceDB: p.PlatformName, ID: p.PromoteID}] + if !found { + continue + } + row.Status = "正常" + if (ch.Status != nil && *ch.Status == -2) || (ch.LockStatus != nil && *ch.LockStatus == -2) { + row.Status = "解除合作" + } else if (ch.Status != nil && *ch.Status == -1) || (ch.LockStatus != nil && *ch.LockStatus < 0) { + row.Status = "锁定" + } + if f.Status != "" && row.Status != f.Status { + continue + } + login := ch.LastLoginTime + if login == nil || *login == 0 { + login = ch.CreateTime + } + if login != nil { + days := (now - *login) / 86400 + if days < 0 { + days = 0 + } + row.NoLoginDays = &days + } + if i, ok := rejections[p.ID]; ok { + row.Rejected = true + row.RejectReason = rejected[i].RejectReason + row.RejectedAt = rejected[i].ReviewAt + } + rows = append(rows, row) + } + out.Total = int64(len(rows)) + o := business.NewListOptions(opts...) + start := min(o.Offset, len(rows)) + end := len(rows) + if o.Limit > 0 { + end = min(end, start+o.Limit) + } + out.Rows = rows[start:end] + return out, nil +} + +// Event tables are written transactionally by OA in each platform database. +type channelEventPO struct { + ID, PromoteID uint64 + PromoteAccount, Kind string + PromoteLevel int + OccurredAt time.Time + OldBusierID, NewBusierID uint64 + OldBusierAccount, NewBusierAccount, Operator, Reason string + FromPool, ToPool bool +} + +func (channelEventPO) TableName() string { return "tab_promote_business_event" } +func eventSources(sources []string) []string { + if len(sources) == 0 { + return []string{"lx", "hw", "yq"} + } + return sources +} +func (r *activationPoolRepo) Churn(ctx context.Context, f business.ActivationPoolFilter, opts ...business.ListOption) (*business.ActivationChurnPage, error) { + out := &business.ActivationChurnPage{Rows: []*business.ActivationChurn{}, Available: true} + o := business.NewListOptions(opts...) + for _, source := range eventSources(f.Sources) { + db, err := r.databases.Database(source) + if err != nil || db == nil { + out.Available = false + continue + } + db = db.WithContext(ctx) + ok, err := businessTableAvailable(db, "tab_promote_business_event") + if err != nil { + return nil, err + } + if !ok { + out.Available = false + continue + } + q := db.Model(&channelEventPO{}).Where("kind='transfer' AND from_pool=?", false) + if f.Name != "" { + q = q.Where("promote_account LIKE ?", "%"+f.Name+"%") + } + if f.Target != "" { + if strings.Contains("渠道激活池", f.Target) { + q = q.Where("(to_pool=? OR new_busier_account LIKE ?)", true, "%"+f.Target+"%") + } else { + q = q.Where("to_pool=? AND new_busier_account LIKE ?", false, "%"+f.Target+"%") + } + } + if f.Begin != nil { + q = q.Where("occurred_at>=?", *f.Begin) + } + if f.End != nil { + q = q.Where("occurred_at b.ID + }) + start := min(o.Offset, len(out.Rows)) + end := len(out.Rows) + if o.Limit > 0 { + end = min(end, start+o.Limit) + } + out.Rows = out.Rows[start:end] + return out, nil +} diff --git a/internal/data/business/activation_test.go b/internal/data/business/activation_test.go index 3fbb61b..d0b9648 100644 --- a/internal/data/business/activation_test.go +++ b/internal/data/business/activation_test.go @@ -37,7 +37,7 @@ func prepareActivations(t *testing.T) (*gorm.DB, *gorm.DB, business.ActivationRe end := time.Unix(2000, 0).UTC() channel := uint64(100) amount := "30.30" - pos := []activationPO{{ID: 1, PromoteID: &channel, PlatformName: "lx", PromoteAccount: "渠道一", BusierAccount: "Own", BindAt: &start, ReachAt: &end, Status: "reached", AccumAmount: &amount}, {ID: 2, PlatformName: "hw", BusierAccount: "own", Status: "pending"}, {ID: 3, PlatformName: "lx", BusierAccount: "Other", Status: "in_progress"}, {ID: 4, PlatformName: "lx", BusierAccount: "Own", Status: "rejected", DeletedAt: &end}} + pos := []activationPO{{ID: 1, PromoteID: &channel, PlatformName: "lx", PromoteAccount: "渠道一", BusierAccount: "Own", BindAt: &start, ReachAt: &end, UnbindAt: &end, Status: "reached", AccumAmount: &amount}, {ID: 2, PlatformName: "hw", BusierAccount: "own", Status: "pending"}, {ID: 3, PlatformName: "lx", BusierAccount: "Other", Status: "in_progress"}, {ID: 4, PlatformName: "lx", BusierAccount: "Own", Status: "rejected", DeletedAt: &end}} if err := oa.Create(&pos).Error; err != nil { t.Fatal(err) } @@ -53,7 +53,7 @@ func prepareActivations(t *testing.T) (*gorm.DB, *gorm.DB, business.ActivationRe t.Fatal(err) } } - return db, oa, NewActivationRepo(testData{db}, activationDatabases{oa}) + return db, oa, NewActivationRepo(testData{db}, activationSDKFixtures(t, db, oa)) } func TestActivationListsReadOAAndRestrictAccounts(t *testing.T) { db, _, repo := prepareActivations(t) @@ -67,7 +67,7 @@ func TestActivationListsReadOAAndRestrictAccounts(t *testing.T) { t.Fatalf("OA 列表或排序错误: %+v %v", page, err) } page, err = repo.List(ctx, business.ActivationScope{All: true}, business.ActivationFilter{PlatformNames: []string{"lx"}, Statuses: []string{"reached"}}) - if err != nil || page.Total != 1 || page.Rows[0].AccumAmount != "30.30" { + if err != nil || page.Total != 1 || page.Rows[0].AccumAmount != "33.60" { t.Fatalf("平台或状态筛选错误: %+v %v", page, err) } if _, err := repo.Find(ctx, 4); err != business.ErrActivationNotFound { @@ -87,7 +87,11 @@ func TestActivationListsReadOAAndRestrictAccounts(t *testing.T) { } } func TestActivationSpendTreePlatformAndInclusiveBounds(t *testing.T) { - db, _, repo := prepareActivations(t) + _, _, repo := prepareActivations(t) + db, err := repo.(*activationRepo).databases.Database("lx") + if err != nil { + t.Fatal(err) + } end := int64(2000) query := business.ActivationSpendQuery{PlatformName: "lx", PromoteID: 100, Start: 1000, End: &end} items, err := repo.Spend(context.Background(), query) @@ -113,3 +117,38 @@ func TestActivationSpendTreePlatformAndInclusiveBounds(t *testing.T) { t.Fatal("确定没有渠道时应返回真实空流水") } } + +func activationSDKFixtures(t *testing.T, db, oa *gorm.DB) activationTestSources { + t.Helper() + sources := activationTestSources{"oa": oa} + var files []struct{ Name, File string } + if err := db.Raw("PRAGMA database_list").Scan(&files).Error; err != nil { + t.Fatal(err) + } + var file string + for _, f := range files { + if f.Name == "main" { + file = f.File + } + } + for _, source := range []string{"lx", "hw", "yq"} { + sdk := openTestDB(t) + pool, _ := sdk.DB() + pool.SetMaxOpenConns(1) + if err := sdk.Exec("ATTACH DATABASE ? AS sample", file).Error; err != nil { + t.Fatal(err) + } + for _, table := range []string{"tab_promote", "tab_spend", "tab_user"} { + if db.Migrator().HasTable(table) { + if err := sdk.Exec("CREATE TABLE "+table+" AS SELECT * FROM sample."+table+" WHERE source_db=?", source).Error; err != nil { + t.Fatal(err) + } + } + } + if err := sdk.Exec("DETACH DATABASE sample").Error; err != nil { + t.Fatal(err) + } + sources[source] = sdk + } + return sources +} diff --git a/internal/data/business/alert_governance.go b/internal/data/business/alert_governance.go index 05ebd81..78fe34c 100644 --- a/internal/data/business/alert_governance.go +++ b/internal/data/business/alert_governance.go @@ -55,10 +55,13 @@ func toBizAlertExemption(p alertExemptionPO) *business.AlertExemption { return &business.AlertExemption{ID: p.ID, ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}, ChannelName: p.ChannelName, OwnerID: p.OwnerID, AddedBy: p.AddedBy, AddedName: p.AddedName, Reason: p.Reason, AddedAt: p.AddedAt, RemovedBy: p.RemovedBy, RemovedName: p.RemovedName, RemoveReason: p.RemoveReason, RemovedAt: p.RemovedAt} } -type alertGovernanceRepo struct{ data Provider } +type alertGovernanceRepo struct { + data Provider + databases PlatformProvider +} -func NewAlertGovernanceRepo(data Provider) business.AlertGovernanceRepo { - return &alertGovernanceRepo{data} +func NewAlertGovernanceRepo(data Provider, databases PlatformProvider) business.AlertGovernanceRepo { + return &alertGovernanceRepo{data: data, databases: databases} } func (r *alertGovernanceRepo) Candidates(ctx context.Context, scope business.UserScope, keyword string, limit int) ([]*business.AlertChannelOption, error) { @@ -196,47 +199,46 @@ func alertLostQuery(db *gorm.DB, sourceColumn, idColumn string, window business. return db.Table("tab_promote lp").Select("1").Where("lp.source_db="+sourceColumn+" AND lp.id="+idColumn). Where("(COALESCE(NULLIF(lp.last_login_time,0),lp.create_time)>0 AND COALESCE(NULLIF(lp.last_login_time,0),lp.create_time)?", cursor).Order("id").Limit(500).Find(&batch).Error; err != nil { - return nil, business.ErrStorage + for start := 0; start < len(entries); start += 500 { + batch := entries[start:min(start+500, len(entries))] + pairs := [][]any{} + for _, e := range batch { + pairs = append(pairs, []any{e.SourceDB, e.PromoteID}) } - if len(batch) == 0 { - break + var pools []activationPoolPO + if err := oa.Where("(platform_name,promote_id) IN ?", pairs).Find(&pools).Error; err != nil { + return nil, business.ErrActivationSource } - keys := make([]business.ChannelKey, 0, len(batch)) - for _, p := range batch { - keys = append(keys, business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}) + byKey := map[business.ChannelKey]activationPoolPO{} + for _, p := range pools { + byKey[business.ChannelKey{SourceDB: p.PlatformName, ID: p.PromoteID}] = p } - states, err := loadAlertLoss(db, keys, window) - if err != nil { - return nil, err - } - for _, p := range batch { - if lost := states[business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}]; lost != nil && *lost { - result = append(result, toBizAlertExemption(p)) + for _, e := range batch { + if p, ok := byKey[business.ChannelKey{SourceDB: e.SourceDB, ID: e.PromoteID}]; ok && (p.DeletedAt == nil || !p.CreatedAt.Before(e.AddedAt)) { + result = append(result, toBizAlertExemption(e)) } } - cursor = batch[len(batch)-1].ID } return result, nil } diff --git a/internal/data/business/alert_review_test.go b/internal/data/business/alert_review_test.go index fd61477..0f9b8be 100644 --- a/internal/data/business/alert_review_test.go +++ b/internal/data/business/alert_review_test.go @@ -41,7 +41,7 @@ func exerciseAlertOwnerRecheck(t *testing.T, db *gorm.DB) { } write(true) oldExemption := &business.AlertExemption{ChannelKey: business.ChannelKey{SourceDB: "lx", ID: 1}, OwnerID: 7, AddedBy: 7, AddedName: "原商务", Reason: "原商务的豁免", AddedAt: now} - if err := NewAlertGovernanceRepo(testData{db}).AddExemption(ctx, scope, oldExemption.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return oldExemption, nil }); err != nil { + if err := NewAlertGovernanceRepo(testData{db}, activationDatabases{db}).AddExemption(ctx, scope, oldExemption.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return oldExemption, nil }); err != nil { t.Fatal(err) } @@ -72,7 +72,7 @@ func exerciseAlertOwnerRecheck(t *testing.T, db *gorm.DB) { if len(histories) != 2 || histories[0].BusinessID != 7 || histories[0].AlertContent != "换绑前" || histories[1].BusinessID != 8 { t.Fatalf("旧商务预警不应被转移给新商务: %+v", histories) } - governance := NewAlertGovernanceRepo(testData{db}) + governance := NewAlertGovernanceRepo(testData{db}, activationDatabases{db}) called := false build := func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { called = true diff --git a/internal/data/business/alert_v2_test.go b/internal/data/business/alert_v2_test.go index 2232712..a183b10 100644 --- a/internal/data/business/alert_v2_test.go +++ b/internal/data/business/alert_v2_test.go @@ -97,7 +97,7 @@ func TestAlertV2ExemptionHistoryAndAutomaticRemoval(t *testing.T) { if err := db.Exec("INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,status) VALUES(7,'lx',7,'0')").Error; err != nil { t.Fatal(err) } - r := NewAlertGovernanceRepo(testData{db}) + r := NewAlertGovernanceRepo(testData{db}, activationDatabases{db}) ctx := context.Background() entry := &business.AlertExemption{ChannelKey: business.ChannelKey{SourceDB: "lx", ID: 1}, ChannelName: "父渠道", OwnerID: 7, AddedBy: 7, AddedName: "一级商务", AddedAt: now, Reason: "长期合作"} if err := r.AddExemption(ctx, business.UserScope{All: true}, entry.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return entry, nil }); err != nil { @@ -118,7 +118,16 @@ func TestAlertV2ExemptionHistoryAndAutomaticRemoval(t *testing.T) { if err != nil || written { t.Fatalf("免预警仍生成: %v %v", written, err) } - // 父渠道自身近 30 天流水仅 100 元，满足激活池条件；子渠道流水不冒充它自身的流水。 + // 本地达到流失阈值不等于 OA 已入池；只有收到 OA 入池记录才移出豁免。 + if err := db.AutoMigrate(&activationPoolPO{}); err != nil { + t.Fatal(err) + } + if rows, err := r.LostExemptions(ctx, business.AlertLossAt(now)); err != nil || len(rows) != 0 { + t.Fatalf("未入池误清理: %v %v", rows, err) + } + if err := db.Create(&activationPoolPO{ID: 1, PromoteID: 1, PlatformName: "lx", CreatedAt: now}).Error; err != nil { + t.Fatal(err) + } uc := business.NewAlertGovernanceUsecase(r, nil) if err := uc.Reconcile(ctx, now); err != nil { t.Fatal(err) diff --git a/internal/data/business/dashboard.go b/internal/data/business/dashboard.go index 8872611..ac492f7 100644 --- a/internal/data/business/dashboard.go +++ b/internal/data/business/dashboard.go @@ -7,9 +7,14 @@ import ( "kra/internal/biz/business" ) -type dashboardRepo struct{ data Provider } +type dashboardRepo struct { + data Provider + databases PlatformProvider +} -func NewDashboardRepo(data Provider) business.DashboardRepo { return &dashboardRepo{data: data} } +func NewDashboardRepo(data Provider, databases PlatformProvider) business.DashboardRepo { + return &dashboardRepo{data: data, databases: databases} +} 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} @@ -90,5 +95,8 @@ func (r *dashboardRepo) Overview(ctx context.Context, scope business.AnalyticsSc 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 + if err := r.events(ctx, scope, sources, w, out); err != nil { + return nil, err + } return out, nil } diff --git a/internal/data/business/dashboard_events.go b/internal/data/business/dashboard_events.go new file mode 100644 index 0000000..0eefcc6 --- /dev/null +++ b/internal/data/business/dashboard_events.go @@ -0,0 +1,68 @@ +package business + +import ( + "context" + "kra/internal/biz/business" +) + +func (r *dashboardRepo) events(ctx context.Context, scope business.AnalyticsScope, sources []string, w business.DashboardWindow, out *business.DashboardData) error { + out.EventsAvailable = true + for _, source := range eventSources(sources) { + db, err := r.databases.Database(source) + if err != nil || db == nil { + out.EventsAvailable = false + continue + } + db = db.WithContext(ctx) + ok, err := businessTableAvailable(db, "tab_promote_business_event") + if err != nil { + return err + } + if !ok { + out.EventsAvailable = false + continue + } + var coverage []channelEventPO + if err := db.Where("kind='coverage'").Order("occurred_at,id").Limit(1).Find(&coverage).Error; err != nil { + return business.ErrActivationSource + } + if len(coverage) > 0 && w.PreviousBegin.Before(coverage[0].OccurredAt) { + out.EventsAvailable = false + out.EventsIncomplete = true + } + q := db.Model(&channelEventPO{}).Where("kind IN ? AND occurred_at>=? AND occurred_at=?) AS current,COUNT(DISTINCT promote_id) AS count", w.Begin).Group("kind,current").Scan(&rows).Error; err != nil { + return business.ErrActivationSource + } + for _, row := range rows { + target := &out.Previous + if row.Current { + target = &out.Current + } + if row.Kind == "locked" { + target.LockedChannels += row.Count + } else { + target.TerminatedChannels += row.Count + } + } + } + return nil +} diff --git a/internal/data/business/management_test.go b/internal/data/business/management_test.go index 70dea5c..6dcdb3f 100644 --- a/internal/data/business/management_test.go +++ b/internal/data/business/management_test.go @@ -167,7 +167,7 @@ func TestPerformanceOwnerAggregationAndDrilldownUseSameScope(t *testing.T) { 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)}) + data, err := NewDashboardRepo(testData{db}, activationDatabases{}).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) } diff --git a/internal/data/business/migrations.go b/internal/data/business/migrations.go index 2083e9f..81b7ba9 100644 --- a/internal/data/business/migrations.go +++ b/internal/data/business/migrations.go @@ -7,7 +7,12 @@ import ( ) func Migrations() []migration.Step { - return []migration.Step{swmigration.StateMigration(), {ID: "202611060023_business_binding_status_index", Migrate: migrateBindingStatusIndex}, {ID: "202611060022_business_promote_id_capacity", Migrate: migratePromoteIDCapacity}, {ID: "202611050020_business_alert_v2", Migrate: migrateAlertV2}, {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.Step{swmigration.StateMigration(), {ID: "202611060024_business_oa_remark", Migrate: func(db *gorm.DB) error { + if db.Migrator().HasTable(&accountPO{}) && !db.Migrator().HasColumn(&accountPO{}, "remark") { + return db.Migrator().AddColumn(&accountPO{}, "Remark") + } + return nil + }}, {ID: "202611060023_business_binding_status_index", Migrate: migrateBindingStatusIndex}, {ID: "202611060022_business_promote_id_capacity", Migrate: migratePromoteIDCapacity}, {ID: "202611050020_business_alert_v2", Migrate: migrateAlertV2}, {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{}) diff --git a/internal/data/data.go b/internal/data/data.go index 2199d3d..c52de72 100644 --- a/internal/data/data.go +++ b/internal/data/data.go @@ -44,6 +44,7 @@ var ProviderSet = wire.NewSet( databusiness.NewAlertRuleRepo, databusiness.NewAlertRecordRepo, databusiness.NewActivationRepo, + databusiness.NewActivationPoolRepo, databusiness.NewChannelRepo, databusiness.NewChannelWriteRepo, databusiness.NewBusinessPortraitRepo, diff --git a/internal/integration/oa/activation.go b/internal/integration/oa/activation.go new file mode 100644 index 0000000..e3411b2 --- /dev/null +++ b/internal/integration/oa/activation.go @@ -0,0 +1,73 @@ +package oa + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/go-kratos/kratos/v3/errors" + "kra/internal/biz/business" + "kra/internal/config" +) + +type activationGateway struct { + config *config.Store + client *http.Client +} + +func NewActivationGateway(store *config.Store) business.ActivationGateway { + return &activationGateway{config: store, client: &http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}} +} +func (g *activationGateway) Apply(ctx context.Context, a business.ActivationApplication) error { + cfg := g.config.Snapshot() + if cfg == nil || cfg.BusinessOA == nil || cfg.BusinessOA.URL == "" || cfg.BusinessOA.Key == "" { + return business.ErrActivationSource + } + body, err := json.Marshal(struct { + UserID string `json:"userId"` + PoolID string `json:"poolId"` + BusierID string `json:"busierId"` + SourceDB string `json:"sourceDb"` + }{strconv.FormatUint(uint64(a.UserID), 10), strconv.FormatUint(a.PoolID, 10), strconv.FormatUint(a.BusierID, 10), a.SourceDB}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(cfg.BusinessOA.URL, "/")+"/internal/business/activation/apply", bytes.NewReader(body)) + if err != nil { + return business.ErrActivationSource + } + stamp := strconv.FormatInt(time.Now().Unix(), 10) + mac := hmac.New(sha256.New, []byte(cfg.BusinessOA.Key)) + mac.Write([]byte(req.Method + "\n" + req.URL.EscapedPath() + "\n" + stamp + "\n")) + mac.Write(body) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Business-Timestamp", stamp) + req.Header.Set("X-Business-Signature", hex.EncodeToString(mac.Sum(nil))) + res, err := g.client.Do(req) + if err != nil { + return business.ErrActivationSource + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return business.ErrActivationSource + } + var result struct { + Code *int `json:"code"` + Message string `json:"msg"` + } + if err := json.NewDecoder(io.LimitReader(res.Body, 65536)).Decode(&result); err != nil || result.Code == nil { + return business.ErrActivationSource + } + if *result.Code != 0 { + return errors.BadRequest("SW_ACTIVATION_APPLY", result.Message) + } + return nil +} diff --git a/internal/integration/oa/activation_test.go b/internal/integration/oa/activation_test.go new file mode 100644 index 0000000..dc3943d --- /dev/null +++ b/internal/integration/oa/activation_test.go @@ -0,0 +1,51 @@ +package oa + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "kra/internal/biz/business" + "kra/internal/config" + "net/http" + "net/http/httptest" + "testing" +) + +func TestActivationGatewayPreservesIDsAndSignsOAPath(t *testing.T) { + key := "test-key" + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + body, _ := io.ReadAll(r.Body) + mac := hmac.New(sha256.New, []byte(key)) + mac.Write([]byte(r.Method + "\n" + r.URL.EscapedPath() + "\n" + r.Header.Get("X-Business-Timestamp") + "\n")) + mac.Write(body) + if hex.EncodeToString(mac.Sum(nil)) != r.Header.Get("X-Business-Signature") || r.URL.Path != "/oa/internal/business/activation/apply" { + t.Error("invalid signature/path") + } + var values map[string]string + if err := json.Unmarshal(body, &values); err != nil { + t.Error(err) + } + if values["poolId"] != "9007199254740993" || values["busierId"] != "42" || values["userId"] != "7" { + t.Errorf("identity precision: %v", values) + } + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"code":0,"msg":"申请成功"}`) + })) + defer server.Close() + store := config.NewStore(&config.Config{BusinessOA: &config.BusinessOA{URL: server.URL + "/oa/", Key: key}}) + if err := NewActivationGateway(store).Apply(context.Background(), business.ActivationApplication{UserID: 7, PoolID: 9007199254740993, BusierID: 42, SourceDB: "lx"}); err != nil { + t.Fatal(err) + } + store.Replace(&config.Config{}) + if err := NewActivationGateway(store).Apply(context.Background(), business.ActivationApplication{}); err != business.ErrActivationSource { + t.Fatal("missing configuration must fail closed") + } + if calls != 1 { + t.Fatal("unexpected request") + } +} diff --git a/internal/integration/provider.go b/internal/integration/provider.go index ba11cb8..7d1a4fc 100644 --- a/internal/integration/provider.go +++ b/internal/integration/provider.go @@ -11,6 +11,7 @@ import ( "kra/internal/integration/email" "kra/internal/integration/gameplatform" mqintegration "kra/internal/integration/mq" + "kra/internal/integration/oa" paymentintegration "kra/internal/integration/payment" "kra/internal/integration/sms" "kra/internal/integration/storage" @@ -23,6 +24,7 @@ import ( ) var ProviderSet = wire.NewSet( + oa.NewActivationGateway, gameplatform.NewPasswords, gameplatform.NewPasswordEncoder, gameplatform.NewPasswordSMS, diff --git a/internal/modules/business/activation_surface.go b/internal/modules/business/activation_surface.go new file mode 100644 index 0000000..ff7ed1d --- /dev/null +++ b/internal/modules/business/activation_surface.go @@ -0,0 +1,41 @@ +package business + +import ( + "gorm.io/gorm" + datasystem "kra/internal/data/system" + "kra/internal/modules/surface" + "kra/pkg/module" +) + +func migrateActivationSurface(db *gorm.DB) error { + if !db.Migrator().HasTable("sys_authorities") { + return nil + } + var roots int64 + if err := db.Table("sys_authorities").Where("authority_id=888 AND deleted_at IS NULL").Count(&roots).Error; err != nil { + return err + } + if roots == 0 { + return nil + } + if db.Migrator().HasTable("sys_base_menus") { + if err := db.Table("sys_base_menus").Where("name=?", "sw_ChannelActivationList").Update("title", "激活列表").Error; err != nil { + return err + } + } + apis := surface.APIsForPrefix("/sw/promoteActivePool/", 0) + if err := datasystem.EnsureAdminSurface(db, module.Surface{APIs: apis}, 888); err != nil { + return err + } + // 申请绑定与流失记录对各级商务开放，升级旧的仅三级菜单授权。 + if db.Migrator().HasTable("sys_authority_menus") && db.Migrator().HasTable("sys_base_menus") { + if err := db.Exec(`INSERT INTO sys_authority_menus(sys_authority_authority_id,sys_base_menu_id) + SELECT a.authority_id,m.id FROM sys_authorities a CROSS JOIN sys_base_menus m + WHERE a.business_level IN (1,2,3) AND a.deleted_at IS NULL AND m.deleted_at IS NULL + AND m.name IN ('sw_ChannelActivationList','sw_BusMgmt') + AND NOT EXISTS (SELECT 1 FROM sys_authority_menus x WHERE x.sys_authority_authority_id=a.authority_id AND x.sys_base_menu_id=m.id)`).Error; err != nil { + return err + } + } + return datasystem.GrantMenuAPIs(db, "sw_ChannelActivationList", apis) +} diff --git a/internal/modules/business/activation_surface_test.go b/internal/modules/business/activation_surface_test.go new file mode 100644 index 0000000..07a0aa2 --- /dev/null +++ b/internal/modules/business/activation_surface_test.go @@ -0,0 +1,43 @@ +package business + +import ( + "github.com/glebarez/sqlite" + "gorm.io/gorm" + datasystem "kra/internal/data/system" + "path/filepath" + "testing" +) + +func TestActivationUpgradeGivesAllBusinessLevelsApplicationAccess(t *testing.T) { + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "surface.db")), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + pool, _ := db.DB() + defer pool.Close() + for _, step := range datasystem.Migrations() { + if err := step.Migrate(db); err != nil { + t.Fatal(err) + } + } + for _, sql := range []string{"INSERT INTO sys_authorities(authority_id,business_level) VALUES(888,NULL),(41,1),(57,2),(83,3),(100,NULL)", "INSERT INTO sys_base_menus(id,name,title) VALUES(1,'sw_BusMgmt','商务管理'),(2,'sw_ChannelActivationList','渠道激活')"} { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + for i := 0; i < 2; i++ { + if err := migrateActivationSurface(db); err != nil { + t.Fatal(err) + } + } + var count int64 + if err := db.Table("sys_authority_menus").Where("sys_base_menu_id=2 AND sys_authority_authority_id IN (41,57,83)").Count(&count).Error; err != nil || count != 3 { + t.Fatalf("business menu access %d: %v", count, err) + } + if err := db.Table("casbin_rule").Where("v0 IN ('41','57','83') AND v1='/sw/promoteActivePool/apply' AND v2='POST'").Count(&count).Error; err != nil || count != 3 { + t.Fatalf("application grants %d: %v", count, err) + } + if err := db.Table("casbin_rule").Where("v0='100' AND v1='/sw/promoteActivePool/apply'").Count(&count).Error; err != nil || count != 0 { + t.Fatal("unrelated role received grant") + } +} diff --git a/internal/modules/business/definition.go b/internal/modules/business/definition.go index a70edee..e6d84de 100644 --- a/internal/modules/business/definition.go +++ b/internal/modules/business/definition.go @@ -106,6 +106,7 @@ func Definition() module.Definition { }) }}) migrations = append(migrations, migration.Step{ID: "202611050021_business_alert_surface", Migrate: migrateAlertSurface}) + migrations = append(migrations, migration.Step{ID: "202611060025_business_activation_surface", Migrate: migrateActivationSurface}) definition := module.Definition{Name: "business", Migrations: migrations, Prerequisites: databusiness.CheckPrerequisites, ValidateSchema: databusiness.CheckSchema, TimedTasks: append([]module.TimedTask{alertDetectionTask}, tasks...), Surface: module.Surface{ Menus: []module.Menu{ alertExemptionMenu, @@ -119,7 +120,7 @@ func Definition() module.Definition { {Name: "sw_RegistrationDetails", Path: "RegistrationDetails", ParentName: "sw_DataAnalysis", Component: "view/sw/RegistrationDetails.vue", Title: "注册明细", Icon: databusiness.MenuIcon("sw_RegistrationDetails"), Sort: 3}, {Name: "sw_RechargeDetails", Path: "RechargeDetails", ParentName: "sw_DataAnalysis", Component: "view/sw/RechargeDetails.vue", Title: "充值明细", Icon: databusiness.MenuIcon("sw_RechargeDetails"), Sort: 2}, {Name: "sw_ChannelList", Path: "ChannelList", ParentName: "sw_BusMgmt", Component: "view/sw/ChannelList.vue", Title: "渠道列表", Icon: databusiness.MenuIcon("sw_ChannelList"), Sort: 2}, - {Name: "sw_ChannelActivationList", Path: "ChannelActivationList", ParentName: "sw_BusMgmt", Component: "view/sw/ChannelActivationList.vue", Title: "渠道激活", Icon: databusiness.MenuIcon("sw_ChannelActivationList"), Sort: 3}, + {Name: "sw_ChannelActivationList", Path: "ChannelActivationList", ParentName: "sw_BusMgmt", Component: "view/sw/ChannelActivationList.vue", Title: "激活列表", Icon: databusiness.MenuIcon("sw_ChannelActivationList"), Sort: 3}, {Name: "sw_AlertList", Path: "AlertList", ParentName: "sw_AlertCenter", Component: "view/sw/AlertList.vue", Title: "预警列表", Icon: databusiness.MenuIcon("sw_AlertList"), Sort: 1}, {Name: "sw_AlertCenter", Path: "sw-AlertCenter", Component: "view/routerHolder.vue", Title: "预警中心", Icon: databusiness.MenuIcon("sw_AlertCenter"), Sort: 4}, {Name: "sw_AlertRules", Path: "AlertRules", ParentName: "sw_AlertCenter", Component: "view/sw/AlertRules.vue", Title: "预警规则", Icon: databusiness.MenuIcon("sw_AlertRules"), Sort: 2}, diff --git a/internal/modules/catalog_test.go b/internal/modules/catalog_test.go index a161e98..292c281 100644 --- a/internal/modules/catalog_test.go +++ b/internal/modules/catalog_test.go @@ -25,16 +25,16 @@ func TestCatalogContainsBuiltInModulesInDependencyOrder(t *testing.T) { t.Fatalf("definition[%d] = %q, want %q", index, got, name) } } - if got := catalog.MigrationSteps(); len(got) != 39 { - t.Fatalf("module migrations = %d, want 39", len(got)) + if got := catalog.MigrationSteps(); len(got) != 41 { + t.Fatalf("module migrations = %d, want 41", len(got)) } var menus, apis int for _, definition := range catalog.Definitions { menus += len(definition.Surface.Menus) apis += len(definition.Surface.APIs) } - if menus != 21 || apis != 90 { - t.Fatalf("admin surface = %d menus/%d APIs, want 21/90", menus, apis) + if menus != 21 || apis != 94 { + t.Fatalf("admin surface = %d menus/%d APIs, want 21/94", menus, apis) } if got := catalog.DefaultTimedTasks(); len(got) != 5 { t.Fatalf("default timed tasks = %d, want 5", len(got)) diff --git a/internal/routecatalog/catalog.go b/internal/routecatalog/catalog.go index e3edd6c..d538dd5 100644 --- a/internal/routecatalog/catalog.go +++ b/internal/routecatalog/catalog.go @@ -64,6 +64,10 @@ var routes = map[string]routeValue{ "GET /sw/promote/bindOptions": {group: "渠道管理", description: "查询可用游戏商务绑定"}, "GET /sw/promote/game-distribution": {group: "渠道管理", description: "查询推广游戏累计分布"}, "POST /sw/promote": {group: "渠道管理", description: "创建源平台渠道", audit: true}, + "GET /sw/promoteActivePool/pool": {group: "渠道激活", description: "查询可申请渠道"}, + "GET /sw/promoteActivePool/options": {group: "渠道激活", description: "查询本人已审核游戏商务"}, + "POST /sw/promoteActivePool/apply": {group: "渠道激活", description: "申请激活池渠道"}, + "GET /sw/promoteActivePool/churn": {group: "渠道激活", description: "查询渠道流失记录"}, "GET /sw/promoteActivePool/bindList": {group: "渠道激活", description: "查询 OA 渠道激活绑定"}, "GET /sw/promoteActivePool/bindSpendDetail": {group: "渠道激活", description: "查询绑定期间游戏流水"}, "GET /sw/alertRecord/list": {group: "预警记录", description: "查询预警列表及状态统计"}, diff --git a/internal/server/handler/activation_pool.go b/internal/server/handler/activation_pool.go new file mode 100644 index 0000000..564855e --- /dev/null +++ b/internal/server/handler/activation_pool.go @@ -0,0 +1,54 @@ +package handler + +import ( + "github.com/gin-gonic/gin" + "github.com/go-kratos/kratos/v3/errors" + "kra/internal/service/dto" +) + +func (h *Activation) Pool(c *gin.Context) { + var req dto.ActivationPoolQuery + if c.ShouldBindQuery(&req) != nil { + Fail(c, "申请绑定查询参数不正确") + return + } + result, err := h.service.Pool(c.Request.Context(), &req) + if err != nil { + Fail(c, errors.FromError(err).Message) + return + } + OKWithData(c, result) +} +func (h *Activation) Options(c *gin.Context) { + result, err := h.service.Options(c.Request.Context()) + if err != nil { + Fail(c, errors.FromError(err).Message) + return + } + OKWithData(c, result) +} +func (h *Activation) Apply(c *gin.Context) { + var req dto.ActivationApplyRequest + if c.ShouldBindJSON(&req) != nil { + Fail(c, "激活申请参数不正确") + return + } + if err := h.service.Apply(c.Request.Context(), &req); err != nil { + Fail(c, errors.FromError(err).Message) + return + } + OKWithData(c, nil) +} +func (h *Activation) Churn(c *gin.Context) { + var req dto.ActivationPoolQuery + if c.ShouldBindQuery(&req) != nil { + Fail(c, "流失记录查询参数不正确") + return + } + result, err := h.service.Churn(c.Request.Context(), &req) + if err != nil { + Fail(c, errors.FromError(err).Message) + return + } + OKWithData(c, result) +} diff --git a/internal/server/router/routes.go b/internal/server/router/routes.go index c24893f..ae9edca 100644 --- a/internal/server/router/routes.go +++ b/internal/server/router/routes.go @@ -49,6 +49,10 @@ func (r *Routes) RegisterRoutes(public, private *gin.RouterGroup, engine *gin.En }, func() { RegisterChannel(private, r.handlers.Channel) }, func() { + private.GET("/sw/promoteActivePool/pool", r.handlers.Activation.Pool) + private.GET("/sw/promoteActivePool/options", r.handlers.Activation.Options) + private.POST("/sw/promoteActivePool/apply", r.handlers.Activation.Apply) + private.GET("/sw/promoteActivePool/churn", r.handlers.Activation.Churn) private.GET("/sw/promoteActivePool/bindList", r.handlers.Activation.List) private.GET("/sw/promoteActivePool/bindSpendDetail", r.handlers.Activation.SpendDetail) }, diff --git a/internal/service/business/activation.go b/internal/service/business/activation.go index 6db6ed1..eeb524b 100644 --- a/internal/service/business/activation.go +++ b/internal/service/business/activation.go @@ -12,10 +12,13 @@ type ActivationUsecase interface { List(context.Context, business.ActivationFilter, ...business.ListOption) (*business.ActivationPage, error) SpendDetail(context.Context, uint64) (*business.ActivationSpendDetail, error) } -type ActivationService struct{ uc ActivationUsecase } +type ActivationService struct { + uc ActivationUsecase + pool *business.ActivationPoolUsecase +} -func NewActivationService(uc *business.ActivationUsecase) *ActivationService { - return &ActivationService{uc: uc} +func NewActivationService(uc *business.ActivationUsecase, pool *business.ActivationPoolUsecase) *ActivationService { + return &ActivationService{uc: uc, pool: pool} } func splitQueryValues(values []string) []string { @@ -57,7 +60,7 @@ func (s *ActivationService) List(ctx context.Context, req *dto.ActivationQuery) } out := &dto.ActivationPage{Rows: []*dto.ActivationResponse{}, Total: result.Total, Available: result.Available} for _, r := range result.Rows { - out.Rows = append(out.Rows, &dto.ActivationResponse{ID: r.ID, PromoteAccount: r.PromoteAccount, PlatformName: r.PlatformName, PlatformTitle: business.PlatformName(r.PlatformName), BusierAccount: r.BusierAccount, ApplicantOA: r.ApplicantOA, BindAt: localTime(r.BindAt), UnbindAt: localTime(r.UnbindAt), ReachAt: localTime(r.ReachAt), EndAt: localTime(r.EndAt()), AccumAmount: r.AccumAmount, Status: r.Status, StatusText: business.ActivationStatusLabel(r.Status)}) + out.Rows = append(out.Rows, &dto.ActivationResponse{CreatedAt: localTime(r.CreatedAt), RegisteredPlayers: r.RegisteredPlayers, MetricsAvailable: r.MetricsAvailable, ID: r.ID, PromoteAccount: r.PromoteAccount, PlatformName: r.PlatformName, PlatformTitle: business.PlatformName(r.PlatformName), BusierAccount: r.BusierAccount, ApplicantOA: r.ApplicantOA, BindAt: localTime(r.BindAt), UnbindAt: localTime(r.UnbindAt), ReachAt: localTime(r.ReachAt), EndAt: localTime(r.EndAt()), AccumAmount: r.AccumAmount, Status: r.Status, StatusText: business.ActivationStatusLabel(r.Status)}) } return out, nil } @@ -72,7 +75,12 @@ func (s *ActivationService) SpendDetail(ctx context.Context, id uint64) (*dto.Ac a := result.Activation out := &dto.ActivationSpendDetail{PromoteAccount: a.PromoteAccount, BindAt: localTime(a.BindAt), EndAt: localTime(a.EndAt()), List: []*dto.ActivationSpendItem{}, TotalAmount: result.TotalAmount} for _, item := range result.Items { - out.List = append(out.List, &dto.ActivationSpendItem{PlatformTitle: business.PlatformName(a.PlatformName), GameName: item.GameName, TotalAmount: item.Amount}) + out.List = append(out.List, &dto.ActivationSpendItem{PlatformTitle: business.PlatformName(func() string { + if item.PlatformName != "" { + return item.PlatformName + } + return a.PlatformName + }()), GameName: item.GameName, TotalAmount: item.Amount}) } return out, nil } diff --git a/internal/service/business/activation_pool.go b/internal/service/business/activation_pool.go new file mode 100644 index 0000000..7a4698c --- /dev/null +++ b/internal/service/business/activation_pool.go @@ -0,0 +1,112 @@ +package business + +import ( + "context" + "kra/internal/biz/business" + "kra/internal/service/dto" + "strings" + "time" +) + +func activationPoolFilter(req *dto.ActivationPoolQuery) (business.ActivationPoolFilter, []business.ListOption, error) { + f := business.ActivationPoolFilter{} + if req == nil { + return f, nil, business.ErrActivationInput + } + for _, source := range splitQueryValues(req.Sources) { + value, err := business.NormalizePlatform(source) + if err != nil { + return f, nil, err + } + f.Sources = append(f.Sources, value) + } + f.Name = strings.TrimSpace(req.Name) + f.Target = strings.TrimSpace(req.Target) + f.Status = req.Status + f.ChannelType = req.ChannelType + if f.ChannelType != "" && f.ChannelType != "公会" && f.ChannelType != "个人" { + return f, nil, business.ErrActivationInput + } + if f.Status != "" && f.Status != "正常" && f.Status != "锁定" && f.Status != "解除合作" { + return f, nil, business.ErrActivationInput + } + loc := time.FixedZone("Asia/Shanghai", 8*3600) + if req.Begin != "" { + v, err := time.ParseInLocation("2006-01-02", req.Begin, loc) + if err != nil { + return f, nil, business.ErrActivationInput + } + f.Begin = &v + } + if req.End != "" { + v, err := time.ParseInLocation("2006-01-02", req.End, loc) + if err != nil { + return f, nil, business.ErrActivationInput + } + f.End = &v + } + if f.Begin != nil && f.End != nil && f.Begin.After(*f.End) { + return f, nil, business.ErrActivationInput + } + page, size := req.Page, req.PageSize + if page < 1 { + page = 1 + } + if size < 1 { + size = 10 + } + if size > 100 { + size = 100 + } + if page > int(^uint(0)>>1)/size { + return f, nil, business.ErrActivationInput + } + return f, []business.ListOption{business.ListOffset((page - 1) * size), business.ListLimit(size)}, nil +} +func (s *ActivationService) Pool(ctx context.Context, req *dto.ActivationPoolQuery) (*dto.ActivationPoolPage, error) { + f, opts, err := activationPoolFilter(req) + if err != nil { + return nil, err + } + page, err := s.pool.Pool(ctx, f, opts...) + if err != nil { + return nil, err + } + out := &dto.ActivationPoolPage{Rows: []*dto.ActivationPoolRow{}, Total: page.Total, Available: page.Available} + for _, r := range page.Rows { + out.Rows = append(out.Rows, &dto.ActivationPoolRow{ID: r.ID, SourceDB: r.SourceDB, PlatformTitle: business.PlatformName(r.SourceDB), Name: r.Name, ChannelType: r.ChannelType, Status: r.Status, NoLoginDays: r.NoLoginDays, Rejected: r.Rejected, RejectReason: r.RejectReason, RejectedAt: localTime(r.RejectedAt)}) + } + return out, nil +} +func (s *ActivationService) Options(ctx context.Context) ([]*dto.ActivationBindingOption, error) { + rows, err := s.pool.Options(ctx) + if err != nil { + return nil, err + } + out := []*dto.ActivationBindingOption{} + for _, r := range rows { + out = append(out, &dto.ActivationBindingOption{ID: r.ID, SourceDB: r.SourceDB, Account: r.Account, Level: r.Level}) + } + return out, nil +} +func (s *ActivationService) Apply(ctx context.Context, req *dto.ActivationApplyRequest) error { + if req == nil || req.PoolID == 0 || req.BusierID == 0 { + return business.ErrActivationInput + } + return s.pool.Apply(ctx, req.PoolID, req.BusierID, req.SourceDB) +} +func (s *ActivationService) Churn(ctx context.Context, req *dto.ActivationPoolQuery) (*dto.ActivationChurnPage, error) { + f, opts, err := activationPoolFilter(req) + if err != nil { + return nil, err + } + page, err := s.pool.Churn(ctx, f, opts...) + if err != nil { + return nil, err + } + out := &dto.ActivationChurnPage{Rows: []*dto.ActivationChurnRow{}, Total: page.Total, Available: page.Available} + for _, r := range page.Rows { + out.Rows = append(out.Rows, &dto.ActivationChurnRow{ID: r.ID, SourceDB: r.SourceDB, PlatformTitle: business.PlatformName(r.SourceDB), Name: r.Name, From: r.From, To: r.To, Reason: r.Reason, Operator: r.Operator, At: *localTime(&r.At), ToPool: r.ToPool}) + } + return out, nil +} diff --git a/internal/service/dto/activation.go b/internal/service/dto/activation.go index af5121e..ed3e515 100644 --- a/internal/service/dto/activation.go +++ b/internal/service/dto/activation.go @@ -9,19 +9,22 @@ type ActivationQuery struct { BusierAccount string `form:"busierAccount"` } type ActivationResponse struct { - ID uint64 `json:"id,string"` - PromoteAccount string `json:"promoteAccount"` - PlatformName string `json:"platformName"` - PlatformTitle string `json:"platformTitle"` - BusierAccount string `json:"busierAccount"` - ApplicantOA string `json:"applicantOa"` - BindAt *string `json:"bindAt"` - UnbindAt *string `json:"unbindAt"` - ReachAt *string `json:"reachAt"` - EndAt *string `json:"endAt"` - AccumAmount string `json:"accumAmount"` - Status string `json:"status"` - StatusText string `json:"statusText"` + CreatedAt *string `json:"createdAt"` + RegisteredPlayers *int64 `json:"registeredPlayers"` + MetricsAvailable bool `json:"metricsAvailable"` + ID uint64 `json:"id,string"` + PromoteAccount string `json:"promoteAccount"` + PlatformName string `json:"platformName"` + PlatformTitle string `json:"platformTitle"` + BusierAccount string `json:"busierAccount"` + ApplicantOA string `json:"applicantOa"` + BindAt *string `json:"bindAt"` + UnbindAt *string `json:"unbindAt"` + ReachAt *string `json:"reachAt"` + EndAt *string `json:"endAt"` + AccumAmount string `json:"accumAmount"` + Status string `json:"status"` + StatusText string `json:"statusText"` } type ActivationPage struct { Rows []*ActivationResponse `json:"rows"` diff --git a/internal/service/dto/activation_pool.go b/internal/service/dto/activation_pool.go new file mode 100644 index 0000000..756966e --- /dev/null +++ b/internal/service/dto/activation_pool.go @@ -0,0 +1,58 @@ +package dto + +type ActivationPoolQuery struct { + Page int `form:"pageNum"` + PageSize int `form:"pageSize"` + Sources []string `form:"platformNames"` + Name string `form:"promoteAccount"` + ChannelType string `form:"channelType"` + Status string `form:"status"` + Target string `form:"target"` + Begin string `form:"begin"` + End string `form:"end"` +} +type ActivationPoolRow struct { + ID uint64 `json:"id,string"` + SourceDB string `json:"sourceDb"` + PlatformTitle string `json:"platformTitle"` + Name string `json:"name"` + ChannelType string `json:"channelType"` + Status string `json:"status"` + NoLoginDays *int64 `json:"noLoginDays"` + Rejected bool `json:"rejected"` + RejectReason string `json:"rejectReason"` + RejectedAt *string `json:"rejectedAt"` +} +type ActivationPoolPage struct { + Rows []*ActivationPoolRow `json:"rows"` + Total int64 `json:"total"` + Available bool `json:"available"` +} +type ActivationBindingOption struct { + ID uint64 `json:"id,string"` + SourceDB string `json:"sourceDb"` + Account string `json:"account"` + Level int `json:"level"` +} +type ActivationApplyRequest struct { + PoolID uint64 `json:"poolId,string"` + BusierID uint64 `json:"busierId,string"` + SourceDB string `json:"sourceDb"` +} +type ActivationChurnRow struct { + ID uint64 `json:"id,string"` + SourceDB string `json:"sourceDb"` + PlatformTitle string `json:"platformTitle"` + Name string `json:"name"` + From string `json:"from"` + To string `json:"to"` + Reason string `json:"reason"` + Operator string `json:"operator"` + At string `json:"at"` + ToPool bool `json:"toPool"` +} +type ActivationChurnPage struct { + Rows []*ActivationChurnRow `json:"rows"` + Total int64 `json:"total"` + Available bool `json:"available"` +} diff --git a/web/src/api/activation.js b/web/src/api/activation.js index fb40a5a..6fb1d21 100644 --- a/web/src/api/activation.js +++ b/web/src/api/activation.js @@ -2,3 +2,8 @@ import service from '@/utils/businessRequest' export const listChannelActivations = (params) => service({ url: '/sw/promoteActivePool/bindList', method: 'get', params }) export const getActivationSpend = (id, options = {}) => service({ ...options, url: '/sw/promoteActivePool/bindSpendDetail', method: 'get', params: { id } }) + +export const listActivationPool = (params, options = {}) => service({ ...options, url: '/sw/promoteActivePool/pool', method: 'get', params }) +export const listActivationOptions = () => service({ url: '/sw/promoteActivePool/options', method: 'get' }) +export const applyActivation = (data) => service({ url: '/sw/promoteActivePool/apply', method: 'post', data }) +export const listActivationChurn = (params, options = {}) => service({ ...options, url: '/sw/promoteActivePool/churn', method: 'get', params }) diff --git a/web/src/view/sw/ChannelActivationList.vue b/web/src/view/sw/ChannelActivationList.vue index b6f6805..74c5e78 100644 --- a/web/src/view/sw/ChannelActivationList.vue +++ b/web/src/view/sw/ChannelActivationList.vue @@ -1,10 +1,14 @@ diff --git a/web/src/view/sw/components/ActivationResources.vue b/web/src/view/sw/components/ActivationResources.vue new file mode 100644 index 0000000..88dca71 --- /dev/null +++ b/web/src/view/sw/components/ActivationResources.vue @@ -0,0 +1,97 @@ + + +