政策周期修正

This commit is contained in:
zhx 2026-07-19 23:21:31 +08:00
parent 764caa2193
commit d27fba74bc
42 changed files with 1191 additions and 177 deletions

View File

@ -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,6 +1520,8 @@ 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"`
// 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
}
@ -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" +

View File

@ -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 {

View File

@ -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

View File

@ -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)

View File

@ -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"

View File

@ -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"

View File

@ -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 == "" {

View File

@ -422,10 +422,13 @@ 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"`
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"`
@ -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"`

View File

@ -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)
}

View File

@ -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,

View File

@ -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) {

View File

@ -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 可直接筛出“只允许手动结算”的政策,后续工资结算页会复用同一查询口径。

View File

@ -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"`

View File

@ -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),

View File

@ -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 ""

View File

@ -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")
}
}

View File

@ -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"`

View File

@ -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,

View File

@ -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
}

View File

@ -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)
}

View File

@ -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},

View File

@ -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;

View File

@ -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;

View File

@ -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,

View File

@ -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),

View File

@ -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)
}
}

View File

@ -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)

View File

@ -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 或 triggergateway 用当前主播身份区域读取 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(),

View File

@ -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='房间背景图素材表';

View File

@ -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'

View File

@ -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',

View File

@ -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;

View File

@ -29,6 +29,9 @@ type HostSalarySettlementBatchResult struct {
// HostSalaryPolicy 是 wallet-service 结算读取的后台政策运行快照。
type HostSalaryPolicy struct {
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
}

View File

@ -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 收款人名下主播的预收入工资,经理中心团队工资卡片使用。

View File

@ -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)

View File

@ -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",

View File

@ -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,

View File

@ -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
}

View File

@ -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
}

View File

@ -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 {

View File

@ -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,

View File

@ -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)