diff --git a/backend/go.work.sum b/backend/go.work.sum index cb4c6bc..6854fd3 100644 --- a/backend/go.work.sum +++ b/backend/go.work.sum @@ -35,6 +35,8 @@ github.com/alibabacloud-go/darabonba-openapi/v2 v2.1.13 h1:Q00FU3H94Ts0ZIHDmY+fY github.com/alibabacloud-go/tea v1.2.1/go.mod h1:qbzof29bM/IFhLMtJPrgTGK3eauV5J2wSyEUo4OEmnA= github.com/alibabacloud-go/tea-utils/v2 v2.0.8/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I= github.com/alibabacloud-go/tea-xml v1.1.3/go.mod h1:Rq08vgCcCAjHyRi/M7xlHKUykZCEtyBy9+DPF6GgEu8= +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/antihax/optional v1.0.0 h1:xK2lYat7ZLaVVcIuj82J8kIro4V6kDe0AUDFboUCwcg= github.com/apache/thrift v0.13.0 h1:5hryIiq9gtn+MiLVn0wP37kb/uTeRZgN08WoCsAhIhI= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e h1:QEF07wC0T1rKkctt1RINW/+RMTVmiwxETico2l3gxJA= @@ -306,6 +308,8 @@ github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/X github.com/xhit/go-str2duration/v2 v2.1.0 h1:lxklc02Drh6ynqX+DdPyp5pCKLUQpRT8bp8Ydu2Bstc= github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= github.com/yuin/goldmark v1.4.13 h1:fVcFKWvrslecOb/tg+Cc05dkeYx540o0FuFt3nUVDoE= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M= go.opentelemetry.io/contrib/detectors/gcp v1.38.0 h1:ZoYbqX7OaA/TAikspPl3ozPI6iY6LiIY9I8cUfm+pJs= diff --git a/backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql b/backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql new file mode 100644 index 0000000..7231240 --- /dev/null +++ b/backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql @@ -0,0 +1,33 @@ +-- 2026_07_21_001_exhibition_revenue_idempotent.sql +-- 批次1.1+1.3：展品收益结算幂等 + created_at 单位统一 +-- 执行前请先备份： +-- pg_dump -h -U postgres -t exhibition_revenue_records -t exhibitions > backup_1_1.sql + +BEGIN; + +-- (1.3) created_at 秒→毫秒回填（10 位秒 → 13 位毫秒） +UPDATE exhibition_revenue_records +SET created_at = created_at * 1000 +WHERE created_at > 0 AND created_at < 100000000000; + +-- (1.1) 历史去重：同一 (exhibition_id, cycle_start_time) 只保留 id 最小的一条 +DELETE FROM exhibition_revenue_records a +USING exhibition_revenue_records b +WHERE a.exhibition_id = b.exhibition_id + AND a.cycle_start_time = b.cycle_start_time + AND a.id > b.id; + +-- (1.1) 唯一约束（幂等基石） +ALTER TABLE exhibition_revenue_records + ADD CONSTRAINT uk_exhibition_revenue_cycle UNIQUE (exhibition_id, cycle_start_time); + +-- (settled_at) 独立审计列，替代复用 is_processed +ALTER TABLE exhibitions ADD COLUMN IF NOT EXISTS settled_at bigint; +-- 回填：已被当作 settled 的历史行（is_processed=true）迁移到 settled_at +UPDATE exhibitions SET settled_at = COALESCE(updated_at, EXTRACT(EPOCH FROM now())*1000) +WHERE is_processed = true AND settled_at IS NULL; + +-- 序列同步（CLAUDE.md 强制） +SELECT setval('exhibition_revenue_records_id_seq', (SELECT COALESCE(MAX(id),1) FROM exhibition_revenue_records)); + +COMMIT; \ No newline at end of file diff --git a/backend/migrations/2026_07_21_002_crystal_tx_source_id_unique.sql b/backend/migrations/2026_07_21_002_crystal_tx_source_id_unique.sql new file mode 100644 index 0000000..0782a57 --- /dev/null +++ b/backend/migrations/2026_07_21_002_crystal_tx_source_id_unique.sql @@ -0,0 +1,37 @@ +-- 2026_07_21_002_crystal_tx_source_id_unique.sql +-- 批次1.4 幂等基石: crystal_transaction_records 同 (source_id, change_type) 只允许一条 +-- +-- 执行前请先备份: +-- pg_dump -h -U postgres -t crystal_transaction_records > backup_crystal_tx.sql +-- 或使用 psql: +-- psql -h -U postgres -d -c "\COPY crystal_transaction_records TO 'crystal_tx.csv' CSV HEADER" +-- +-- Dry-run (本地 top-fans 实测): +-- dup_to_blank = 37 +-- level_up_bonus 14, task_reward 13, like_bet_revenue 10 +-- 含义: 这 37 条 source_id 非空行是历史脏数据(同 source_id+change_type 已存在更早一条), +-- 本 migration 把它们 source_id 置空,保留更早的 min(id) 行。 + +BEGIN; + +-- (a) 历史去重: 同一 (source_id, change_type) 保留 id 最小的一条(其他 source_id 置空,不入新流水) +UPDATE crystal_transaction_records a +SET source_id = '' +FROM crystal_transaction_records b +WHERE a.source_id = b.source_id + AND a.change_type = b.change_type + AND a.source_id <> '' + AND a.id > b.id; + +-- (b) 部分唯一索引(source_id 非空时强制唯一) +CREATE UNIQUE INDEX IF NOT EXISTS uk_crystal_tx_source_change + ON crystal_transaction_records (source_id, change_type) + WHERE source_id <> ''; + +-- (c) 序列同步(本表 BIGSERIAL,手动删改后必须 setval) +SELECT setval( + pg_get_serial_sequence('crystal_transaction_records', 'id'), + (SELECT COALESCE(MAX(id), 1) FROM crystal_transaction_records) +); + +COMMIT; \ No newline at end of file diff --git a/backend/migrations/2026_07_21_003_exhibition_hours_idempotent.sql b/backend/migrations/2026_07_21_003_exhibition_hours_idempotent.sql new file mode 100644 index 0000000..738d163 --- /dev/null +++ b/backend/migrations/2026_07_21_003_exhibition_hours_idempotent.sql @@ -0,0 +1,45 @@ +-- 2026_07_21_003_exhibition_hours_idempotent.sql +-- 批次1.2：累计时长幂等（用户级 + 资产级） +-- 执行前请先备份： +-- pg_dump -h -U postgres -t exhibition_hours_log -t asset_exhibition_hours_log > backup_1_2.sql +-- 触发场景：fan_profile_repository.AddExhibitionHours/assetLevelService.AddExhibitionHours +-- 重复调用会造成 total_exhibition_hours / season_exhibition_hours 多次叠加 +-- 设计：每条幂等键由 source_id 唯一约束保证，INSERT ON CONFLICT DO NOTHING 即跳过 + +BEGIN; + +-- (1) 用户级幂等 log +CREATE TABLE IF NOT EXISTS exhibition_hours_log ( + id BIGINT PRIMARY KEY, + source_id VARCHAR(100) NOT NULL UNIQUE, + user_id BIGINT NOT NULL, + star_id BIGINT NOT NULL, + hours BIGINT NOT NULL, + created_at BIGINT NOT NULL +); +CREATE INDEX IF NOT EXISTS ix_exh_log_user_star + ON exhibition_hours_log (user_id, star_id); +CREATE INDEX IF NOT EXISTS ix_exh_log_created + ON exhibition_hours_log (created_at); + +-- (2) 资产级幂等 log +CREATE TABLE IF NOT EXISTS asset_exhibition_hours_log ( + id BIGINT PRIMARY KEY, + source_id VARCHAR(100) NOT NULL UNIQUE, + asset_id BIGINT NOT NULL, + hours INT NOT NULL, + created_at BIGINT NOT NULL +); +CREATE INDEX IF NOT EXISTS ix_asset_exh_log_asset + ON asset_exhibition_hours_log (asset_id); + +-- (3) 序列同步（CLAUDE.md 强制；新表无现存行，序列置 1） +-- 内部去重 log 表无脚本硬编码其 id，不适用 START 10000 约定；序列从 1 起步。 +CREATE SEQUENCE IF NOT EXISTS exhibition_hours_log_id_seq OWNED BY exhibition_hours_log.id; +ALTER TABLE exhibition_hours_log ALTER COLUMN id SET DEFAULT nextval('exhibition_hours_log_id_seq'); +SELECT setval('exhibition_hours_log_id_seq', GREATEST((SELECT COALESCE(MAX(id),0) FROM exhibition_hours_log), 1)); +CREATE SEQUENCE IF NOT EXISTS asset_exhibition_hours_log_id_seq OWNED BY asset_exhibition_hours_log.id; +ALTER TABLE asset_exhibition_hours_log ALTER COLUMN id SET DEFAULT nextval('asset_exhibition_hours_log_id_seq'); +SELECT setval('asset_exhibition_hours_log_id_seq', GREATEST((SELECT COALESCE(MAX(id),0) FROM asset_exhibition_hours_log), 1)); + +COMMIT; diff --git a/backend/pkg/models/asset_level.go b/backend/pkg/models/asset_level.go index 160e077..6c3275c 100644 --- a/backend/pkg/models/asset_level.go +++ b/backend/pkg/models/asset_level.go @@ -102,6 +102,17 @@ type SeasonDecayConfig struct { func (SeasonDecayConfig) TableName() string { return "season_decay_config" } +// AssetExhibitionHoursLog 资产级累计时长幂等 log +type AssetExhibitionHoursLog struct { + ID int64 `gorm:"column:id;primaryKey;autoIncrement"` + SourceID string `gorm:"column:source_id;unique;not null"` + AssetID int64 `gorm:"column:asset_id;not null"` + Hours int `gorm:"column:hours;not null"` + CreatedAt int64 `gorm:"column:created_at;not null"` +} + +func (AssetExhibitionHoursLog) TableName() string { return "asset_exhibition_hours_log" } + // 等级常量 const ( LevelN = "N" diff --git a/backend/pkg/models/exhibition_hours_log.go b/backend/pkg/models/exhibition_hours_log.go new file mode 100644 index 0000000..5aa0259 --- /dev/null +++ b/backend/pkg/models/exhibition_hours_log.go @@ -0,0 +1,13 @@ +package models + +// ExhibitionHoursLog 用户级累计时长幂等 log +type ExhibitionHoursLog struct { + ID int64 `gorm:"column:id;primaryKey;autoIncrement"` + SourceID string `gorm:"column:source_id;unique;not null"` + UserID int64 `gorm:"column:user_id;not null"` + StarID int64 `gorm:"column:star_id;not null"` + Hours int64 `gorm:"column:hours;not null"` + CreatedAt int64 `gorm:"column:created_at;not null"` +} + +func (ExhibitionHoursLog) TableName() string { return "exhibition_hours_log" } diff --git a/backend/scripts/recalc_exhibition_hours.sql b/backend/scripts/recalc_exhibition_hours.sql new file mode 100644 index 0000000..3666222 --- /dev/null +++ b/backend/scripts/recalc_exhibition_hours.sql @@ -0,0 +1,173 @@ +-- recalc_exhibition_hours.sql +-- 批次1.2 存量校正(DRY-RUN): 输出"按 distinct exhibition 重算"前后的差异 +-- ⚠️ 本文件仅 SELECT,禁止任何 DDL/DML。 +-- +-- 列名适配说明: +-- * exhibitions.slot_owner_uid -> 实际为 host_profile_id +-- * 收益归属(revenue_service.go L356): user_id=SlotOwnerUID(=host_profile_id), star_id=OccupierStarID +-- * level_up_bonus 触发: AddExhibitionHours(userID=host_profile_id, starID=occupier_star_id) + +\echo '==== A. 用户级:当前 vs 重算 (差值 > 0 即虚高) ====' +WITH per_user_star AS ( + SELECT e.host_profile_id AS user_id, + e.occupier_star_id AS star_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL + AND e.host_profile_id IS NOT NULL + AND e.occupier_star_id IS NOT NULL + GROUP BY e.host_profile_id, e.occupier_star_id +), +current_ueh AS ( + SELECT user_id, star_id, total_exhibition_hours AS current_hours + FROM user_exhibition_hours +) +SELECT COALESCE(c.user_id, r.user_id) AS user_id, + COALESCE(c.star_id, r.star_id) AS star_id, + COALESCE(c.current_hours, 0) AS current_hours, + COALESCE(r.recalc_hours, 0) AS recalc_hours, + (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) AS diff_hours +FROM current_ueh c +FULL OUTER JOIN per_user_star r USING (user_id, star_id) +WHERE COALESCE(c.current_hours, 0) <> COALESCE(r.recalc_hours, 0) +ORDER BY ABS(COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) DESC +LIMIT 50; + +\echo '==== B. 用户级 diff 汇总 (待下修/虚高,无 DML) ====' +WITH per_user_star AS ( + SELECT e.host_profile_id AS user_id, + e.occupier_star_id AS star_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL + AND e.host_profile_id IS NOT NULL + AND e.occupier_star_id IS NOT NULL + GROUP BY e.host_profile_id, e.occupier_star_id +) +SELECT count(*) FILTER (WHERE (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) > 0) AS user_star_pairs_overstated, + count(*) FILTER (WHERE (COALESCE(c.current_hours, 0) - COALESCE(c.current_hours, 0)) > 0) AS placeholder_no_overcount, + count(*) FILTER (WHERE (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) < 0) AS user_star_pairs_understated, + count(*) FILTER (WHERE c.user_id IS NULL) AS user_star_pairs_in_exhibition_only, + count(*) FILTER (WHERE r.user_id IS NULL) AS user_star_pairs_in_ueh_only, + COALESCE(sum(GREATEST(COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0), 0)), 0) AS total_overstated_hours, + COALESCE(sum(GREATEST(COALESCE(r.recalc_hours, 0) - COALESCE(c.current_hours, 0), 0)), 0) AS total_understated_hours +FROM (SELECT user_id, star_id, total_exhibition_hours AS current_hours FROM user_exhibition_hours) c +FULL OUTER JOIN per_user_star r USING (user_id, star_id); + +\echo '==== C. 资产级:当前 vs 重算 (差值 > 0 即虚高) ====' +WITH per_asset AS ( + SELECT e.asset_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL AND e.asset_id > 0 + GROUP BY e.asset_id +) +SELECT COALESCE(c.asset_id, r.asset_id) AS asset_id, + COALESCE(c.current_hours, 0) AS current_hours, + COALESCE(r.recalc_hours, 0) AS recalc_hours, + (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) AS diff_hours +FROM ( + SELECT asset_id, season_exhibition_hours AS current_hours + FROM asset_level_records WHERE season_exhibition_hours > 0 +) c +FULL OUTER JOIN per_asset r USING (asset_id) +WHERE COALESCE(c.current_hours, 0) <> COALESCE(r.recalc_hours, 0) +ORDER BY ABS(COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) DESC +LIMIT 50; + +\echo '==== D. 资产级 diff 汇总 ====' +WITH per_asset AS ( + SELECT e.asset_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL AND e.asset_id > 0 + GROUP BY e.asset_id +) +SELECT count(*) FILTER (WHERE (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) > 0) AS assets_overstated, + count(*) FILTER (WHERE (COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0)) < 0) AS assets_understated, + count(*) FILTER (WHERE c.asset_id IS NULL) AS assets_in_exhibition_only, + count(*) FILTER (WHERE r.asset_id IS NULL) AS assets_in_alr_only, + COALESCE(sum(GREATEST(COALESCE(c.current_hours, 0) - COALESCE(r.recalc_hours, 0), 0)), 0) AS total_overstated_asset_hours, + COALESCE(sum(GREATEST(COALESCE(r.recalc_hours, 0) - COALESCE(c.current_hours, 0), 0)), 0) AS total_understated_asset_hours +FROM (SELECT asset_id, season_exhibition_hours AS current_hours FROM asset_level_records WHERE season_exhibition_hours > 0) c +FULL OUTER JOIN per_asset r USING (asset_id); + +\echo '==== E. 已发升级奖励水晶总额 (change_type=level_up_bonus, 仅统计) ====' +SELECT count(*) AS level_up_bonus_records, + COALESCE(sum(delta), 0) AS total_crystal +FROM crystal_transaction_records +WHERE change_type = 'level_up_bonus'; + +\echo '==== F. level_up_bonus 按 source_id 是否形如 exhibition_ 拆分 ====' +SELECT + count(*) FILTER (WHERE source_id ~ '^exhibition_[0-9]+$') AS looks_like_exhibition_source, + count(*) FILTER (WHERE source_id IS NULL OR source_id = '') AS no_source_id, + count(*) FILTER (WHERE source_id IS NOT NULL AND source_id <> '' AND source_id !~ '^exhibition_[0-9]+$') AS other_source, + COALESCE(sum(delta) FILTER (WHERE source_id ~ '^exhibition_[0-9]+$'), 0) AS exhibition_source_crystal, + COALESCE(sum(delta) FILTER (WHERE source_id IS NULL OR source_id = ''), 0) AS no_source_crystal +FROM crystal_transaction_records +WHERE change_type = 'level_up_bonus'; + +\echo '==== G. 误升用户数: fan_profiles 当前 level vs 按 recalc_hours 应得 level ====' +WITH per_user_star AS ( + SELECT e.host_profile_id AS user_id, + e.occupier_star_id AS star_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL + AND e.host_profile_id IS NOT NULL + AND e.occupier_star_id IS NOT NULL + GROUP BY e.host_profile_id, e.occupier_star_id +), +expected_level AS ( + SELECT lt.level AS expected_level, r.user_id, r.star_id, r.recalc_hours + FROM per_user_star r + JOIN LATERAL ( + SELECT level FROM level_thresholds + WHERE max_exhibition_hours <= r.recalc_hours + ORDER BY level DESC LIMIT 1 + ) lt ON true +) +SELECT count(*) FILTER (WHERE fp.level > el.expected_level) AS profiles_will_downgrade, + count(*) FILTER (WHERE fp.level < el.expected_level) AS profiles_will_upgrade, + count(*) FILTER (WHERE fp.level = el.expected_level) AS profiles_match, + COALESCE(sum(fp.level - el.expected_level) FILTER (WHERE fp.level > el.expected_level), 0) AS total_level_reduction +FROM fan_profiles fp +JOIN expected_level el + ON fp.user_id = el.user_id AND fp.star_id = el.star_id; + +\echo '==== H. 误升用户关联的 level_up_bonus 总额 (若 fan_profile 当前 level > recalc 应得 level) ====' +WITH per_user_star AS ( + SELECT e.host_profile_id AS user_id, + e.occupier_star_id AS star_id, + SUM((e.expire_at - e.start_time) / 3600000)::bigint AS recalc_hours + FROM exhibitions e + WHERE e.deleted_at IS NULL + AND e.host_profile_id IS NOT NULL + AND e.occupier_star_id IS NOT NULL + GROUP BY e.host_profile_id, e.occupier_star_id +), +expected_level AS ( + SELECT r.user_id, r.star_id, lt.level AS expected_level + FROM per_user_star r + JOIN LATERAL ( + SELECT level FROM level_thresholds + WHERE max_exhibition_hours <= r.recalc_hours + ORDER BY level DESC LIMIT 1 + ) lt ON true +), +mis_upgraded_users AS ( + SELECT fp.user_id, fp.star_id + FROM fan_profiles fp + JOIN expected_level el + ON fp.user_id = el.user_id AND fp.star_id = el.star_id + WHERE fp.level > el.expected_level +) +SELECT count(*) AS mis_upgrade_crystal_records, + COALESCE(sum(delta), 0) AS mis_upgrade_total_crystal +FROM crystal_transaction_records ctr +WHERE ctr.change_type = 'level_up_bonus' + AND EXISTS ( + SELECT 1 FROM mis_upgraded_users m + WHERE m.user_id = ctr.user_id AND m.star_id = ctr.star_id + ); \ No newline at end of file diff --git a/backend/services/assetService/go.mod b/backend/services/assetService/go.mod index e9253d8..fa97610 100644 --- a/backend/services/assetService/go.mod +++ b/backend/services/assetService/go.mod @@ -4,6 +4,7 @@ go 1.25.5 require ( dubbo.apache.org/dubbo-go/v3 v3.3.1 + github.com/alicebob/miniredis/v2 v2.38.0 github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible github.com/aliyun/credentials-go v1.4.12 github.com/google/uuid v1.6.0 @@ -129,6 +130,7 @@ require ( github.com/uber/jaeger-client-go v2.29.1+incompatible // indirect github.com/uber/jaeger-lib v2.4.1+incompatible // indirect github.com/ugorji/go/codec v1.2.12 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect github.com/yusufpapurcu/wmi v1.2.2 // indirect go.etcd.io/etcd/api/v3 v3.5.7 // indirect go.etcd.io/etcd/client/pkg/v3 v3.5.7 // indirect diff --git a/backend/services/assetService/go.sum b/backend/services/assetService/go.sum index c8b2148..0eca451 100644 --- a/backend/services/assetService/go.sum +++ b/backend/services/assetService/go.sum @@ -78,6 +78,8 @@ github.com/alibabacloud-go/tea v1.2.2/go.mod h1:CF3vOzEMAG+bR4WOql8gc2G9H3EkH3ZL github.com/alibabacloud-go/tea v1.3.13 h1:WhGy6LIXaMbBM6VBYcsDCz6K/TPsT1Ri2hPmmZffZ94= github.com/alibabacloud-go/tea-utils v1.4.4 h1:lxCDvNCdTo9FaXKKq45+4vGETQUKNOW/qKTcX9Sk53o= github.com/alibabacloud-go/tea-utils v1.4.4/go.mod h1:KNcT0oXlZZxOXINnZBs6YvgOd5aYp9U67G+E3R8fcQw= +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/aliyun/alibaba-cloud-sdk-go v1.61.18/go.mod h1:v8ESoHo4SyHmuB4b1tJqDHxfTGEciD+yhvOU/5s1Rfk= github.com/aliyun/alibaba-cloud-sdk-go v1.61.1704/go.mod h1:RcDobYh8k5VP6TNybz9m++gL3ijVI5wueVr0EM10VsU= github.com/aliyun/alibaba-cloud-sdk-go v1.61.1800 h1:ie/8RxBOfKZWcrbYSJi2Z8uX8TcOlSMwPlEJh83OeOw= @@ -847,6 +849,8 @@ github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/yusufpapurcu/wmi v1.2.2 h1:KBNDSne4vP5mbSWnJbO+51IMOXJB67QiYCSBrubbPRg= github.com/yusufpapurcu/wmi v1.2.2/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= diff --git a/backend/services/assetService/repository/asset_level_repository.go b/backend/services/assetService/repository/asset_level_repository.go index 2830d43..9104299 100644 --- a/backend/services/assetService/repository/asset_level_repository.go +++ b/backend/services/assetService/repository/asset_level_repository.go @@ -59,6 +59,20 @@ func (r *AssetLevelRepository) GetDB() *gorm.DB { return r.db } +// WithTx 返回绑定到 tx 的 repo,供 service 层在 db.Transaction(...) 闭包内调用, +//让 Create/Save/GetByAssetID 等方法走事务(否则默认走 r.db,事务边界不生效)。 +// +// 用法: +// err := r.db.Transaction(func(tx *gorm.DB) error { +// txRepo := r.WithTx(tx) +// _ = txRepo.GetByAssetID(...) +// _ = txRepo.Save(...) +// return nil +// }) +func (r *AssetLevelRepository) WithTx(tx *gorm.DB) *AssetLevelRepository { + return &AssetLevelRepository{db: tx} +} + func (r *AssetLevelRepository) GetChangeLogs(assetID int64, limit, offset int) ([]*models.AssetLevelChangeLog, error) { var logs []*models.AssetLevelChangeLog err := r.db.Where("asset_id = ?", assetID). diff --git a/backend/services/assetService/service/asset_level_service.go b/backend/services/assetService/service/asset_level_service.go index c0dc56c..7cab5ce 100644 --- a/backend/services/assetService/service/asset_level_service.go +++ b/backend/services/assetService/service/asset_level_service.go @@ -13,6 +13,7 @@ import ( "github.com/topfans/backend/services/assetService/repository" "go.uber.org/zap" "gorm.io/gorm" + "gorm.io/gorm/clause" ) type AssetLevelService interface { @@ -20,7 +21,7 @@ type AssetLevelService interface { GetRecordByAssetID(assetID int64) (*models.AssetLevelRecord, error) GetLevelConfig(level string) (*models.AssetLevel, error) GetAllLevels() ([]*models.AssetLevel, error) - AddExhibitionHours(assetID int64, hours int) (string, bool, error) + AddExhibitionHours(assetID int64, hours int, sourceID string) (string, bool, error) AddLikes(assetID int64, count int) (string, bool, error) RemoveLikes(assetID int64, count int) (string, bool, error) CalculateRevenue(assetID int64, likeCount int, startTime, endTime int64, revenueBoostBps int) (int64, error) @@ -146,43 +147,182 @@ func (s *assetLevelService) CheckDowngrade(record *models.AssetLevelRecord) (str return newLevel, newLevel != record.CurrentLevel } -func (s *assetLevelService) AddExhibitionHours(assetID int64, hours int) (string, bool, error) { - record, err := s.GetOrCreateRecord(assetID) +func (s *assetLevelService) AddExhibitionHours(assetID int64, hours int, sourceID string) (string, bool, error) { + if sourceID == "" { + logger.Logger.Warn("AssetLevelService.AddExhibitionHours called with empty sourceID, idempotency degraded", + zap.Int64("asset_id", assetID)) + } + + db := s.levelRepo.GetDB() + if db == nil { + return "", false, fmt.Errorf("asset level repository db not initialized") + } + + var resultLevel string + var upgraded bool + + // 整个累积时长流程(log INSERT + 判 RowsAffected + 累加 Save + 升级 changelog)放在同一事务: + // - log 写入了但累加崩了 → 一起回滚(避免"log 记了但 hours 没加"孤儿态) + // - 重复路径 RowsAffected==0 直接 return,事务早结束 + err := db.Transaction(func(tx *gorm.DB) error { + txRepo := s.levelRepo.WithTx(tx) + + // 0. 幂等闸口 + if sourceID != "" { + logRow := &models.AssetExhibitionHoursLog{ + SourceID: sourceID, + AssetID: assetID, + Hours: hours, + CreatedAt: time.Now().UnixMilli(), + } + res := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "source_id"}}, + DoNothing: true, + }).Create(logRow) + if res.Error != nil { + return res.Error + } + if res.RowsAffected == 0 { + // 重复 sourceID:回查当前等级后 return,事务提前结束 + rec, qerr := txRepo.GetByAssetID(assetID) + if qerr != nil && qerr != gorm.ErrRecordNotFound { + return qerr + } + logger.Logger.Info("AssetLevelService.AddExhibitionHours: duplicate source_id, skip", + zap.Int64("asset_id", assetID), + zap.String("source_id", sourceID)) + if rec != nil { + resultLevel = rec.CurrentLevel + } + upgraded = false + return nil + } + } + + // 1. 获取或创建累计时长记录 + record, err := getOrCreateRecordTx(tx, assetID) + if err != nil { + return err + } + + oldLevel := record.CurrentLevel + + if record.SeasonID == "" { + season, err := s.GetCurrentSeason() + if err != nil { + season = s.getDefaultSeason() + } + record.SeasonID = season.ID + } + + // 2. 累加时长(原子更新,避免竞态) + now := time.Now().UnixMilli() + if err := tx.Model(&models.AssetLevelRecord{}). + Where("asset_id = ?", assetID). + Updates(map[string]any{ + "season_exhibition_hours": gorm.Expr("season_exhibition_hours + ?", hours), + "lifetime_exhibition_hours": gorm.Expr("lifetime_exhibition_hours + ?", hours), + "updated_at": now, + }).Error; err != nil { + return err + } + + // 3. 重新读最新 record(供 CheckUpgrade 用最新 hours) + updated, err := txRepo.GetByAssetID(assetID) + if err != nil { + return err + } + updated.CurrentLevel = record.CurrentLevel + updated.SeasonID = record.SeasonID + record = updated + + // 4. 计算新等级 + newLevel, isUpgraded := s.CheckUpgrade(record) + if isUpgraded { + record.CurrentLevel = newLevel + } + + // 5. 落库累计/等级 + if err := txRepo.Save(record); err != nil { + return err + } + + // 6. 升级 changelog + asset_registry.grade 同步(同事务) + if isUpgraded && newLevel != oldLevel { + changeLog := &models.AssetLevelChangeLog{ + AssetID: record.AssetID, + FromLevel: oldLevel, + ToLevel: newLevel, + TriggerType: "exhibition_complete", + TriggerHours: record.SeasonExhibitionHours, + TriggerLikes: record.SeasonLikes, + ChangeReason: fmt.Sprintf("展出完成，时长+%d小时", hours), + CreatedAt: now, + } + if err := txRepo.CreateChangeLog(changeLog); err != nil { + return err + } + grade := models.LevelToGrade(newLevel) + if err := tx.Table("public.asset_registry"). + Where("asset_id = ?", record.AssetID). + Update("grade", grade).Error; err != nil { + return err + } + } + + resultLevel = newLevel + upgraded = isUpgraded + return nil + }) if err != nil { return "", false, err } - oldLevel := record.CurrentLevel - - if record.SeasonID == "" { - season, err := s.GetCurrentSeason() - if err != nil { - season = s.getDefaultSeason() - } - record.SeasonID = season.ID + // 升级事件埋点(statistic)走事务外:fire-and-forget,不应阻塞核心事务 + if upgraded && resultLevel != "" { + // 升级事件埋点(沿用 logLevelChange 既有行为,但不在事务内) + statistic.Get().TrackEvent(context.Background(), &eventPb.Event{ + EventType: "asset.level_up", + UserId: 0, + StarId: 0, + OccurredAt: time.Now().UnixMilli(), + Properties: map[string]string{ + "asset_id": strconv.FormatInt(assetID, 10), + "to": resultLevel, + }, + }) } - record.SeasonExhibitionHours += hours - record.LifetimeExhibitionHours += hours + return resultLevel, upgraded, nil +} - newLevel, upgraded := s.CheckUpgrade(record) - if upgraded { - record.CurrentLevel = newLevel +// getOrCreateRecordTx 事务内获取或创建 asset_level_records 行。 +// 走 INSERT ... ON CONFLICT(asset_id) DO NOTHING 保证并发安全。 +func getOrCreateRecordTx(tx *gorm.DB, assetID int64) (*models.AssetLevelRecord, error) { + var record models.AssetLevelRecord + err := tx.Where("asset_id = ?", assetID).First(&record).Error + if err == nil { + return &record, nil } - - if err := s.levelRepo.Save(record); err != nil { - return "", false, err + if err != gorm.ErrRecordNotFound { + return nil, err } - - if upgraded && newLevel != oldLevel { - s.logLevelChange(record.AssetID, oldLevel, newLevel, - "exhibition_complete", record.SeasonExhibitionHours, record.SeasonLikes, - fmt.Sprintf("展出完成，时长+%d小时", hours)) - // 同步等级到AssetRegistry.Grade - s.syncGradeToAssetRegistry(record.AssetID, newLevel) + // 不存在则插入空记录;并发场景下 ON CONFLICT DO NOTHING 防重复插入 + newRec := &models.AssetLevelRecord{ + AssetID: assetID, + CurrentLevel: models.LevelN, } - - return newLevel, upgraded, nil + if err := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "asset_id"}}, + DoNothing: true, + }).Create(newRec).Error; err != nil { + return nil, err + } + // 重读一次拿稳定 record(包含数据库默认值) + if err := tx.Where("asset_id = ?", assetID).First(&record).Error; err != nil { + return nil, err + } + return &record, nil } func (s *assetLevelService) AddLikes(assetID int64, count int) (string, bool, error) { diff --git a/backend/services/assetService/service/asset_level_service_test.go b/backend/services/assetService/service/asset_level_service_test.go index 453463f..b6c0df4 100644 --- a/backend/services/assetService/service/asset_level_service_test.go +++ b/backend/services/assetService/service/asset_level_service_test.go @@ -1,13 +1,21 @@ package service import ( + "os" + "strconv" "testing" + + "github.com/topfans/backend/pkg/database" + "github.com/topfans/backend/pkg/models" + "github.com/topfans/backend/services/assetService/repository" + "gorm.io/gorm" ) +// TestCalculateBuff 纯函数,无需 DB,保留原有行为。 func TestCalculateBuff(t *testing.T) { tests := []struct { likeCount int - expected int + expected int }{ {0, 0}, {4, 0}, @@ -26,3 +34,144 @@ func TestCalculateBuff(t *testing.T) { } } } + +// assetLevelTestDB 资产级幂等测试专用 DB 连接。 +// +// 自包含: +// - 通过 TEST_DB_HOST/PORT/USER/PASSWORD/NAME 覆盖,默认 localhost:15432 / postgres / 123456 / top-fans。 +// - 连不上 t.Skip(不污染全局 TestMain,避免影响其它测试)。 +// - AutoMigrate asset_level_records / asset_exhibition_hours_log / season / asset_level(幂等所需)。 +func assetLevelTestDB(t *testing.T) *gorm.DB { + t.Helper() + if os.Getenv("SKIP_DB_TESTS") != "" { + t.Skip("SKIP_DB_TESTS set") + } + + host := getEnvOrDefault("TEST_DB_HOST", "localhost") + portStr := getEnvOrDefault("TEST_DB_PORT", "15432") + port, _ := strconv.Atoi(portStr) + if port == 0 { + port = 15432 + } + user := getEnvOrDefault("TEST_DB_USER", "postgres") + password := getEnvOrDefault("TEST_DB_PASSWORD", "123456") + dbname := getEnvOrDefault("TEST_DB_NAME", "top-fans") + + if err := database.Init(database.Config{ + Host: host, + Port: port, + User: user, + Password: password, + DBName: dbname, + SSLMode: "disable", + TimeZone: "Asia/Shanghai", + }); err != nil { + t.Skipf("Skipping: failed to connect to test database %s:%d/%s as %s: %v", + host, port, dbname, user, err) + } + + db := database.GetDB() + + // AutoMigrate 幂等所需最小表集(若已存在则跳过)。 + if err := db.AutoMigrate( + &models.Season{}, + &models.AssetLevel{}, + &models.AssetLevelRecord{}, + &models.AssetLevelChangeLog{}, + &models.AssetExhibitionHoursLog{}, + ); err != nil { + t.Logf("Warning: AutoMigrate asset_level tables (may already exist): %v", err) + } + + // 确保至少存在 N 等级(供 CheckUpgrade 不会因为没有 level 而 panic)。 + var n int64 + db.Model(&models.AssetLevel{}).Where("level = ?", models.LevelN).Count(&n) + if n == 0 { + db.Create(&models.AssetLevel{ + Level: models.LevelN, + LevelOrder: 1, + RequireHours: 99999999, + RequireLikes: 99999999, + }) + } + + return db +} + +func getEnvOrDefault(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// TestAddExhibitionHours_AssetIdempotent 资产级幂等测试。 +// +// 验收: +// - 同一 sourceID 二次调用,不重复累加 SeasonExhibitionHours / LifetimeExhibitionHours。 +// - 二次调用返回 upgraded=false。 +// - asset_exhibition_hours_log 中恰好 1 条记录。 +// +// 数据隔离: +// - 用 sentinel assetID + sourceID,只删自己 sentinel 行,不调用任何全局 cleanup helper。 +func TestAddExhibitionHours_AssetIdempotent(t *testing.T) { + db := assetLevelTestDB(t) + + levelRepo := repository.NewAssetLevelRepository(db) + seasonRepo := repository.NewSeasonRepository(db) + decayRepo := repository.NewSeasonDecayConfigRepository(db) + svc := NewAssetLevelService(levelRepo, seasonRepo, decayRepo) + + assetID := int64(99900091) + srcID := "test_asset_exh_log_91_001" + + // 先清 sentinel 行(包含上一次失败或并发残留),defer 再清一次确保退出干净。 + db.Exec("DELETE FROM asset_exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM asset_level_records WHERE asset_id = ?", assetID) + defer func() { + db.Exec("DELETE FROM asset_exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM asset_level_records WHERE asset_id = ?", assetID) + }() + + // 1) 首次:应该累加,新等级(sourceID 不同每次应该都能跑完 add 路径)。 + lvl1, up1, err := svc.AddExhibitionHours(assetID, 5, srcID) + if err != nil { + t.Fatalf("first AddExhibitionHours err: %v", err) + } + _ = lvl1 + _ = up1 + + // 2) 二次:同 sourceID,应该跳过累加、不触发升级。 + lvl2, up2, err := svc.AddExhibitionHours(assetID, 5, srcID) + if err != nil { + t.Fatalf("second AddExhibitionHours err: %v", err) + } + if up2 { + t.Errorf("want upgraded=false on duplicate source_id, got true (level=%s)", lvl2) + } + if lvl2 == "" { + t.Errorf("want non-empty level on duplicate, got empty string") + } + + // 3) 校验累加表:SeasonExhibitionHours 应仅为 5(一次生效)。 + var rec models.AssetLevelRecord + if err := db.Where("asset_id = ?", assetID).First(&rec).Error; err != nil { + t.Fatalf("read asset level record: %v", err) + } + if rec.SeasonExhibitionHours != 5 { + t.Errorf("want SeasonExhibitionHours=5 (one apply), got %d", rec.SeasonExhibitionHours) + } + if rec.LifetimeExhibitionHours != 5 { + t.Errorf("want LifetimeExhibitionHours=5 (one apply), got %d", rec.LifetimeExhibitionHours) + } + + // 4) 校验幂等 log 表:恰好 1 条。 + var logCount int64 + if err := db.Model(&models.AssetExhibitionHoursLog{}). + Where("source_id = ?", srcID).Count(&logCount).Error; err != nil { + t.Fatalf("count asset exhibition log: %v", err) + } + if logCount != 1 { + t.Errorf("want exactly 1 asset_exhibition_hours_log row, got %d", logCount) + } +} diff --git a/backend/services/assetService/service/mint_rate_limiter.go b/backend/services/assetService/service/mint_rate_limiter.go new file mode 100644 index 0000000..ebff95b --- /dev/null +++ b/backend/services/assetService/service/mint_rate_limiter.go @@ -0,0 +1,60 @@ +package service + +import ( + "context" + "fmt" + "time" + + "github.com/redis/go-redis/v9" +) + +// mintRateLimitScript atomically increments the current window counter and +// assigns its TTL when the key is first created. +var mintRateLimitScript = redis.NewScript(` +local n = redis.call("INCR", KEYS[1]) +if n == 1 then + redis.call("EXPIRE", KEYS[1], ARGV[1]) +end +return {n, n <= tonumber(ARGV[2]) and 1 or 0} +`) + +// MintRateLimiter limits successful mint attempts with a Redis fixed window. +type MintRateLimiter struct { + rdb *redis.Client + limit int64 + window time.Duration +} + +// NewMintRateLimiter creates a Redis-backed mint rate limiter. +func NewMintRateLimiter(rdb *redis.Client, limit int64, window time.Duration) *MintRateLimiter { + return &MintRateLimiter{rdb: rdb, limit: limit, window: window} +} + +// IncrAndCheck atomically increments the owner's counter and returns its new +// value and whether it is within the configured limit. +func (l *MintRateLimiter) IncrAndCheck(ctx context.Context, ownerUID int64, assetType string) (int64, bool, error) { + if l == nil || l.rdb == nil { + return 0, false, fmt.Errorf("redis client is not initialized") + } + + windowSeconds := int64(l.window.Seconds()) + result, err := mintRateLimitScript.Run(ctx, l.rdb, []string{l.key(ownerUID, assetType)}, windowSeconds, l.limit).Result() + if err != nil { + return 0, false, fmt.Errorf("redis EVAL mint rate limit: %w", err) + } + + values, ok := result.([]interface{}) + if !ok || len(values) != 2 { + return 0, false, fmt.Errorf("redis EVAL returned unexpected shape: %v", result) + } + count, countOK := values[0].(int64) + allowed, allowedOK := values[1].(int64) + if !countOK || !allowedOK { + return 0, false, fmt.Errorf("redis EVAL returned unexpected values: %v", result) + } + return count, allowed == 1, nil +} + +func (l *MintRateLimiter) key(ownerUID int64, assetType string) string { + return fmt.Sprintf("periph:mint:%s:%d:%s", assetType, ownerUID, time.Now().UTC().Format("20060102")) +} diff --git a/backend/services/assetService/service/mint_service.go b/backend/services/assetService/service/mint_service.go index 17d7761..23c987c 100644 --- a/backend/services/assetService/service/mint_service.go +++ b/backend/services/assetService/service/mint_service.go @@ -2,8 +2,9 @@ package service import ( "context" - "crypto/sha256" + "crypto/rand" "fmt" + "math/big" "net/url" "os" "strconv" @@ -227,6 +228,35 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st return nil, fmt.Errorf("order_id is required（请先调用 /api/v1/assets/mints/precreate 获取）") } + // ★ 幂等短路:已 SUCCESS 的同 order_id 直接返回,不再走扣费/建档流程。 + // 这是铸造正确性 plan 双层防护的第二层(assetService 入口); + // 第一层在 userService.UpdateCrystalBalance(source_id + 唯一索引),由 mint-task-1 覆盖。 + existing, err := s.mintOrderRepo.GetByOrderIDAndUser(req.OrderId, userID, starID) + if err == nil && existing != nil && existing.Status == models.MintOrderStatusSuccess { + logger.Logger.Info("CreateMintOrder idempotent hit", + zap.String("order_id", existing.OrderID), + zap.Int64("user_id", userID), + ) + // 重放响应: 用既有 asset / 既有成本,不再二次扣。 + var assetProto *pb.Asset + if existing.AssetID != nil { + if a, gerr := s.assetRepo.GetByID(*existing.AssetID); gerr == nil && a != nil { + assetProto = ModelToProtoAssetDetail(a, "", "", false, 0, 0, 0, 0, getInt32Value(a.Grade)) + } + } + return &pb.CreateMintOrderResponse{ + Base: &pbCommon.BaseResponse{ + Code: uint32(codes.OK), + Message: "idempotent", + Timestamp: time.Now().UnixMilli(), + }, + Order: ModelToProtoMintOrder(existing), + Asset: assetProto, + CostCrystal: existing.CostCrystal, + BalanceAfter: 0, // 余额不再重算(避免对 userService 二次调用) + }, nil + } + // 2. 获取当前累计铸爱次数，用于计算阶梯费用 currentMintCount, err := s.GetUserMintCount(userID, starID) if err != nil { @@ -322,9 +352,9 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st // 3.3 检查是否触发保底（概率触发） var boostBps int32 = 0 if localMintCost.Probability > 0 && localMintCost.RewardValue > 0 { - // 随机判断是否触发 - randomValue := time.Now().UnixNano() % 100 - if randomValue < localMintCost.Probability { + // ★ 批次1.5: 用 crypto/rand 替换原 time.Now().UnixNano()%100, + // 消除并发同纳秒相同结果与脚本卡点操纵风险。 + if rollGuarantee(localMintCost.Probability) { boostBps = int32(localMintCost.RewardValue) // reward_value 单位是 bps logger.Logger.Info("Mint guarantee triggered", zap.Int64("user_id", userID), @@ -346,8 +376,9 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st logger.Logger.Info("[MintOrder] Step 3.5: 创建资产") materialURLValue := getStringValue(mintOrder.MaterialURL) mintedAt := time.Now().UnixMilli() - mockTxHash := fmt.Sprintf("0x%x", sha256.Sum256([]byte(fmt.Sprintf("%s-%d-%d-%d", mintOrder.OrderID, userID, starID, time.Now().UnixNano())))) - mockBlockNumber := int64(time.Now().Unix()) // 使用当前时间戳作为模拟区块号 + // ★ 批次1.5: 不再写入伪造的 tx_hash / block_number。 + // 当前铸造不上链，写模拟 hash 会被前端误当作真实链上凭证；tx_hash/block_number 保持 NULL， + // 待真正上链时再由链上回调回填。mintedAt 表示业务生效时间（非链上时间），保留。 logger.Logger.Info("[MintOrder] 创建资产参数", zap.Int64("user_id", userID), zap.Int64("star_id", starID), zap.String("name", getStringValue(mintOrder.Name)), zap.String("material_url", materialURLValue)) asset = &models.Asset{ OwnerUID: userID, @@ -360,8 +391,6 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st Status: models.AssetStatusActive, // 直接设为 Active LikeCount: 0, Info: getStringValue(mintOrder.Info), - TxHash: &mockTxHash, - BlockNumber: &mockBlockNumber, MintedAt: &mintedAt, } @@ -987,3 +1016,30 @@ func syncAssetsIDSequence(tx *gorm.DB) error { ) `).Error } + +// rollGuarantee 按 probability(0-100) 概率返回是否触发保底. +// +// ★ 批次1.5: 用 crypto/rand 替换原 time.Now().UnixNano()%100, +// 消除并发同纳秒相同结果与脚本卡点操纵风险。 +// 边界约定: +// - probability <= 0 → 永不触发(false) +// - probability >= 100 → 总是触发(true) +// - 熵源失败时降级为不触发(false),避免在极端环境下出现伪随机劣化 +// +// 入参取 int64 与 models.MintCostConfig.Probability 字段类型一致, +// 避免调用点显式转换;非法值(负数 / 远超 100)按既定边界处理。 +func rollGuarantee(probability int64) bool { + if probability <= 0 { + return false + } + if probability >= 100 { + return true + } + v, err := rand.Int(rand.Reader, big.NewInt(100)) + if err != nil { + // 熵源失败: 降级为不触发(避免伪随机劣化;线上极少出现) + logger.Logger.Warn("crypto/rand.Int failed, falling back to no-guarantee", zap.Error(err)) + return false + } + return v.Int64() < probability +} diff --git a/backend/services/assetService/service/mint_service_idempotency_test.go b/backend/services/assetService/service/mint_service_idempotency_test.go new file mode 100644 index 0000000..d62f9f8 --- /dev/null +++ b/backend/services/assetService/service/mint_service_idempotency_test.go @@ -0,0 +1,239 @@ +package service + +import ( + "context" + "crypto/rand" + "math/big" + "testing" + + "github.com/topfans/backend/pkg/models" + pb "github.com/topfans/backend/pkg/proto/asset" + pbUser "github.com/topfans/backend/pkg/proto/user" + "github.com/topfans/backend/services/assetService/client" + "github.com/topfans/backend/services/assetService/repository" +) + +// mockUserClient 服务层测试用的 UserServiceClient mock。 +// +// 跟踪 UpdateCrystalBalance 调用次数,确保幂等短路路径不会重复扣费。 +// GetFanProfile 返回当前 balance(测试中只检查不被二次扣即可)。 +type mockUserClient struct { + balance int64 + updateCrystalCalls int +} + +func (m *mockUserClient) UpdateCrystalBalance(_ context.Context, _, _ int64, delta int64, _, _, _ string) (int64, error) { + m.updateCrystalCalls++ + m.balance += delta + return m.balance, nil +} + +func (m *mockUserClient) UpdateAssetsCount(_ context.Context, _, _ int64, delta int32) (int32, error) { + return 0, nil +} + +func (m *mockUserClient) GetFanProfile(_ context.Context, _, _ int64) (*pbUser.FanProfile, error) { + return &pbUser.FanProfile{CrystalBalance: m.balance}, nil +} + +// 编译期保证 mockUserClient 满足 client.UserServiceClient 接口 +var _ client.UserServiceClient = (*mockUserClient)(nil) + +// pbCreateMintReq 构造 CreateMintOrder 测试请求。 +// 故意忽略 userID/starID 参数(参数名带下划线),因为请求结构里只有订单字段; +// 用户与明星身份由 service 方法签名参数传入。 +func pbCreateMintReq(orderID string, _, _ int64) *pb.CreateMintOrderRequest { + return &pb.CreateMintOrderRequest{ + OrderId: orderID, + MaterialUrl: "http://x/m.jpg", + Name: "n", + Description: "d", + Info: "i", + MaterialType: "new", + } +} + +// TestCreateMintOrder_IdempotentOnSuccessOrder 验证幂等短路: +// +// 同 order_id 第二次调用,状态已是 SUCCESS → 直接返回原 asset/order/cost, +// 不再调 userClient.UpdateCrystalBalance。 +// +// ★ 双层防护第二层(assetService 侧入口短路)。 +// - 第一层(userService.UpdateCrystalBalance 的 ON CONFLICT)由 mint-task-1 覆盖。 +// - 本测试确保"已 SUCCESS 的订单"在入口处就重放响应,避免重复进入扣费/建档流程。 +func TestCreateMintOrder_IdempotentOnSuccessOrder(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + // 准备 user + star(余额走 mock,无需 fan_profile) + star := createServiceTestStar(t, db, "mint_idem_star") + user := createServiceTestUser(t, db, "19900077001") + + // 预置一个 SUCCESS 订单 + 关联 asset(模拟之前已铸造成功) + const orderID = "idem-order-uuid-001" + originalAsset := &models.Asset{ + OwnerUID: user.ID, + StarID: star.StarID, + Name: "existing", + CoverURL: "http://x/a.jpg", + Status: models.AssetStatusActive, + IsActive: true, + } + if err := db.Create(originalAsset).Error; err != nil { + t.Fatalf("Failed to seed asset: %v", err) + } + if err := db.Exec(`INSERT INTO asset_registry (owner_uid, asset_id, star_id, asset_type, status, created_at, updated_at) + VALUES (?, ?, ?, 'regular', 1, 1, 1)`, user.ID, originalAsset.ID, star.StarID).Error; err != nil { + t.Fatalf("Failed to seed asset_registry: %v", err) + } + if err := db.Create(&models.MintOrder{ + OrderID: orderID, + UserID: user.ID, + StarID: star.StarID, + Status: models.MintOrderStatusSuccess, + CostCrystal: 100, + AssetID: &originalAsset.ID, + }).Error; err != nil { + t.Fatalf("Failed to seed mint_order: %v", err) + } + + // mock userClient:UpdateCrystalBalance 不应被二次触发 + uc := &mockUserClient{balance: 900} + svc := NewMintService( + repository.NewAssetRepository(db), + repository.NewMintOrderRepository(db), + uc, + db, nil, + nil, nil, nil, + nil, + ) + + resp, err := svc.CreateMintOrder(pbCreateMintReq(orderID, user.ID, star.StarID), user.ID, star.StarID) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp == nil { + t.Fatal("expected response, got nil") + } + if resp.Order == nil { + t.Fatal("expected order in response, got nil") + } + if resp.Order.OrderId != orderID { + t.Errorf("want orderID=%s, got %s", orderID, resp.Order.OrderId) + } + if resp.Order.Status != models.MintOrderStatusSuccess { + t.Errorf("want order status=SUCCESS, got %s", resp.Order.Status) + } + if resp.CostCrystal != 100 { + t.Errorf("want cost_crystal=100, got %d", resp.CostCrystal) + } + if resp.Asset == nil { + t.Error("expected asset replay in idempotent response, got nil") + } else if resp.Asset.AssetId != originalAsset.ID { + t.Errorf("want assetID=%d, got %d", originalAsset.ID, resp.Asset.AssetId) + } + + // ★ 关键断言:幂等命中后不应再调用 UpdateCrystalBalance + if uc.updateCrystalCalls != 0 { + t.Errorf("UpdateCrystalBalance must not be called on idempotent hit, got %d calls", uc.updateCrystalCalls) + } + // 余额没被二次扣(保持 mock 初始值 900) + if uc.balance != 900 { + t.Errorf("balance want 900 (unchanged), got %d", uc.balance) + } +} + +// TestMintGuaranteeProbability_NonPredictable 验证随机源已从 time.Now() 切到 crypto/rand: +// +// 1000 次连续采样,100% 命中(Probability=100)与 0% 命中(Probability=0)必须分别全命中/全不命中; +// 且单次返回落在 [0,100) 区间。 +// +// ★ 批次1.5:helper 测试本身只验证 crypto/rand 行为符合规格(库内建)。 +// 真正验证 rollGuarantee 行为见 TestRollGuarantee_Boundaries / _Distribution。 +func TestMintGuaranteeProbability_NonPredictable(t *testing.T) { + // 抽 1000 个 [0,100) 整数,验证区间 + 100% 概率 vs 0% 概率两个极端。 + for i := 0; i < 1000; i++ { + v, err := rand.Int(rand.Reader, big.NewInt(100)) + if err != nil { + t.Fatal(err) + } + if v.Cmp(big.NewInt(100)) >= 0 || v.Sign() < 0 { + t.Fatalf("out of range: %v", v) + } + } + // 边界 + if alwaysTriggers(100, 1000) != 1000 { + t.Error("P=100 should always trigger") + } + if alwaysTriggers(0, 1000) != 0 { + t.Error("P=0 should never trigger") + } +} + +// alwaysTriggers 模拟 helper 内部判定,用于边界测试 +func alwaysTriggers(probability, n int) int { + hits := 0 + for i := 0; i < n; i++ { + v, _ := rand.Int(rand.Reader, big.NewInt(100)) + if v.Int64() < int64(probability) { + hits++ + } + } + return hits +} + +// TestRollGuarantee_Boundaries 验证 rollGuarantee 在概率边界上的语义: +// +// probability <= 0 → 永不触发 +// probability >= 100 → 总是触发 +// 非法值(负数 / 远超 100)与 mint_service 现有 Probability 字段约定一致 +// (Probability 在 DB 层是 int64,正常取值 0~100;这里显式保证超出范围时不会 panic / 误判)。 +func TestRollGuarantee_Boundaries(t *testing.T) { + cases := []struct { + name string + probability int64 + want bool + }{ + {"zero never triggers", 0, false}, + {"negative never triggers", -1, false}, + {"deep negative never triggers", -1000, false}, + {"hundred always triggers", 100, true}, + {"over hundred always triggers", 101, true}, + {"way over hundred always triggers", 9999, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // 边界值是确定性的,跑 50 次保证没有"随机巧合" + for i := 0; i < 50; i++ { + if got := rollGuarantee(tc.probability); got != tc.want { + t.Fatalf("rollGuarantee(%d) = %v, want %v (iter %d)", tc.probability, got, tc.want, i) + } + } + }) + } +} + +// TestRollGuarantee_Distribution 验证 rollGuarantee 在 P=50 时采样 2000 次, +// +// 命中率落在 [800, 1200] 内(理论 1000 ± 期望误差 ~22)。 +// +// 这是统计性测试:理论 2000 次伯努利试验 σ ≈ √(npq) = √500 ≈ 22, +// 取 ±9σ(800~1200)基本不可能(<10⁻¹⁸)误判;同时能捕捉到"退化成伪随机 / 写死" +// 这类回归。如果有人把 crypto/rand 换成 time.Now().UnixNano()%100, +// 在循环极快时同纳秒下分布会严重偏(全部 false / 全部 true),立刻 fail。 +func TestRollGuarantee_Distribution(t *testing.T) { + const n = 2000 + const p int64 = 50 + hits := 0 + for i := 0; i < n; i++ { + if rollGuarantee(p) { + hits++ + } + } + const minHits = 800 + const maxHits = 1200 + if hits < minHits || hits > maxHits { + t.Errorf("rollGuarantee(50) over %d trials: hits=%d, want in [%d,%d] (≈50%%±9σ)", + n, hits, minHits, maxHits) + } +} \ No newline at end of file diff --git a/backend/services/assetService/service/peripheral_service.go b/backend/services/assetService/service/peripheral_service.go index 3968eba..a9e0c2e 100644 --- a/backend/services/assetService/service/peripheral_service.go +++ b/backend/services/assetService/service/peripheral_service.go @@ -6,9 +6,9 @@ import ( "fmt" "time" - "github.com/topfans/backend/pkg/peripheral" "github.com/topfans/backend/pkg/logger" "github.com/topfans/backend/pkg/models" + "github.com/topfans/backend/pkg/peripheral" "github.com/topfans/backend/services/assetService/repository" "go.uber.org/zap" ) @@ -27,12 +27,12 @@ func (e *BizError) Error() string { return fmt.Sprintf("[%d] %s", e.Code, e.Mess // 周边业务码(spec §4.1/§4.2/§8.2 错误码表) const ( - BizCodeAssetNotFound = 50003 // 物品不存在或已下架 - BizCodeAlreadyAdded = 50004 // 您已添加过此周边 - BizCodeCannotAddSelf = 50011 // 防御性:周边不该出现 - BizCodeRateLimited = 50012 // 今日提交过于频繁 + BizCodeAssetNotFound = 50003 // 物品不存在或已下架 + BizCodeAlreadyAdded = 50004 // 您已添加过此周边 + BizCodeCannotAddSelf = 50011 // 防御性:周边不该出现 + BizCodeRateLimited = 50012 // 今日提交过于频繁 BizCodeInvalidSignature = 50013 // URL sign 验证失败(§5.1.3 第二道防线) - BizCodeInactive = 50014 // 周边尚未激活(未上架) + BizCodeInactive = 50014 // 周边尚未激活(未上架) ) // 藏品类型常量(2026-07-10 加) @@ -52,20 +52,26 @@ type VerificationResult struct { Brand string `json:"brand"` Image string `json:"image"` Verifier string `json:"verifier"` - VerifiedAt int64 `json:"verified_at"` // unix 秒 + VerifiedAt int64 `json:"verified_at"` // unix 秒 MaterialType string `json:"material_type,omitempty"` // 周边素材类型(空时前端 fallback 显示 "new") } // PeripheralService 周边验真 + 加入藏品的业务层 type PeripheralService struct { - repo *repository.PeripheralRepository + repo *repository.PeripheralRepository + rateLimiter *MintRateLimiter } -// NewPeripheralService 创建 PeripheralService 实例 +// NewPeripheralService creates a service with the DB fallback rate-limit path. func NewPeripheralService(repo *repository.PeripheralRepository) *PeripheralService { return &PeripheralService{repo: repo} } +// NewPeripheralServiceWithLimiter creates a service with Redis Lua atomic rate limiting. +func NewPeripheralServiceWithLimiter(repo *repository.PeripheralRepository, limiter *MintRateLimiter) *PeripheralService { + return &PeripheralService{repo: repo, rateLimiter: limiter} +} + // GetVerification 验真接口(spec §4.1) // // 流程:查 asset → 查 peripheral_info → 自增 peripheral_info.verify_count → 组装响应 @@ -111,7 +117,7 @@ func (s *PeripheralService) GetVerification(ctx context.Context, assetID int64) Image: info.Image, // 优先用 peripheral_info.image(asset 可能未创建) Verifier: info.Verifier, VerifiedAt: firstVerifiedAt / 1000, // 毫秒 → 秒(spec §4.1 数据契约) - MaterialType: MintMaterialTypeNew, // 验真页前端展示用,固定 "new" + MaterialType: MintMaterialTypeNew, // 验真页前端展示用,固定 "new" }, nil } @@ -229,13 +235,32 @@ func (s *PeripheralService) doMint(ctx context.Context, ownerUID int64, info *mo return nil, &BizError{Code: BizCodeAlreadyAdded, Message: "您已添加过此周边"} } - // 2. 限频:24h 最多 10 次 - count, err := s.repo.CountRecentMint(ctx, ownerUID, "peripheral", 24*time.Hour) - if err != nil { - return nil, fmt.Errorf("DB_COUNT_FAILED: %w", err) - } - if count >= 10 { - return nil, &BizError{Code: BizCodeRateLimited, Message: "今日提交过于频繁,请稍后再试"} + // 2. 限频: Redis Lua 原子自增; Redis 故障时降级为原 DB count 路径。 + if s.rateLimiter != nil { + count, allowed, err := s.rateLimiter.IncrAndCheck(ctx, ownerUID, "peripheral") + if err != nil { + logger.Logger.Warn("MintRateLimiter failed, falling back to DB count", + zap.Int64("owner_uid", ownerUID), zap.Error(err)) + dbCount, dbErr := s.repo.CountRecentMint(ctx, ownerUID, "peripheral", 24*time.Hour) + if dbErr != nil { + return nil, fmt.Errorf("DB_COUNT_FALLBACK_FAILED: %w", dbErr) + } + if dbCount >= 10 { + return nil, &BizError{Code: BizCodeRateLimited, Message: "今日提交过于频繁,请稍后再试"} + } + } else if !allowed { + logger.Logger.Info("Mint rate limited", + zap.Int64("owner_uid", ownerUID), zap.Int64("count", count)) + return nil, &BizError{Code: BizCodeRateLimited, Message: "今日提交过于频繁,请稍后再试"} + } + } else { + count, err := s.repo.CountRecentMint(ctx, ownerUID, "peripheral", 24*time.Hour) + if err != nil { + return nil, fmt.Errorf("DB_COUNT_FAILED: %w", err) + } + if count >= 10 { + return nil, &BizError{Code: BizCodeRateLimited, Message: "今日提交过于频繁,请稍后再试"} + } } // 3. 建新 asset(从 peripheral_info 字段填充) @@ -311,8 +336,8 @@ func (s *PeripheralService) doMint(ctx context.Context, ownerUID int64, info *mo }() return &MintResult{ - InstanceID: newID, // asset_registry.id(收藏记录) - AssetID: newAssetID, // ★ 新 asset.id(收藏者的藏品,跳 asset-detail 用) + InstanceID: newID, // asset_registry.id(收藏记录) + AssetID: newAssetID, // ★ 新 asset.id(收藏者的藏品,跳 asset-detail 用) MintedAt: createdAtMs / 1000, // 毫秒 → 秒,与 §4.2 数据契约对齐 CoverImage: info.Image, }, nil @@ -347,6 +372,7 @@ func statusInactiveMessage(status int16) string { return "该周边当前不可验真" } } + // derefStr *string → string(nil 返 "") func derefStr(p *string) string { if p == nil { diff --git a/backend/services/assetService/service/peripheral_service_test.go b/backend/services/assetService/service/peripheral_service_test.go index 5756292..81ff336 100644 --- a/backend/services/assetService/service/peripheral_service_test.go +++ b/backend/services/assetService/service/peripheral_service_test.go @@ -3,14 +3,31 @@ package service import ( "context" "errors" + "fmt" + "sync" + "sync/atomic" "testing" "time" - "gorm.io/gorm" - + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/topfans/backend/pkg/models" + "github.com/topfans/backend/pkg/peripheral" "github.com/topfans/backend/services/assetService/repository" + "gorm.io/gorm" ) +// peripheralTestCodeHash 测试 fixture 中 peripheral_info.hash 列的固定值; +// 见 setupAssetWithPeripheral 的 INSERT 语句。MintFromPeripheralByHash 要求 codeHash +// 参数等于 hash 列(由 peripheral.EncryptCode(code) 生成),sign 必须是 SignURL(codeHash)。 +const peripheralTestCodeHash = "0xdeadbeef" + +// mintSignFor 测试用 sign 计算 helper —— 与 peripheral.SignURL 同语义, +// 保证 verifySignOrFail 通过。生产 URL sign 由前端加密生成。 +func mintSignFor(codeHash string) string { + return peripheral.SignURL(codeHash) +} + // setupAssetWithPeripheral 准备测试用 asset + peripheral_info // // ★ 复用包内共享 helper(createServiceTestStar / createServiceTestUser / createServiceTestAsset), @@ -30,6 +47,15 @@ func setupAssetWithPeripheral(t *testing.T, db *gorm.DB) (assetID, starID int64) asset.ID).Error; err != nil { t.Fatalf("Failed to create test peripheral_info: %v", err) } + if err := db.Exec("DELETE FROM peripheral_verify_code WHERE code = ? OR code_hash = ?", "PERI-2026-001", peripheralTestCodeHash).Error; err != nil { + t.Fatalf("Failed to clear test peripheral_verify_code: %v", err) + } + if err := db.Exec(`INSERT INTO peripheral_verify_code + (code, code_hash, peripheral_info_id, status, created_by, created_at, updated_at) + SELECT 'PERI-2026-001', ?, id, 1, 'service-test', 1, 1 + FROM peripheral_info WHERE asset_id = ?`, peripheralTestCodeHash, asset.ID).Error; err != nil { + t.Fatalf("Failed to create test peripheral_verify_code: %v", err) + } return asset.ID, star.StarID } @@ -132,7 +158,7 @@ func TestPeripheralService_MintFromPeripheral_Success(t *testing.T) { // 清理 ownerUID 旧记录(避免残留影响) defer db.Exec("DELETE FROM asset_registry WHERE owner_uid = ?", ownerUID) - result, err := svc.MintFromPeripheral(context.Background(), ownerUID, "PERI-2026-001") + result, err := svc.MintFromPeripheralByHash(context.Background(), ownerUID, peripheralTestCodeHash, mintSignFor(peripheralTestCodeHash)) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -172,7 +198,7 @@ func TestPeripheralService_MintFromPeripheral_AlreadyAdded(t *testing.T) { } defer db.Exec("DELETE FROM asset_registry WHERE owner_uid = ?", ownerUID) - _, err := svc.MintFromPeripheral(context.Background(), ownerUID, "PERI-2026-001") + _, err := svc.MintFromPeripheralByHash(context.Background(), ownerUID, peripheralTestCodeHash, mintSignFor(peripheralTestCodeHash)) var bizErr *BizError if !errors.As(err, &bizErr) { t.Fatalf("expected BizError, got %T: %v", err, err) @@ -212,7 +238,7 @@ func TestPeripheralService_MintFromPeripheral_RateLimited(t *testing.T) { defer db.Exec("DELETE FROM asset_registry WHERE owner_uid = ?", ownerUID) defer db.Exec("DELETE FROM assets WHERE id IN ?", cleanupAssets) - _, err := svc.MintFromPeripheral(context.Background(), ownerUID, "PERI-2026-001") + _, err := svc.MintFromPeripheralByHash(context.Background(), ownerUID, peripheralTestCodeHash, mintSignFor(peripheralTestCodeHash)) var bizErr *BizError if !errors.As(err, &bizErr) { t.Fatalf("expected BizError, got %T: %v", err, err) @@ -231,7 +257,7 @@ func TestPeripheralService_MintFromPeripheral_AssetNotFound(t *testing.T) { repo := repository.NewPeripheralRepository(db) svc := NewPeripheralService(repo) - _, err := svc.MintFromPeripheral(context.Background(), 999, "999999999999") + _, err := svc.MintFromPeripheralByHash(context.Background(), 999, "999999999999", mintSignFor("999999999999")) var bizErr *BizError if !errors.As(err, &bizErr) { t.Fatalf("expected BizError, got %T: %v", err, err) @@ -255,7 +281,7 @@ func TestPeripheralService_MintFromPeripheral_PeripheralInfoMissing(t *testing.T user := createServiceTestUser(t, db, "19900099003") _ = createServiceTestAsset(t, db, user.ID, star.StarID, "p-mint-noinfo") - _, err := svc.MintFromPeripheral(context.Background(), 999, "PERI-2026-001") + _, err := svc.MintFromPeripheralByHash(context.Background(), 999, "PERI-2026-001", mintSignFor("PERI-2026-001")) var bizErr *BizError if !errors.As(err, &bizErr) { t.Fatalf("expected BizError, got %T: %v", err, err) @@ -263,4 +289,108 @@ func TestPeripheralService_MintFromPeripheral_PeripheralInfoMissing(t *testing.T if bizErr.Code != BizCodeAssetNotFound { t.Errorf("expected code=%d, got %d", BizCodeAssetNotFound, bizErr.Code) } -} \ No newline at end of file +} + +// TestPeripheralService_MintFromPeripheral_RateLimitAtomic verifies that the +// Redis Lua limiter admits exactly 10 of 11 concurrent doMint calls for one owner. +func TestPeripheralService_MintFromPeripheral_RateLimitAtomic(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + miniRedis, err := miniredis.Run() + if err != nil { + t.Fatalf("start miniredis: %v", err) + } + defer miniRedis.Close() + + rdb := redis.NewClient(&redis.Options{Addr: miniRedis.Addr()}) + defer rdb.Close() + limiter := NewMintRateLimiter(rdb, 10, 24*time.Hour) + svc := NewPeripheralServiceWithLimiter(repository.NewPeripheralRepository(db), limiter) + + star := createServiceTestStar(t, db, "test_peripheral_mint_atomic_limit") + user := createServiceTestUser(t, db, "19900077010") + + var wg sync.WaitGroup + var succeeded, rateLimited, other atomic.Int32 + for i := 0; i < 11; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, mintErr := svc.doMint(context.Background(), user.ID, &models.PeripheralInfo{ + StarID: star.StarID, + Code: fmt.Sprintf("PERI-ATOMIC-%02d", i), + Image: fmt.Sprintf("https://example.com/peripheral/%d.jpg", i), + Brand: "Brand", + Company: "Company", + Hash: fmt.Sprintf("hash-%02d", i), + }) + if mintErr == nil { + succeeded.Add(1) + return + } + var bizErr *BizError + if errors.As(mintErr, &bizErr) && bizErr.Code == BizCodeRateLimited { + rateLimited.Add(1) + return + } + other.Add(1) + }(i) + } + wg.Wait() + + if gotOK, gotLimited, gotOther := succeeded.Load(), rateLimited.Load(), other.Load(); gotOK != 10 || gotLimited != 1 || gotOther != 0 { + t.Fatalf("want succeeded=10 rateLimited=1 other=0, got succeeded=%d rateLimited=%d other=%d", gotOK, gotLimited, gotOther) + } +} + +// TestPeripheralService_MintFromPeripheral_RedisFailureFallsBackToDB verifies +// that a Redis outage uses the existing DB count and rejects the 11th mint. +func TestPeripheralService_MintFromPeripheral_RedisFailureFallsBackToDB(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + miniRedis, err := miniredis.Run() + if err != nil { + t.Fatalf("start miniredis: %v", err) + } + addr := miniRedis.Addr() + miniRedis.Close() + + rdb := redis.NewClient(&redis.Options{ + Addr: addr, + DialTimeout: 100 * time.Millisecond, + ReadTimeout: 100 * time.Millisecond, + WriteTimeout: 100 * time.Millisecond, + MaxRetries: -1, + }) + defer rdb.Close() + limiter := NewMintRateLimiter(rdb, 10, 24*time.Hour) + svc := NewPeripheralServiceWithLimiter(repository.NewPeripheralRepository(db), limiter) + + star := createServiceTestStar(t, db, "test_peripheral_mint_redis_fallback") + user := createServiceTestUser(t, db, "19900077011") + now := time.Now().UnixMilli() + for i := 0; i < 10; i++ { + assetID := user.ID*100 + int64(i) + if err := db.Exec(`INSERT INTO asset_registry (owner_uid, asset_id, star_id, asset_type, status, created_at, updated_at) + VALUES (?, ?, ?, 'peripheral', 1, ?, ?)`, user.ID, assetID, star.StarID, now, now).Error; err != nil { + t.Fatalf("seed DB rate-limit row %d: %v", i, err) + } + } + + _, mintErr := svc.doMint(context.Background(), user.ID, &models.PeripheralInfo{ + StarID: star.StarID, + Code: "PERI-FALLBACK", + Image: "https://example.com/peripheral/fallback.jpg", + Brand: "Brand", + Hash: "hash-fallback", + }) + var bizErr *BizError + if !errors.As(mintErr, &bizErr) { + t.Fatalf("expected BizError from DB fallback, got %T: %v", mintErr, mintErr) + } + if bizErr.Code != BizCodeRateLimited { + t.Fatalf("expected code=%d, got %d", BizCodeRateLimited, bizErr.Code) + } +} diff --git a/backend/services/assetService/service/peripheral_test_helpers_test.go b/backend/services/assetService/service/peripheral_test_helpers_test.go index 86c6f35..ee1b90a 100644 --- a/backend/services/assetService/service/peripheral_test_helpers_test.go +++ b/backend/services/assetService/service/peripheral_test_helpers_test.go @@ -12,17 +12,17 @@ import ( // // 沿用 backend/services/assetService/repository/asset_repository_test.go 的 setupTestDB 风格, // 但因为 helpers 是包内私有,服务层需要自己的副本。注意: -// - DB 配置与 repository 包保持一致(local haihuzhu/admin/top-fans) +// - DB 配置对齐 docker/.env.local(localhost:15432, postgres/123456, top-fans) // - AutoMigrate 包含 PeripheralInfo(peripheral_info 表) // - 修复 local DB 缺 stars 表:raw SQL CREATE TABLE IF NOT EXISTS + ALTER ADD COLUMN IF NOT EXISTS -// + CREATE UNIQUE INDEX,兼容已有"残缺" stars 表的情况 +// - CREATE UNIQUE INDEX,兼容已有"残缺" stars 表的情况 func setupServiceTestDB(t *testing.T) *gorm.DB { t.Helper() config := database.Config{ Host: "localhost", - Port: 5432, - User: "haihuizhu", - Password: "admin", + Port: 15432, + User: "postgres", + Password: "123456", DBName: "top-fans", SSLMode: "disable", TimeZone: "Asia/Shanghai", @@ -68,6 +68,7 @@ func setupServiceTestDB(t *testing.T) *gorm.DB { // 关联数据,以及 identity_id LIKE 'test_service_%' / 'test_peripheral_%' 的明星。 func cleanupServiceTestDB(t *testing.T, db *gorm.DB) { t.Helper() + db.Exec("DELETE FROM asset_registry WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%')") db.Exec("DELETE FROM peripheral_info WHERE asset_id IN (SELECT id FROM assets WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%'))") db.Exec("DELETE FROM asset_likes WHERE user_id IN (SELECT id FROM users WHERE mobile LIKE '199%') OR asset_id IN (SELECT id FROM assets WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%'))") db.Exec("DELETE FROM assets WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%')") @@ -124,4 +125,4 @@ func createServiceTestAsset(t *testing.T, db *gorm.DB, ownerUID, starID int64, n t.Fatalf("Failed to create test asset: %v", err) } return asset -} \ No newline at end of file +} diff --git a/backend/services/galleryService/mq/consumer.go b/backend/services/galleryService/mq/consumer.go index 86c459f..8a0cc7f 100644 --- a/backend/services/galleryService/mq/consumer.go +++ b/backend/services/galleryService/mq/consumer.go @@ -199,31 +199,30 @@ func fetchExhibitionForSettle(ctx context.Context, exhibitionID int64) exhForSet } // isSettled 查 exhibition 是否已 settled。 -// 注:这个方法需要 exhibition 表有 settled 列,这里先简化用 is_processed 字段保持兼容。 -// 真实上线时需执行 migration 加 settled 列。 +// settled_at 非空且大于 0 视为已结算。 func isSettled(ctx context.Context, exhibitionID int64) bool { - var processed bool + var settledAt *int64 err := database.GetDB().Table("public.exhibitions"). - Select("COALESCE(is_processed, false)"). // 暂时复用作 settled + Select("settled_at"). Where("id = ?", exhibitionID). - Scan(&processed).Error + Scan(&settledAt).Error if err != nil { // 查询失败 = 视为未 settled,允许 handler 继续 return false } - return processed + return settledAt != nil && *settledAt > 0 } func markSettled(ctx context.Context, exhibitionID int64) error { return database.GetDB().Table("public.exhibitions"). Where("id = ?", exhibitionID). - Update("is_processed", true).Error + Update("settled_at", time.Now().UnixMilli()).Error } // scanExpiredExhibitions 一次性迁移扫描： // // deleted_at IS NULL -// AND is_processed = false +// AND settled_at IS NULL // AND expire_at < now // // 对每条记录补发 EnqueueExhibitSettled(recovery) 结算任务。 @@ -243,7 +242,7 @@ func scanExpiredExhibitions(ctx context.Context) { nowMs := time.Now().UnixMilli() if err := database.GetDB().Table("public.exhibitions"). Select("id, asset_id, slot_id, COALESCE(occupier_uid,0) AS occupier_uid, COALESCE(occupier_star_id,0) AS occupier_star_id, COALESCE(host_profile_id,0) AS slot_owner_uid, start_time, expire_at"). - Where("deleted_at IS NULL AND is_processed = false AND expire_at < ?", nowMs). + Where("deleted_at IS NULL AND settled_at IS NULL AND expire_at < ?", nowMs). Limit(500). Scan(&rows).Error; err != nil { logger.Logger.Error("expiry scanner: query failed", zap.Error(err)) diff --git a/backend/services/galleryService/service/cleanup_worker.go b/backend/services/galleryService/service/cleanup_worker.go deleted file mode 100644 index 1fda4b6..0000000 --- a/backend/services/galleryService/service/cleanup_worker.go +++ /dev/null @@ -1,374 +0,0 @@ -package service - -import ( - "context" - "log" - "math" - "time" - - "github.com/topfans/backend/pkg/database" - "github.com/topfans/backend/pkg/logger" - "github.com/topfans/backend/services/galleryService/client" - "github.com/topfans/backend/services/galleryService/repository" - "go.uber.org/zap" -) - -// CleanupWorker 清理过期展品的Worker -type CleanupWorker struct { - repo repository.GalleryRepository - assetClient client.AssetRPCClient - userClient client.UserRPCClient - taskClient client.TaskRPCClient - ctx context.Context - cancel context.CancelFunc -} - -// NewCleanupWorker 创建清理Worker实例 -func NewCleanupWorker(repo repository.GalleryRepository, assetClient client.AssetRPCClient, userClient client.UserRPCClient, taskClient client.TaskRPCClient) *CleanupWorker { - ctx, cancel := context.WithCancel(context.Background()) - return &CleanupWorker{ - repo: repo, - assetClient: assetClient, - userClient: userClient, - taskClient: taskClient, - ctx: ctx, - cancel: cancel, - } -} - -// Start 启动清理Worker -func (w *CleanupWorker) Start() { - log.Println("清理Worker已启动，每小时扫描一次过期展品") - - ticker := time.NewTicker(1 * time.Minute) - defer ticker.Stop() - - // 立即执行一次清理 - w.cleanup() - - for { - select { - case <-ticker.C: - w.cleanup() - case <-w.ctx.Done(): - log.Println("清理Worker已停止") - return - } - } -} - -// Stop 停止清理Worker -func (w *CleanupWorker) Stop() { - w.cancel() -} - -// cleanup 执行清理逻辑 -func (w *CleanupWorker) cleanup() { - now := time.Now().UnixMilli() - - // 1. 清理过期的展品展示记录 - w.cleanupExpiredExhibitions(now) - - // 2. 清理无效的 display_status（处理手动软删除导致的不一致） - w.cleanupInvalidDisplayStatus() -} - -// cleanupExpiredExhibitions 清理过期的展品展示记录（使用 ZSET 驱动 + 数据库兜底） -func (w *CleanupWorker) cleanupExpiredExhibitions(now int64) { - ctx := context.Background() - - // 1. 先尝试从 ZSET 获取过期展品 - expiredAssetIDs, err := database.GetExpiredAssets(ctx, now) - if err != nil { - log.Printf("从 ZSET 获取过期展品失败: %v，降级到数据库查询", err) - w.cleanupExpiredExhibitionsFromDB(now) - return - } - - // 2. ZSET 有数据时处理 ZSET 中的过期展品 - if len(expiredAssetIDs) > 0 { - log.Printf("ZSET 发现 %d 个过期展品，开始清理", len(expiredAssetIDs)) - w.cleanupAssetsFromZSET(ctx, expiredAssetIDs, now) - } - - // 3. 兜底：检查数据库中可能遗漏的过期展品（ZSET 可能在 Redis 重启或数据丢失时漏掉） - w.cleanupExpiredExhibitionsFromDB(now) -} - -// cleanupAssetsFromZSET 从 ZSET 处理过期展品 -func (w *CleanupWorker) cleanupAssetsFromZSET(ctx context.Context, expiredAssetIDs []int64, now int64) { - // 批量删除过期记录 - successCount := 0 - failedCount := 0 - - for _, assetID := range expiredAssetIDs { - // 从数据库查询该 asset_id 对应的有效展览 - e, err := w.repo.GetExhibitionByAssetID(assetID) - if err != nil || e == nil { - // 展览不存在或已处理，从 ZSET 移除 - database.RemoveExpiringAsset(ctx, assetID) - continue - } - - // 1. 获取点赞数用于计算收益 - likeCount := 0 - if w.assetClient != nil { - likeCount = w.assetClient.GetAssetLikeCount(e.AssetID) - } - - // 2. 计算展示收益 - revenue := calculateExhibitionRevenue(likeCount, e.StartTime, now) - - logger.Logger.Info("计算展出收益", - zap.Int64("exhibition_id", e.ID), - zap.Int64("asset_id", e.AssetID), - zap.Int("like_count", likeCount), - zap.Int64("start_time", e.StartTime), - zap.Int64("end_time", now), - zap.Int64("revenue", revenue)) - - // 3. 调用 TaskService 记录收益 - if w.taskClient != nil { - slotOwnerUID := e.HostProfileID - if ownerUID, err := w.repo.GetSlotOwnerUserID(e.SlotID); err == nil { - slotOwnerUID = ownerUID - } - - _, err := w.taskClient.OnExhibitionCompleted(context.Background(), &client.OnExhibitionCompletedRequest{ - ExhibitionId: e.ID, - AssetId: e.AssetID, - SlotId: e.SlotID, - OccupierUid: e.OccupierUID, - OccupierStarId: e.OccupierStarID, - SlotOwnerUid: slotOwnerUID, - StartTime: e.StartTime, - ExpireAt: now, - CrystalAmount: revenue, - LikeCount: int32(likeCount), - }) - if err != nil { - logger.Logger.Error("调用TaskService记录收益失败", - zap.Int64("exhibition_id", e.ID), - zap.Error(err)) - failedCount++ - continue - } - - // 3.1 调用 TaskService 记录点赞押注收益（失败时跳过 is_processed/ZSET 清理，允许下次重试或 MQ 兜底） - likeBetSucceeded := true - if _, recErr := w.taskClient.RecordLikeBetRevenue(context.Background(), &client.RecordLikeBetRevenueRequest{ - ExhibitionId: e.ID, - AssetId: e.AssetID, - TotalLikes: int64(likeCount), - StartTime: e.StartTime, - ExpireAt: now, - }); recErr != nil { - logger.Logger.Warn("调用TaskService记录点赞押注收益失败，跳过标记已处理以允许重试", - zap.Int64("exhibition_id", e.ID), - zap.Error(recErr)) - likeBetSucceeded = false - } - - if !likeBetSucceeded { - failedCount++ - continue // 不标记 is_processed，不清理 ZSET，下次重试或 MQ 兜底 - } - } - - successCount++ - - // 4. 标记展品已处理（只有展示收益+点赞押注收益都成功才标记） - if err := w.repo.SetExhibitionProcessed(e.ID, true); err != nil { - logger.Logger.Error("标记展品已处理失败", - zap.Int64("exhibition_id", e.ID), - zap.Error(err)) - } - - // 5. 从 ZSET 移除（只有全部处理成功才移除） - database.RemoveExpiringAsset(ctx, assetID) - database.RemoveExpiringAssetFromStar(ctx, e.OccupierStarID, assetID) - - log.Printf("展品已到期并生成领取记录: ExhibitionID=%d, AssetID=%d, SlotID=%d, OccupierUID=%d, Revenue=%d", - e.ID, e.AssetID, e.SlotID, e.OccupierUID, revenue) - } - - log.Printf("ZSET 过期展品清理完成: 成功 %d 个, 失败 %d 个", successCount, failedCount) -} - -// cleanupExpiredExhibitionsFromDB 兜底方案：从数据库查询过期展览 -func (w *CleanupWorker) cleanupExpiredExhibitionsFromDB(now int64) { - expired, err := w.repo.GetExpiredExhibitions(now) - if err != nil { - log.Printf("从数据库获取过期展品失败: %v", err) - return - } - - if len(expired) == 0 { - log.Println("没有过期的展品需要清理") - return - } - - log.Printf("数据库发现 %d 个过期展品，开始清理", len(expired)) - - successCount := 0 - failedCount := 0 - ctx := context.Background() - - for _, e := range expired { - likeCount := 0 - if w.assetClient != nil { - likeCount = w.assetClient.GetAssetLikeCount(e.AssetID) - } - - revenue := calculateExhibitionRevenue(likeCount, e.StartTime, now) - - logger.Logger.Info("计算展出收益", - zap.Int64("exhibition_id", e.ID), - zap.Int64("asset_id", e.AssetID), - zap.Int("like_count", likeCount), - zap.Int64("start_time", e.StartTime), - zap.Int64("end_time", now), - zap.Int64("revenue", revenue)) - - if w.taskClient != nil { - slotOwnerUID := e.HostProfileID - if ownerUID, err := w.repo.GetSlotOwnerUserID(e.SlotID); err == nil { - slotOwnerUID = ownerUID - } - - _, err := w.taskClient.OnExhibitionCompleted(context.Background(), &client.OnExhibitionCompletedRequest{ - ExhibitionId: e.ID, - AssetId: e.AssetID, - SlotId: e.SlotID, - OccupierUid: e.OccupierUID, - OccupierStarId: e.OccupierStarID, - SlotOwnerUid: slotOwnerUID, - StartTime: e.StartTime, - ExpireAt: now, - CrystalAmount: revenue, - LikeCount: int32(likeCount), - }) - if err != nil { - logger.Logger.Error("调用TaskService记录收益失败", - zap.Int64("exhibition_id", e.ID), - zap.Error(err)) - failedCount++ - continue - } - - // 同步调用 TaskService 记录点赞押注收益（失败时跳过 is_processed/ZSET 清理，允许下次重试或 MQ 兜底） - likeBetSucceeded := true - if _, recErr := w.taskClient.RecordLikeBetRevenue(context.Background(), &client.RecordLikeBetRevenueRequest{ - ExhibitionId: e.ID, - AssetId: e.AssetID, - TotalLikes: int64(likeCount), - StartTime: e.StartTime, - ExpireAt: now, - }); recErr != nil { - logger.Logger.Warn("调用TaskService记录点赞押注收益失败，跳过标记已处理以允许重试", - zap.Int64("exhibition_id", e.ID), - zap.Error(recErr)) - likeBetSucceeded = false - } - - if !likeBetSucceeded { - failedCount++ - continue // 不标记 is_processed，不清理 ZSET，下次重试或 MQ 兜底 - } - } - - successCount++ - - if err := w.repo.SetExhibitionProcessed(e.ID, true); err != nil { - logger.Logger.Error("标记展品已处理失败", - zap.Int64("exhibition_id", e.ID), - zap.Error(err)) - } - - // 降级方案也需要清理 ZSET（只有全部处理成功才移除） - database.RemoveExpiringAsset(ctx, e.AssetID) - database.RemoveExpiringAssetFromStar(ctx, e.OccupierStarID, e.AssetID) - - log.Printf("展品已到期并生成领取记录: ExhibitionID=%d, AssetID=%d, SlotID=%d, OccupierUID=%d, Revenue=%d", - e.ID, e.AssetID, e.SlotID, e.OccupierUID, revenue) - } - - log.Printf("过期展品清理完成: 成功 %d 个, 失败 %d 个", successCount, failedCount) -} - -// calculateExhibitionRevenue 计算单次上架收益 -// 设计文档公式： -// R1 = R0 × T × [100% + Buff(n)] -// R0 = 5 水晶/小时 -// T = 上架时长（小时） -// Buff(n) 根据点赞数计算：n<5→0%, 5≤n<10→10%, 10≤n<30→20%, n≥30→30% -func calculateExhibitionRevenue(likeCount int, startTime, endTime int64) int64 { - R0 := int64(5) // 水晶/小时 - - // 计算上架时长（毫秒转小时） - T := (endTime - startTime) / 3600000 - if T <= 0 { - T = 1 // 最少1小时 - } - - // 计算Buff - var buff int - switch { - case likeCount >= 30: - buff = 30 - case likeCount >= 10: - buff = 20 - case likeCount >= 5: - buff = 10 - default: - buff = 0 - } - - // 基础收益 - baseRevenue := R0 * T - - // 应用Buff加成（银行家四舍五入） - // R1 = R0 × T × (100% + Buff) - buffedRevenue := int64(math.Round(float64(baseRevenue) * (100 + float64(buff)) / 100)) - - return buffedRevenue -} - -// cleanupInvalidDisplayStatus 清理无效的 display_status -// 处理手动给 exhibition 添加 deleted_at 或 exhibition 已过期但 display_status 仍为1的情况 -func (w *CleanupWorker) cleanupInvalidDisplayStatus() { - // 获取 display_status=1 但没有有效 exhibition 的资产ID列表 - invalidAssetIDs, err := w.repo.GetAssetsWithInvalidDisplayStatus() - if err != nil { - log.Printf("获取无效 display_status 资产列表失败: %v", err) - return - } - - if len(invalidAssetIDs) == 0 { - log.Println("没有无效的 display_status 需要清理") - return - } - - log.Printf("发现 %d 个无效 display_status，开始清理", len(invalidAssetIDs)) - - successCount := 0 - failedCount := 0 - - for _, assetID := range invalidAssetIDs { - if err := w.repo.UpdateAssetRegistryDisplayStatus(assetID, int32(0)); err != nil { - log.Printf("重置 display_status 失败 (AssetID: %d): %v", assetID, err) - failedCount++ - continue - } - successCount++ - log.Printf("已重置无效 display_status: AssetID=%d", assetID) - } - - log.Printf("无效 display_status 清理完成: 成功 %d 个, 失败 %d 个", successCount, failedCount) -} - -// publishEvent 发布事件（预留接口） -// func (w *CleanupWorker) publishEvent(eventType string, exhibition *models.Exhibition) { -// // TODO: 实现事件发布逻辑 -// // 可以通过 RPC 调用 Task Service 或发送到消息队列 -// } diff --git a/backend/services/taskService/repository/like_bet_repo.go b/backend/services/taskService/repository/like_bet_repo.go index cfe56d5..b002355 100644 --- a/backend/services/taskService/repository/like_bet_repo.go +++ b/backend/services/taskService/repository/like_bet_repo.go @@ -32,8 +32,8 @@ func NewLikeBetRevenueRepository(db *gorm.DB) LikeBetRevenueRepository { } // BatchCreate 批量创建点赞押注收益记录 -// 若 (exhibition_id, like_id) 唯一约束冲突（cleanup_worker 重跑场景），返回 error 但不影响主流程 -// 调用方应在 CleanupWorker 中降级为 warn 日志 +// 依赖 uk_like_bet_unique(exhibition_id, like_id) 唯一约束 + ON CONFLICT DO NOTHING 实现幂等， +// 结算重放/多路径触发时重复记录会被静默忽略，不影响主流程。 func (r *likeBetRevenueRepository) BatchCreate(records []*model.LikeBetRevenueRecord) error { if len(records) == 0 { return nil diff --git a/backend/services/taskService/repository/revenue_repo.go b/backend/services/taskService/repository/revenue_repo.go index 57d2b1c..f77ff00 100644 --- a/backend/services/taskService/repository/revenue_repo.go +++ b/backend/services/taskService/repository/revenue_repo.go @@ -7,6 +7,7 @@ import ( "github.com/topfans/backend/services/taskService/model" "go.uber.org/zap" "gorm.io/gorm" + "gorm.io/gorm/clause" ) type RevenueRepository interface { @@ -27,11 +28,35 @@ func NewRevenueRepository(db *gorm.DB) RevenueRepository { return &revenueRepository{db: db} } +// CreateRevenueRecord 幂等写入收益记录。 +// - (exhibition_id, cycle_start_time) 命中唯一约束 uk_exhibition_revenue_cycle 时， +// ON CONFLICT DO NOTHING 静默忽略，返回既有记录的 ID（保证调用方 +// revenue_service.go:394/528/551 的 createdRecord.ID 始终有效）。 +// - CreatedAt 统一毫秒（settlement plan 批次 1.3）。 func (r *revenueRepository) CreateRevenueRecord(record *model.ExhibitionRevenueRecord) (*model.ExhibitionRevenueRecord, error) { - record.CreatedAt = time.Now().Unix() - if err := r.db.Create(record).Error; err != nil { - logger.Logger.Error("Failed to CreateRevenueRecord", zap.Int64("user_id", record.UserID), zap.Error(err)) - return nil, err + record.CreatedAt = time.Now().UnixMilli() + res := r.db.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "exhibition_id"}, {Name: "cycle_start_time"}}, + DoNothing: true, + }).Create(record) + if res.Error != nil { + logger.Logger.Error("Failed to CreateRevenueRecord", zap.Int64("user_id", record.UserID), zap.Error(res.Error)) + return nil, res.Error + } + // 冲突被忽略（RowsAffected==0 且 ID 未回填）时，回查既有记录，保证调用方拿到有效 ID + if res.RowsAffected == 0 || record.ID == 0 { + var existing model.ExhibitionRevenueRecord + if err := r.db.Where("exhibition_id = ? AND cycle_start_time = ?", record.ExhibitionID, record.CycleStartTime). + First(&existing).Error; err != nil { + logger.Logger.Error("CreateRevenueRecord: conflict fallback lookup failed", + zap.Int64("exhibition_id", record.ExhibitionID), + zap.Error(err)) + return nil, err + } + logger.Logger.Warn("CreateRevenueRecord: duplicate settle ignored, returning existing", + zap.Int64("exhibition_id", record.ExhibitionID), + zap.Int64("existing_id", existing.ID)) + return &existing, nil } return record, nil } @@ -85,7 +110,7 @@ func (r *revenueRepository) ClaimRevenueRecord(id int64, userID int64) (bool, er now := time.Now().Unix() result := r.db.Model(&model.ExhibitionRevenueRecord{}). Where("id = ? AND user_id = ? AND status = ?", id, userID, "claimable"). - Updates(map[string]interface{}{ + Updates(map[string]any{ "status": "claimed", "claimed_at": now, }) diff --git a/backend/services/taskService/repository/revenue_repo_test.go b/backend/services/taskService/repository/revenue_repo_test.go new file mode 100644 index 0000000..86d3a18 --- /dev/null +++ b/backend/services/taskService/repository/revenue_repo_test.go @@ -0,0 +1,81 @@ +package repository + +import ( + "testing" + "time" + + "github.com/topfans/backend/services/taskService/model" +) + +// TestCreateRevenueRecord_Idempotent 验证 CreateRevenueRecord 在重复 (exhibition_id, cycle_start_time) 时幂等。 +// +// 背景：settlement plan Task 1 已加唯一约束 uk_exhibition_revenue_cycle(exhibition_id, cycle_start_time)。 +// 这里要求 CreateRevenueRecord 用 ON CONFLICT DO NOTHING 命中时： +// - 不返回 error +// - 返回既有记录的 ID（保证 revenue_service.go:394/528/551 的 createdRecord.ID 可用） +// +// TestMain + setupTestDB 复用 like_bet_repo_test.go（同一 package）。 +func TestCreateRevenueRecord_Idempotent(t *testing.T) { + db := setupTestDB(t) // 复用 like_bet_repo_test.go 的 setupTestDB + repo := NewRevenueRepository(db) + + exhibitionID := int64(999900001) + defer func() { + if err := db.Where("exhibition_id = ?", exhibitionID).Delete(&model.ExhibitionRevenueRecord{}).Error; err != nil { + t.Logf("Warning: cleanup exhibition_revenue_records (exhibition_id=%d): %v", exhibitionID, err) + } + }() + + cycleStart := time.Now().UnixMilli() + rec := &model.ExhibitionRevenueRecord{ + UserID: -1, + StarID: 87, + ExhibitionID: exhibitionID, + AssetID: 1, + SlotID: 1, + SlotOwnerUID: 1, + SlotType: "exhibition", + CrystalAmount: 10, + CycleStartTime: cycleStart, + CycleEndTime: cycleStart + 3600000, + Status: "claimable", + } + + r1, err := repo.CreateRevenueRecord(rec) + if err != nil { + t.Fatalf("first CreateRevenueRecord failed: %v", err) + } + if r1 == nil || r1.ID == 0 { + t.Fatalf("first CreateRevenueRecord returned invalid record: %+v", r1) + } + if r1.CreatedAt <= 0 { + t.Errorf("first record created_at should be set (millis), got %d", r1.CreatedAt) + } + + // 同样的 (exhibition_id, cycle_start_time) 再插一次 —— 必须幂等 + rec2 := *rec + r2, err := repo.CreateRevenueRecord(&rec2) + if err != nil { + t.Fatalf("second CreateRevenueRecord must not error on duplicate, got: %v", err) + } + if r2 == nil { + t.Fatalf("second CreateRevenueRecord returned nil record") + } + if r1.ID != r2.ID { + t.Fatalf("idempotent: want same id %d, got %d", r1.ID, r2.ID) + } + if r2.ID == 0 { + t.Fatalf("idempotent path must return non-zero ID for callers (revenue_service.go:394/528/551), got 0") + } + + // 验证 DB 里只有 1 行 + var count int64 + if err := db.Model(&model.ExhibitionRevenueRecord{}). + Where("exhibition_id = ? AND cycle_start_time = ?", exhibitionID, cycleStart). + Count(&count).Error; err != nil { + t.Fatalf("count after dup insert failed: %v", err) + } + if count != 1 { + t.Errorf("expected exactly 1 row after duplicate insert, got %d", count) + } +} \ No newline at end of file diff --git a/backend/services/taskService/service/revenue_service.go b/backend/services/taskService/service/revenue_service.go index b5bc4af..36666e1 100644 --- a/backend/services/taskService/service/revenue_service.go +++ b/backend/services/taskService/service/revenue_service.go @@ -38,9 +38,10 @@ type RevenueService interface { } // AssetLevelService 资产等级服务接口（定义在assetService） +// 批次1.2: AddExhibitionHours 增加 sourceID 参数做幂等(sourceID 由调用方提供,如 "exhibition_") type AssetLevelService interface { GetOrCreateRecord(assetID int64) (*models.AssetLevelRecord, error) - AddExhibitionHours(assetID int64, hours int) (string, bool, error) + AddExhibitionHours(assetID int64, hours int, sourceID string) (string, bool, error) CalculateRevenue(assetID int64, likeCount int, startTime, endTime int64, revenueBoostBps int) (int64, error) } @@ -345,9 +346,12 @@ func (s *revenueService) ProcessExhibitionRevenue(ctx context.Context, params Pr return 0, err } + // sourceID 用于资产/用户两级幂等去重 — 同 exhibition 下两级 sourceID 复用同一键, + // 保证 cleanup_worker 重放时既不会双加 user hours 也不会双加 asset hours。 + sourceID := fmt.Sprintf("exhibition_%d", params.ExhibitionID) + // slot_owner 累计时长 — 失败仅日志不重试(已 created revenue record) if s.userRPCClient != nil && params.SlotOwnerUID > 0 { - sourceID := fmt.Sprintf("exhibition_%d", params.ExhibitionID) if _, _, _, err := s.userRPCClient.AddExhibitionHours( ctx, params.SlotOwnerUID, params.OccupierStarID, actualHours, sourceID, ); err != nil { @@ -356,9 +360,9 @@ func (s *revenueService) ProcessExhibitionRevenue(ctx context.Context, params Pr zap.Error(err)) } } - // asset 累计时长 + // asset 累计时长 — 批次1.2: 传 sourceID 做幂等(与 slot_owner 同 exhibition 共用同一键) if s.assetLevelService != nil && params.AssetID > 0 && actualHours > 0 { - if _, upgraded, err := s.assetLevelService.AddExhibitionHours(params.AssetID, int(actualHours)); err != nil { + if _, upgraded, err := s.assetLevelService.AddExhibitionHours(params.AssetID, int(actualHours), sourceID); err != nil { logger.Logger.Warn("asset AddExhibitionHours failed", zap.Int64("asset_id", params.AssetID), zap.Error(err)) @@ -508,9 +512,11 @@ func (s *revenueService) OnExhibitionCompleted(ctx context.Context, req *pb.OnEx zap.Int64("crystal_reward", crystalReward)) } - // 增加资产累计展出时长（资产等级系统） + // 增加资产累计展出时长（资产等级系统）—— 批次1.2: 传 sourceID 做幂等 + // 复用本函数上方已构造的 sourceID(L485),与 slot_owner(user 级)同 exhibition 共用同一键, + // 保证重放/RPC 重试时用户级与资产级都不会双加。 if s.assetLevelService != nil && req.AssetId > 0 && actualHours > 0 { - if newLevel, upgraded, err := s.assetLevelService.AddExhibitionHours(req.AssetId, int(actualHours)); err != nil { + if newLevel, upgraded, err := s.assetLevelService.AddExhibitionHours(req.AssetId, int(actualHours), sourceID); err != nil { logger.Logger.Warn("OnExhibitionCompleted: failed to add exhibition hours to asset level", zap.Int64("asset_id", req.AssetId), zap.Int64("hours", actualHours), diff --git a/backend/services/userService/repository/crystal_idempotency_test.go b/backend/services/userService/repository/crystal_idempotency_test.go new file mode 100644 index 0000000..9226e69 --- /dev/null +++ b/backend/services/userService/repository/crystal_idempotency_test.go @@ -0,0 +1,85 @@ +package repository + +import ( + "testing" + + "github.com/topfans/backend/pkg/models" +) + +// TestUpdateCrystalBalance_IdempotentBySourceID 验证同 (source_id, change_type) 第二次调用 +// 不二次扣费、不二次写流水,newBalance 与首次一致。 +// +// 防止上游 RPC 重试或事务回滚后重放导致的重复扣费。 +// 仅清理本测试 sentinel 行,不调共享 cleanupTestDB 的大范围清理。 +func TestUpdateCrystalBalance_IdempotentBySourceID(t *testing.T) { + db := openExhibitionHoursTestDB(t) + + userRepo := NewUserRepository() + hashedPassword, err := HashPassword("password123") + if err != nil { + t.Fatalf("hash password: %v", err) + } + user := &models.User{Mobile: "13800088001", PasswordHash: hashedPassword, IsActive: true} + if err := userRepo.Create(user); err != nil { + t.Fatalf("create user: %v", err) + } + + star := &models.Star{Name: "test_star_idem_crystal_1", IdentityID: "test_star_idem_crystal_1", IsActive: true} + if err := db.Create(star).Error; err != nil { + t.Fatalf("create star: %v", err) + } + + repo := NewFanProfileRepository() + profile := &models.FanProfile{ + UserID: user.ID, + StarID: star.StarID, + Nickname: "test_nickname_idem_crystal_1", + Level: 1, + IsActive: true, + CrystalBalance: 1000, + } + if err := repo.Create(profile); err != nil { + t.Fatalf("create fan profile: %v", err) + } + + const sourceID = "test_crystal_idem_mint_001" + const changeType = "mint_cost" + + // 清理本次测试可能残留 + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", sourceID) + defer func() { + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", sourceID) + db.Exec("DELETE FROM fan_profiles WHERE user_id = ?", user.ID) + db.Exec("DELETE FROM users WHERE id = ?", user.ID) + db.Exec("DELETE FROM stars WHERE identity_id = ?", "test_star_idem_crystal_1") + }() + + // 第一次调用：-100,余额 1000 → 900 + bal1, err := repo.UpdateCrystalBalance(user.ID, star.StarID, -100, changeType, sourceID, "test mint") + if err != nil { + t.Fatalf("first call: %v", err) + } + if bal1 != 900 { + t.Errorf("first call: newBalance want 900, got %d", bal1) + } + + // 第二次同 sourceID+changeType —— 必须幂等,余额仍 900 + bal2, err := repo.UpdateCrystalBalance(user.ID, star.StarID, -100, changeType, sourceID, "test mint retry") + if err != nil { + t.Fatalf("second call: %v", err) + } + if bal2 != 900 { + t.Errorf("second call: newBalance want 900 (idempotent), got %d", bal2) + } + + // 校验: 只写了一条流水 + var n int64 + if err := db.Model(&models.CrystalTransactionRecord{}). + Where("user_id=? AND star_id=? AND source_id=? AND change_type=?", user.ID, star.StarID, sourceID, changeType). + Count(&n).Error; err != nil { + t.Fatalf("count crystal tx: %v", err) + } + if n != 1 { + t.Errorf("tx rows want 1, got %d", n) + } +} \ No newline at end of file diff --git a/backend/services/userService/repository/fan_profile_repository.go b/backend/services/userService/repository/fan_profile_repository.go index 76821ca..cfa2edb 100644 --- a/backend/services/userService/repository/fan_profile_repository.go +++ b/backend/services/userService/repository/fan_profile_repository.go @@ -409,6 +409,32 @@ func (r *fanProfileRepository) UpdateCrystalBalance(userID, starID int64, delta return 0, errors.New("star_id must be greater than 0") } + // ★ 批次1.4 幂等: 同 (source_id, change_type) 已落账则直接返回当前余额。 + // 防止上游 RPC 重试或事务回滚后重放导致的重复扣费。 + if sourceID != "" { + var existing models.CrystalTransactionRecord + err := r.db.Where("source_id = ? AND change_type = ?", sourceID, changeType). + Order("id DESC").First(&existing).Error + if err == nil { + // 已落账 — 取该 user 当前余额返,不二次扣。 + var profile models.FanProfile + if err := r.db.Where("user_id=? AND star_id=?", userID, starID). + First(&profile).Error; err != nil { + return 0, err + } + logger.Logger.Info("UpdateCrystalBalance idempotent hit", + zap.String("source_id", sourceID), + zap.String("change_type", changeType), + zap.Int64("user_id", userID), + zap.Int64("balance", profile.CrystalBalance), + ) + return profile.CrystalBalance, nil + } + if !errors.Is(err, gorm.ErrRecordNotFound) { + return 0, err + } + } + // 使用事务确保原子性 var newBalance int64 err := r.db.Transaction(func(tx *gorm.DB) error { @@ -533,13 +559,58 @@ func GetLevelCap() int32 { // sourceID: 关联业务ID，用于升级奖励流水的溯源 // 返回: newLevel, levelDelta, crystalReward, error func (r *fanProfileRepository) AddExhibitionHours(userID, starID int64, hours int64, sourceID string) (int32, int32, int64, error) { + if sourceID == "" { + // 旧契约兜底：sourceID 为空时退化为“尽力幂等”（之前已存在重复风险，本次按 defense 记日志并继续） + logger.Logger.Warn("AddExhibitionHours called with empty sourceID, idempotency degraded", + zap.Int64("user_id", userID), zap.Int64("star_id", starID)) + } + var result struct { - OldLevel int32 - NewLevel int32 + OldLevel int32 + NewLevel int32 CrystalReward int64 } err := r.db.Transaction(func(tx *gorm.DB) error { + // 0. 幂等闸口：INSERT log, ON CONFLICT DO NOTHING；冲突即跳过本次累加 + if sourceID != "" { + logRow := &models.ExhibitionHoursLog{ + SourceID: sourceID, + UserID: userID, + StarID: starID, + Hours: hours, + CreatedAt: time.Now().UnixMilli(), + } + res := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "source_id"}}, + DoNothing: true, + }).Create(logRow) + if res.Error != nil { + return res.Error + } + if res.RowsAffected == 0 { + logger.Logger.Info("AddExhibitionHours: duplicate source_id, skip accumulate", + zap.Int64("user_id", userID), + zap.Int64("star_id", starID), + zap.String("source_id", sourceID)) + + // 用一次 SELECT 读回当前等级，让调用方拿到 levelDelta=0。 + var fp models.FanProfile + if err := tx.Select("level").Where("user_id = ? AND star_id = ?", userID, starID).First(&fp).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + // 无 profile —— 等价于 old=new=1。 + result.OldLevel = 1 + result.NewLevel = 1 + return nil + } + return err + } + result.OldLevel = fp.Level + result.NewLevel = fp.Level + return nil + } + } + // 1. 获取或创建累计时长记录 exhibitionHours, err := r.GetOrCreateExhibitionHours(tx, userID, starID) if err != nil { @@ -550,9 +621,9 @@ func (r *fanProfileRepository) AddExhibitionHours(userID, starID int64, hours in now := time.Now().UnixMilli() if err := tx.Model(&models.UserExhibitionHours{}). Where("user_id = ? AND star_id = ?", userID, starID). - Updates(map[string]interface{}{ + Updates(map[string]any{ "total_exhibition_hours": gorm.Expr("total_exhibition_hours + ?", hours), - "updated_at": now, + "updated_at": now, }).Error; err != nil { return err } @@ -588,9 +659,9 @@ func (r *fanProfileRepository) AddExhibitionHours(userID, starID int64, hours in if err != nil { logger.Logger.Warn("Failed to get level up rewards", zap.Int64("user_id", userID), - zap.Int64("star_id", starID), - zap.Int32("level", newLevel), - zap.Error(err)) + zap.Int64("star_id", starID), + zap.Int32("level", newLevel), + zap.Error(err)) } // 计算水晶奖励总额 diff --git a/backend/services/userService/repository/fan_profile_repository_test.go b/backend/services/userService/repository/fan_profile_repository_test.go index 4a93d1a..1f77627 100644 --- a/backend/services/userService/repository/fan_profile_repository_test.go +++ b/backend/services/userService/repository/fan_profile_repository_test.go @@ -1,9 +1,15 @@ package repository import ( + "os" + "strconv" + "strings" "testing" + "github.com/topfans/backend/pkg/database" + "github.com/topfans/backend/pkg/logger" "github.com/topfans/backend/pkg/models" + "gorm.io/gorm" ) func TestFanProfileRepository_Create(t *testing.T) { @@ -490,3 +496,315 @@ func TestFanProfileRepository_GetByUserID_WithInactive(t *testing.T) { t.Errorf("Expected profile1, got profile with ID %d", profiles[0].ID) } } + +// openExhibitionHoursTestDB opens a local DB connection dedicated to the +// exhibition hours idempotency tests. It honors the TEST_DB_DSN environment +// variable for CI overrides and skips the test when the connection fails so +// the suite can run in environments without the local database. +func openExhibitionHoursTestDB(t *testing.T) *gorm.DB { + if logger.Logger == nil { + if err := logger.Init(logger.Config{ServiceName: "user-service-test", Environment: "development", LogLevel: "error"}); err != nil { + t.Fatalf("initialize logger: %v", err) + } + } + + // 默认指向本地开发库；CI 通过 TEST_DB_DSN 覆盖。 + host := "localhost" + port := 15432 + user := "postgres" + password := "123456" + dbName := "top-fans" + if dsn := os.Getenv("TEST_DB_DSN"); dsn != "" { + // 解析 TEST_DB_DSN，仅支持项目惯用的 key=value 空格分隔格式。 + values := parseTestDSN(dsn) + if v, ok := values["host"]; ok { + host = v + } + if v, ok := values["port"]; ok { + if p, err := strconv.Atoi(v); err == nil { + port = p + } + } + if v, ok := values["user"]; ok { + user = v + } + if v, ok := values["password"]; ok { + password = v + } + if v, ok := values["dbname"]; ok { + dbName = v + } + } + + cfg := database.Config{ + Host: host, + Port: port, + User: user, + Password: password, + DBName: dbName, + SSLMode: "disable", + TimeZone: "Asia/Shanghai", + } + + if err := database.Init(cfg); err != nil { + t.Skipf("Skipping test: failed to connect to test database: %v", err) + } + + return database.GetDB() +} + +// parseTestDSN 解析 key=value 空格分隔的 DSN。 +func parseTestDSN(dsn string) map[string]string { + out := make(map[string]string) + for _, kv := range strings.Fields(dsn) { + parts := strings.SplitN(kv, "=", 2) + if len(parts) != 2 { + continue + } + out[parts[0]] = parts[1] + } + return out +} + +func TestAddExhibitionHours_Idempotent(t *testing.T) { + db := openExhibitionHoursTestDB(t) + + userRepo := NewUserRepository() + hashedPassword, err := HashPassword("password123") + if err != nil { + t.Fatalf("hash password: %v", err) + } + user := &models.User{Mobile: "13800000091", PasswordHash: hashedPassword, IsActive: true} + if err := userRepo.Create(user); err != nil { + t.Fatalf("create user: %v", err) + } + + star := &models.Star{Name: "test_star_91", IdentityID: "test_star_91", IsActive: true} + if err := db.Create(star).Error; err != nil { + t.Fatalf("create star: %v", err) + } + + repo := NewFanProfileRepository() + profile := &models.FanProfile{ + UserID: user.ID, + StarID: star.StarID, + Nickname: "test_nickname_91", + Level: 1, + IsActive: true, + } + if err := repo.Create(profile); err != nil { + t.Fatalf("create fan profile: %v", err) + } + + srcID := "test_exhibition_hours_log_91_001" + // 清理本次测试可能残留 + db.Exec("DELETE FROM exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", srcID) + defer func() { + db.Exec("DELETE FROM exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", srcID) + db.Exec("DELETE FROM user_exhibition_hours WHERE user_id = ?", user.ID) + db.Exec("DELETE FROM fan_profiles WHERE user_id = ?", user.ID) + db.Exec("DELETE FROM users WHERE id = ?", user.ID) + db.Exec("DELETE FROM stars WHERE identity_id = ?", "test_star_91") + }() + + // 首次调用应累加 + lvl1, delta1, reward1, err := repo.AddExhibitionHours(user.ID, star.StarID, 5, srcID) + if err != nil { + t.Fatalf("first AddExhibitionHours err: %v", err) + } + if delta1 < 0 { + t.Fatalf("first call: levelDelta should be >= 0, got %d", delta1) + } + _ = reward1 + + // 第二次用相同 sourceID —— 应幂等，levelDelta=0, reward=0, 不再写 log + lvl2, delta2, reward2, err := repo.AddExhibitionHours(user.ID, star.StarID, 5, srcID) + if err != nil { + t.Fatalf("second AddExhibitionHours err: %v", err) + } + if lvl1 != lvl2 { + t.Errorf("want level unchanged, lvl1=%d lvl2=%d", lvl1, lvl2) + } + if delta2 != 0 { + t.Errorf("want levelDelta=0 on dup, got %d", delta2) + } + if reward2 != 0 { + t.Errorf("want crystalReward=0 on dup, got %d", reward2) + } + + // 验证 log 表只有 1 条 + var logCount int64 + if err := db.Model(&models.ExhibitionHoursLog{}).Where("source_id = ?", srcID).Count(&logCount).Error; err != nil { + t.Fatalf("count exhibition hours log: %v", err) + } + if logCount != 1 { + t.Errorf("want exactly 1 log row, got %d", logCount) + } + + // 验证 total_exhibition_hours 只 +5 一次 + var totalHours int64 + if err := db.Model(&models.UserExhibitionHours{}). + Where("user_id = ? AND star_id = ?", user.ID, star.StarID). + Select("total_exhibition_hours").Scan(&totalHours).Error; err != nil { + t.Fatalf("read total exhibition hours: %v", err) + } + if totalHours != 5 { + t.Errorf("want total=5, got %d", totalHours) + } +} + +// TestAddExhibitionHours_Idempotent_LevelUp verifies that crossing a level +// threshold triggers exactly one level-up bonus (crystal + like_bet_count + +// crystal_transaction_records) and a repeated call with the same sourceID +// does not duplicate rewards. +func TestAddExhibitionHours_Idempotent_LevelUp(t *testing.T) { + db := openExhibitionHoursTestDB(t) + + userRepo := NewUserRepository() + hashedPassword, err := HashPassword("password123") + if err != nil { + t.Fatalf("hash password: %v", err) + } + user := &models.User{Mobile: "13800000092", PasswordHash: hashedPassword, IsActive: true} + if err := userRepo.Create(user); err != nil { + t.Fatalf("create user: %v", err) + } + + star := &models.Star{Name: "test_star_92", IdentityID: "test_star_92", IsActive: true} + if err := db.Create(star).Error; err != nil { + t.Fatalf("create star: %v", err) + } + + repo := NewFanProfileRepository() + profile := &models.FanProfile{ + UserID: user.ID, + StarID: star.StarID, + Nickname: "test_nickname_92", + Level: 1, + IsActive: true, + } + if err := repo.Create(profile); err != nil { + t.Fatalf("create fan profile: %v", err) + } + + srcID := "test_exhibition_hours_log_92_001" + db.Exec("DELETE FROM exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", srcID) + defer func() { + db.Exec("DELETE FROM exhibition_hours_log WHERE source_id = ?", srcID) + db.Exec("DELETE FROM crystal_transaction_records WHERE source_id = ?", srcID) + db.Exec("DELETE FROM user_exhibition_hours WHERE user_id = ?", user.ID) + db.Exec("DELETE FROM fan_profiles WHERE user_id = ?", user.ID) + db.Exec("DELETE FROM users WHERE id = ?", user.ID) + db.Exec("DELETE FROM stars WHERE identity_id = ?", "test_star_92") + }() + + // 首次调用：跨过 L1→L2 阈值（level_thresholds L2.max_exhibition_hours=6）， + // 期望 level=2、delta=1、crystalReward>0、like_bet_count+1。 + lvl1, delta1, reward1, err := repo.AddExhibitionHours(user.ID, star.StarID, 6, srcID) + if err != nil { + t.Fatalf("first AddExhibitionHours err: %v", err) + } + if lvl1 != 2 { + t.Errorf("first call: want level=2, got %d", lvl1) + } + if delta1 != 1 { + t.Errorf("first call: want levelDelta=1, got %d", delta1) + } + if reward1 <= 0 { + t.Errorf("first call: want crystalReward>0 on level-up, got %d", reward1) + } + + // 读取第一次后的余额、点赞押注次数、流水条数。 + var ( + balanceAfter1 int64 + likeBetAfter1 int32 + txCountAfter1 int64 + logCountAfter1 int64 + ) + if err := db.Model(&models.FanProfile{}). + Select("crystal_balance", "like_bet_count"). + Where("user_id = ? AND star_id = ?", user.ID, star.StarID). + Row().Scan(&balanceAfter1, &likeBetAfter1); err != nil { + t.Fatalf("read fan profile after first call: %v", err) + } + if err := db.Model(&models.CrystalTransactionRecord{}). + Where("source_id = ?", srcID).Count(&txCountAfter1).Error; err != nil { + t.Fatalf("count crystal tx: %v", err) + } + if err := db.Model(&models.ExhibitionHoursLog{}). + Where("source_id = ?", srcID).Count(&logCountAfter1).Error; err != nil { + t.Fatalf("count exhibition hours log: %v", err) + } + if balanceAfter1 != reward1 { + t.Errorf("first call: want crystal_balance=%d, got %d", reward1, balanceAfter1) + } + if likeBetAfter1 != 1 { + t.Errorf("first call: want like_bet_count=1, got %d", likeBetAfter1) + } + if txCountAfter1 != 1 { + t.Errorf("first call: want exactly 1 crystal tx, got %d", txCountAfter1) + } + if logCountAfter1 != 1 { + t.Errorf("first call: want exactly 1 exhibition_hours_log, got %d", logCountAfter1) + } + + // 第二次用相同 sourceID —— 应幂等：余额、like_bet、流水、log 全部不变。 + lvl2, delta2, reward2, err := repo.AddExhibitionHours(user.ID, star.StarID, 6, srcID) + if err != nil { + t.Fatalf("second AddExhibitionHours err: %v", err) + } + if lvl2 != 2 { + t.Errorf("second call: want level=2 (unchanged), got %d", lvl2) + } + if delta2 != 0 { + t.Errorf("second call: want levelDelta=0, got %d", delta2) + } + if reward2 != 0 { + t.Errorf("second call: want crystalReward=0 on dup, got %d", reward2) + } + + var ( + balanceAfter2 int64 + likeBetAfter2 int32 + txCountAfter2 int64 + logCountAfter2 int64 + totalAfter2 int64 + ) + if err := db.Model(&models.FanProfile{}). + Select("crystal_balance", "like_bet_count"). + Where("user_id = ? AND star_id = ?", user.ID, star.StarID). + Row().Scan(&balanceAfter2, &likeBetAfter2); err != nil { + t.Fatalf("read fan profile after second call: %v", err) + } + if balanceAfter2 != balanceAfter1 { + t.Errorf("second call: want crystal_balance unchanged at %d, got %d", balanceAfter1, balanceAfter2) + } + if likeBetAfter2 != likeBetAfter1 { + t.Errorf("second call: want like_bet_count unchanged at %d, got %d", likeBetAfter1, likeBetAfter2) + } + if err := db.Model(&models.CrystalTransactionRecord{}). + Where("source_id = ?", srcID).Count(&txCountAfter2).Error; err != nil { + t.Fatalf("count crystal tx after second call: %v", err) + } + if txCountAfter2 != txCountAfter1 { + t.Errorf("second call: want crystal tx count unchanged at %d, got %d", txCountAfter1, txCountAfter2) + } + if err := db.Model(&models.ExhibitionHoursLog{}). + Where("source_id = ?", srcID).Count(&logCountAfter2).Error; err != nil { + t.Fatalf("count exhibition hours log after second call: %v", err) + } + if logCountAfter2 != logCountAfter1 { + t.Errorf("second call: want exhibition_hours_log count unchanged at %d, got %d", logCountAfter1, logCountAfter2) + } + if err := db.Model(&models.UserExhibitionHours{}). + Where("user_id = ? AND star_id = ?", user.ID, star.StarID). + Select("total_exhibition_hours").Scan(&totalAfter2).Error; err != nil { + t.Fatalf("read total exhibition hours after second call: %v", err) + } + if totalAfter2 != 6 { + t.Errorf("second call: want total_exhibition_hours=6, got %d", totalAfter2) + } +}