From d27fba74bccccf3034abf54e1f56ff8ad5d0056b Mon Sep 17 00:00:00 2001 From: zhx Date: Sun, 19 Jul 2026 23:21:31 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=BF=E7=AD=96=E5=91=A8=E6=9C=9F=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/proto/wallet/v1/wallet.pb.go | 38 ++++- api/proto/wallet/v1/wallet.proto | 4 + deploy/rocketmq/broker.local.conf | 10 ++ server/admin/cmd/server/main.go | 25 +++- .../admin/configs/config.tencent.example.yaml | 9 ++ server/admin/configs/config.yaml | 9 ++ server/admin/internal/config/config.go | 50 +++++++ server/admin/internal/model/models.go | 11 +- .../financewithdrawal/dingtalk_notifier.go | 100 +++++++++++++ .../modules/financewithdrawal/handler.go | 9 +- .../modules/financewithdrawal/service.go | 31 +++- .../modules/hostagencypolicy/handler.go | 1 + .../modules/hostagencypolicy/request.go | 1 + .../modules/hostagencypolicy/response.go | 4 + .../modules/hostagencypolicy/service.go | 103 ++++++++------ .../modules/hostagencypolicy/service_test.go | 38 ++++- .../modules/hostsalarysettlement/dto.go | 2 + .../modules/hostsalarysettlement/service.go | 56 +++++--- .../internal/platform/dingtalkrobot/client.go | 132 ++++++++++++++++++ .../host_agency_salary_policy_repository.go | 22 +-- server/admin/internal/repository/seed.go | 3 +- ...5_move_activity_template_to_ops_center.sql | 11 ++ .../106_host_salary_policy_cycle_versions.sql | 53 +++++++ .../activity_template_runtime_repository.go | 2 +- .../financewithdrawal/dingtalk_notifier.go | 5 +- .../internal/transport/http/response_test.go | 20 ++- .../http/userapi/agency_center_handler.go | 1 + .../userapi/host_center_policy_handler.go | 10 +- .../deploy/mysql/initdb/001_room_service.sql | 4 +- .../003_room_vip_media_and_decorations.sql | 27 ++++ .../mysql/initdb/001_wallet_service.sql | 91 +++++++++++- .../004_host_salary_policy_cycle_versions.sql | 94 +++++++++++++ .../internal/domain/ledger/host_salary.go | 7 +- .../internal/service/wallet/host_salary.go | 26 +++- .../service/wallet/ports/repository.go | 2 +- .../internal/service/wallet/service_test.go | 103 ++++++++++++-- .../storage/mysql/host_salary_settlement.go | 91 +++++++----- .../storage/mysql/host_salary_stats.go | 27 ++-- .../internal/storage/mysql/schema.go | 76 ++++++++++ .../internal/testutil/mysqltest/mysqltest.go | 29 +++- .../internal/transport/grpc/host_salary.go | 6 +- tools/databi-real-flow/main.go | 25 +++- 42 files changed, 1191 insertions(+), 177 deletions(-) create mode 100644 deploy/rocketmq/broker.local.conf create mode 100644 server/admin/internal/modules/financewithdrawal/dingtalk_notifier.go create mode 100644 server/admin/internal/platform/dingtalkrobot/client.go create mode 100644 server/admin/migrations/105_move_activity_template_to_ops_center.sql create mode 100644 server/admin/migrations/106_host_salary_policy_cycle_versions.sql create mode 100644 services/wallet-service/deploy/mysql/migrations/004_host_salary_policy_cycle_versions.sql diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index 9a10d601..85e31eba 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -1385,6 +1385,8 @@ type HostSalaryPolicy struct { EffectiveFromMs int64 `protobuf:"varint,9,opt,name=effective_from_ms,json=effectiveFromMs,proto3" json:"effective_from_ms,omitempty"` EffectiveToMs int64 `protobuf:"varint,10,opt,name=effective_to_ms,json=effectiveToMs,proto3" json:"effective_to_ms,omitempty"` Levels []*HostSalaryPolicyLevel `protobuf:"bytes,11,rep,name=levels,proto3" json:"levels,omitempty"` + CycleKey string `protobuf:"bytes,12,opt,name=cycle_key,json=cycleKey,proto3" json:"cycle_key,omitempty"` + PolicyVersion uint64 `protobuf:"varint,13,opt,name=policy_version,json=policyVersion,proto3" json:"policy_version,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1496,6 +1498,20 @@ func (x *HostSalaryPolicy) GetLevels() []*HostSalaryPolicyLevel { return nil } +func (x *HostSalaryPolicy) GetCycleKey() string { + if x != nil { + return x.CycleKey + } + return "" +} + +func (x *HostSalaryPolicy) GetPolicyVersion() uint64 { + if x != nil { + return x.PolicyVersion + } + return 0 +} + type GetActiveHostSalaryPolicyRequest struct { state protoimpl.MessageState `protogen:"open.v1"` RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` @@ -1504,8 +1520,10 @@ type GetActiveHostSalaryPolicyRequest struct { SettlementMode string `protobuf:"bytes,4,opt,name=settlement_mode,json=settlementMode,proto3" json:"settlement_mode,omitempty"` SettlementTriggerMode string `protobuf:"bytes,5,opt,name=settlement_trigger_mode,json=settlementTriggerMode,proto3" json:"settlement_trigger_mode,omitempty"` NowMs int64 `protobuf:"varint,6,opt,name=now_ms,json=nowMs,proto3" json:"now_ms,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // cycle_key 精确选择该工资周期已绑定的不可变政策版本；空值仅供旧客户端回退当前 UTC 月。 + CycleKey string `protobuf:"bytes,7,opt,name=cycle_key,json=cycleKey,proto3" json:"cycle_key,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *GetActiveHostSalaryPolicyRequest) Reset() { @@ -1580,6 +1598,13 @@ func (x *GetActiveHostSalaryPolicyRequest) GetNowMs() int64 { return 0 } +func (x *GetActiveHostSalaryPolicyRequest) GetCycleKey() string { + if x != nil { + return x.CycleKey + } + return "" +} + type GetActiveHostSalaryPolicyResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Found bool `protobuf:"varint,1,opt,name=found,proto3" json:"found,omitempty"` @@ -26443,7 +26468,7 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\x17agency_salary_usd_minor\x18\x05 \x01(\x03R\x14agencySalaryUsdMinor\x12\x16\n" + "\x06status\x18\x06 \x01(\tR\x06status\x12\x1d\n" + "\n" + - "sort_order\x18\a \x01(\x05R\tsortOrder\"\xe9\x03\n" + + "sort_order\x18\a \x01(\x05R\tsortOrder\"\xad\x04\n" + "\x10HostSalaryPolicy\x12\x1b\n" + "\tpolicy_id\x18\x01 \x01(\x04R\bpolicyId\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12\x1b\n" + @@ -26456,7 +26481,9 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\x11effective_from_ms\x18\t \x01(\x03R\x0feffectiveFromMs\x12&\n" + "\x0feffective_to_ms\x18\n" + " \x01(\x03R\reffectiveToMs\x12>\n" + - "\x06levels\x18\v \x03(\v2&.hyapp.wallet.v1.HostSalaryPolicyLevelR\x06levels\"\xf1\x01\n" + + "\x06levels\x18\v \x03(\v2&.hyapp.wallet.v1.HostSalaryPolicyLevelR\x06levels\x12\x1b\n" + + "\tcycle_key\x18\f \x01(\tR\bcycleKey\x12%\n" + + "\x0epolicy_version\x18\r \x01(\x04R\rpolicyVersion\"\x8e\x02\n" + " GetActiveHostSalaryPolicyRequest\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x19\n" + @@ -26464,7 +26491,8 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\tregion_id\x18\x03 \x01(\x03R\bregionId\x12'\n" + "\x0fsettlement_mode\x18\x04 \x01(\tR\x0esettlementMode\x126\n" + "\x17settlement_trigger_mode\x18\x05 \x01(\tR\x15settlementTriggerMode\x12\x15\n" + - "\x06now_ms\x18\x06 \x01(\x03R\x05nowMs\"t\n" + + "\x06now_ms\x18\x06 \x01(\x03R\x05nowMs\x12\x1b\n" + + "\tcycle_key\x18\a \x01(\tR\bcycleKey\"t\n" + "!GetActiveHostSalaryPolicyResponse\x12\x14\n" + "\x05found\x18\x01 \x01(\bR\x05found\x129\n" + "\x06policy\x18\x02 \x01(\v2!.hyapp.wallet.v1.HostSalaryPolicyR\x06policy\"\x9a\x02\n" + diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index 1f78c2ef..bf359a93 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -201,6 +201,8 @@ message HostSalaryPolicy { int64 effective_from_ms = 9; int64 effective_to_ms = 10; repeated HostSalaryPolicyLevel levels = 11; + string cycle_key = 12; + uint64 policy_version = 13; } message GetActiveHostSalaryPolicyRequest { @@ -210,6 +212,8 @@ message GetActiveHostSalaryPolicyRequest { string settlement_mode = 4; string settlement_trigger_mode = 5; int64 now_ms = 6; + // cycle_key 精确选择该工资周期已绑定的不可变政策版本；空值仅供旧客户端回退当前 UTC 月。 + string cycle_key = 7; } message GetActiveHostSalaryPolicyResponse { diff --git a/deploy/rocketmq/broker.local.conf b/deploy/rocketmq/broker.local.conf new file mode 100644 index 00000000..9b8fd75c --- /dev/null +++ b/deploy/rocketmq/broker.local.conf @@ -0,0 +1,10 @@ +brokerClusterName=DefaultCluster +brokerName=broker-a +brokerId=0 +listenPort=10911 +brokerIP1=127.0.0.1 +deleteWhen=04 +fileReservedTime=48 +brokerRole=ASYNC_MASTER +flushDiskType=ASYNC_FLUSH +autoCreateTopicEnable=true diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index 87d3995f..66c28af7 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -89,6 +89,7 @@ import ( vipconfigmodule "hyapp-admin-server/internal/modules/vipconfig" weeklystarmodule "hyapp-admin-server/internal/modules/weeklystar" wheelmodule "hyapp-admin-server/internal/modules/wheel" + "hyapp-admin-server/internal/platform/dingtalkrobot" "hyapp-admin-server/internal/platform/logging" "hyapp-admin-server/internal/platform/tencentcos" "hyapp-admin-server/internal/repository" @@ -357,6 +358,10 @@ func main() { ) luckyGiftHandler := luckygiftmodule.New(luckygiftadmin.NewGRPC(luckyGiftConn), cfg.LuckyGiftService.RequestTimeout, auditHandler). BindUserClient(userclient.NewGRPC(userConn)) + withdrawalReviewNotifier, err := newWithdrawalReviewNotifier(cfg.FinanceNotifications.DingTalk) + if err != nil { + fatalRuntime("create_finance_dingtalk_notifier_failed", err) + } handlers := router.Handlers{ Audit: auditHandler, Auth: authmodule.New(store, auth, auditHandler, cfg), @@ -385,7 +390,8 @@ func main() { financeordermodule.WithLegacyCoinSellerRegionResolver(func(ctx context.Context, appCode string, countryCode string) (int64, bool, error) { return paymentmodule.RegionIDForCountryCode(ctx, moneyRegionSources, appCode, countryCode) })), - FinanceWithdrawal: financewithdrawalmodule.NewWithSources(store, walletclient.NewGRPC(walletConn), activityclient.NewGRPC(activityConn), auditHandler, withdrawalsource.New(cfg.WithdrawalSources)), + FinanceWithdrawal: financewithdrawalmodule.NewWithSources(store, walletclient.NewGRPC(walletConn), activityclient.NewGRPC(activityConn), auditHandler, withdrawalsource.New(cfg.WithdrawalSources)). + BindReviewPendingNotifier(withdrawalReviewNotifier), ExternalAdmin: externaladminmodule.New(db, userDB, externaladminmodule.Config{ SessionTTL: cfg.ExternalAdmin.SessionTTL, MaxLoginFailures: cfg.ExternalAdmin.MaxLoginFailures, @@ -474,6 +480,23 @@ func main() { } } +func newWithdrawalReviewNotifier(cfg config.DingTalkRobotConfig) (financewithdrawalmodule.ReviewPendingNotifier, error) { + if !cfg.Enabled || strings.TrimSpace(cfg.WebhookURL) == "" { + return nil, nil + } + client, err := dingtalkrobot.New(dingtalkrobot.Config{ + WebhookURL: cfg.WebhookURL, + Secret: cfg.Secret, + AtMobiles: cfg.AtMobiles, + AtAll: cfg.AtAll, + RequestTimeout: cfg.RequestTimeout, + }) + if err != nil { + return nil, err + } + return financewithdrawalmodule.NewDingTalkReviewPendingNotifier(client), nil +} + func fatalRuntime(msg string, err error) { slog.Error(msg, "error", err.Error()) os.Exit(1) diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index 81350cf2..38e9ff32 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -137,6 +137,15 @@ statistics_service: base_url: "http://10.2.1.16:13010" # 漏斗冷查询允许短时索引页预热；正常查询应远低于该上限，超时只负责隔离异常请求。 request_timeout: "10s" +finance_notifications: + dingtalk: + enabled: true + # 生产凭证通过 HYAPP_ADMIN_FINANCE_DINGTALK_* 注入，示例配置不保存机器人 access_token。 + webhook_url: "" + secret: "" + at_mobiles: [] + at_all: false + request_timeout: "3s" dashboard_external_sources: - enabled: true type: "mysql" diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 309a417d..340a923d 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -135,6 +135,15 @@ statistics_service: base_url: "http://127.0.0.1:13010" # 漏斗冷查询允许短时索引页预热；正常查询应远低于该上限，超时只负责隔离异常请求。 request_timeout: "10s" +finance_notifications: + dingtalk: + enabled: true + # 真实 webhook/access_token 和可选签名密钥只通过 HYAPP_ADMIN_FINANCE_DINGTALK_* 注入。 + webhook_url: "" + secret: "" + at_mobiles: [] + at_all: false + request_timeout: "3s" dashboard_external_sources: - enabled: false type: "mysql" diff --git a/server/admin/internal/config/config.go b/server/admin/internal/config/config.go index 5bc49416..830ca64d 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -50,6 +50,7 @@ type Config struct { LuckyGiftService LuckyGiftServiceConfig `yaml:"lucky_gift_service"` GameService GameServiceConfig `yaml:"game_service"` StatisticsService StatisticsServiceConfig `yaml:"statistics_service"` + FinanceNotifications FinanceNotificationsConfig `yaml:"finance_notifications"` DashboardExternalSources []DashboardExternalSourceConfig `yaml:"dashboard_external_sources"` LegacyCoinSellerRechargeSources []LegacyCoinSellerRechargeSourceConfig `yaml:"legacy_coin_seller_recharge_sources"` WithdrawalSources []WithdrawalSourceConfig `yaml:"withdrawal_sources"` @@ -116,6 +117,21 @@ type StatisticsServiceConfig struct { RequestTimeout time.Duration `yaml:"request_timeout"` } +// FinanceNotificationsConfig 只保存审核协作通知配置；提现状态和资金终态仍由 Admin/Wallet 的持久事实决定。 +type FinanceNotificationsConfig struct { + DingTalk DingTalkRobotConfig `yaml:"dingtalk"` +} + +// DingTalkRobotConfig 的 webhook 和签名密钥属于运行时凭证，可提交配置只能保留空值。 +type DingTalkRobotConfig struct { + Enabled bool `yaml:"enabled"` + WebhookURL string `yaml:"webhook_url"` + Secret string `yaml:"secret"` + AtMobiles []string `yaml:"at_mobiles"` + AtAll bool `yaml:"at_all"` + RequestTimeout time.Duration `yaml:"request_timeout"` +} + // WithdrawalSourceConfig 描述接入统一双审工作台的外部钱包系统。 // submit_token 只允许来源 App 创建自己的申请；callback_token 用于 admin 终审时回调来源钱包，二者分离避免任一方向泄漏后扩大权限。 type WithdrawalSourceConfig struct { @@ -361,6 +377,12 @@ func Default() Config { BaseURL: "http://127.0.0.1:13010", RequestTimeout: 3 * time.Second, }, + FinanceNotifications: FinanceNotificationsConfig{ + DingTalk: DingTalkRobotConfig{ + Enabled: true, + RequestTimeout: 3 * time.Second, + }, + }, DashboardExternalSources: []DashboardExternalSourceConfig{ { Enabled: false, @@ -586,6 +608,13 @@ func (cfg *Config) Normalize() { if cfg.StatisticsService.RequestTimeout <= 0 { cfg.StatisticsService.RequestTimeout = 3 * time.Second } + cfg.FinanceNotifications.DingTalk.WebhookURL = strings.TrimSpace(cfg.FinanceNotifications.DingTalk.WebhookURL) + cfg.FinanceNotifications.DingTalk.Secret = strings.TrimSpace(cfg.FinanceNotifications.DingTalk.Secret) + cfg.FinanceNotifications.DingTalk.AtMobiles = compactStrings(cfg.FinanceNotifications.DingTalk.AtMobiles) + if cfg.FinanceNotifications.DingTalk.RequestTimeout <= 0 { + cfg.FinanceNotifications.DingTalk.RequestTimeout = 3 * time.Second + } + cfg.applyFinanceDingTalkEnvOverrides() for index := range cfg.DashboardExternalSources { source := &cfg.DashboardExternalSources[index] // 外接 App 大屏的数据源类型在配置层归一化；查询层只根据类型读取对应的外部聚合事实，不回退到 hyapp statistics-service。 @@ -1194,6 +1223,27 @@ func (cfg *Config) applyWithdrawalSourceEnvOverrides() { } } +func (cfg *Config) applyFinanceDingTalkEnvOverrides() { + dingTalk := &cfg.FinanceNotifications.DingTalk + if value := strings.TrimSpace(os.Getenv("HYAPP_ADMIN_FINANCE_DINGTALK_ENABLED")); value != "" { + dingTalk.Enabled = parseBoolEnv(value) + } + if value := strings.TrimSpace(firstEnv("HYAPP_ADMIN_FINANCE_DINGTALK_WEBHOOK_URL", "FINANCE_DINGTALK_WEBHOOK_URL")); value != "" { + // webhook query 中包含 access_token，只允许运行环境覆盖，不能复制到 Admin YAML 或日志。 + dingTalk.WebhookURL = value + dingTalk.Enabled = true + } + if value := strings.TrimSpace(firstEnv("HYAPP_ADMIN_FINANCE_DINGTALK_SECRET", "FINANCE_DINGTALK_SECRET")); value != "" { + dingTalk.Secret = value + } + if value := strings.TrimSpace(os.Getenv("HYAPP_ADMIN_FINANCE_DINGTALK_AT_MOBILES")); value != "" { + dingTalk.AtMobiles = compactStrings(strings.Split(value, ",")) + } + if value := strings.TrimSpace(os.Getenv("HYAPP_ADMIN_FINANCE_DINGTALK_AT_ALL")); value != "" { + dingTalk.AtAll = parseBoolEnv(value) + } +} + func envAppKey(appCode string) string { appCode = strings.ToUpper(strings.TrimSpace(appCode)) if appCode == "" { diff --git a/server/admin/internal/model/models.go b/server/admin/internal/model/models.go index 3b000c32..81a3e386 100644 --- a/server/admin/internal/model/models.go +++ b/server/admin/internal/model/models.go @@ -422,11 +422,14 @@ func (PolicyPublishItem) TableName() string { type HostAgencySalaryPolicy struct { ID uint `gorm:"primaryKey" json:"id"` - AppCode string `gorm:"size:32;uniqueIndex:uk_admin_host_agency_salary_policy_name;index:idx_admin_host_agency_salary_policy_region,not null;default:lalu" json:"appCode"` + AppCode string `gorm:"size:32;uniqueIndex:uk_admin_host_agency_salary_policy_name;uniqueIndex:uk_admin_host_agency_salary_policy_version;index:idx_admin_host_agency_salary_policy_region;index:idx_admin_host_agency_salary_policy_cycle,not null;default:lalu" json:"appCode"` Name string `gorm:"size:120;uniqueIndex:uk_admin_host_agency_salary_policy_name;not null" json:"name"` // RegionID 是第一阶段的政策适用边界；结算时按用户所属区域命中唯一 active 政策。 - RegionID int64 `gorm:"uniqueIndex:uk_admin_host_agency_salary_policy_name;index:idx_admin_host_agency_salary_policy_region,not null" json:"regionId"` - Status string `gorm:"size:24;index:idx_admin_host_agency_salary_policy_region,not null;default:disabled" json:"status"` + RegionID int64 `gorm:"uniqueIndex:uk_admin_host_agency_salary_policy_name;uniqueIndex:uk_admin_host_agency_salary_policy_version;index:idx_admin_host_agency_salary_policy_region;index:idx_admin_host_agency_salary_policy_cycle,not null" json:"regionId"` + // CycleKey 是工资归属月；PolicyVersion 使用政策主键形成稳定版本号，发布后两者均不可修改。 + CycleKey string `gorm:"column:cycle_key;size:7;uniqueIndex:uk_admin_host_agency_salary_policy_version;index:idx_admin_host_agency_salary_policy_cycle,not null" json:"cycleKey"` + PolicyVersion uint64 `gorm:"column:policy_version;uniqueIndex:uk_admin_host_agency_salary_policy_version;not null;default:0" json:"policyVersion"` + Status string `gorm:"size:24;index:idx_admin_host_agency_salary_policy_region,not null;default:disabled" json:"status"` // SettlementMode 只配置结算节奏，等级金额仍按累计值保存，实际发放由后续结算任务计算差额。 SettlementMode string `gorm:"size:24;not null;default:daily" json:"settlementMode"` // SettlementTriggerMode 区分自动任务与后台人工结算；手动政策发布后仍可被查询，但不会被 cron 自动发薪。 @@ -434,7 +437,7 @@ type HostAgencySalaryPolicy struct { // 比例字段使用 decimal 字符串承载，避免金币、钻石、美元换算出现 float 精度误差。 GiftCoinToDiamondRatio string `gorm:"column:gift_coin_to_diamond_ratio;type:decimal(18,6);not null;default:1.000000" json:"giftCoinToDiamondRatio"` ResidualDiamondToUSDRate string `gorm:"column:residual_diamond_to_usd_rate;type:decimal(24,12);not null;default:0.000000000000" json:"residualDiamondToUsdRate"` - // 时间范围采用 epoch ms 半开区间，EffectiveToMS=0 表示长期有效。 + // 时间范围保留为运营展示元数据；工资结算只按 CycleKey 绑定版本，不能再用执行时间选政策。 EffectiveFromMS int64 `gorm:"column:effective_from_ms;index:idx_admin_host_agency_salary_policy_region,not null;default:0" json:"effectiveFromMs"` EffectiveToMS int64 `gorm:"column:effective_to_ms;not null;default:0" json:"effectiveToMs"` Description string `gorm:"size:255;not null;default:''" json:"description"` diff --git a/server/admin/internal/modules/financewithdrawal/dingtalk_notifier.go b/server/admin/internal/modules/financewithdrawal/dingtalk_notifier.go new file mode 100644 index 00000000..e161d677 --- /dev/null +++ b/server/admin/internal/modules/financewithdrawal/dingtalk_notifier.go @@ -0,0 +1,100 @@ +package financewithdrawal + +import ( + "context" + "fmt" + "strings" + "time" + + "hyapp-admin-server/internal/model" + "hyapp-admin-server/internal/repository" +) + +type markdownSender interface { + SendMarkdown(ctx context.Context, title string, text string) error +} + +// ReviewPendingNotifier 只通知“当前待谁处理”，不参与提现创建、冻结或审核事务。 +type ReviewPendingNotifier interface { + NotifyReviewPending(ctx context.Context, application model.UserWithdrawalApplication, stage repository.WithdrawalApplicationReviewStage) error +} + +type dingTalkReviewPendingNotifier struct { + sender markdownSender +} + +func NewDingTalkReviewPendingNotifier(sender markdownSender) ReviewPendingNotifier { + if sender == nil { + return nil + } + return &dingTalkReviewPendingNotifier{sender: sender} +} + +func (n *dingTalkReviewPendingNotifier) NotifyReviewPending(ctx context.Context, application model.UserWithdrawalApplication, stage repository.WithdrawalApplicationReviewStage) error { + if n == nil || n.sender == nil { + return nil + } + title, text := withdrawalReviewPendingMarkdown(application, stage) + return n.sender.SendMarkdown(ctx, title, text) +} + +func withdrawalReviewPendingMarkdown(application model.UserWithdrawalApplication, stage repository.WithdrawalApplicationReviewStage) (string, string) { + stageName := "运营审核阶段" + pageURL := "https://admin-acc.global-interaction.com/finance/?view=withdrawalOperations" + if stage == repository.WithdrawalApplicationReviewStageFinance { + stageName = "财务审核阶段" + pageURL = "https://admin-acc.global-interaction.com/finance/?view=withdrawalFinance" + } + title := fmt.Sprintf("用户提现待处理【%s】", stageName) + lines := []string{ + fmt.Sprintf("### 用户提现待处理【%s】", stageName), + "", + fmt.Sprintf("- 当前阶段：**%s**", stageName), + fmt.Sprintf("- 申请ID：%d", application.ID), + fmt.Sprintf("- APP：%s", withdrawalMarkdownValue(application.AppCode)), + fmt.Sprintf("- 来源系统：%s", withdrawalMarkdownValue(application.SourceSystem)), + fmt.Sprintf("- 来源单号：%s", withdrawalMarkdownValue(application.SourceApplicationID)), + fmt.Sprintf("- 用户ID：%s", withdrawalMarkdownValue(application.UserID)), + fmt.Sprintf("- 提现金额：$ %s", withdrawalMarkdownValue(application.WithdrawAmount)), + fmt.Sprintf("- 提现方式：%s", withdrawalMarkdownValue(application.WithdrawMethod)), + fmt.Sprintf("- 提现地址：%s", withdrawalMarkdownValue(application.WithdrawAddress)), + fmt.Sprintf("- 申请时间：%s", formatWithdrawalNotifyMillis(application.CreatedAtMS)), + } + if stage == repository.WithdrawalApplicationReviewStageFinance { + // 财务提醒保留运营审核快照，接收人无需先打开后台就能确认这是一张已完成初审的申请。 + lines = append(lines, + fmt.Sprintf("- 运营审核人：%s", withdrawalMarkdownValue(application.OperationsReviewerName)), + fmt.Sprintf("- 运营审核备注：%s", withdrawalMarkdownValue(application.OperationsAuditRemark)), + fmt.Sprintf("- 运营通过时间：%s", formatWithdrawalNotifyMillis(int64Value(application.OperationsReviewedAtMS))), + ) + } + lines = append(lines, "- 后台地址："+pageURL) + return title, strings.Join(lines, "\n") +} + +func formatWithdrawalNotifyMillis(value int64) string { + if value <= 0 { + return "-" + } + // 协作消息按运营团队使用的北京时间展示；接口和数据库仍以 Unix 毫秒 UTC 事实为准。 + return time.UnixMilli(value).In(time.FixedZone("UTC+8", 8*60*60)).Format("2006-01-02 15:04:05") +} + +func int64Value(value *int64) int64 { + if value == nil { + return 0 + } + return *value +} + +func withdrawalMarkdownValue(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "-" + } + runes := []rune(value) + if len(runes) > 240 { + value = string(runes[:240]) + "..." + } + return strings.NewReplacer("\r", " ", "\n", " ", "\t", " ").Replace(value) +} diff --git a/server/admin/internal/modules/financewithdrawal/handler.go b/server/admin/internal/modules/financewithdrawal/handler.go index d4742da1..203beb44 100644 --- a/server/admin/internal/modules/financewithdrawal/handler.go +++ b/server/admin/internal/modules/financewithdrawal/handler.go @@ -32,6 +32,13 @@ func NewWithSources(store *repository.Store, wallet walletclient.Client, activit return &Handler{service: NewServiceWithSources(store, wallet, activity, sources), audit: audit, sources: sources} } +func (h *Handler) BindReviewPendingNotifier(notifier ReviewPendingNotifier) *Handler { + if h != nil && h.service != nil { + h.service.BindReviewPendingNotifier(notifier) + } + return h +} + // CreateExternalApplication 是来源钱包的服务间入口，不使用后台 JWT。 // 每个 App 使用独立 submit token，且业务唯一键保证来源端持久重试不会生成重复审核单。 func (h *Handler) CreateExternalApplication(c *gin.Context) { @@ -55,7 +62,7 @@ func (h *Handler) CreateExternalApplication(c *gin.Context) { response.Unauthorized(c, "提现来源鉴权失败") return } - item, err := h.service.CreateExternalApplication(externalWithdrawalApplicationInput{ + item, err := h.service.CreateExternalApplication(c.Request.Context(), externalWithdrawalApplicationInput{ AppCode: req.AppCode, SourceSystem: req.SourceSystem, SourceApplicationID: req.SourceApplicationID, diff --git a/server/admin/internal/modules/financewithdrawal/service.go b/server/admin/internal/modules/financewithdrawal/service.go index 1052ab02..8d15856a 100644 --- a/server/admin/internal/modules/financewithdrawal/service.go +++ b/server/admin/internal/modules/financewithdrawal/service.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "math/big" "strconv" "strings" @@ -38,6 +39,7 @@ type Service struct { wallet walletclient.Client activity activityclient.Client sources *withdrawalsource.Registry + notifier ReviewPendingNotifier now func() time.Time } @@ -69,7 +71,14 @@ func NewServiceWithSources(store *repository.Store, wallet walletclient.Client, return service } -func (s *Service) CreateExternalApplication(input externalWithdrawalApplicationInput) (*withdrawalApplicationDTO, error) { +func (s *Service) BindReviewPendingNotifier(notifier ReviewPendingNotifier) *Service { + if s != nil { + s.notifier = notifier + } + return s +} + +func (s *Service) CreateExternalApplication(ctx context.Context, input externalWithdrawalApplicationInput) (*withdrawalApplicationDTO, error) { if s == nil || s.store == nil || s.sources == nil { return nil, errors.New("external withdrawal intake is not configured") } @@ -118,6 +127,8 @@ func (s *Service) CreateExternalApplication(input externalWithdrawalApplicationI if err != nil { return nil, err } + // 外部 Likei/Aslan 申请直接进入 Admin，不经过 Gateway 的创建通知；落库成功后在这里补齐运营阶段提醒。 + s.notifyReviewPending(ctx, *application, repository.WithdrawalApplicationReviewStageOperations) dto := withdrawalApplicationDTOFromModel(*application) return &dto, nil } @@ -276,10 +287,28 @@ func (s *Service) auditApplication(ctx context.Context, actor shared.Actor, id u if err != nil { return nil, err } + if stage == repository.WithdrawalApplicationReviewStageOperations && decision == model.WithdrawalApplicationStatusApproved { + // 运营通过已经提交到数据库，随后独立提醒财务；机器人异常不能把已转交的申请回滚到运营队列。 + s.notifyReviewPending(ctx, *updated, repository.WithdrawalApplicationReviewStageFinance) + } dto := withdrawalApplicationDTOFromModel(*updated) return &dto, nil } +func (s *Service) notifyReviewPending(ctx context.Context, application model.UserWithdrawalApplication, stage repository.WithdrawalApplicationReviewStage) { + if s == nil || s.notifier == nil { + return + } + if err := s.notifier.NotifyReviewPending(ctx, application, stage); err != nil { + slog.Warn("withdrawal_review_dingtalk_notify_failed", + "application_id", application.ID, + "app_code", application.AppCode, + "review_stage", stage, + "error", err, + ) + } +} + // executeOperationsRejection 把拒绝拆成 claim -> 外部幂等动作 -> finalize 三段。 // claim 是独立短事务，因此 wallet 已释放后即使 notice 或最终 admin 提交失败，申请仍停在 rejecting，不能被改点通过或进入财务。 func (s *Service) executeOperationsRejection(ctx context.Context, actor shared.Actor, id uint, appCode string, remark string, requestID string, commandID string, nowMS int64) (*withdrawalApplicationDTO, error) { diff --git a/server/admin/internal/modules/hostagencypolicy/handler.go b/server/admin/internal/modules/hostagencypolicy/handler.go index e5d72fbe..65cb6950 100644 --- a/server/admin/internal/modules/hostagencypolicy/handler.go +++ b/server/admin/internal/modules/hostagencypolicy/handler.go @@ -29,6 +29,7 @@ func (h *Handler) ListPolicies(c *gin.Context) { items, total, err := h.service.List(appctx.FromContext(c.Request.Context()), repository.HostAgencySalaryPolicyListOptions{ Keyword: options.Keyword, RegionID: queryInt64(c, "region_id", "regionId"), + CycleKey: firstQuery(c, "cycle_key", "cycleKey"), Status: options.Status, SettlementMode: firstQuery(c, "settlement_mode", "settlementMode"), // trigger mode 可直接筛出“只允许手动结算”的政策，后续工资结算页会复用同一查询口径。 diff --git a/server/admin/internal/modules/hostagencypolicy/request.go b/server/admin/internal/modules/hostagencypolicy/request.go index a9afb456..40f6181d 100644 --- a/server/admin/internal/modules/hostagencypolicy/request.go +++ b/server/admin/internal/modules/hostagencypolicy/request.go @@ -3,6 +3,7 @@ package hostagencypolicy type policyRequest struct { Name string `json:"name"` RegionID int64 `json:"region_id"` + CycleKey string `json:"cycle_key"` Status string `json:"status"` SettlementMode string `json:"settlement_mode"` SettlementTriggerMode string `json:"settlement_trigger_mode"` diff --git a/server/admin/internal/modules/hostagencypolicy/response.go b/server/admin/internal/modules/hostagencypolicy/response.go index d0ecf127..9060dd63 100644 --- a/server/admin/internal/modules/hostagencypolicy/response.go +++ b/server/admin/internal/modules/hostagencypolicy/response.go @@ -11,6 +11,8 @@ type policyDTO struct { AppCode string `json:"app_code"` Name string `json:"name"` RegionID int64 `json:"region_id"` + CycleKey string `json:"cycle_key"` + PolicyVersion uint64 `json:"policy_version"` Status string `json:"status"` SettlementMode string `json:"settlement_mode"` SettlementTriggerMode string `json:"settlement_trigger_mode"` @@ -56,6 +58,8 @@ func policyFromModel(item model.HostAgencySalaryPolicy) policyDTO { AppCode: item.AppCode, Name: item.Name, RegionID: item.RegionID, + CycleKey: item.CycleKey, + PolicyVersion: item.PolicyVersion, Status: item.Status, SettlementMode: item.SettlementMode, SettlementTriggerMode: firstNonBlank(item.SettlementTriggerMode, settlementTriggerAutomatic), diff --git a/server/admin/internal/modules/hostagencypolicy/service.go b/server/admin/internal/modules/hostagencypolicy/service.go index e2ad2e13..89e09f54 100644 --- a/server/admin/internal/modules/hostagencypolicy/service.go +++ b/server/admin/internal/modules/hostagencypolicy/service.go @@ -69,7 +69,7 @@ func (s *Service) Create(appCode string, actorID uint, req policyRequest) (polic if err != nil { return policyDTO{}, err } - // 启用政策必须先做时间段冲突校验，否则同一区域结算时会匹配到两套工资规则。 + // 启用政策必须先做周期冲突校验，否则同一区域同一月份会出现两套工资规则。 if err := s.ensureNoActiveOverlap(item, 0); err != nil { return policyDTO{}, err } @@ -85,6 +85,10 @@ func (s *Service) Update(appCode string, actorID uint, id uint, req policyReques if err != nil { return policyDTO{}, err } + if item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 { + // 运行侧已经按周期绑定该版本；编辑会让历史结算口径漂移，只能为新周期新建政策。 + return policyDTO{}, errors.New("已发布的周期政策版本不可修改，请为新周期创建政策") + } // 请求体重新走创建同一套校验，编辑和新增保持完全一致的字段约束。 updated, err := policyModelFromRequest(item.AppCode, actorID, req) if err != nil { @@ -92,6 +96,7 @@ func (s *Service) Update(appCode string, actorID uint, id uint, req policyReques } item.Name = updated.Name item.RegionID = updated.RegionID + item.CycleKey = updated.CycleKey item.Status = updated.Status item.SettlementMode = updated.SettlementMode item.SettlementTriggerMode = updated.SettlementTriggerMode @@ -115,14 +120,15 @@ func (s *Service) Update(appCode string, actorID uint, id uint, req policyReques return policyFromModel(item), nil } -func (s *Service) Delete(ctx context.Context, appCode string, id uint) error { +func (s *Service) Delete(_ context.Context, appCode string, id uint) error { // 先按 app_code 查一次，确保删除动作只能落在当前应用上下文内，随后仓储层事务删除主表和等级表。 - if _, err := s.store.GetHostAgencySalaryPolicy(appctx.Normalize(appCode), id); err != nil { + item, err := s.store.GetHostAgencySalaryPolicy(appctx.Normalize(appCode), id) + if err != nil { return err } - // 删除后台配置前先清掉 wallet 运行快照，避免 admin 记录消失后结算仍命中过期政策。 - if err := s.deleteRuntimePolicy(ctx, appctx.Normalize(appCode), id); err != nil { - return err + if item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 { + // 已发布版本及其周期绑定是历史结算依据；停用或删除都不能破坏已存在月份的可追溯性。 + return errors.New("已发布的周期政策版本不可删除") } return s.store.DeleteHostAgencySalaryPolicy(appctx.Normalize(appCode), id) } @@ -133,6 +139,13 @@ func (s *Service) Publish(ctx context.Context, appCode string, actorID uint, id if err != nil { return policyDTO{}, err } + if item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 { + // 重复点击发布按幂等成功返回；不再覆盖运行表，也不生成新的周期绑定。 + return policyFromModel(item), nil + } + if item.Status != policyStatusActive { + return policyDTO{}, errors.New("工资政策启用后才能发布周期版本") + } if err := s.ensureNoActiveOverlap(item, id); err != nil { return policyDTO{}, err } @@ -157,13 +170,12 @@ func (s *Service) ensureNoActiveOverlap(item model.HostAgencySalaryPolicy, exclu if item.Status != policyStatusActive { return nil } - // 时间段判断采用半开区间：[effective_from_ms, effective_to_ms)，effective_to_ms=0 表示长期有效。 - overlap, err := s.store.HasOverlappingActiveHostAgencySalaryPolicy(item.AppCode, item.RegionID, item.EffectiveFromMS, item.EffectiveToMS, excludeID) + overlap, err := s.store.HasActiveHostAgencySalaryPolicyForCycle(item.AppCode, item.RegionID, item.CycleKey, excludeID) if err != nil { return err } if overlap { - return errors.New("同一区域同一时间只能存在一个启用政策") + return errors.New("同一区域同一工资周期只能存在一个启用政策") } return nil } @@ -181,26 +193,22 @@ func (s *Service) publishRuntimePolicy(ctx context.Context, item model.HostAgenc } defer func() { _ = tx.Rollback() }() - // wallet-service 运行表以 admin policy_id 为主键；每次发布整包覆盖主表和等级，保证结算读取到一套完整梯度。 + policyVersion := item.PolicyVersion + if policyVersion == 0 { + // 仅兼容迁移前创建但尚未发布的草稿；新记录在 admin 仓储创建事务中固定为 policy ID。 + policyVersion = uint64(item.ID) + } + // policy 主表和等级表只允许首次插入；任何重复主键或同周期绑定冲突都会回滚整包发布，禁止覆盖历史版本。 if _, err := tx.ExecContext(ctx, ` INSERT INTO host_agency_salary_policies ( - app_code, policy_id, name, region_id, status, settlement_mode, settlement_trigger_mode, + app_code, policy_id, cycle_key, policy_version, name, region_id, status, settlement_mode, settlement_trigger_mode, gift_coin_to_diamond_ratio, residual_diamond_to_usd_rate, effective_from_ms, effective_to_ms, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON DUPLICATE KEY UPDATE - name = VALUES(name), - region_id = VALUES(region_id), - status = VALUES(status), - settlement_mode = VALUES(settlement_mode), - settlement_trigger_mode = VALUES(settlement_trigger_mode), - gift_coin_to_diamond_ratio = VALUES(gift_coin_to_diamond_ratio), - residual_diamond_to_usd_rate = VALUES(residual_diamond_to_usd_rate), - effective_from_ms = VALUES(effective_from_ms), - effective_to_ms = VALUES(effective_to_ms), - updated_at_ms = VALUES(updated_at_ms)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, item.AppCode, item.ID, + item.CycleKey, + policyVersion, item.Name, item.RegionID, item.Status, @@ -215,9 +223,6 @@ func (s *Service) publishRuntimePolicy(ctx context.Context, item model.HostAgenc ); err != nil { return err } - if _, err := tx.ExecContext(ctx, `DELETE FROM host_agency_salary_policy_levels WHERE app_code = ? AND policy_id = ?`, item.AppCode, item.ID); err != nil { - return err - } for _, level := range item.Levels { _, hostSalaryMinor, err := parseFixedDecimal(level.HostSalaryUSD, 2, true, "主播美元工资") if err != nil { @@ -248,22 +253,13 @@ func (s *Service) publishRuntimePolicy(ctx context.Context, item model.HostAgenc return err } } - return tx.Commit() -} - -func (s *Service) deleteRuntimePolicy(ctx context.Context, appCode string, policyID uint) error { - if s == nil || s.walletDB == nil { - return errors.New("wallet mysql is not configured") - } - tx, err := s.walletDB.BeginTx(ctx, nil) - if err != nil { - return err - } - defer func() { _ = tx.Rollback() }() - if _, err := tx.ExecContext(ctx, `DELETE FROM host_agency_salary_policy_levels WHERE app_code = ? AND policy_id = ?`, appCode, policyID); err != nil { - return err - } - if _, err := tx.ExecContext(ctx, `DELETE FROM host_agency_salary_policies WHERE app_code = ? AND policy_id = ?`, appCode, policyID); err != nil { + // 周期绑定的复合主键是最终防线：同一区域同一周期只能首次绑定一次，后续月份发布不会影响它。 + if _, err := tx.ExecContext(ctx, ` + INSERT INTO host_salary_policy_cycle_bindings ( + app_code, region_id, cycle_key, policy_id, policy_version, created_at_ms + ) VALUES (?, ?, ?, ?, ?, ?)`, + item.AppCode, item.RegionID, item.CycleKey, item.ID, policyVersion, publishedAtMS, + ); err != nil { return err } return tx.Commit() @@ -274,10 +270,14 @@ func policyModelFromRequest(appCode string, actorID uint, req policyRequest) (mo if name == "" || len([]rune(name)) > 120 { return model.HostAgencySalaryPolicy{}, errors.New("政策名称不正确") } - // 当前阶段保持单区域政策：一个区域同一时段只能命中一条 active 规则，后续结算查询也以 region_id 做索引。 + // 当前阶段保持单区域政策：一个区域同一工资周期只能命中一条 active 规则。 if req.RegionID <= 0 { return model.HostAgencySalaryPolicy{}, errors.New("请选择适用区域") } + cycleKey, err := normalizePolicyCycleKey(req.CycleKey, req.EffectiveFromMS) + if err != nil { + return model.HostAgencySalaryPolicy{}, err + } status := normalizeStatus(req.Status) if status == "" { // 新建默认停用，避免运营保存草稿时立即影响结算口径。 @@ -329,6 +329,7 @@ func policyModelFromRequest(appCode string, actorID uint, req policyRequest) (mo AppCode: appCode, Name: name, RegionID: req.RegionID, + CycleKey: cycleKey, Status: status, SettlementMode: settlementMode, SettlementTriggerMode: settlementTriggerMode, @@ -525,6 +526,22 @@ func nonZeroMS(value int64, fallback int64) int64 { return fallback } +func normalizePolicyCycleKey(raw string, effectiveFromMS int64) (string, error) { + cycleKey := strings.TrimSpace(raw) + if cycleKey == "" && effectiveFromMS > 0 { + // 兼容尚未升级的后台页面：旧请求仍可从生效开始时间推导 UTC 归属月，新页面应显式提交 cycle_key。 + cycleKey = time.UnixMilli(effectiveFromMS).UTC().Format("2006-01") + } + if len(cycleKey) != len("2006-01") { + return "", errors.New("请选择工资周期") + } + parsed, err := time.Parse("2006-01", cycleKey) + if err != nil || parsed.UTC().Format("2006-01") != cycleKey { + return "", errors.New("工资周期格式必须为 YYYY-MM") + } + return cycleKey, nil +} + func truncateRunes(value string, limit int) string { if limit <= 0 { return "" diff --git a/server/admin/internal/modules/hostagencypolicy/service_test.go b/server/admin/internal/modules/hostagencypolicy/service_test.go index af8b4426..29c8aee9 100644 --- a/server/admin/internal/modules/hostagencypolicy/service_test.go +++ b/server/admin/internal/modules/hostagencypolicy/service_test.go @@ -6,6 +6,7 @@ func TestPolicyModelFromRequestNormalizesAndSortsLevels(t *testing.T) { item, err := policyModelFromRequest("lalu", 7, policyRequest{ Name: "Lalu 默认政策", RegionID: 101, + CycleKey: "2026-06", Status: "active", SettlementMode: "half-month", SettlementTriggerMode: "manual", @@ -19,7 +20,7 @@ func TestPolicyModelFromRequestNormalizesAndSortsLevels(t *testing.T) { if err != nil { t.Fatalf("policy should be valid: %v", err) } - if item.AppCode != "lalu" || item.CreatedByAdminID != 7 || item.SettlementMode != settlementModeHalfMonth || item.SettlementTriggerMode != settlementTriggerManual { + if item.AppCode != "lalu" || item.CycleKey != "2026-06" || item.CreatedByAdminID != 7 || item.SettlementMode != settlementModeHalfMonth || item.SettlementTriggerMode != settlementTriggerManual { t.Fatalf("policy fields mismatch: %+v", item) } // 工资政策不能单独改送礼入主播周期钻石的比例；该比例只由礼物钻石配置页控制。 @@ -38,6 +39,7 @@ func TestPolicyModelFromRequestRejectsNonIncreasingLevels(t *testing.T) { _, err := policyModelFromRequest("lalu", 1, policyRequest{ Name: "Invalid", RegionID: 101, + CycleKey: "2026-06", SettlementMode: settlementModeDaily, Levels: []levelRequest{ {Level: 1, RequiredDiamonds: 700000, HostSalaryUSD: "3", HostCoinReward: 198000, AgencySalaryUSD: "0.8"}, @@ -53,6 +55,7 @@ func TestPolicyModelFromRequestDefaultsSettlementTriggerMode(t *testing.T) { item, err := policyModelFromRequest("lalu", 1, policyRequest{ Name: "Default trigger", RegionID: 101, + CycleKey: "2026-06", SettlementMode: settlementModeDaily, Levels: []levelRequest{ {Level: 1, RequiredDiamonds: 350000, HostSalaryUSD: "1.5", HostCoinReward: 90000, AgencySalaryUSD: "0.5"}, @@ -70,6 +73,7 @@ func TestPolicyModelFromRequestRejectsInvalidSettlementMode(t *testing.T) { _, err := policyModelFromRequest("lalu", 1, policyRequest{ Name: "Invalid", RegionID: 101, + CycleKey: "2026-06", SettlementMode: "monthly", Levels: []levelRequest{ {Level: 1, RequiredDiamonds: 350000, HostSalaryUSD: "1.5", HostCoinReward: 90000, AgencySalaryUSD: "0.5"}, @@ -84,6 +88,7 @@ func TestPolicyModelFromRequestRejectsInvalidSettlementTriggerMode(t *testing.T) _, err := policyModelFromRequest("lalu", 1, policyRequest{ Name: "Invalid", RegionID: 101, + CycleKey: "2026-06", SettlementMode: settlementModeDaily, SettlementTriggerMode: "cron", Levels: []levelRequest{ @@ -94,3 +99,34 @@ func TestPolicyModelFromRequestRejectsInvalidSettlementTriggerMode(t *testing.T) t.Fatal("expected invalid settlement trigger mode to fail") } } + +func TestPolicyModelFromRequestDerivesUTCCycleForLegacyRequest(t *testing.T) { + item, err := policyModelFromRequest("lalu", 1, policyRequest{ + Name: "Legacy cycle", + RegionID: 101, + EffectiveFromMS: 1782864000000, // 2026-07-01T00:00:00Z + Levels: []levelRequest{ + {Level: 1, RequiredDiamonds: 350000, HostSalaryUSD: "1.5", HostCoinReward: 90000, AgencySalaryUSD: "0.5"}, + }, + }) + if err != nil { + t.Fatalf("legacy request should derive cycle: %v", err) + } + if item.CycleKey != "2026-07" { + t.Fatalf("derived cycle mismatch: %+v", item) + } +} + +func TestPolicyModelFromRequestRejectsInvalidCycle(t *testing.T) { + _, err := policyModelFromRequest("lalu", 1, policyRequest{ + Name: "Invalid cycle", + RegionID: 101, + CycleKey: "2026-13", + Levels: []levelRequest{ + {Level: 1, RequiredDiamonds: 350000, HostSalaryUSD: "1.5", HostCoinReward: 90000, AgencySalaryUSD: "0.5"}, + }, + }) + if err == nil { + t.Fatal("expected invalid cycle to fail") + } +} diff --git a/server/admin/internal/modules/hostsalarysettlement/dto.go b/server/admin/internal/modules/hostsalarysettlement/dto.go index 8b198ddf..1428cc78 100644 --- a/server/admin/internal/modules/hostsalarysettlement/dto.go +++ b/server/admin/internal/modules/hostsalarysettlement/dto.go @@ -61,6 +61,7 @@ type pendingDTO struct { RegionID int64 `json:"region_id"` RegionName string `json:"region_name"` PolicyID uint64 `json:"policy_id"` + PolicyVersion uint64 `json:"policy_version"` PolicyName string `json:"policy_name"` SettlementType string `json:"settlement_type"` TriggerMode string `json:"trigger_mode"` @@ -107,6 +108,7 @@ type settlementDTO struct { AgencyOwner userDTO `json:"agency_owner"` CycleKey string `json:"cycle_key"` PolicyID uint64 `json:"policy_id"` + PolicyVersion uint64 `json:"policy_version"` LevelNo int32 `json:"level_no"` TotalDiamonds int64 `json:"total_diamonds"` HostSalaryUSDMinorDelta int64 `json:"host_salary_usd_minor_delta"` diff --git a/server/admin/internal/modules/hostsalarysettlement/service.go b/server/admin/internal/modules/hostsalarysettlement/service.go index 5a2ef038..7b207f28 100644 --- a/server/admin/internal/modules/hostsalarysettlement/service.go +++ b/server/admin/internal/modules/hostsalarysettlement/service.go @@ -54,6 +54,8 @@ type walletCronClient interface { type hostPolicy struct { PolicyID uint64 + CycleKey string + PolicyVersion uint64 Name string RegionID int64 SettlementMode string @@ -108,7 +110,7 @@ func (s *Service) ListPending(ctx context.Context, appCode string, req pendingQu } req = normalizePendingQuery(req) // 待结算列表只做只读重算，不写 progress、不写钱包；真正入账必须走 Settle 调 wallet-service。 - candidates, err := s.collectPendingCandidates(ctx, strings.TrimSpace(appCode), req, time.Now().UTC().UnixMilli()) + candidates, err := s.collectPendingCandidates(ctx, strings.TrimSpace(appCode), req) if err != nil { return nil, 0, err } @@ -158,7 +160,7 @@ func (s *Service) Settle(ctx context.Context, appCode string, operatorAdminID in RegionID: req.RegionID, CountryID: req.CountryID, CountryCode: req.CountryCode, - }, time.Now().UTC().UnixMilli()) + }) if err != nil { return settleResultDTO{}, err } @@ -229,7 +231,7 @@ func (s *Service) callWalletSettlement(ctx context.Context, appCode string, req } } -func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, req pendingQuery, nowMS int64) ([]hostCandidate, error) { +func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, req pendingQuery) ([]hostCandidate, error) { req = normalizePendingQuery(req) regions, err := s.regionFilter(ctx, appCode, req.RegionID, req.CountryID, req.CountryCode) if err != nil { @@ -255,7 +257,6 @@ func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, userClause = "AND a.user_id = ?" args = append(args, req.UserID) } - args = append(args, nowMS, nowMS) triggerClause := "" if req.TriggerMode != "" { triggerClause = "AND p.settlement_trigger_mode = ?" @@ -284,12 +285,17 @@ func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, %s AND EXISTS ( SELECT 1 - FROM host_agency_salary_policies p - WHERE p.app_code = a.app_code - AND p.region_id = a.region_id + FROM host_salary_policy_cycle_bindings binding + JOIN host_agency_salary_policies p + ON p.app_code = binding.app_code + AND p.policy_id = binding.policy_id + AND p.region_id = binding.region_id + AND p.cycle_key = binding.cycle_key + AND p.policy_version = binding.policy_version + WHERE binding.app_code = a.app_code + AND binding.region_id = a.region_id + AND binding.cycle_key = a.cycle_key AND p.status = 'active' - AND p.effective_from_ms <= ? - AND (p.effective_to_ms = 0 OR p.effective_to_ms > ?) %s %s ) @@ -319,10 +325,10 @@ func (s *Service) collectPendingCandidates(ctx context.Context, appCode string, ); err != nil { return nil, err } - cacheKey := fmt.Sprintf("%d:%s:%s", item.RegionID, req.SettlementType, req.TriggerMode) + cacheKey := fmt.Sprintf("%d:%s:%s:%s", item.RegionID, item.CycleKey, req.SettlementType, req.TriggerMode) policy, ok := policyCache[cacheKey] if !ok { - policy, ok, err = s.resolvePolicy(ctx, appCode, item.RegionID, req.SettlementType, req.TriggerMode, nowMS) + policy, ok, err = s.resolvePolicy(ctx, appCode, item.RegionID, item.CycleKey, req.SettlementType, req.TriggerMode) if err != nil { return nil, err } @@ -380,7 +386,7 @@ func (s *Service) List(ctx context.Context, appCode string, req query) ([]settle } rows, err := s.walletDB.QueryContext(ctx, ` SELECT settlement_id, command_id, transaction_id, settlement_type, trigger_mode, settlement_role, user_id, agency_owner_user_id, - cycle_key, policy_id, level_no, total_diamonds, host_salary_usd_minor_delta, + cycle_key, policy_id, policy_version, level_no, total_diamonds, host_salary_usd_minor_delta, host_coin_reward_delta, agency_salary_usd_minor_delta, residual_usd_minor_delta, status, reason, created_at_ms FROM host_salary_settlement_records @@ -411,6 +417,7 @@ func (s *Service) List(ctx context.Context, appCode string, req query) ([]settle &agencyOwnerUserID, &item.CycleKey, &item.PolicyID, + &item.PolicyVersion, &item.LevelNo, &item.TotalDiamonds, &item.HostSalaryUSDMinorDelta, @@ -515,25 +522,31 @@ func settlementWhere(appCode string, req query) (string, []any) { return "WHERE " + strings.Join(conditions, " AND "), args } -func (s *Service) resolvePolicy(ctx context.Context, appCode string, regionID int64, settlementType string, triggerMode string, nowMS int64) (hostPolicy, bool, error) { - conditions := []string{"app_code = ?", "region_id = ?", "status = 'active'", "effective_from_ms <= ?", "(effective_to_ms = 0 OR effective_to_ms > ?)"} - args := []any{appCode, regionID, nowMS, nowMS} +func (s *Service) resolvePolicy(ctx context.Context, appCode string, regionID int64, cycleKey string, settlementType string, triggerMode string) (hostPolicy, bool, error) { + conditions := []string{"binding.app_code = ?", "binding.region_id = ?", "binding.cycle_key = ?", "p.status = 'active'"} + args := []any{appCode, regionID, cycleKey} if triggerMode != "" { - conditions = append(conditions, "settlement_trigger_mode = ?") + conditions = append(conditions, "p.settlement_trigger_mode = ?") args = append(args, triggerMode) } if settlementType != settlementTypeMonthEnd { - conditions = append(conditions, "settlement_mode = ?") + conditions = append(conditions, "p.settlement_mode = ?") args = append(args, settlementType) } row := s.walletDB.QueryRowContext(ctx, ` - SELECT policy_id, name, region_id, settlement_mode, settlement_trigger_mode, CAST(residual_diamond_to_usd_rate AS CHAR) - FROM host_agency_salary_policies + SELECT p.policy_id, p.cycle_key, p.policy_version, p.name, p.region_id, + p.settlement_mode, p.settlement_trigger_mode, CAST(p.residual_diamond_to_usd_rate AS CHAR) + FROM host_salary_policy_cycle_bindings binding + JOIN host_agency_salary_policies p + ON p.app_code = binding.app_code + AND p.policy_id = binding.policy_id + AND p.region_id = binding.region_id + AND p.cycle_key = binding.cycle_key + AND p.policy_version = binding.policy_version WHERE `+strings.Join(conditions, " AND ")+` - ORDER BY effective_from_ms DESC, policy_id DESC LIMIT 1`, args...) var policy hostPolicy - if err := row.Scan(&policy.PolicyID, &policy.Name, &policy.RegionID, &policy.SettlementMode, &policy.TriggerMode, &policy.ResidualDiamondToUSDRate); err != nil { + if err := row.Scan(&policy.PolicyID, &policy.CycleKey, &policy.PolicyVersion, &policy.Name, &policy.RegionID, &policy.SettlementMode, &policy.TriggerMode, &policy.ResidualDiamondToUSDRate); err != nil { if errors.Is(err, sql.ErrNoRows) { return hostPolicy{}, false, nil } @@ -867,6 +880,7 @@ func pendingDTOFromCandidate(item hostCandidate, host userProfile, agency userPr RegionID: item.RegionID, RegionName: item.RegionName, PolicyID: item.Policy.PolicyID, + PolicyVersion: item.Policy.PolicyVersion, PolicyName: item.Policy.Name, SettlementType: item.SettlementType, TriggerMode: item.Policy.TriggerMode, diff --git a/server/admin/internal/platform/dingtalkrobot/client.go b/server/admin/internal/platform/dingtalkrobot/client.go new file mode 100644 index 00000000..85c92278 --- /dev/null +++ b/server/admin/internal/platform/dingtalkrobot/client.go @@ -0,0 +1,132 @@ +package dingtalkrobot + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +type Config struct { + WebhookURL string + Secret string + AtMobiles []string + AtAll bool + RequestTimeout time.Duration +} + +// Client 位于 admin 自己的 platform 边界，避免独立 module 反向依赖根模块 hyapp/pkg。 +type Client struct { + webhookURL string + secret string + atMobiles []string + atAll bool + httpClient *http.Client +} + +func New(config Config) (*Client, error) { + webhookURL := strings.TrimSpace(config.WebhookURL) + if webhookURL == "" { + return nil, errors.New("dingtalk webhook url is required") + } + parsed, err := url.ParseRequestURI(webhookURL) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { + return nil, errors.New("dingtalk webhook url must be an absolute http url") + } + timeout := config.RequestTimeout + if timeout <= 0 { + timeout = 3 * time.Second + } + return &Client{ + webhookURL: webhookURL, + secret: strings.TrimSpace(config.Secret), + atMobiles: compactStrings(config.AtMobiles), + atAll: config.AtAll, + httpClient: &http.Client{Timeout: timeout}, + }, nil +} + +// SendMarkdown 将钉钉 HTTP 状态和业务 errcode 都视为失败；业务层决定通知失败不能回滚审核事实。 +func (c *Client) SendMarkdown(ctx context.Context, title string, text string) error { + if c == nil || c.httpClient == nil { + return errors.New("dingtalk client is not configured") + } + title, text = strings.TrimSpace(title), strings.TrimSpace(text) + if title == "" || text == "" { + return errors.New("dingtalk markdown title and text are required") + } + body, err := json.Marshal(map[string]any{ + "msgtype": "markdown", + "markdown": map[string]string{"title": title, "text": text}, + "at": map[string]any{"atMobiles": c.atMobiles, "isAtAll": c.atAll}, + }) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.signedWebhookURL(), bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + payload, err := io.ReadAll(io.LimitReader(resp.Body, 4096)) + if err != nil { + return err + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return fmt.Errorf("dingtalk returned http %d", resp.StatusCode) + } + var result struct { + ErrCode int `json:"errcode"` + ErrMsg string `json:"errmsg"` + } + if err := json.Unmarshal(payload, &result); err != nil { + return fmt.Errorf("decode dingtalk response: %w", err) + } + if result.ErrCode != 0 { + return fmt.Errorf("dingtalk returned errcode=%d errmsg=%s", result.ErrCode, result.ErrMsg) + } + return nil +} + +func (c *Client) signedWebhookURL() string { + if c.secret == "" { + return c.webhookURL + } + parsed, err := url.Parse(c.webhookURL) + if err != nil { + return c.webhookURL + } + timestamp := time.Now().UTC().UnixMilli() + payload := fmt.Sprintf("%d\n%s", timestamp, c.secret) + mac := hmac.New(sha256.New, []byte(c.secret)) + _, _ = mac.Write([]byte(payload)) + query := parsed.Query() + query.Set("timestamp", fmt.Sprintf("%d", timestamp)) + query.Set("sign", base64.StdEncoding.EncodeToString(mac.Sum(nil))) + parsed.RawQuery = query.Encode() + return parsed.String() +} + +func compactStrings(values []string) []string { + out := make([]string, 0, len(values)) + for _, value := range values { + if value = strings.TrimSpace(value); value != "" { + out = append(out, value) + } + } + return out +} diff --git a/server/admin/internal/repository/host_agency_salary_policy_repository.go b/server/admin/internal/repository/host_agency_salary_policy_repository.go index 3793c402..b46cf0e3 100644 --- a/server/admin/internal/repository/host_agency_salary_policy_repository.go +++ b/server/admin/internal/repository/host_agency_salary_policy_repository.go @@ -12,6 +12,7 @@ type HostAgencySalaryPolicyListOptions struct { AppCode string Keyword string RegionID int64 + CycleKey string Status string SettlementMode string SettlementTriggerMode string @@ -26,6 +27,9 @@ func (s *Store) ListHostAgencySalaryPolicies(options HostAgencySalaryPolicyListO // 第一阶段仍按单区域保存，区域筛选直接命中主表索引。 db = db.Where("region_id = ?", options.RegionID) } + if cycleKey := strings.TrimSpace(options.CycleKey); cycleKey != "" { + db = db.Where("cycle_key = ?", cycleKey) + } if status := strings.TrimSpace(options.Status); status != "" { db = db.Where("status = ?", status) } @@ -50,7 +54,7 @@ func (s *Store) ListHostAgencySalaryPolicies(options HostAgencySalaryPolicyListO // 等级必须随政策一起预加载并按 level_no 排序，前端展示和后续编辑都依赖稳定顺序。 err := db.Preload("Levels", func(tx *gorm.DB) *gorm.DB { return tx.Order("level_no ASC") - }).Order("region_id ASC, effective_from_ms DESC, id DESC").Offset((page - 1) * pageSize).Limit(pageSize).Find(&items).Error + }).Order("region_id ASC, cycle_key DESC, policy_version DESC, id DESC").Offset((page - 1) * pageSize).Limit(pageSize).Find(&items).Error return items, total, err } @@ -71,6 +75,11 @@ func (s *Store) CreateHostAgencySalaryPolicy(item *model.HostAgencySalaryPolicy) if err := tx.Create(item).Error; err != nil { return err } + // 主键在同一事务内作为不可变版本号；它全局唯一，也让旧数据可以无歧义回填 policy_version=id。 + item.PolicyVersion = uint64(item.ID) + if err := tx.Model(item).UpdateColumn("policy_version", item.PolicyVersion).Error; err != nil { + return err + } for index := range levels { levels[index].PolicyID = item.ID } @@ -134,15 +143,10 @@ func (s *Store) DeleteHostAgencySalaryPolicy(appCode string, id uint) error { }) } -func (s *Store) HasOverlappingActiveHostAgencySalaryPolicy(appCode string, regionID int64, effectiveFromMS int64, effectiveToMS int64, excludeID uint) (bool, error) { - // 与当前政策开始时间相交：已有长期有效或已有结束时间晚于当前开始。 +func (s *Store) HasActiveHostAgencySalaryPolicyForCycle(appCode string, regionID int64, cycleKey string, excludeID uint) (bool, error) { + // 工资政策以周期为唯一选择边界；不同月份允许并存，同一区域同一月份只能有一条 active 配置。 db := s.db.Model(&model.HostAgencySalaryPolicy{}). - Where("app_code = ? AND region_id = ? AND status = ?", strings.TrimSpace(appCode), regionID, "active"). - Where("(effective_to_ms = 0 OR effective_to_ms > ?)", effectiveFromMS) - if effectiveToMS > 0 { - // 当前政策不是长期有效时，再要求已有政策开始时间早于当前结束，形成标准半开区间相交判断。 - db = db.Where("effective_from_ms < ?", effectiveToMS) - } + Where("app_code = ? AND region_id = ? AND cycle_key = ? AND status = ?", strings.TrimSpace(appCode), regionID, strings.TrimSpace(cycleKey), "active") if excludeID > 0 { db = db.Where("id <> ?", excludeID) } diff --git a/server/admin/internal/repository/seed.go b/server/admin/internal/repository/seed.go index 7ec4e46e..af1cf52e 100644 --- a/server/admin/internal/repository/seed.go +++ b/server/admin/internal/repository/seed.go @@ -381,7 +381,8 @@ func (s *Store) seedMenus() error { {ParentID: &paymentID, Title: "三方临时支付链接", Code: "payment-temporary-links", Path: "/payment/temporary-links", Icon: "receipt", PermissionCode: "payment-temporary-link:view", Sort: 70, Visible: true}, {ParentID: &paymentID, Title: "支付内购商品", Code: "payment-recharge-products", Path: "/payment/recharge-products", Icon: "wallet", PermissionCode: "payment-product:view", Sort: 71, Visible: true}, {ParentID: &paymentID, Title: "币商充值订单", Code: "finance-coin-seller-recharge-orders", Path: "/finance/orders/coin-seller-recharges", Icon: "receipt", PermissionCode: "finance-order:coin-seller-recharge:view", Sort: 72, Visible: true}, - {ParentID: &activityID, Title: "活动模版", Code: "activity-template", Path: "/activities/templates", Icon: "campaign", PermissionCode: "activity-template:view", Sort: 68, Visible: true}, + // 活动模版权限继续用于 ops-center 的按钮和接口鉴权；主后台旧二级菜单只保留不可见兼容记录。 + {ParentID: &activityID, Title: "活动模版", Code: "activity-template", Path: "/activities/templates", Icon: "campaign", PermissionCode: "activity-template:view", Sort: 68, Visible: false}, {ParentID: &activityID, Title: "每日任务", Code: "daily-task-list", Path: "/activities/daily-tasks", Icon: "task", PermissionCode: "daily-task:view", Sort: 69, Visible: true}, {ParentID: &activityID, Title: "注册奖励", Code: "registration-reward", Path: "/activities/registration-reward", Icon: "gift", PermissionCode: "registration-reward:view", Sort: 70, Visible: true}, {ParentID: &activityID, Title: "成就配置", Code: "achievement-config", Path: "/activities/achievements", Icon: "military_tech", PermissionCode: "achievement:view", Sort: 71, Visible: true}, diff --git a/server/admin/migrations/105_move_activity_template_to_ops_center.sql b/server/admin/migrations/105_move_activity_template_to_ops_center.sql new file mode 100644 index 00000000..09d614b9 --- /dev/null +++ b/server/admin/migrations/105_move_activity_template_to_ops_center.sql @@ -0,0 +1,11 @@ +SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; + +-- 活动模版工作台迁入 ops-center，权限和旧路径仍保留用于接口鉴权及历史链接兼容。 +-- admin_menus.code 是唯一索引，本次更新只命中一行，不扫描菜单表，也不改角色权限关系。 +SET @now_ms = CAST(UNIX_TIMESTAMP(UTC_TIMESTAMP(3)) * 1000 AS UNSIGNED); + +UPDATE admin_menus +SET visible = FALSE, + updated_at_ms = @now_ms +WHERE code = 'activity-template' + AND visible = TRUE; diff --git a/server/admin/migrations/106_host_salary_policy_cycle_versions.sql b/server/admin/migrations/106_host_salary_policy_cycle_versions.sql new file mode 100644 index 00000000..3a4be4cd --- /dev/null +++ b/server/admin/migrations/106_host_salary_policy_cycle_versions.sql @@ -0,0 +1,53 @@ +SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; +SET time_zone = '+00:00'; + +-- 工资政策是小型配置表；迁移只回填政策行并创建周期索引，不扫描用户、工资进度或钱包流水。 +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'admin_host_agency_salary_policies' AND COLUMN_NAME = 'cycle_key') = 0, + 'ALTER TABLE admin_host_agency_salary_policies ADD COLUMN cycle_key VARCHAR(7) NOT NULL DEFAULT '''' COMMENT ''政策归属工资周期，UTC yyyy-MM'' AFTER region_id, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'admin_host_agency_salary_policies' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE admin_host_agency_salary_policies ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''不可变政策版本，等于政策 ID'' AFTER cycle_key, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +-- 老记录优先按原有效期起点归档；没有起点时依次使用发布时间、创建时间，全部按 UTC 计算。 +UPDATE admin_host_agency_salary_policies +SET cycle_key = DATE_FORMAT(FROM_UNIXTIME( + COALESCE(NULLIF(effective_from_ms, 0), NULLIF(published_at_ms, 0), created_at_ms) / 1000 + ), '%Y-%m'), + policy_version = id +WHERE cycle_key = '' OR policy_version = 0; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'admin_host_agency_salary_policies' + AND INDEX_NAME = 'uk_admin_host_agency_salary_policy_version') = 0, + 'ALTER TABLE admin_host_agency_salary_policies ADD UNIQUE INDEX uk_admin_host_agency_salary_policy_version (app_code, region_id, cycle_key, policy_version), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'admin_host_agency_salary_policies' + AND INDEX_NAME = 'idx_admin_host_agency_salary_policy_cycle') = 0, + 'ALTER TABLE admin_host_agency_salary_policies ADD INDEX idx_admin_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/services/activity-service/internal/storage/mysql/activity_template_runtime_repository.go b/services/activity-service/internal/storage/mysql/activity_template_runtime_repository.go index bc28496c..c7eb9a4a 100644 --- a/services/activity-service/internal/storage/mysql/activity_template_runtime_repository.go +++ b/services/activity-service/internal/storage/mysql/activity_template_runtime_repository.go @@ -647,7 +647,7 @@ func (r *Repository) PrepareActivityTemplateTaskClaim(ctx context.Context, runti app_code, claim_id, command_id, template_id, template_code, version_no, task_day, task_key, user_id, earning_region_id, attribution_at_ms, reward_resource_group_id, reward_snapshot_id, reward_snapshot_json, wallet_command_id, status, attempt_count, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?, ?)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?, ?)`, claim.AppCode, claim.ClaimID, claim.CommandID, claim.TemplateID, claim.TemplateCode, claim.VersionNo, claim.TaskDay, claim.TaskKey, claim.UserID, claim.EarningRegionID, claim.AttributionAtMS, claim.RewardResourceGroupID, claim.RewardSnapshot.SnapshotID, diff --git a/services/gateway-service/internal/financewithdrawal/dingtalk_notifier.go b/services/gateway-service/internal/financewithdrawal/dingtalk_notifier.go index bfbcac9b..fdb3130d 100644 --- a/services/gateway-service/internal/financewithdrawal/dingtalk_notifier.go +++ b/services/gateway-service/internal/financewithdrawal/dingtalk_notifier.go @@ -26,15 +26,16 @@ func (n *dingTalkNotifier) NotifyApplicationCreated(ctx context.Context, applica return nil } return n.client.SendMarkdown(ctx, dingtalkrobot.MarkdownMessage{ - Title: "用户提现运营审核待处理", + Title: "用户提现运营审核待处理【运营审核阶段】", Text: withdrawalApplicationCreatedMarkdown(application), }) } func withdrawalApplicationCreatedMarkdown(application Application) string { return strings.Join([]string{ - "### 用户提现运营审核待处理", + "### 用户提现运营审核待处理【运营审核阶段】", "", + "- 当前阶段：**运营审核阶段**", fmt.Sprintf("- 申请ID：%d", application.ID), fmt.Sprintf("- APP：%s", markdownValue(application.AppCode)), fmt.Sprintf("- 用户ID：%d", application.UserID), diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 4da70a8d..ba758987 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -6762,6 +6762,7 @@ func TestUpdateAgencyProfileForwardsOwnerCommand(t *testing.T) { } func TestHostCenterPlatformPolicyUsesCurrentHostRegion(t *testing.T) { + expectedCycle := time.Now().UTC().Format("2006-01") hostClient := &fakeUserHostClient{hostProfile: &userv1.HostProfile{ UserId: 42, Status: "active", @@ -6773,6 +6774,8 @@ func TestHostCenterPlatformPolicyUsesCurrentHostRegion(t *testing.T) { Found: true, Policy: &walletv1.HostSalaryPolicy{ PolicyId: 9001, + CycleKey: expectedCycle, + PolicyVersion: 9001, Name: "Host Growth Policy", RegionId: 30, Status: "active", @@ -6789,7 +6792,7 @@ func TestHostCenterPlatformPolicyUsesCurrentHostRegion(t *testing.T) { }, hostSalaryProgressResp: &walletv1.GetHostSalaryProgressResponse{Progress: &walletv1.HostSalaryProgress{ HostUserId: 42, - CycleKey: "2026-06", + CycleKey: expectedCycle, RegionId: 30, AgencyOwnerUserId: 99, TotalDiamonds: 1200, @@ -6813,10 +6816,10 @@ func TestHostCenterPlatformPolicyUsesCurrentHostRegion(t *testing.T) { if hostClient.lastHost == nil || hostClient.lastHost.GetUserId() != 42 { t.Fatalf("host profile request mismatch: %+v", hostClient.lastHost) } - if walletClient.lastHostSalaryPolicy == nil || walletClient.lastHostSalaryPolicy.GetRegionId() != 30 || walletClient.lastHostSalaryPolicy.GetSettlementTriggerMode() != "" || walletClient.lastHostSalaryPolicy.GetAppCode() == "" || walletClient.lastHostSalaryPolicy.GetRequestId() == "" { + if walletClient.lastHostSalaryPolicy == nil || walletClient.lastHostSalaryPolicy.GetRegionId() != 30 || walletClient.lastHostSalaryPolicy.GetCycleKey() != expectedCycle || walletClient.lastHostSalaryPolicy.GetSettlementTriggerMode() != "" || walletClient.lastHostSalaryPolicy.GetAppCode() == "" || walletClient.lastHostSalaryPolicy.GetRequestId() == "" { t.Fatalf("host salary policy request mismatch: %+v", walletClient.lastHostSalaryPolicy) } - if walletClient.lastHostSalaryProgress == nil || walletClient.lastHostSalaryProgress.GetHostUserId() != 42 || walletClient.lastHostSalaryProgress.GetAppCode() == "" || walletClient.lastHostSalaryProgress.GetRequestId() == "" { + if walletClient.lastHostSalaryProgress == nil || walletClient.lastHostSalaryProgress.GetHostUserId() != 42 || walletClient.lastHostSalaryProgress.GetCycleKey() != expectedCycle || walletClient.lastHostSalaryProgress.GetAppCode() == "" || walletClient.lastHostSalaryProgress.GetRequestId() == "" { t.Fatalf("host salary progress request mismatch: %+v", walletClient.lastHostSalaryProgress) } var response httpkit.ResponseEnvelope @@ -6829,10 +6832,10 @@ func TestHostCenterPlatformPolicyUsesCurrentHostRegion(t *testing.T) { firstLevel := levels[0].(map[string]any) progress := data["progress"].(map[string]any) levelProgress := data["level_progress"].(map[string]any) - if data["found"] != true || data["host_region_id"] != float64(30) || policy["policy_id"] != "9001" || policy["settlement_mode"] != "half_month" || policy["settlement_trigger_mode"] != "manual" || firstLevel["host_salary_usd"] != 15.0 || firstLevel["agency_salary_usd"] != 3.0 { + if data["found"] != true || data["host_region_id"] != float64(30) || policy["policy_id"] != "9001" || policy["cycle_key"] != expectedCycle || policy["policy_version"] != float64(9001) || policy["settlement_mode"] != "half_month" || policy["settlement_trigger_mode"] != "manual" || firstLevel["host_salary_usd"] != 15.0 || firstLevel["agency_salary_usd"] != 3.0 { t.Fatalf("host policy response mismatch: %+v", data) } - if progress["cycle_key"] != "2026-06" || progress["total_diamonds"] != float64(1200) { + if progress["cycle_key"] != expectedCycle || progress["total_diamonds"] != float64(1200) { t.Fatalf("host salary progress response mismatch: %+v", progress) } if levelProgress["current_level"] != float64(1) || levelProgress["current_value"] != float64(1200) || levelProgress["next_level"] != float64(2) || levelProgress["needed_for_next_level"] != float64(1800) { @@ -6927,6 +6930,7 @@ func TestAgencyCenterOverviewAllowsOwnerWithoutManagerProfile(t *testing.T) { } func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { + expectedCycle := time.Now().UTC().Format("2006-01") hostClient := &fakeUserHostClient{ roleSummary: &userv1.UserRoleSummary{UserId: 42, IsAgency: true, IsManager: false, AgencyId: 7001}, getAgencyResp: &userv1.GetAgencyResponse{Agency: &userv1.Agency{ @@ -6942,6 +6946,8 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { Found: true, Policy: &walletv1.HostSalaryPolicy{ PolicyId: 9101, + CycleKey: expectedCycle, + PolicyVersion: 9101, Name: "Agency Region Policy", RegionId: 31, Status: "active", @@ -6973,7 +6979,7 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { if hostClient.lastRoleSummary == nil || hostClient.lastRoleSummary.GetUserId() != 42 || hostClient.lastGetAgency == nil || hostClient.lastGetAgency.GetAgencyId() != 7001 { t.Fatalf("agency owner resolution mismatch: role=%+v agency=%+v", hostClient.lastRoleSummary, hostClient.lastGetAgency) } - if walletClient.lastHostSalaryPolicy == nil || walletClient.lastHostSalaryPolicy.GetRegionId() != 31 || walletClient.lastHostSalaryPolicy.GetSettlementTriggerMode() != "" || walletClient.lastHostSalaryPolicy.GetAppCode() == "" || walletClient.lastHostSalaryPolicy.GetRequestId() == "" { + if walletClient.lastHostSalaryPolicy == nil || walletClient.lastHostSalaryPolicy.GetRegionId() != 31 || walletClient.lastHostSalaryPolicy.GetCycleKey() != expectedCycle || walletClient.lastHostSalaryPolicy.GetSettlementTriggerMode() != "" || walletClient.lastHostSalaryPolicy.GetAppCode() == "" || walletClient.lastHostSalaryPolicy.GetRequestId() == "" { t.Fatalf("agency policy request mismatch: %+v", walletClient.lastHostSalaryPolicy) } if walletClient.lastHostSalaryProgress != nil { @@ -6987,7 +6993,7 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { policy := data["policy"].(map[string]any) levels := policy["levels"].([]any) secondLevel := levels[1].(map[string]any) - if data["found"] != true || data["agency_region_id"] != float64(31) || policy["policy_id"] != "9101" || len(levels) != 2 || secondLevel["agency_salary_usd"] != 7.2 { + if data["found"] != true || data["agency_region_id"] != float64(31) || policy["policy_id"] != "9101" || policy["cycle_key"] != expectedCycle || policy["policy_version"] != float64(9101) || len(levels) != 2 || secondLevel["agency_salary_usd"] != 7.2 { t.Fatalf("agency policy response mismatch: %+v", data) } } diff --git a/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go b/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go index 86cecdcb..fbbd7157 100644 --- a/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go +++ b/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go @@ -126,6 +126,7 @@ func (h *Handler) getAgencyCenterPlatformPolicy(writer http.ResponseWriter, requ AppCode: appcode.FromContext(request.Context()), RegionId: agency.GetRegionId(), SettlementTriggerMode: "", + CycleKey: time.Now().UTC().Format("2006-01"), }) if err != nil { httpkit.WriteRPCError(writer, request, err) diff --git a/services/gateway-service/internal/transport/http/userapi/host_center_policy_handler.go b/services/gateway-service/internal/transport/http/userapi/host_center_policy_handler.go index dfc24d78..4b69398b 100644 --- a/services/gateway-service/internal/transport/http/userapi/host_center_policy_handler.go +++ b/services/gateway-service/internal/transport/http/userapi/host_center_policy_handler.go @@ -3,6 +3,7 @@ package userapi import ( "net/http" "strconv" + "time" userv1 "hyapp.local/api/proto/user/v1" walletv1 "hyapp.local/api/proto/wallet/v1" @@ -15,6 +16,8 @@ const hostProfileStatusActive = "active" type hostCenterPolicyData struct { PolicyID string `json:"policy_id"` + CycleKey string `json:"cycle_key"` + PolicyVersion uint64 `json:"policy_version"` Name string `json:"name"` RegionID int64 `json:"region_id"` Status string `json:"status"` @@ -68,11 +71,13 @@ func (h *Handler) getHostCenterPlatformPolicy(writer http.ResponseWriter, reques return } - // H5 不传 region、policy id 或 trigger；gateway 用当前主播身份区域读取 active 政策，让 automatic/manual 都能展示。 + cycleKey := time.Now().UTC().Format("2006-01") + // 政策和进度共用同一个 UTC 周期锚点，避免月界瞬间两个调用读取到不同月份。 resp, err := h.walletClient.GetActiveHostSalaryPolicy(request.Context(), &walletv1.GetActiveHostSalaryPolicyRequest{ RequestId: httpkit.RequestIDFromContext(request.Context()), AppCode: appcode.FromContext(request.Context()), RegionId: profile.GetRegionId(), + CycleKey: cycleKey, }) if err != nil { httpkit.WriteRPCError(writer, request, err) @@ -84,6 +89,7 @@ func (h *Handler) getHostCenterPlatformPolicy(writer http.ResponseWriter, reques RequestId: httpkit.RequestIDFromContext(request.Context()), AppCode: appcode.FromContext(request.Context()), HostUserId: profile.GetUserId(), + CycleKey: cycleKey, }) if err != nil { httpkit.WriteRPCError(writer, request, err) @@ -132,6 +138,8 @@ func hostCenterPolicyFromProto(policy *walletv1.HostSalaryPolicy) *hostCenterPol } return &hostCenterPolicyData{ PolicyID: strconv.FormatUint(policy.GetPolicyId(), 10), + CycleKey: policy.GetCycleKey(), + PolicyVersion: policy.GetPolicyVersion(), Name: policy.GetName(), RegionID: policy.GetRegionId(), Status: policy.GetStatus(), diff --git a/services/room-service/deploy/mysql/initdb/001_room_service.sql b/services/room-service/deploy/mysql/initdb/001_room_service.sql index ce5983ef..b026570c 100644 --- a/services/room-service/deploy/mysql/initdb/001_room_service.sql +++ b/services/room-service/deploy/mysql/initdb/001_room_service.sql @@ -117,6 +117,8 @@ CREATE TABLE IF NOT EXISTS room_background_images ( background_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '背景图 ID', room_id VARCHAR(64) NOT NULL COMMENT '房间 ID', image_url VARCHAR(1024) NOT NULL COMMENT '背景图 URL', + -- utf8mb4 的 1024 字符 URL 无法直接进入 InnoDB 3072 字节唯一键；生成列保留完整 URL 唯一性，避免前缀索引误判。 + image_url_sha256 BINARY(32) GENERATED ALWAYS AS (UNHEX(SHA2(image_url, 256))) STORED COMMENT '背景图 URL 唯一性哈希', command_id VARCHAR(128) NULL COMMENT '新版素材保存幂等键；历史行保持 NULL', object_key VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'COS 对象 key', content_type VARCHAR(64) NOT NULL DEFAULT '' COMMENT '服务端嗅探 MIME', @@ -135,7 +137,7 @@ CREATE TABLE IF NOT EXISTS room_background_images ( created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', PRIMARY KEY (background_id), - UNIQUE KEY uk_room_background_url (app_code, room_id, image_url), + UNIQUE KEY uk_room_background_url (app_code, room_id, image_url_sha256), UNIQUE KEY uk_room_background_command (app_code, room_id, command_id), KEY idx_room_background_list (app_code, room_id, created_at_ms DESC, background_id DESC) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='房间背景图素材表'; diff --git a/services/room-service/deploy/mysql/migrations/003_room_vip_media_and_decorations.sql b/services/room-service/deploy/mysql/migrations/003_room_vip_media_and_decorations.sql index fe38217d..c03d7bf2 100644 --- a/services/room-service/deploy/mysql/migrations/003_room_vip_media_and_decorations.sql +++ b/services/room-service/deploy/mysql/migrations/003_room_vip_media_and_decorations.sql @@ -21,6 +21,15 @@ BEGIN SELECT 1 FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'room_background_images' ) THEN + -- 先移除旧 URL 唯一键，否则将 image_url 扩到 1024 字符时，utf8mb4 最大键长会超过 InnoDB 3072 字节限制。 + IF EXISTS ( + SELECT 1 FROM information_schema.statistics + WHERE table_schema = DATABASE() AND table_name = 'room_background_images' + AND index_name = 'uk_room_background_url' + ) THEN + ALTER TABLE room_background_images DROP INDEX uk_room_background_url, ALGORITHM=INPLACE, LOCK=NONE; + END IF; + IF EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'room_background_images' @@ -43,6 +52,13 @@ BEGIN IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'room_background_images' AND column_name = 'command_id') THEN ALTER TABLE room_background_images ADD COLUMN command_id VARCHAR(128) NULL COMMENT '新版素材保存幂等键；历史行保持 NULL', ALGORITHM=INSTANT; END IF; + IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'room_background_images' AND column_name = 'image_url_sha256') THEN + -- 不使用 URL 前缀唯一键；前 700 余字符相同的不同 URL 仍必须能同时保存。 + ALTER TABLE room_background_images + ADD COLUMN image_url_sha256 BINARY(32) GENERATED ALWAYS AS (UNHEX(SHA2(image_url, 256))) STORED COMMENT '背景图 URL 唯一性哈希', + -- MySQL 8.4 新增 STORED 生成列只支持 COPY；该表为低频素材表，复制期间的写锁是可控的一次性迁移成本。 + ALGORITHM=COPY; + END IF; IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'room_background_images' AND column_name = 'object_key') THEN ALTER TABLE room_background_images ADD COLUMN object_key VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'COS 对象 key', ALGORITHM=INSTANT; END IF; @@ -83,6 +99,17 @@ BEGIN ALTER TABLE room_background_images ADD COLUMN media_status VARCHAR(32) NOT NULL DEFAULT '' COMMENT 'active 表示已通过专用上传入口', ALGORITHM=INSTANT; END IF; + IF NOT EXISTS ( + SELECT 1 FROM information_schema.statistics + WHERE table_schema = DATABASE() AND table_name = 'room_background_images' + AND index_name = 'uk_room_background_url' + ) THEN + ALTER TABLE room_background_images + ADD UNIQUE KEY uk_room_background_url (app_code, room_id, image_url_sha256), + ALGORITHM=INPLACE, + LOCK=NONE; + END IF; + IF NOT EXISTS ( SELECT 1 FROM information_schema.statistics WHERE table_schema = DATABASE() AND table_name = 'room_background_images' diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index 44088acf..8b469a90 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -1,4 +1,5 @@ SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; +SET time_zone = '+00:00'; -- 使用 utf8mb4 连接字符集，保证中文注释、中文名称和 emoji 在导入时不被降级。 @@ -288,6 +289,8 @@ CREATE TABLE IF NOT EXISTS host_period_diamond_entries ( CREATE TABLE IF NOT EXISTS host_agency_salary_policies ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', policy_id BIGINT UNSIGNED NOT NULL COMMENT '工资政策 ID，来自后台配置', + cycle_key VARCHAR(7) NOT NULL COMMENT '政策归属工资周期，UTC yyyy-MM', + policy_version BIGINT UNSIGNED NOT NULL COMMENT '不可变政策版本，等于后台政策 ID', name VARCHAR(120) NOT NULL DEFAULT '' COMMENT '政策名称', region_id BIGINT NOT NULL COMMENT '适用区域 ID', status VARCHAR(24) NOT NULL DEFAULT 'disabled' COMMENT '状态：active/disabled', @@ -300,9 +303,21 @@ CREATE TABLE IF NOT EXISTS host_agency_salary_policies ( created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', PRIMARY KEY (app_code, policy_id), - KEY idx_host_agency_salary_policy_active (app_code, region_id, status, settlement_mode, settlement_trigger_mode, effective_from_ms, effective_to_ms) + KEY idx_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status, settlement_mode, settlement_trigger_mode) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播代理工资政策运行快照表'; +-- 周期绑定一旦插入不可更新；复合主键保证新月份政策无法覆盖历史月份结算口径。 +CREATE TABLE IF NOT EXISTS host_salary_policy_cycle_bindings ( + app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', + region_id BIGINT NOT NULL COMMENT '适用区域 ID', + cycle_key VARCHAR(7) NOT NULL COMMENT '工资周期键，UTC yyyy-MM', + policy_id BIGINT UNSIGNED NOT NULL COMMENT '绑定的不可变政策 ID', + policy_version BIGINT UNSIGNED NOT NULL COMMENT '绑定的不可变政策版本', + created_at_ms BIGINT NOT NULL COMMENT '首次绑定时间，UTC epoch ms', + PRIMARY KEY (app_code, region_id, cycle_key), + UNIQUE KEY uk_host_salary_policy_cycle_version (app_code, policy_id, policy_version) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播工资周期不可变政策绑定表'; + CREATE TABLE IF NOT EXISTS host_agency_salary_policy_levels ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', policy_id BIGINT UNSIGNED NOT NULL COMMENT '工资政策 ID', @@ -333,6 +348,7 @@ CREATE TABLE IF NOT EXISTS host_salary_settlement_progress ( residual_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '月底剩余钻石已转主播美元金额，单位美分', month_end_cleared_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '月底清算完成时间，0 表示未清算', last_policy_id BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '最后一次使用的政策 ID', + last_policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '最后一次使用的不可变政策版本', version BIGINT NOT NULL DEFAULT 1 COMMENT '版本号', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', @@ -352,6 +368,7 @@ CREATE TABLE IF NOT EXISTS host_salary_settlement_records ( agency_owner_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '代理收款用户 ID', cycle_key VARCHAR(16) NOT NULL COMMENT '工资周期键，UTC yyyy-MM', policy_id BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '使用的政策 ID', + policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '使用的不可变政策版本', level_no INT NOT NULL DEFAULT 0 COMMENT '本次结算后的等级', total_diamonds BIGINT NOT NULL DEFAULT 0 COMMENT '结算时周期累计钻石', host_salary_usd_minor_delta BIGINT NOT NULL DEFAULT 0 COMMENT '本次主播美元工资增量，单位美分', @@ -436,6 +453,78 @@ PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND COLUMN_NAME = 'cycle_key') = 0, + 'ALTER TABLE host_agency_salary_policies ADD COLUMN cycle_key VARCHAR(7) NOT NULL DEFAULT '''' COMMENT ''政策归属工资周期，UTC yyyy-MM'' AFTER policy_id', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE host_agency_salary_policies ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''不可变政策版本，等于后台政策 ID'' AFTER cycle_key', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +-- 配置表行数很小；只回填未版本化旧政策，不扫描工资账户或结算流水。 +UPDATE host_agency_salary_policies +SET cycle_key = DATE_FORMAT(FROM_UNIXTIME(COALESCE(NULLIF(effective_from_ms, 0), created_at_ms) / 1000), '%Y-%m'), + policy_version = policy_id +WHERE cycle_key = '' OR policy_version = 0; + +-- 旧库同一区域同月若有多条快照，选择生效时间和 policy_id 最大的一条作为一次性迁移绑定；之后主键禁止覆盖。 +INSERT IGNORE INTO host_salary_policy_cycle_bindings ( + app_code, region_id, cycle_key, policy_id, policy_version, created_at_ms +) +SELECT policy.app_code, policy.region_id, policy.cycle_key, policy.policy_id, policy.policy_version, policy.created_at_ms +FROM host_agency_salary_policies policy +WHERE policy.cycle_key <> '' + AND policy.status = 'active' + AND NOT EXISTS ( + SELECT 1 + FROM host_agency_salary_policies newer + WHERE newer.app_code = policy.app_code + AND newer.region_id = policy.region_id + AND newer.cycle_key = policy.cycle_key + AND newer.status = 'active' + AND ( + newer.effective_from_ms > policy.effective_from_ms + OR (newer.effective_from_ms = policy.effective_from_ms AND newer.policy_id > policy.policy_id) + ) + ); + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND INDEX_NAME = 'idx_host_agency_salary_policy_cycle') = 0, + 'ALTER TABLE host_agency_salary_policies ADD INDEX idx_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status, settlement_mode, settlement_trigger_mode)', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_progress' AND COLUMN_NAME = 'last_policy_version') = 0, + 'ALTER TABLE host_salary_settlement_progress ADD COLUMN last_policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''最后一次使用的不可变政策版本'' AFTER last_policy_id', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_records' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE host_salary_settlement_records ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''使用的不可变政策版本'' AFTER policy_id', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + SET @ddl := IF( (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_progress' AND COLUMN_NAME = 'last_settled_total_diamonds') = 0, 'ALTER TABLE host_salary_settlement_progress ADD COLUMN last_settled_total_diamonds BIGINT NOT NULL DEFAULT 0 COMMENT ''上次结算任务已处理到的周期累计钻石，避免未升档时重复空跑'' AFTER settled_agency_salary_usd_minor', diff --git a/services/wallet-service/deploy/mysql/migrations/004_host_salary_policy_cycle_versions.sql b/services/wallet-service/deploy/mysql/migrations/004_host_salary_policy_cycle_versions.sql new file mode 100644 index 00000000..27c56687 --- /dev/null +++ b/services/wallet-service/deploy/mysql/migrations/004_host_salary_policy_cycle_versions.sql @@ -0,0 +1,94 @@ +SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; +SET time_zone = '+00:00'; + +USE hyapp_wallet; + +-- 工资政策和等级是小型配置表；新增列及索引只触碰配置数据，不扫描周期账户或钱包流水。 +-- 结算进度/记录可能持续增长，因此只新增带默认值的审计列并使用在线 DDL，不执行历史全表回填。 +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND COLUMN_NAME = 'cycle_key') = 0, + 'ALTER TABLE host_agency_salary_policies ADD COLUMN cycle_key VARCHAR(7) NOT NULL DEFAULT '''' COMMENT ''政策归属工资周期，UTC yyyy-MM'' AFTER policy_id, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE host_agency_salary_policies ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''不可变政策版本，等于后台政策 ID'' AFTER cycle_key, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +-- 老政策没有显式周期时按原有效期起点归档；起点为 0 的历史行退回创建月份。会话已固定 UTC，避免服务器时区漂移。 +UPDATE host_agency_salary_policies +SET cycle_key = DATE_FORMAT(FROM_UNIXTIME(COALESCE(NULLIF(effective_from_ms, 0), created_at_ms) / 1000), '%Y-%m'), + policy_version = policy_id +WHERE cycle_key = '' OR policy_version = 0; + +CREATE TABLE IF NOT EXISTS host_salary_policy_cycle_bindings ( + app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', + region_id BIGINT NOT NULL COMMENT '适用区域 ID', + cycle_key VARCHAR(7) NOT NULL COMMENT '工资周期键，UTC yyyy-MM', + policy_id BIGINT UNSIGNED NOT NULL COMMENT '绑定的不可变政策 ID', + policy_version BIGINT UNSIGNED NOT NULL COMMENT '绑定的不可变政策版本', + created_at_ms BIGINT NOT NULL COMMENT '首次绑定时间，UTC epoch ms', + PRIMARY KEY (app_code, region_id, cycle_key), + UNIQUE KEY uk_host_salary_policy_cycle_version (app_code, policy_id, policy_version) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播工资周期不可变政策绑定表'; + +-- 同一区域同月若存在多条旧快照，只选原查询会命中的最新一条；INSERT IGNORE 保证迁移可重放且不会覆盖既有绑定。 +INSERT IGNORE INTO host_salary_policy_cycle_bindings ( + app_code, region_id, cycle_key, policy_id, policy_version, created_at_ms +) +SELECT policy.app_code, policy.region_id, policy.cycle_key, policy.policy_id, policy.policy_version, policy.created_at_ms +FROM host_agency_salary_policies policy +WHERE policy.cycle_key <> '' + AND policy.status = 'active' + AND NOT EXISTS ( + SELECT 1 + FROM host_agency_salary_policies newer + WHERE newer.app_code = policy.app_code + AND newer.region_id = policy.region_id + AND newer.cycle_key = policy.cycle_key + AND newer.status = 'active' + AND ( + newer.effective_from_ms > policy.effective_from_ms + OR (newer.effective_from_ms = policy.effective_from_ms AND newer.policy_id > policy.policy_id) + ) + ); + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND INDEX_NAME = 'idx_host_agency_salary_policy_cycle') = 0, + 'ALTER TABLE host_agency_salary_policies ADD INDEX idx_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status, settlement_mode, settlement_trigger_mode), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_progress' AND COLUMN_NAME = 'last_policy_version') = 0, + 'ALTER TABLE host_salary_settlement_progress ADD COLUMN last_policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''最后一次使用的不可变政策版本'' AFTER last_policy_id, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_salary_settlement_records' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE host_salary_settlement_records ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''使用的不可变政策版本'' AFTER policy_id, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/services/wallet-service/internal/domain/ledger/host_salary.go b/services/wallet-service/internal/domain/ledger/host_salary.go index 66e28345..5d3f9f4d 100644 --- a/services/wallet-service/internal/domain/ledger/host_salary.go +++ b/services/wallet-service/internal/domain/ledger/host_salary.go @@ -28,7 +28,10 @@ type HostSalarySettlementBatchResult struct { // HostSalaryPolicy 是 wallet-service 结算读取的后台政策运行快照。 type HostSalaryPolicy struct { - PolicyID uint64 + PolicyID uint64 + // CycleKey 与 PolicyVersion 共同标识已发布的不可变周期快照；结算只能按账户 cycle_key 精确命中。 + CycleKey string + PolicyVersion uint64 Name string RegionID int64 Status string @@ -68,7 +71,7 @@ type HostSalaryProgress struct { type TeamHostSalaryStatsQuery struct { AgencyOwnerUserIDs []int64 CycleKeys []string - // NowMs 用于解析各区域当前生效的工资政策做档位估算。 + // NowMs 只保留调用兼容；工资估算必须按每个 CycleKey 读取绑定版本，不能按查询执行时间选政策。 NowMs int64 } diff --git a/services/wallet-service/internal/service/wallet/host_salary.go b/services/wallet-service/internal/service/wallet/host_salary.go index a558d464..2636e976 100644 --- a/services/wallet-service/internal/service/wallet/host_salary.go +++ b/services/wallet-service/internal/service/wallet/host_salary.go @@ -35,6 +35,9 @@ func (s *Service) ProcessHostSalarySettlementBatch(ctx context.Context, command if strings.TrimSpace(command.CycleKey) == "" { command.CycleKey = hostSalaryCycleKey(command.NowMs) } + if !validHostSalaryCycleKey(command.CycleKey) { + return ledger.HostSalarySettlementBatchResult{}, xerr.New(xerr.InvalidArgument, "cycle_key must use YYYY-MM") + } ctx = appcode.WithContext(ctx, command.AppCode) return s.repository.ProcessHostSalarySettlementBatch(ctx, command) } @@ -44,8 +47,17 @@ func hostSalaryCycleKey(nowMs int64) string { return time.UnixMilli(nowMs).UTC().Format("2006-01") } -// GetActiveHostSalaryPolicy 读取主播所在区域当前生效的工资政策，返回的 policy 与工资结算使用同一张 runtime 表。 -func (s *Service) GetActiveHostSalaryPolicy(ctx context.Context, appCode string, regionID int64, settlementMode string, triggerMode string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) { +func validHostSalaryCycleKey(cycleKey string) bool { + cycleKey = strings.TrimSpace(cycleKey) + if len(cycleKey) != len("2006-01") { + return false + } + parsed, err := time.Parse("2006-01", cycleKey) + return err == nil && parsed.UTC().Format("2006-01") == cycleKey +} + +// GetActiveHostSalaryPolicy 按工资周期读取主播所在区域已绑定的不可变政策版本。 +func (s *Service) GetActiveHostSalaryPolicy(ctx context.Context, appCode string, regionID int64, settlementMode string, triggerMode string, cycleKey string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) { if regionID <= 0 { return ledger.HostSalaryPolicy{}, false, xerr.New(xerr.InvalidArgument, "region_id is required") } @@ -63,10 +75,18 @@ func (s *Service) GetActiveHostSalaryPolicy(ctx context.Context, appCode string, if nowMs <= 0 { nowMs = s.now().UTC().UnixMilli() } + cycleKey = strings.TrimSpace(cycleKey) + if cycleKey == "" { + // 旧 H5 未传周期时只允许展示层回退当前 UTC 月；结算批次始终显式携带自己的 cycle_key。 + cycleKey = hostSalaryCycleKey(nowMs) + } + if !validHostSalaryCycleKey(cycleKey) { + return ledger.HostSalaryPolicy{}, false, xerr.New(xerr.InvalidArgument, "cycle_key must use YYYY-MM") + } appCode = appcode.Normalize(appCode) ctx = appcode.WithContext(ctx, appCode) // H5 展示可以不传 trigger mode，此时 automatic/manual 政策都可展示；结算任务仍会传具体 trigger 精确匹配。 - return s.repository.GetActiveHostSalaryPolicy(ctx, regionID, settlementMode, triggerMode, nowMs) + return s.repository.GetActiveHostSalaryPolicy(ctx, regionID, settlementMode, triggerMode, cycleKey) } // GetTeamHostSalaryStats 聚合一组 Agency 收款人名下主播的预收入工资，经理中心团队工资卡片使用。 diff --git a/services/wallet-service/internal/service/wallet/ports/repository.go b/services/wallet-service/internal/service/wallet/ports/repository.go index 3f065710..5bba12bd 100644 --- a/services/wallet-service/internal/service/wallet/ports/repository.go +++ b/services/wallet-service/internal/service/wallet/ports/repository.go @@ -52,7 +52,7 @@ type BalanceStore interface { // HostSalaryStore 管理主播/公会工资政策、进度和批量结算。 type HostSalaryStore interface { - GetActiveHostSalaryPolicy(ctx context.Context, regionID int64, settlementMode string, triggerMode string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) + GetActiveHostSalaryPolicy(ctx context.Context, regionID int64, settlementMode string, triggerMode string, cycleKey string) (ledger.HostSalaryPolicy, bool, error) GetHostSalaryProgress(ctx context.Context, userID int64, cycleKey string) (ledger.HostSalaryProgress, error) GetTeamHostSalaryStats(ctx context.Context, query ledger.TeamHostSalaryStatsQuery) ([]ledger.TeamHostSalaryCycleStat, error) GetHostRevenueStats(ctx context.Context, query ledger.HostRevenueStatsQuery) (ledger.HostRevenueStats, error) diff --git a/services/wallet-service/internal/service/wallet/service_test.go b/services/wallet-service/internal/service/wallet/service_test.go index e76850ad..4365850c 100644 --- a/services/wallet-service/internal/service/wallet/service_test.go +++ b/services/wallet-service/internal/service/wallet/service_test.go @@ -777,14 +777,15 @@ func TestGetActiveHostSalaryPolicyReadsRuntimePolicy(t *testing.T) { }) svc := walletservice.New(repository) - policy, found, err := svc.GetActiveHostSalaryPolicy(context.Background(), "lalu", 8801, "", "", nowMs) + cycleKey := time.UnixMilli(nowMs).UTC().Format("2006-01") + policy, found, err := svc.GetActiveHostSalaryPolicy(context.Background(), "lalu", 8801, "", "", cycleKey, nowMs) if err != nil { t.Fatalf("GetActiveHostSalaryPolicy failed: %v", err) } if !found { t.Fatal("expected active host salary policy") } - if policy.PolicyID != 901 || policy.Name != "current host policy" || policy.SettlementMode != ledger.HostSalarySettlementModeHalfMonth || policy.RegionID != 8801 { + if policy.PolicyID != 901 || policy.PolicyVersion != 901 || policy.CycleKey != cycleKey || policy.Name != "current host policy" || policy.SettlementMode != ledger.HostSalarySettlementModeHalfMonth || policy.RegionID != 8801 { t.Fatalf("policy mismatch: %+v", policy) } if policy.SettlementTriggerMode != ledger.HostSalarySettlementTriggerManual { @@ -793,7 +794,7 @@ func TestGetActiveHostSalaryPolicyReadsRuntimePolicy(t *testing.T) { if len(policy.Levels) != 2 || policy.Levels[1].LevelNo != 2 || policy.Levels[1].HostSalaryUSDMinor != 4500 || policy.Levels[1].AgencySalaryUSDMinor != 900 { t.Fatalf("policy levels mismatch: %+v", policy.Levels) } - if _, found, err := svc.GetActiveHostSalaryPolicy(context.Background(), "lalu", 8801, "", ledger.HostSalarySettlementTriggerAutomatic, nowMs); err != nil { + if _, found, err := svc.GetActiveHostSalaryPolicy(context.Background(), "lalu", 8801, "", ledger.HostSalarySettlementTriggerAutomatic, cycleKey, nowMs); err != nil { t.Fatalf("GetActiveHostSalaryPolicy with trigger failed: %v", err) } else if found { t.Fatal("automatic trigger query should not match manual policy") @@ -1192,27 +1193,109 @@ func TestHostSalaryMonthEndCreditsResidualAndClearsCycle(t *testing.T) { } } +// TestHostSalaryMonthEndUsesBoundPreviousCyclePolicy 验证次月执行上月月结时只能读取上月绑定版本，不能使用执行月份的新政策。 +func TestHostSalaryMonthEndUsesBoundPreviousCyclePolicy(t *testing.T) { + repository := mysqltest.NewRepository(t) + repository.SetBalance(10001, 500) + repository.SetGiftPrice("rose", "salary-cycle-version", 200, 10, 10) + previousCycle := time.Now().UTC().AddDate(0, -1, 0).Format("2006-01") + currentCycle := time.Now().UTC().Format("2006-01") + repository.SetHostSalaryPolicy(ledger.HostSalaryPolicy{ + PolicyID: 401, + CycleKey: previousCycle, + PolicyVersion: 401, + Name: "previous cycle policy", + RegionID: 8801, + Status: "active", + SettlementMode: ledger.HostSalarySettlementModeDaily, + Levels: []ledger.HostSalaryPolicyLevel{ + {LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSDMinor: 150, SortOrder: 1}, + }, + }) + repository.SetHostSalaryPolicy(ledger.HostSalaryPolicy{ + PolicyID: 402, + CycleKey: currentCycle, + PolicyVersion: 402, + Name: "current cycle policy", + RegionID: 8801, + Status: "active", + SettlementMode: ledger.HostSalarySettlementModeDaily, + Levels: []ledger.HostSalaryPolicyLevel{ + {LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSDMinor: 900, SortOrder: 1}, + }, + }) + svc := walletservice.New(repository) + receipt, err := svc.DebitGift(context.Background(), ledger.DebitGiftCommand{ + CommandID: "cmd-salary-cycle-version", + RoomID: "room-1", + SenderUserID: 10001, + TargetUserID: 10002, + GiftID: "rose", + GiftCount: 1, + PriceVersion: "salary-cycle-version", + TargetIsHost: true, + TargetHostRegionID: 8801, + }) + if err != nil { + t.Fatalf("cycle version gift failed: %v", err) + } + repository.MoveHostPeriodCycle(10002, receipt.HostPeriodCycleKey, previousCycle) + + result, err := svc.ProcessHostSalarySettlementBatch(context.Background(), ledger.HostSalarySettlementBatchCommand{ + AppCode: "lalu", + RunID: "salary-cycle-version-month-end", + WorkerID: "salary-test", + BatchSize: 10, + SettlementType: ledger.HostSalarySettlementTypeMonthEnd, + CycleKey: previousCycle, + NowMs: time.Now().UnixMilli(), + }) + if err != nil { + t.Fatalf("previous cycle month-end settlement failed: %v", err) + } + if result.SuccessCount != 1 || result.FailureCount != 0 { + t.Fatalf("previous cycle month-end result mismatch: %+v", result) + } + assertBalance(t, svc, 10002, ledger.AssetHostSalaryUSD, 150) + if got := repository.CountRows("host_salary_settlement_records", "user_id = ? AND cycle_key = ? AND policy_id = ? AND policy_version = ? AND host_salary_usd_minor_delta = ?", int64(10002), previousCycle, uint64(401), uint64(401), int64(150)); got != 1 { + t.Fatalf("month-end must persist previous cycle policy version, got %d", got) + } +} + // TestHostSalaryFullLifecycleFromPublishedPolicyToNextCycle 走通“已发布政策 -> 主播收礼 -> 多次增量结算 -> 月底清算 -> 下周期重新起算”的真实账务链路。 func TestHostSalaryFullLifecycleFromPublishedPolicyToNextCycle(t *testing.T) { repository := mysqltest.NewRepository(t) repository.SetBalance(10001, 2000) repository.SetGiftPrice("rose", "salary-full-lifecycle", 120, 10, 10) - // 管理后台发布后 wallet-service 消费的是 host_agency_salary_policies 运行时快照；这里直接 seed 该表来锁定送礼和结算链路。 + previousCycle := time.Now().UTC().AddDate(0, -1, 0).Format("2006-01") + currentCycle := time.Now().UTC().Format("2006-01") + levels := []ledger.HostSalaryPolicyLevel{ + {LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSDMinor: 150, HostCoinReward: 90, AgencySalaryUSDMinor: 50, SortOrder: 1}, + {LevelNo: 2, RequiredDiamonds: 400, HostSalaryUSDMinor: 300, HostCoinReward: 198, AgencySalaryUSDMinor: 80, SortOrder: 2}, + } + // 每个月分别发布不可变版本；相同金额也不能跨周期复用同一条 runtime 政策。 repository.SetHostSalaryPolicy(ledger.HostSalaryPolicy{ PolicyID: 301, - Name: "published daily salary", + CycleKey: previousCycle, + Name: "previous cycle daily salary", RegionID: 8801, Status: "active", SettlementMode: ledger.HostSalarySettlementModeDaily, ResidualDiamondToUSDRate: "0.01", - Levels: []ledger.HostSalaryPolicyLevel{ - {LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSDMinor: 150, HostCoinReward: 90, AgencySalaryUSDMinor: 50, SortOrder: 1}, - {LevelNo: 2, RequiredDiamonds: 400, HostSalaryUSDMinor: 300, HostCoinReward: 198, AgencySalaryUSDMinor: 80, SortOrder: 2}, - }, + Levels: levels, + }) + repository.SetHostSalaryPolicy(ledger.HostSalaryPolicy{ + PolicyID: 304, + CycleKey: currentCycle, + Name: "current cycle daily salary", + RegionID: 8801, + Status: "active", + SettlementMode: ledger.HostSalarySettlementModeDaily, + ResidualDiamondToUSDRate: "0.01", + Levels: levels, }) svc := walletservice.New(repository) ctx := context.Background() - previousCycle := time.Now().UTC().AddDate(0, -1, 0).Format("2006-01") first, err := svc.DebitGift(ctx, ledger.DebitGiftCommand{ CommandID: "cmd-full-lifecycle-gift-1", diff --git a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go index 0c05d8b4..8cbe86d3 100644 --- a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go +++ b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go @@ -34,6 +34,7 @@ type hostSalaryProgress struct { ResidualUSDMinor int64 MonthEndClearedAtMS int64 LastPolicyID uint64 + LastPolicyVersion uint64 Version int64 } @@ -48,6 +49,7 @@ type hostSalarySettlementMetadata struct { CycleKey string `json:"cycle_key"` RegionID int64 `json:"region_id"` PolicyID uint64 `json:"policy_id"` + PolicyVersion uint64 `json:"policy_version"` PolicyName string `json:"policy_name"` LevelNo int `json:"level_no"` RequiredDiamonds int64 `json:"required_diamonds"` @@ -151,7 +153,7 @@ func (r *Repository) listHostSalaryCandidates(ctx context.Context, command ledge args = append(args, id) } } - args = append(args, command.NowMs, command.NowMs, command.TriggerMode) + args = append(args, command.TriggerMode) args = append(args, policyModeArgs...) // 日结/半月结只扫描“本周期钻石比上次结算更多”的主播，避免 cron 重跑时反复锁定无新增收益的账户。 // 月底批次必须扫描全部未关闭周期账户，因为它还负责剩余钻石折美元和周期逻辑关闭。 @@ -173,12 +175,17 @@ func (r *Repository) listHostSalaryCandidates(ctx context.Context, command ledge %s AND EXISTS ( SELECT 1 - FROM host_agency_salary_policies p - WHERE p.app_code = a.app_code - AND p.region_id = a.region_id + FROM host_salary_policy_cycle_bindings binding + JOIN host_agency_salary_policies p + ON p.app_code = binding.app_code + AND p.policy_id = binding.policy_id + AND p.region_id = binding.region_id + AND p.cycle_key = binding.cycle_key + AND p.policy_version = binding.policy_version + WHERE binding.app_code = a.app_code + AND binding.region_id = a.region_id + AND binding.cycle_key = a.cycle_key AND p.status = 'active' - AND p.effective_from_ms <= ? - AND (p.effective_to_ms = 0 OR p.effective_to_ms > ?) AND p.settlement_trigger_mode = ? %s ) @@ -225,8 +232,8 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led return true, tx.Commit() } - // 政策按周期账户里的区域快照解析；该快照由同周期最近一次主播收礼写入。 - policy, ok, err := r.resolveHostSalaryPolicy(ctx, tx, account.RegionID, command.SettlementType, command.TriggerMode, command.NowMs) + // 区域来自周期账户快照，政策则沿 (app, region, cycle_key) 绑定读取；执行时间不参与版本选择。 + policy, ok, err := r.resolveHostSalaryPolicy(ctx, tx, account.RegionID, account.CycleKey, command.SettlementType, command.TriggerMode) if err != nil || !ok { return false, err } @@ -268,6 +275,7 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led nextProgress := progress nextProgress.LastPolicyID = policy.PolicyID + nextProgress.LastPolicyVersion = policy.PolicyVersion if shouldSettleHost && (hostSalaryDelta > 0 || hostCoinDelta > 0 || residualDelta > 0) { nextProgress.SettledLevelNo = maxInt(nextProgress.SettledLevelNo, level.LevelNo) nextProgress.SettledHostSalaryUSDMinor = maxInt64(nextProgress.SettledHostSalaryUSDMinor, level.HostSalaryUSDMinor) @@ -304,6 +312,7 @@ func (r *Repository) processHostSalaryCandidate(ctx context.Context, command led CycleKey: account.CycleKey, RegionID: account.RegionID, PolicyID: policy.PolicyID, + PolicyVersion: policy.PolicyVersion, PolicyName: policy.Name, LevelNo: level.LevelNo, RequiredDiamonds: level.RequiredDiamonds, @@ -373,8 +382,9 @@ func (r *Repository) lockHostSalaryProgress(ctx context.Context, tx *sql.Tx, use INSERT IGNORE INTO host_salary_settlement_progress ( app_code, user_id, cycle_key, settled_level_no, settled_host_salary_usd_minor, settled_host_coin_reward, settled_agency_salary_usd_minor, last_settled_total_diamonds, - residual_usd_minor, month_end_cleared_at_ms, last_policy_id, version, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, 0, 0, 0, 0, 0, 0, 0, 0, 1, ?, ?)`, + residual_usd_minor, month_end_cleared_at_ms, last_policy_id, last_policy_version, + version, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, ?, ?)`, appcode.FromContext(ctx), userID, cycleKey, nowMs, nowMs, ); err != nil { return hostSalaryProgress{}, err @@ -382,23 +392,23 @@ func (r *Repository) lockHostSalaryProgress(ctx context.Context, tx *sql.Tx, use row := tx.QueryRowContext(ctx, ` SELECT settled_level_no, settled_host_salary_usd_minor, settled_host_coin_reward, settled_agency_salary_usd_minor, last_settled_total_diamonds, residual_usd_minor, - month_end_cleared_at_ms, last_policy_id, version + month_end_cleared_at_ms, last_policy_id, last_policy_version, version FROM host_salary_settlement_progress WHERE app_code = ? AND user_id = ? AND cycle_key = ? FOR UPDATE`, appcode.FromContext(ctx), userID, cycleKey) var progress hostSalaryProgress - if err := row.Scan(&progress.SettledLevelNo, &progress.SettledHostSalaryUSDMinor, &progress.SettledHostCoinReward, &progress.SettledAgencySalaryUSDMinor, &progress.LastSettledTotalDiamonds, &progress.ResidualUSDMinor, &progress.MonthEndClearedAtMS, &progress.LastPolicyID, &progress.Version); err != nil { + if err := row.Scan(&progress.SettledLevelNo, &progress.SettledHostSalaryUSDMinor, &progress.SettledHostCoinReward, &progress.SettledAgencySalaryUSDMinor, &progress.LastSettledTotalDiamonds, &progress.ResidualUSDMinor, &progress.MonthEndClearedAtMS, &progress.LastPolicyID, &progress.LastPolicyVersion, &progress.Version); err != nil { return hostSalaryProgress{}, err } return progress, nil } -// GetActiveHostSalaryPolicy 返回当前 app、区域和触发模式下正在生效的主播工资政策；这是 H5 展示平台政策的只读入口。 -func (r *Repository) GetActiveHostSalaryPolicy(ctx context.Context, regionID int64, settlementMode string, triggerMode string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) { +// GetActiveHostSalaryPolicy 返回 app、区域和工资周期已绑定的不可变政策版本；这是 H5 展示平台政策的只读入口。 +func (r *Repository) GetActiveHostSalaryPolicy(ctx context.Context, regionID int64, settlementMode string, triggerMode string, cycleKey string) (ledger.HostSalaryPolicy, bool, error) { if r == nil || r.db == nil { return ledger.HostSalaryPolicy{}, false, xerr.New(xerr.Unavailable, "mysql repository is not configured") } - return r.queryActiveHostSalaryPolicy(ctx, r.db, regionID, settlementMode, triggerMode, nowMs) + return r.queryHostSalaryPolicyByCycle(ctx, r.db, regionID, cycleKey, settlementMode, triggerMode) } // GetHostSalaryProgress 读取主播工资周期钻石账户；没有任何收礼时返回空累计，前端据此展示距离首档还差多少钻石。 @@ -425,45 +435,50 @@ func (r *Repository) GetHostSalaryProgress(ctx context.Context, userID int64, cy return progress, nil } -func (r *Repository) resolveHostSalaryPolicy(ctx context.Context, tx *sql.Tx, regionID int64, settlementType string, triggerMode string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) { +func (r *Repository) resolveHostSalaryPolicy(ctx context.Context, tx *sql.Tx, regionID int64, cycleKey string, settlementType string, triggerMode string) (ledger.HostSalaryPolicy, bool, error) { settlementMode := "" if settlementType != ledger.HostSalarySettlementTypeMonthEnd { settlementMode = settlementType } - return r.queryActiveHostSalaryPolicy(ctx, tx, regionID, settlementMode, triggerMode, nowMs) + return r.queryHostSalaryPolicyByCycle(ctx, tx, regionID, cycleKey, settlementMode, triggerMode) } -func (r *Repository) queryActiveHostSalaryPolicy(ctx context.Context, q hostSalaryPolicyQuerier, regionID int64, settlementMode string, triggerMode string, nowMs int64) (ledger.HostSalaryPolicy, bool, error) { +func (r *Repository) queryHostSalaryPolicyByCycle(ctx context.Context, q hostSalaryPolicyQuerier, regionID int64, cycleKey string, settlementMode string, triggerMode string) (ledger.HostSalaryPolicy, bool, error) { modeClause := "" triggerClause := "" - args := []any{appcode.FromContext(ctx), regionID, nowMs, nowMs} + args := []any{appcode.FromContext(ctx), regionID, cycleKey} if strings.TrimSpace(triggerMode) != "" { // H5 展示不传 trigger 时 automatic/manual 都可展示；结算任务传 trigger 时仍精确匹配。 - triggerClause = "AND settlement_trigger_mode = ?" + triggerClause = "AND p.settlement_trigger_mode = ?" args = append(args, triggerMode) } if strings.TrimSpace(settlementMode) != "" { // SQL 里 trigger 条件排在 mode 前面，所以参数也按 trigger -> mode 追加；否则 manual half-month 会被查成 trigger=half_month。 - modeClause = "AND settlement_mode = ?" + modeClause = "AND p.settlement_mode = ?" args = append(args, settlementMode) } query := fmt.Sprintf(` - SELECT policy_id, name, region_id, status, settlement_mode, settlement_trigger_mode, - CAST(gift_coin_to_diamond_ratio AS CHAR), - CAST(residual_diamond_to_usd_rate AS CHAR), - effective_from_ms, effective_to_ms - FROM host_agency_salary_policies - WHERE app_code = ? - AND region_id = ? - AND status = 'active' - AND effective_from_ms <= ? - AND (effective_to_ms = 0 OR effective_to_ms > ?) + SELECT p.policy_id, p.cycle_key, p.policy_version, p.name, p.region_id, p.status, + p.settlement_mode, p.settlement_trigger_mode, + CAST(p.gift_coin_to_diamond_ratio AS CHAR), + CAST(p.residual_diamond_to_usd_rate AS CHAR), + p.effective_from_ms, p.effective_to_ms + FROM host_salary_policy_cycle_bindings binding + JOIN host_agency_salary_policies p + ON p.app_code = binding.app_code + AND p.policy_id = binding.policy_id + AND p.region_id = binding.region_id + AND p.cycle_key = binding.cycle_key + AND p.policy_version = binding.policy_version + WHERE binding.app_code = ? + AND binding.region_id = ? + AND binding.cycle_key = ? + AND p.status = 'active' %s %s - ORDER BY effective_from_ms DESC, policy_id DESC LIMIT 1`, triggerClause, modeClause) var policy ledger.HostSalaryPolicy - err := q.QueryRowContext(ctx, query, args...).Scan(&policy.PolicyID, &policy.Name, &policy.RegionID, &policy.Status, &policy.SettlementMode, &policy.SettlementTriggerMode, &policy.GiftCoinToDiamondRatio, &policy.ResidualDiamondToUSDRate, &policy.EffectiveFromMs, &policy.EffectiveToMs) + err := q.QueryRowContext(ctx, query, args...).Scan(&policy.PolicyID, &policy.CycleKey, &policy.PolicyVersion, &policy.Name, &policy.RegionID, &policy.Status, &policy.SettlementMode, &policy.SettlementTriggerMode, &policy.GiftCoinToDiamondRatio, &policy.ResidualDiamondToUSDRate, &policy.EffectiveFromMs, &policy.EffectiveToMs) if err != nil { if errors.Is(err, sql.ErrNoRows) { return ledger.HostSalaryPolicy{}, false, nil @@ -564,10 +579,10 @@ func (r *Repository) insertHostSalarySettlementRecord(ctx context.Context, tx *s _, err := tx.ExecContext(ctx, ` INSERT INTO host_salary_settlement_records ( app_code, settlement_id, command_id, transaction_id, settlement_type, trigger_mode, settlement_role, user_id, - agency_owner_user_id, cycle_key, policy_id, level_no, total_diamonds, + agency_owner_user_id, cycle_key, policy_id, policy_version, level_no, total_diamonds, host_salary_usd_minor_delta, host_coin_reward_delta, agency_salary_usd_minor_delta, residual_usd_minor_delta, status, reason, created_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'succeeded', '', ?)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'succeeded', '', ?)`, appcode.FromContext(ctx), hostSalarySettlementID(metadata.AppCode, commandID), commandID, @@ -579,6 +594,7 @@ func (r *Repository) insertHostSalarySettlementRecord(ctx context.Context, tx *s metadata.AgencyOwnerUserID, metadata.CycleKey, metadata.PolicyID, + metadata.PolicyVersion, metadata.LevelNo, metadata.TotalDiamonds, metadata.HostSalaryUSDMinorDelta, @@ -602,6 +618,7 @@ func (r *Repository) updateHostSalaryProgress(ctx context.Context, tx *sql.Tx, u residual_usd_minor = ?, month_end_cleared_at_ms = ?, last_policy_id = ?, + last_policy_version = ?, version = version + 1, updated_at_ms = ? WHERE app_code = ? AND user_id = ? AND cycle_key = ? AND version = ?`, @@ -613,6 +630,7 @@ func (r *Repository) updateHostSalaryProgress(ctx context.Context, tx *sql.Tx, u progress.ResidualUSDMinor, progress.MonthEndClearedAtMS, progress.LastPolicyID, + progress.LastPolicyVersion, nowMs, appcode.FromContext(ctx), userID, @@ -736,7 +754,7 @@ func hostSalarySettlementID(appCode string, commandID string) string { func hostSalarySettlementRequestHash(metadata hostSalarySettlementMetadata) string { // 请求哈希覆盖金额、等级和周期口径；同 commandID 但 payload 改变会被交易层识别为冲突。 - return stableHash(fmt.Sprintf("host_salary|%s|%s|%s|%s|%d|%d|%s|%d|%d|%d|%d|%d|%d|%d", + return stableHash(fmt.Sprintf("host_salary|%s|%s|%s|%s|%d|%d|%s|%d|%d|%d|%d|%d|%d|%d|%d", appcode.Normalize(metadata.AppCode), metadata.SettlementType, metadata.TriggerMode, @@ -745,6 +763,7 @@ func hostSalarySettlementRequestHash(metadata hostSalarySettlementMetadata) stri metadata.AgencyOwnerUserID, metadata.CycleKey, metadata.PolicyID, + metadata.PolicyVersion, metadata.LevelNo, metadata.TotalDiamonds, metadata.HostSalaryUSDMinorDelta, diff --git a/services/wallet-service/internal/storage/mysql/host_salary_stats.go b/services/wallet-service/internal/storage/mysql/host_salary_stats.go index e4c72f89..c5b684b8 100644 --- a/services/wallet-service/internal/storage/mysql/host_salary_stats.go +++ b/services/wallet-service/internal/storage/mysql/host_salary_stats.go @@ -2,6 +2,7 @@ package mysql import ( "context" + "fmt" "strings" "hyapp/pkg/appcode" @@ -19,8 +20,8 @@ func (r *Repository) GetTeamHostSalaryStats(ctx context.Context, query ledger.Te } owners := uniquePositiveInt64(query.AgencyOwnerUserIDs) stats := make([]ledger.TeamHostSalaryCycleStat, 0, len(query.CycleKeys)) - // 政策按区域跨周期复用同一份解析结果；估算统一用“当前生效政策”，不回放历史政策版本。 - policyByRegion := map[int64]*ledger.HostSalaryPolicy{} + // 同一区域不同月份可以绑定不同不可变版本；缓存键必须含 cycle_key，禁止把新月份政策套到历史周期。 + policyByRegionCycle := map[string]*ledger.HostSalaryPolicy{} for _, cycleKey := range query.CycleKeys { cycleKey = strings.TrimSpace(cycleKey) if cycleKey == "" { @@ -32,7 +33,7 @@ func (r *Repository) GetTeamHostSalaryStats(ctx context.Context, query ledger.Te stats = append(stats, stat) continue } - if err := r.aggregateTeamHostSalaryCycle(ctx, owners, cycleKey, query.NowMs, policyByRegion, &stat); err != nil { + if err := r.aggregateTeamHostSalaryCycle(ctx, owners, cycleKey, policyByRegionCycle, &stat); err != nil { return nil, err } stats = append(stats, stat) @@ -49,7 +50,7 @@ type teamHostSalaryRow struct { MonthEndClearedAtMS int64 } -func (r *Repository) aggregateTeamHostSalaryCycle(ctx context.Context, owners []int64, cycleKey string, nowMs int64, policyByRegion map[int64]*ledger.HostSalaryPolicy, stat *ledger.TeamHostSalaryCycleStat) error { +func (r *Repository) aggregateTeamHostSalaryCycle(ctx context.Context, owners []int64, cycleKey string, policyByRegionCycle map[string]*ledger.HostSalaryPolicy, stat *ledger.TeamHostSalaryCycleStat) error { // 先把整个结果集读完再解析政策：政策查询和主查询共用连接池，游标未关时嵌套发查询 // 在池被打满的场景下会互相等待造成饥饿；行结构很小，整批驻留内存成本可忽略。 hostRows, err := r.listTeamHostSalaryRows(ctx, owners, cycleKey) @@ -63,10 +64,9 @@ func (r *Repository) aggregateTeamHostSalaryCycle(ctx context.Context, owners [] return err } estimated := settledTotal - // month_end 已清算的周期以真实发放为准，不再用“当前政策”重估： - // 政策在周期关闭后上调时，重估会显示从未发放过的金额，误导经理对账。 + // month_end 已清算的周期以真实发放为准；未关闭周期也只能用自己 cycle_key 已绑定的版本估算。 if row.MonthEndClearedAtMS == 0 { - policy, err := r.resolveTeamSalaryPolicy(ctx, row.RegionID, nowMs, policyByRegion) + policy, err := r.resolveTeamSalaryPolicy(ctx, row.RegionID, cycleKey, policyByRegionCycle) if err != nil { return err } @@ -138,20 +138,21 @@ func (r *Repository) listTeamHostSalaryRows(ctx context.Context, owners []int64, return hostRows, rows.Err() } -func (r *Repository) resolveTeamSalaryPolicy(ctx context.Context, regionID int64, nowMs int64, policyByRegion map[int64]*ledger.HostSalaryPolicy) (*ledger.HostSalaryPolicy, error) { - if cached, ok := policyByRegion[regionID]; ok { +func (r *Repository) resolveTeamSalaryPolicy(ctx context.Context, regionID int64, cycleKey string, policyByRegionCycle map[string]*ledger.HostSalaryPolicy) (*ledger.HostSalaryPolicy, error) { + cacheKey := fmt.Sprintf("%d:%s", regionID, cycleKey) + if cached, ok := policyByRegionCycle[cacheKey]; ok { return cached, nil } - // 展示估算不区分结算模式和触发方式，取区域当前最新生效政策，与 H5 平台政策展示口径一致。 - policy, found, err := r.queryActiveHostSalaryPolicy(ctx, r.db, regionID, "", "", nowMs) + // 展示估算不区分结算模式和触发方式，但仍必须沿周期绑定读取精确版本。 + policy, found, err := r.queryHostSalaryPolicyByCycle(ctx, r.db, regionID, cycleKey, "", "") if err != nil { return nil, err } if !found { // 区域没有生效政策时记 nil，主播预收入退化为已结算累计，不做档位估算。 - policyByRegion[regionID] = nil + policyByRegionCycle[cacheKey] = nil return nil, nil } - policyByRegion[regionID] = &policy + policyByRegionCycle[cacheKey] = &policy return &policy, nil } diff --git a/services/wallet-service/internal/storage/mysql/schema.go b/services/wallet-service/internal/storage/mysql/schema.go index 27117f8e..9377b6db 100644 --- a/services/wallet-service/internal/storage/mysql/schema.go +++ b/services/wallet-service/internal/storage/mysql/schema.go @@ -50,6 +50,72 @@ func isDuplicateKeyNameError(err error) bool { } func ensureHostSalarySettlementSchema(ctx context.Context, db *sql.DB) error { + // 工资政策表是小型配置表；启动兼容只补列、回填配置和建立点查索引，不扫描工资账户或钱包流水。 + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_agency_salary_policies + ADD COLUMN settlement_trigger_mode VARCHAR(24) NOT NULL DEFAULT 'automatic' COMMENT '结算触发方式：automatic/manual' AFTER settlement_mode`); err != nil && !isDuplicateColumnError(err) { + return err + } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_agency_salary_policies + ADD COLUMN cycle_key VARCHAR(7) NOT NULL DEFAULT '' COMMENT '政策归属工资周期，UTC yyyy-MM' AFTER policy_id`); err != nil && !isDuplicateColumnError(err) { + return err + } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_agency_salary_policies + ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '不可变政策版本，等于后台政策 ID' AFTER cycle_key`); err != nil && !isDuplicateColumnError(err) { + return err + } + if _, err := db.ExecContext(ctx, ` + CREATE TABLE IF NOT EXISTS host_salary_policy_cycle_bindings ( + app_code VARCHAR(32) NOT NULL DEFAULT 'lalu', + region_id BIGINT NOT NULL, + cycle_key VARCHAR(7) NOT NULL, + policy_id BIGINT UNSIGNED NOT NULL, + policy_version BIGINT UNSIGNED NOT NULL, + created_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, region_id, cycle_key), + UNIQUE KEY uk_host_salary_policy_cycle_version (app_code, policy_id, policy_version) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播工资周期不可变政策绑定表'`); err != nil { + return err + } + if _, err := db.ExecContext(ctx, ` + UPDATE host_agency_salary_policies + SET cycle_key = DATE_FORMAT( + DATE_ADD('1970-01-01 00:00:00', INTERVAL (COALESCE(NULLIF(effective_from_ms, 0), created_at_ms) DIV 1000) SECOND), + '%Y-%m' + ), + policy_version = policy_id + WHERE cycle_key = '' OR policy_version = 0`); err != nil { + return err + } + if _, err := db.ExecContext(ctx, ` + INSERT IGNORE INTO host_salary_policy_cycle_bindings ( + app_code, region_id, cycle_key, policy_id, policy_version, created_at_ms + ) + SELECT policy.app_code, policy.region_id, policy.cycle_key, policy.policy_id, policy.policy_version, policy.created_at_ms + FROM host_agency_salary_policies policy + WHERE policy.cycle_key <> '' + AND policy.status = 'active' + AND NOT EXISTS ( + SELECT 1 + FROM host_agency_salary_policies newer + WHERE newer.app_code = policy.app_code + AND newer.region_id = policy.region_id + AND newer.cycle_key = policy.cycle_key + AND newer.status = 'active' + AND ( + newer.effective_from_ms > policy.effective_from_ms + OR (newer.effective_from_ms = policy.effective_from_ms AND newer.policy_id > policy.policy_id) + ) + )`); err != nil { + return err + } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_agency_salary_policies + ADD INDEX idx_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status, settlement_mode, settlement_trigger_mode)`); err != nil && !isDuplicateKeyNameError(err) { + return err + } if _, err := db.ExecContext(ctx, ` ALTER TABLE host_salary_settlement_records @@ -62,6 +128,16 @@ func ensureHostSalarySettlementSchema(ctx context.Context, db *sql.DB) error { ADD COLUMN settlement_role VARCHAR(24) NOT NULL DEFAULT 'all' COMMENT '结算身份：all/host/agency' AFTER trigger_mode`); err != nil && !isDuplicateColumnError(err) { return err } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_progress + ADD COLUMN last_policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '最后一次使用的不可变政策版本' AFTER last_policy_id`); err != nil && !isDuplicateColumnError(err) { + return err + } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_salary_settlement_records + ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '使用的不可变政策版本' AFTER policy_id`); err != nil && !isDuplicateColumnError(err) { + return err + } return nil } diff --git a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go index d1e4965d..02aa91dc 100644 --- a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go +++ b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go @@ -915,6 +915,16 @@ func (r *Repository) SetHostSalaryPolicy(policy ledger.HostSalaryPolicy) { if policy.PolicyID == 0 { policy.PolicyID = 1 } + if strings.TrimSpace(policy.CycleKey) == "" { + if policy.EffectiveFromMs > 0 { + policy.CycleKey = time.UnixMilli(policy.EffectiveFromMs).UTC().Format("2006-01") + } else { + policy.CycleKey = time.Now().UTC().Format("2006-01") + } + } + if policy.PolicyVersion == 0 { + policy.PolicyVersion = policy.PolicyID + } if policy.Status == "" { policy.Status = "active" } @@ -933,10 +943,12 @@ func (r *Repository) SetHostSalaryPolicy(policy ledger.HostSalaryPolicy) { } _, err := r.schema.DB.ExecContext(context.Background(), ` INSERT INTO host_agency_salary_policies ( - policy_id, name, region_id, status, settlement_mode, settlement_trigger_mode, gift_coin_to_diamond_ratio, + policy_id, cycle_key, policy_version, name, region_id, status, settlement_mode, settlement_trigger_mode, gift_coin_to_diamond_ratio, residual_diamond_to_usd_rate, effective_from_ms, effective_to_ms, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE + cycle_key = VALUES(cycle_key), + policy_version = VALUES(policy_version), name = VALUES(name), region_id = VALUES(region_id), status = VALUES(status), @@ -947,7 +959,7 @@ func (r *Repository) SetHostSalaryPolicy(policy ledger.HostSalaryPolicy) { effective_from_ms = VALUES(effective_from_ms), effective_to_ms = VALUES(effective_to_ms), updated_at_ms = VALUES(updated_at_ms) - `, policy.PolicyID, policy.Name, policy.RegionID, policy.Status, policy.SettlementMode, policy.SettlementTriggerMode, policy.GiftCoinToDiamondRatio, policy.ResidualDiamondToUSDRate, policy.EffectiveFromMs, policy.EffectiveToMs, nowMs, nowMs) + `, policy.PolicyID, policy.CycleKey, policy.PolicyVersion, policy.Name, policy.RegionID, policy.Status, policy.SettlementMode, policy.SettlementTriggerMode, policy.GiftCoinToDiamondRatio, policy.ResidualDiamondToUSDRate, policy.EffectiveFromMs, policy.EffectiveToMs, nowMs, nowMs) if err != nil { r.t.Fatalf("seed host salary policy failed: %v", err) } @@ -969,6 +981,17 @@ func (r *Repository) SetHostSalaryPolicy(policy ledger.HostSalaryPolicy) { r.t.Fatalf("seed host salary policy level %d failed: %v", level.LevelNo, err) } } + // 测试 helper 同时建立周期绑定，模拟后台一次完整发布；生产发布路径不允许 ON DUPLICATE 覆盖。 + if _, err := r.schema.DB.ExecContext(context.Background(), ` + INSERT INTO host_salary_policy_cycle_bindings ( + region_id, cycle_key, policy_id, policy_version, created_at_ms + ) VALUES (?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + policy_id = VALUES(policy_id), policy_version = VALUES(policy_version)`, + policy.RegionID, policy.CycleKey, policy.PolicyID, policy.PolicyVersion, nowMs, + ); err != nil { + r.t.Fatalf("seed host salary policy cycle binding failed: %v", err) + } } func initDBPath(t testing.TB) string { diff --git a/services/wallet-service/internal/transport/grpc/host_salary.go b/services/wallet-service/internal/transport/grpc/host_salary.go index 50049a4d..adaef2a0 100644 --- a/services/wallet-service/internal/transport/grpc/host_salary.go +++ b/services/wallet-service/internal/transport/grpc/host_salary.go @@ -10,10 +10,10 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) -// GetActiveHostSalaryPolicy 读取当前生效的主播工资政策，供 gateway 按当前主播区域展示平台规则。 +// GetActiveHostSalaryPolicy 按周期读取主播工资政策，供 gateway 展示与该周期结算完全一致的规则。 func (s *Server) GetActiveHostSalaryPolicy(ctx context.Context, req *walletv1.GetActiveHostSalaryPolicyRequest) (*walletv1.GetActiveHostSalaryPolicyResponse, error) { ctx = appcode.WithContext(ctx, req.GetAppCode()) - policy, found, err := s.svc.GetActiveHostSalaryPolicy(ctx, req.GetAppCode(), req.GetRegionId(), req.GetSettlementMode(), req.GetSettlementTriggerMode(), req.GetNowMs()) + policy, found, err := s.svc.GetActiveHostSalaryPolicy(ctx, req.GetAppCode(), req.GetRegionId(), req.GetSettlementMode(), req.GetSettlementTriggerMode(), req.GetCycleKey(), req.GetNowMs()) if err != nil { return nil, xerr.ToGRPCError(err) } @@ -80,6 +80,8 @@ func hostSalaryPolicyToProto(policy ledger.HostSalaryPolicy) *walletv1.HostSalar } return &walletv1.HostSalaryPolicy{ PolicyId: policy.PolicyID, + CycleKey: policy.CycleKey, + PolicyVersion: policy.PolicyVersion, Name: policy.Name, RegionId: policy.RegionID, Status: policy.Status, diff --git a/tools/databi-real-flow/main.go b/tools/databi-real-flow/main.go index aa1a4fb3..20d393f3 100644 --- a/tools/databi-real-flow/main.go +++ b/tools/databi-real-flow/main.go @@ -217,20 +217,22 @@ func (r *runner) ensureUserMasterData(ctx context.Context) error { } func (r *runner) ensureSalaryConfig(ctx context.Context) error { - now := time.Now().UnixMilli() + nowUTC := time.Now().UTC() + now := nowUTC.UnixMilli() + cycleKey := nowUTC.Format("2006-01") + policyID := uint64(990000000000 + nowUTC.Year()*100 + int(nowUTC.Month())) _, err := r.walletDB.ExecContext(ctx, ` INSERT INTO host_agency_salary_policies ( - app_code, policy_id, name, region_id, status, settlement_mode, settlement_trigger_mode, + app_code, policy_id, cycle_key, policy_version, name, region_id, status, settlement_mode, settlement_trigger_mode, gift_coin_to_diamond_ratio, residual_diamond_to_usd_rate, effective_from_ms, effective_to_ms, created_at_ms, updated_at_ms - ) VALUES (?, 990001, 'Databi Local Daily Policy', ?, 'active', 'daily', 'automatic', 1.000000, 0.000000000000, 0, 0, ?, ?) + ) VALUES (?, ?, ?, ?, 'Databi Local Daily Policy', ?, 'active', 'daily', 'automatic', 1.000000, 0.000000000000, 0, 0, ?, ?) ON DUPLICATE KEY UPDATE - region_id = VALUES(region_id), status = 'active', settlement_mode = 'daily', settlement_trigger_mode = 'automatic', gift_coin_to_diamond_ratio = VALUES(gift_coin_to_diamond_ratio), updated_at_ms = VALUES(updated_at_ms)`, - r.cfg.appCode, regionID, now, now) + r.cfg.appCode, policyID, cycleKey, policyID, regionID, now, now) if err != nil { return fmt.Errorf("seed salary policy: %w", err) } @@ -238,17 +240,26 @@ func (r *runner) ensureSalaryConfig(ctx context.Context) error { INSERT INTO host_agency_salary_policy_levels ( app_code, policy_id, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order, created_at_ms, updated_at_ms - ) VALUES (?, 990001, 1, 100, 150, 0, 50, 'active', 1, ?, ?) + ) VALUES (?, ?, 1, 100, 150, 0, 50, 'active', 1, ?, ?) ON DUPLICATE KEY UPDATE required_diamonds = VALUES(required_diamonds), host_salary_usd_minor = VALUES(host_salary_usd_minor), agency_salary_usd_minor = VALUES(agency_salary_usd_minor), status = 'active', updated_at_ms = VALUES(updated_at_ms)`, - r.cfg.appCode, now, now) + r.cfg.appCode, policyID, now, now) if err != nil { return fmt.Errorf("seed salary policy level: %w", err) } + _, err = r.walletDB.ExecContext(ctx, ` + INSERT INTO host_salary_policy_cycle_bindings ( + app_code, region_id, cycle_key, policy_id, policy_version, created_at_ms + ) VALUES (?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE policy_id = VALUES(policy_id), policy_version = VALUES(policy_version)`, + r.cfg.appCode, regionID, cycleKey, policyID, policyID, now) + if err != nil { + return fmt.Errorf("seed salary policy cycle binding: %w", err) + } _, err = r.walletDB.ExecContext(ctx, `DELETE FROM coin_seller_salary_exchange_rate_tiers WHERE app_code = ? AND region_id = ? AND sort_order = 990001`, r.cfg.appCode, regionID)