830 lines
29 KiB
Go
830 lines
29 KiB
Go
package policyconfig
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"math"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/appctx"
|
||
"hyapp-admin-server/internal/integration/activityclient"
|
||
activityv1 "hyapp.local/api/proto/activity/v1"
|
||
)
|
||
|
||
const (
|
||
policyStatusActive = "active"
|
||
policyStatusDisabled = "disabled"
|
||
|
||
publishStatusDraft = "draft"
|
||
publishStatusPublished = "published"
|
||
publishStatusFailed = "failed"
|
||
|
||
policyTaskRewardAssetCoin = "COIN"
|
||
|
||
defaultPolicyTemplateCode = "first_google70000_coin_seller_92000_100000_v1"
|
||
defaultPolicyTemplateVersion = "v1"
|
||
)
|
||
|
||
type Service struct {
|
||
adminDB *sql.DB
|
||
walletDB *sql.DB
|
||
userDB *sql.DB
|
||
activity activityclient.Client
|
||
}
|
||
|
||
type listTemplatesQuery struct {
|
||
Keyword string
|
||
Status string
|
||
Page int
|
||
PageSize int
|
||
}
|
||
|
||
type listInstancesQuery struct {
|
||
Keyword string
|
||
Status string
|
||
Page int
|
||
PageSize int
|
||
}
|
||
|
||
type policyInstanceRow struct {
|
||
InstanceID uint64
|
||
AppCode string
|
||
InstanceCode string
|
||
TemplateCode string
|
||
TemplateVersion string
|
||
RegionScope string
|
||
Status string
|
||
EffectiveFromMS int64
|
||
EffectiveToMS int64
|
||
}
|
||
|
||
type compiledPolicy struct {
|
||
HostPointRatioPPM int64
|
||
PointsPerUSD int64
|
||
WithdrawFeeBPS int64
|
||
TaskRewardAssetType string
|
||
RuleJSON json.RawMessage
|
||
}
|
||
|
||
func NewService(adminDB *sql.DB, walletDB *sql.DB, userDB *sql.DB, activity activityclient.Client) *Service {
|
||
return &Service{adminDB: adminDB, walletDB: walletDB, userDB: userDB, activity: activity}
|
||
}
|
||
|
||
func (s *Service) ListTemplates(ctx context.Context, query listTemplatesQuery) ([]templateDTO, int64, error) {
|
||
if s == nil || s.adminDB == nil {
|
||
return nil, 0, errors.New("admin database is not configured")
|
||
}
|
||
where, args := templateWhere(query)
|
||
var total int64
|
||
if err := s.adminDB.QueryRowContext(ctx, `SELECT COUNT(*) FROM admin_policy_templates t JOIN admin_policy_template_versions v ON v.template_code = t.template_code AND v.template_version = t.template_version `+where, args...).Scan(&total); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
page, pageSize := normalizePage(query.Page, query.PageSize)
|
||
args = append(args, pageSize, (page-1)*pageSize)
|
||
rows, err := s.adminDB.QueryContext(ctx, `
|
||
SELECT t.template_id, v.version_id, t.template_code, v.template_version, t.name, v.status,
|
||
COALESCE(CAST(v.rule_json AS CHAR), '{}'), t.description, v.created_at_ms, v.updated_at_ms
|
||
FROM admin_policy_templates t
|
||
JOIN admin_policy_template_versions v ON v.template_code = t.template_code AND v.template_version = t.template_version `+where+`
|
||
ORDER BY t.updated_at_ms DESC, v.updated_at_ms DESC
|
||
LIMIT ? OFFSET ?`, args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
items := make([]templateDTO, 0)
|
||
for rows.Next() {
|
||
item, err := scanTemplate(rows)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, total, rows.Err()
|
||
}
|
||
|
||
func (s *Service) CreateTemplate(ctx context.Context, actorID uint, req templateRequest) (templateDTO, error) {
|
||
if s == nil || s.adminDB == nil {
|
||
return templateDTO{}, errors.New("admin database is not configured")
|
||
}
|
||
templateCode := normalizeCode(req.TemplateCode)
|
||
version := normalizeVersion(req.TemplateVersion)
|
||
name := strings.TrimSpace(req.Name)
|
||
status := normalizeStatus(req.Status)
|
||
if templateCode == "" || version == "" || name == "" {
|
||
return templateDTO{}, errors.New("template_code, template_version and name are required")
|
||
}
|
||
ruleJSON, err := normalizeRuleJSON(req.RuleJSON)
|
||
if err != nil {
|
||
return templateDTO{}, err
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
tx, err := s.adminDB.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return templateDTO{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_templates (
|
||
template_code, template_version, name, status, rule_json, description,
|
||
created_by_admin_id, updated_by_admin_id, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
name = VALUES(name), status = VALUES(status), rule_json = VALUES(rule_json),
|
||
description = VALUES(description), updated_by_admin_id = VALUES(updated_by_admin_id), updated_at_ms = VALUES(updated_at_ms)`,
|
||
templateCode, version, name, status, string(ruleJSON), strings.TrimSpace(req.Description), actorID, actorID, nowMS, nowMS,
|
||
); err != nil {
|
||
return templateDTO{}, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_template_versions (
|
||
template_code, template_version, status, rule_json,
|
||
created_by_admin_id, updated_by_admin_id, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
status = VALUES(status), rule_json = VALUES(rule_json),
|
||
updated_by_admin_id = VALUES(updated_by_admin_id), updated_at_ms = VALUES(updated_at_ms)`,
|
||
templateCode, version, status, string(ruleJSON), actorID, actorID, nowMS, nowMS,
|
||
); err != nil {
|
||
return templateDTO{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return templateDTO{}, err
|
||
}
|
||
return s.getTemplate(ctx, templateCode, version)
|
||
}
|
||
|
||
func (s *Service) ListInstances(ctx context.Context, appCode string, query listInstancesQuery) ([]instanceDTO, int64, error) {
|
||
if s == nil || s.adminDB == nil {
|
||
return nil, 0, errors.New("admin database is not configured")
|
||
}
|
||
appCode = appctx.Normalize(appCode)
|
||
where, args := instanceWhere(appCode, query)
|
||
var total int64
|
||
if err := s.adminDB.QueryRowContext(ctx, `SELECT COUNT(*) FROM admin_policy_instances `+where, args...).Scan(&total); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
page, pageSize := normalizePage(query.Page, query.PageSize)
|
||
args = append(args, pageSize, (page-1)*pageSize)
|
||
rows, err := s.adminDB.QueryContext(ctx, `
|
||
SELECT instance_id, app_code, instance_code, template_code, template_version, region_scope, status,
|
||
effective_from_ms, effective_to_ms, publish_status, publish_error, last_published_at_ms,
|
||
created_at_ms, updated_at_ms
|
||
FROM admin_policy_instances `+where+`
|
||
ORDER BY updated_at_ms DESC, instance_id DESC
|
||
LIMIT ? OFFSET ?`, args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
items := make([]instanceDTO, 0)
|
||
for rows.Next() {
|
||
item, err := scanInstanceDTO(rows)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, total, rows.Err()
|
||
}
|
||
|
||
func (s *Service) CreateInstance(ctx context.Context, appCode string, actorID uint, req instanceRequest) (instanceDTO, error) {
|
||
if s == nil || s.adminDB == nil {
|
||
return instanceDTO{}, errors.New("admin database is not configured")
|
||
}
|
||
appCode = appctx.Normalize(appCode)
|
||
instanceCode := normalizeCode(req.InstanceCode)
|
||
templateCode := normalizeCode(req.TemplateCode)
|
||
version := normalizeVersion(req.TemplateVersion)
|
||
if instanceCode == "" || templateCode == "" || version == "" {
|
||
return instanceDTO{}, errors.New("instance_code, template_code and template_version are required")
|
||
}
|
||
if _, err := s.getTemplate(ctx, templateCode, version); err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
status := normalizeStatus(req.Status)
|
||
regionScope := normalizeRegionScope(req.RegionScope)
|
||
if err := validateEffectiveRange(req.EffectiveFromMS, req.EffectiveToMS); err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
effectiveFromMS := req.EffectiveFromMS
|
||
if effectiveFromMS <= 0 {
|
||
effectiveFromMS = nowMS
|
||
}
|
||
result, err := s.adminDB.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_instances (
|
||
app_code, instance_code, template_code, template_version, region_scope, status,
|
||
effective_from_ms, effective_to_ms, publish_status, created_by_admin_id, updated_by_admin_id,
|
||
created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode, instanceCode, templateCode, version, regionScope, status,
|
||
effectiveFromMS, req.EffectiveToMS, publishStatusDraft, actorID, actorID, nowMS, nowMS,
|
||
)
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
id, err := result.LastInsertId()
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
return s.getInstanceDTO(ctx, appCode, uint64(id))
|
||
}
|
||
|
||
func (s *Service) UpdateInstanceStatus(ctx context.Context, appCode string, actorID uint, instanceID uint64, status string) (instanceDTO, error) {
|
||
if s == nil || s.adminDB == nil {
|
||
return instanceDTO{}, errors.New("admin database is not configured")
|
||
}
|
||
appCode = appctx.Normalize(appCode)
|
||
status = normalizeStatus(status)
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
instance, err := s.getInstanceRow(ctx, appCode, instanceID)
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
ruleJSON, err := s.getTemplateRuleJSONSnapshot(ctx, instance.TemplateCode, instance.TemplateVersion)
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
compiled, err := compilePolicy(ruleJSON)
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
result, err := s.adminDB.ExecContext(ctx, `
|
||
UPDATE admin_policy_instances
|
||
SET status = ?, updated_by_admin_id = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND instance_id = ?`,
|
||
status, actorID, nowMS, appCode, instanceID,
|
||
)
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
affected, err := result.RowsAffected()
|
||
if err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
if affected == 0 {
|
||
return instanceDTO{}, errors.New("policy instance not found")
|
||
}
|
||
instance.Status = status
|
||
if err := s.syncWalletRuntimeStatus(ctx, instance, status, nowMS); err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
if err := s.publishActivityRuntime(ctx, instance, compiled, actorID, nowMS); err != nil {
|
||
return instanceDTO{}, err
|
||
}
|
||
return s.getInstanceDTO(ctx, appCode, instanceID)
|
||
}
|
||
|
||
func (s *Service) PublishInstance(ctx context.Context, appCode string, actorID uint, instanceID uint64) (publishDTO, error) {
|
||
if s == nil || s.adminDB == nil || s.walletDB == nil || s.userDB == nil || s.activity == nil {
|
||
return publishDTO{}, errors.New("policy publish database is not configured")
|
||
}
|
||
appCode = appctx.Normalize(appCode)
|
||
instance, err := s.getInstanceRow(ctx, appCode, instanceID)
|
||
if err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
if instance.Status != policyStatusActive {
|
||
return publishDTO{}, errors.New("only active policy instance can be published")
|
||
}
|
||
ruleJSON, err := s.getTemplateRuleJSON(ctx, instance.TemplateCode, instance.TemplateVersion)
|
||
if err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
compiled, err := compilePolicy(ruleJSON)
|
||
if err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
regionIDs, err := s.resolveRegionScope(ctx, instance.AppCode, instance.RegionScope)
|
||
if err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
targetCount := len(regionIDs) + 1
|
||
jobID, err := s.createPublishJob(ctx, instance, actorID, targetCount, nowMS)
|
||
if err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
if err := s.publishWalletRuntime(ctx, instance, compiled, regionIDs, actorID, nowMS); err != nil {
|
||
_ = s.finishPublishJob(ctx, jobID, publishStatusFailed, targetCount, 0, targetCount, err.Error(), nowMS)
|
||
_ = s.markInstancePublishFailed(ctx, instance.AppCode, instance.InstanceID, err.Error(), nowMS)
|
||
return publishDTO{}, err
|
||
}
|
||
if err := s.publishActivityRuntime(ctx, instance, compiled, actorID, nowMS); err != nil {
|
||
_ = s.finishPublishJob(ctx, jobID, publishStatusFailed, targetCount, len(regionIDs), 1, err.Error(), nowMS)
|
||
_ = s.markInstancePublishFailed(ctx, instance.AppCode, instance.InstanceID, err.Error(), nowMS)
|
||
return publishDTO{}, err
|
||
}
|
||
if err := s.insertPublishItems(ctx, jobID, instance.AppCode, regionIDs, true, nowMS); err != nil {
|
||
_ = s.finishPublishJob(ctx, jobID, publishStatusFailed, targetCount, targetCount, 0, err.Error(), nowMS)
|
||
_ = s.markInstancePublishFailed(ctx, instance.AppCode, instance.InstanceID, err.Error(), nowMS)
|
||
return publishDTO{}, err
|
||
}
|
||
if err := s.finishPublishJob(ctx, jobID, "succeeded", targetCount, targetCount, 0, "", nowMS); err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
if _, err := s.adminDB.ExecContext(ctx, `
|
||
UPDATE admin_policy_instances
|
||
SET publish_status = ?, publish_error = '', last_published_at_ms = ?, updated_by_admin_id = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND instance_id = ?`,
|
||
publishStatusPublished, nowMS, actorID, nowMS, instance.AppCode, instance.InstanceID,
|
||
); err != nil {
|
||
return publishDTO{}, err
|
||
}
|
||
return publishDTO{
|
||
JobID: jobID,
|
||
InstanceID: instance.InstanceID,
|
||
AppCode: instance.AppCode,
|
||
InstanceCode: instance.InstanceCode,
|
||
TemplateCode: instance.TemplateCode,
|
||
TemplateVersion: instance.TemplateVersion,
|
||
Status: publishStatusPublished,
|
||
TargetCount: targetCount,
|
||
PublishedRegionID: regionIDs,
|
||
PublishedAtMS: nowMS,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) getTemplate(ctx context.Context, templateCode string, version string) (templateDTO, error) {
|
||
row := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT t.template_id, v.version_id, t.template_code, v.template_version, t.name, v.status,
|
||
COALESCE(CAST(v.rule_json AS CHAR), '{}'), t.description, v.created_at_ms, v.updated_at_ms
|
||
FROM admin_policy_templates t
|
||
JOIN admin_policy_template_versions v ON v.template_code = t.template_code AND v.template_version = t.template_version
|
||
WHERE t.template_code = ? AND v.template_version = ?`,
|
||
templateCode, version,
|
||
)
|
||
item, err := scanTemplate(row)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return templateDTO{}, errors.New("policy template version not found")
|
||
}
|
||
return item, err
|
||
}
|
||
|
||
func (s *Service) getTemplateRuleJSON(ctx context.Context, templateCode string, version string) (json.RawMessage, error) {
|
||
var raw string
|
||
if err := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT COALESCE(CAST(rule_json AS CHAR), '{}')
|
||
FROM admin_policy_template_versions
|
||
WHERE template_code = ? AND template_version = ? AND status = ?`,
|
||
templateCode, version, policyStatusActive,
|
||
).Scan(&raw); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, errors.New("active policy template version not found")
|
||
}
|
||
return nil, err
|
||
}
|
||
return normalizeRuleJSON(json.RawMessage(raw))
|
||
}
|
||
|
||
func (s *Service) getTemplateRuleJSONSnapshot(ctx context.Context, templateCode string, version string) (json.RawMessage, error) {
|
||
var raw string
|
||
if err := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT COALESCE(CAST(rule_json AS CHAR), '{}')
|
||
FROM admin_policy_template_versions
|
||
WHERE template_code = ? AND template_version = ?`,
|
||
templateCode, version,
|
||
).Scan(&raw); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, errors.New("policy template version not found")
|
||
}
|
||
return nil, err
|
||
}
|
||
return normalizeRuleJSON(json.RawMessage(raw))
|
||
}
|
||
|
||
func (s *Service) getInstanceDTO(ctx context.Context, appCode string, instanceID uint64) (instanceDTO, error) {
|
||
row := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT instance_id, app_code, instance_code, template_code, template_version, region_scope, status,
|
||
effective_from_ms, effective_to_ms, publish_status, publish_error, last_published_at_ms,
|
||
created_at_ms, updated_at_ms
|
||
FROM admin_policy_instances
|
||
WHERE app_code = ? AND instance_id = ?`,
|
||
appCode, instanceID,
|
||
)
|
||
item, err := scanInstanceDTO(row)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return instanceDTO{}, errors.New("policy instance not found")
|
||
}
|
||
return item, err
|
||
}
|
||
|
||
func (s *Service) getInstanceRow(ctx context.Context, appCode string, instanceID uint64) (policyInstanceRow, error) {
|
||
row := s.adminDB.QueryRowContext(ctx, `
|
||
SELECT instance_id, app_code, instance_code, template_code, template_version, region_scope, status,
|
||
effective_from_ms, effective_to_ms
|
||
FROM admin_policy_instances
|
||
WHERE app_code = ? AND instance_id = ?`,
|
||
appCode, instanceID,
|
||
)
|
||
var item policyInstanceRow
|
||
err := row.Scan(&item.InstanceID, &item.AppCode, &item.InstanceCode, &item.TemplateCode, &item.TemplateVersion, &item.RegionScope, &item.Status, &item.EffectiveFromMS, &item.EffectiveToMS)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return policyInstanceRow{}, errors.New("policy instance not found")
|
||
}
|
||
return item, err
|
||
}
|
||
|
||
func (s *Service) resolveRegionScope(ctx context.Context, appCode string, scope string) ([]int64, error) {
|
||
scope = normalizeRegionScope(scope)
|
||
if strings.HasPrefix(scope, "region_id:") {
|
||
value := strings.TrimPrefix(scope, "region_id:")
|
||
regionID, err := strconv.ParseInt(value, 10, 64)
|
||
if err != nil || regionID <= 0 {
|
||
return nil, errors.New("region_scope is invalid")
|
||
}
|
||
return []int64{regionID}, nil
|
||
}
|
||
rows, err := s.userDB.QueryContext(ctx, `
|
||
SELECT region_id
|
||
FROM regions
|
||
WHERE app_code = ? AND status = 'active'
|
||
ORDER BY sort_order ASC, region_id ASC`, appCode)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
regionIDs := []int64{0}
|
||
for rows.Next() {
|
||
var regionID int64
|
||
if err := rows.Scan(®ionID); err != nil {
|
||
return nil, err
|
||
}
|
||
if regionID > 0 {
|
||
regionIDs = append(regionIDs, regionID)
|
||
}
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return regionIDs, nil
|
||
}
|
||
|
||
func (s *Service) createPublishJob(ctx context.Context, item policyInstanceRow, actorID uint, targetCount int, nowMS int64) (uint64, error) {
|
||
result, err := s.adminDB.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_publish_jobs (
|
||
instance_id, app_code, instance_code, template_code, template_version, status,
|
||
target_count, operator_admin_id, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?)`,
|
||
item.InstanceID, item.AppCode, item.InstanceCode, item.TemplateCode, item.TemplateVersion, targetCount, actorID, nowMS, nowMS,
|
||
)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
id, err := result.LastInsertId()
|
||
return uint64(id), err
|
||
}
|
||
|
||
func (s *Service) publishWalletRuntime(ctx context.Context, item policyInstanceRow, compiled compiledPolicy, regionIDs []int64, actorID uint, nowMS int64) error {
|
||
if err := ensureWalletPolicyRuntimeTable(ctx, s.walletDB); err != nil {
|
||
return err
|
||
}
|
||
tx, err := s.walletDB.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
if _, err := tx.ExecContext(ctx, `DELETE FROM wallet_policy_instances WHERE app_code = ? AND instance_code = ?`, item.AppCode, item.InstanceCode); err != nil {
|
||
return err
|
||
}
|
||
for _, regionID := range regionIDs {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wallet_policy_instances (
|
||
app_code, instance_code, template_code, template_version, region_id, status,
|
||
effective_from_ms, effective_to_ms, host_point_asset_type, host_point_ratio_ppm,
|
||
points_per_usd, withdraw_fee_bps, rule_json, published_by_admin_id,
|
||
published_at_ms, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'POINT', ?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?)`,
|
||
item.AppCode, item.InstanceCode, item.TemplateCode, item.TemplateVersion, regionID, item.Status,
|
||
item.EffectiveFromMS, item.EffectiveToMS, compiled.HostPointRatioPPM, compiled.PointsPerUSD,
|
||
compiled.WithdrawFeeBPS, string(compiled.RuleJSON), actorID, nowMS, nowMS, nowMS,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (s *Service) syncWalletRuntimeStatus(ctx context.Context, item policyInstanceRow, status string, nowMS int64) error {
|
||
if s.walletDB == nil {
|
||
return nil
|
||
}
|
||
if err := ensureWalletPolicyRuntimeTable(ctx, s.walletDB); err != nil {
|
||
return err
|
||
}
|
||
// 运行表是 owner service 的最终读取面；后台停用实例必须同步停用旧快照，不能只改 admin 状态。
|
||
_, err := s.walletDB.ExecContext(ctx, `
|
||
UPDATE wallet_policy_instances
|
||
SET status = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND instance_code = ?`,
|
||
status, nowMS, item.AppCode, item.InstanceCode,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func (s *Service) publishActivityRuntime(ctx context.Context, item policyInstanceRow, compiled compiledPolicy, actorID uint, nowMS int64) error {
|
||
if s.activity == nil {
|
||
return nil
|
||
}
|
||
// activity 目前只需要 App 级任务奖励默认资产；具体任务定义仍由 activity-service 自己持久化和快照。
|
||
_, err := s.activity.PublishTaskRewardPolicy(ctx, &activityv1.PublishTaskRewardPolicyRequest{
|
||
Meta: &activityv1.RequestMeta{
|
||
Caller: "admin-server",
|
||
AppCode: item.AppCode,
|
||
SentAtMs: nowMS,
|
||
},
|
||
InstanceCode: item.InstanceCode,
|
||
TemplateCode: item.TemplateCode,
|
||
TemplateVersion: item.TemplateVersion,
|
||
Status: item.Status,
|
||
RewardAssetType: compiled.TaskRewardAssetType,
|
||
RuleJson: string(compiled.RuleJSON),
|
||
OperatorAdminId: int64(actorID),
|
||
PublishedAtMs: nowMS,
|
||
})
|
||
return err
|
||
}
|
||
|
||
func ensureWalletPolicyRuntimeTable(ctx context.Context, db *sql.DB) error {
|
||
_, err := db.ExecContext(ctx, `
|
||
CREATE TABLE IF NOT EXISTS wallet_policy_instances (
|
||
app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
|
||
instance_code VARCHAR(96) NOT NULL COMMENT '后台策略实例编码',
|
||
template_code VARCHAR(96) NOT NULL COMMENT '模板编码快照',
|
||
template_version VARCHAR(64) NOT NULL COMMENT '模板版本快照',
|
||
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '适用地区，0 表示 App 兜底',
|
||
status VARCHAR(24) NOT NULL DEFAULT 'active' COMMENT 'active/disabled',
|
||
effective_from_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'UTC epoch ms',
|
||
effective_to_ms BIGINT NOT NULL DEFAULT 0 COMMENT '0 表示长期有效',
|
||
host_point_asset_type VARCHAR(32) NOT NULL DEFAULT 'POINT' COMMENT '主播收益积分资产',
|
||
host_point_ratio_ppm BIGINT NOT NULL DEFAULT 700000 COMMENT '有效付费礼物转 POINT 比例，ppm',
|
||
points_per_usd BIGINT NOT NULL DEFAULT 100000 COMMENT 'POINT/USD 展示换算比例',
|
||
withdraw_fee_bps INT NOT NULL DEFAULT 500 COMMENT '提现手续费 bps',
|
||
rule_json JSON NOT NULL COMMENT '完整政策快照',
|
||
published_by_admin_id BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
published_at_ms BIGINT NOT NULL DEFAULT 0,
|
||
created_at_ms BIGINT NOT NULL,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, instance_code, region_id),
|
||
KEY idx_wallet_policy_active (app_code, region_id, status, effective_from_ms, effective_to_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='钱包运行侧收益政策实例表'`)
|
||
return err
|
||
}
|
||
|
||
func (s *Service) insertPublishItems(ctx context.Context, jobID uint64, appCode string, regionIDs []int64, activitySucceeded bool, nowMS int64) error {
|
||
tx, err := s.adminDB.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
for _, regionID := range regionIDs {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_publish_items (
|
||
job_id, owner_service, app_code, region_id, status, created_at_ms, updated_at_ms
|
||
) VALUES (?, 'wallet', ?, ?, 'succeeded', ?, ?)`,
|
||
jobID, appCode, regionID, nowMS, nowMS,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
activityStatus := "failed"
|
||
if activitySucceeded {
|
||
activityStatus = "succeeded"
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO admin_policy_publish_items (
|
||
job_id, owner_service, app_code, region_id, status, created_at_ms, updated_at_ms
|
||
) VALUES (?, 'activity', ?, 0, ?, ?, ?)`,
|
||
jobID, appCode, activityStatus, nowMS, nowMS,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (s *Service) finishPublishJob(ctx context.Context, jobID uint64, status string, targetCount int, successCount int, failureCount int, errMsg string, nowMS int64) error {
|
||
_, err := s.adminDB.ExecContext(ctx, `
|
||
UPDATE admin_policy_publish_jobs
|
||
SET status = ?, target_count = ?, success_count = ?, failure_count = ?, error_message = ?, updated_at_ms = ?
|
||
WHERE job_id = ?`,
|
||
status, targetCount, successCount, failureCount, truncate(errMsg, 512), nowMS, jobID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func (s *Service) markInstancePublishFailed(ctx context.Context, appCode string, instanceID uint64, errMsg string, nowMS int64) error {
|
||
_, err := s.adminDB.ExecContext(ctx, `
|
||
UPDATE admin_policy_instances
|
||
SET publish_status = ?, publish_error = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND instance_id = ?`,
|
||
publishStatusFailed, truncate(errMsg, 512), nowMS, appCode, instanceID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func compilePolicy(raw json.RawMessage) (compiledPolicy, error) {
|
||
var decoded map[string]any
|
||
if err := json.Unmarshal(raw, &decoded); err != nil {
|
||
return compiledPolicy{}, errors.New("rule_json is invalid")
|
||
}
|
||
pointsPerUSD := int64FromJSON(decoded["points_per_usd"], 100000)
|
||
host, _ := decoded["host"].(map[string]any)
|
||
hostRatioPercent := floatFromJSON(host["point_ratio_percent"], 70)
|
||
withdrawFeeBPS := int64FromJSON(host["withdraw_fee_bps"], 500)
|
||
ratioPPM := int64(math.Round(hostRatioPercent * 10000))
|
||
tasks, _ := decoded["tasks"].(map[string]any)
|
||
taskRewardAssetType := strings.ToUpper(strings.TrimSpace(stringFromJSON(tasks["reward_asset_type"], policyTaskRewardAssetCoin)))
|
||
if pointsPerUSD <= 0 || ratioPPM <= 0 || ratioPPM > 1000000 || withdrawFeeBPS < 0 || withdrawFeeBPS > 10000 {
|
||
return compiledPolicy{}, errors.New("rule_json host point policy is invalid")
|
||
}
|
||
if taskRewardAssetType != policyTaskRewardAssetCoin && taskRewardAssetType != "POINT" {
|
||
return compiledPolicy{}, errors.New("rule_json task reward asset_type is invalid")
|
||
}
|
||
return compiledPolicy{
|
||
HostPointRatioPPM: ratioPPM,
|
||
PointsPerUSD: pointsPerUSD,
|
||
WithdrawFeeBPS: withdrawFeeBPS,
|
||
TaskRewardAssetType: taskRewardAssetType,
|
||
RuleJSON: raw,
|
||
}, nil
|
||
}
|
||
|
||
func scanTemplate(row interface{ Scan(dest ...any) error }) (templateDTO, error) {
|
||
var item templateDTO
|
||
var rule string
|
||
err := row.Scan(&item.TemplateID, &item.VersionID, &item.TemplateCode, &item.TemplateVersion, &item.Name, &item.Status, &rule, &item.Description, &item.CreatedAtMS, &item.UpdatedAtMS)
|
||
item.RuleJSON = json.RawMessage(rule)
|
||
return item, err
|
||
}
|
||
|
||
func scanInstanceDTO(row interface{ Scan(dest ...any) error }) (instanceDTO, error) {
|
||
var item instanceDTO
|
||
err := row.Scan(&item.InstanceID, &item.AppCode, &item.InstanceCode, &item.TemplateCode, &item.TemplateVersion, &item.RegionScope, &item.Status, &item.EffectiveFromMS, &item.EffectiveToMS, &item.PublishStatus, &item.PublishError, &item.LastPublishedAtMS, &item.CreatedAtMS, &item.UpdatedAtMS)
|
||
return item, err
|
||
}
|
||
|
||
func templateWhere(query listTemplatesQuery) (string, []any) {
|
||
conditions := []string{"1 = 1"}
|
||
args := make([]any, 0)
|
||
if query.Status != "" {
|
||
conditions = append(conditions, "v.status = ?")
|
||
args = append(args, normalizeStatus(query.Status))
|
||
}
|
||
if query.Keyword != "" {
|
||
conditions = append(conditions, "(t.template_code LIKE ? OR t.name LIKE ?)")
|
||
like := "%" + query.Keyword + "%"
|
||
args = append(args, like, like)
|
||
}
|
||
return "WHERE " + strings.Join(conditions, " AND "), args
|
||
}
|
||
|
||
func instanceWhere(appCode string, query listInstancesQuery) (string, []any) {
|
||
conditions := []string{"app_code = ?"}
|
||
args := []any{appCode}
|
||
if query.Status != "" {
|
||
conditions = append(conditions, "status = ?")
|
||
args = append(args, normalizeStatus(query.Status))
|
||
}
|
||
if query.Keyword != "" {
|
||
conditions = append(conditions, "(instance_code LIKE ? OR template_code LIKE ?)")
|
||
like := "%" + query.Keyword + "%"
|
||
args = append(args, like, like)
|
||
}
|
||
return "WHERE " + strings.Join(conditions, " AND "), args
|
||
}
|
||
|
||
func normalizePage(page int, pageSize int) (int, int) {
|
||
if page <= 0 {
|
||
page = 1
|
||
}
|
||
if pageSize <= 0 {
|
||
pageSize = 10
|
||
}
|
||
if pageSize > 100 {
|
||
pageSize = 100
|
||
}
|
||
return page, pageSize
|
||
}
|
||
|
||
func normalizeCode(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
value = strings.ReplaceAll(value, " ", "_")
|
||
return value
|
||
}
|
||
|
||
func normalizeVersion(value string) string {
|
||
return strings.TrimSpace(value)
|
||
}
|
||
|
||
func normalizeStatus(value string) string {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case "", "enabled", policyStatusActive:
|
||
return policyStatusActive
|
||
case "inactive", policyStatusDisabled:
|
||
return policyStatusDisabled
|
||
default:
|
||
return strings.ToLower(strings.TrimSpace(value))
|
||
}
|
||
}
|
||
|
||
func normalizeRegionScope(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return "all_active_regions"
|
||
}
|
||
return value
|
||
}
|
||
|
||
func validateEffectiveRange(fromMS int64, toMS int64) error {
|
||
if fromMS < 0 || toMS < 0 {
|
||
return errors.New("effective time is invalid")
|
||
}
|
||
if toMS > 0 && fromMS > 0 && toMS <= fromMS {
|
||
return errors.New("effective_to_ms must be greater than effective_from_ms")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeRuleJSON(raw json.RawMessage) (json.RawMessage, error) {
|
||
if len(raw) == 0 {
|
||
return nil, errors.New("rule_json is required")
|
||
}
|
||
var decoded map[string]any
|
||
if err := json.Unmarshal(raw, &decoded); err != nil || decoded == nil {
|
||
return nil, errors.New("rule_json must be a JSON object")
|
||
}
|
||
normalized, err := json.Marshal(decoded)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return json.RawMessage(normalized), nil
|
||
}
|
||
|
||
func int64FromJSON(value any, fallback int64) int64 {
|
||
switch typed := value.(type) {
|
||
case float64:
|
||
if typed > 0 {
|
||
return int64(typed)
|
||
}
|
||
case json.Number:
|
||
if parsed, err := typed.Int64(); err == nil && parsed > 0 {
|
||
return parsed
|
||
}
|
||
case string:
|
||
if parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64); err == nil && parsed > 0 {
|
||
return parsed
|
||
}
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func floatFromJSON(value any, fallback float64) float64 {
|
||
switch typed := value.(type) {
|
||
case float64:
|
||
if typed > 0 {
|
||
return typed
|
||
}
|
||
case json.Number:
|
||
if parsed, err := typed.Float64(); err == nil && parsed > 0 {
|
||
return parsed
|
||
}
|
||
case string:
|
||
if parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64); err == nil && parsed > 0 {
|
||
return parsed
|
||
}
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func stringFromJSON(value any, fallback string) string {
|
||
switch typed := value.(type) {
|
||
case string:
|
||
if strings.TrimSpace(typed) != "" {
|
||
return typed
|
||
}
|
||
case json.Number:
|
||
if strings.TrimSpace(typed.String()) != "" {
|
||
return typed.String()
|
||
}
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func truncate(value string, limit int) string {
|
||
value = strings.TrimSpace(value)
|
||
if len(value) <= limit {
|
||
return value
|
||
}
|
||
return value[:limit]
|
||
}
|
||
|
||
func init() {
|
||
_ = defaultPolicyTemplateCode
|
||
_ = defaultPolicyTemplateVersion
|
||
}
|