topfans/docs/superpowers/plans/2026-07-21-exhibition-settlement-idempotency.md

展品收益结算幂等 (批次 1.1 + 1.3) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 让展品收益结算幂等——同一展品同一周期只产生一条 exhibition_revenue_records，消除已实测的重复发放（超发 2,525,254 水晶），并统一 created_at 时间单位。

Architecture: 数据层加 UNIQUE(exhibition_id, cycle_start_time) + 写入 ON CONFLICT DO NOTHING，使所有结算路径在 DB 层天然幂等（防御纵深）。活跃结算路径是 MQ (HandleExhibitionSettled → revenue:exhibition 子任务 → RevenueService → revenue_repo.CreateRevenueRecord)；已废弃的 cleanup_worker.go（死代码）已删除。MQ 的 isSettled/markSettled 从复用 is_processed 切到独立 settled_at 列。

Tech Stack: Go 1.25 (go.work 多模块), GORM, PostgreSQL, asynq(MQ), 本地库 top-fans@localhost:15432。

Global Constraints

• Go 组合式，不引入新依赖；沿用 gorm.io/gorm/clause。
• migration 放 backend/migrations/，破坏性 SQL 前 dry-run + 备份；末尾按 CLAUDE.md 规范 setval 同步序列。
• 时间戳统一毫秒 (time.Now().UnixMilli())。
• 不自动 git commit（仓库规矩：需用户明确指示）。步骤里的 commit 命令仅在用户批准后执行。
• 每个任务结束 go build ./...（在对应模块目录）通过。
• 存量数据为测试期脏数据，可清账；但清理脚本仍走 dry-run→确认→执行。

---

前置（已完成）

• ✅ 已删除死代码 backend/services/galleryService/service/cleanup_worker.go（NewCleanupWorker 全仓无调用方，唯一引用是一条注释）。
• ✅ 已修 backend/services/taskService/repository/like_bet_repo.go 中提及已删 CleanupWorker 的过时注释。
• ⚠️ 遗留缺口：cleanupInvalidDisplayStatus（display_status 定时兜底清理）随死 worker 删除；该兜底此前已随死 worker 停摆（非本次回归）。是否需重新装配到 MQ 路径，见 Task 5（记录为独立后续项，不在本 plan 落地）。

---

File Structure

• backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql — 新建。created_at 单位回填 + 历史去重 + 唯一约束 + exhibitions.settled_at 列 + 序列同步。
• backend/services/taskService/repository/revenue_repo.go — 改。CreateRevenueRecord 加 ON CONFLICT DO NOTHING；冲突时回查既有记录返回（保证调用方 createdRecord.ID 不为 0）；CreatedAt 改 UnixMilli()。
• backend/services/galleryService/mq/consumer.go — 改。isSettled/markSettled 从 is_processed 切到 settled_at。
• backend/scripts/fix_exhibition_revenue_dedup.sql — 新建。存量核对/回收超发水晶的 dry-run + 执行脚本（本 plan 只做收益记录去重侧；时长重算属批次 1.2）。

---

Task 1: Migration — created_at 单位 + 历史去重 + 唯一约束 + settled_at

Files:

• Create: backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql

Interfaces:

• Produces: 约束 uk_exhibition_revenue_cycle UNIQUE(exhibition_id, cycle_start_time)；列 exhibitions.settled_at bigint。Task 2/3 依赖它们。

• Step 1: 写 migration（含 dry-run 注释块）

-- 2026_07_21_001_exhibition_revenue_idempotent.sql
-- 批次1.1+1.3：展品收益结算幂等 + created_at 单位统一
-- 执行前请先备份：
--   pg_dump -h <host> -U postgres -t exhibition_revenue_records -t exhibitions <db> > 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;

• Step 2: dry-run 预检（先看将影响多少行，不提交）

Run（对本地 top-fans 库）：

PGPASSWORD=123456 psql -h localhost -p 15432 -U postgres -d top-fans -tA -c "
SELECT
  (SELECT count(*) FROM exhibition_revenue_records WHERE created_at>0 AND created_at<100000000000) AS created_at_to_fix,
  (SELECT count(*) FROM exhibition_revenue_records) - (SELECT count(DISTINCT (exhibition_id,cycle_start_time)) FROM exhibition_revenue_records) AS dup_rows_to_delete;"

Expected: 打印 created_at_to_fix|dup_rows_to_delete（预期约 6013|5501）。人工确认数字合理后再执行 Step 3。

• Step 3: 执行 migration

Run:

PGPASSWORD=123456 psql -h localhost -p 15432 -U postgres -d top-fans -f backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql

Expected: BEGIN ... UPDATE ... DELETE ... ALTER TABLE ... COMMIT，无 error。

• Step 4: 验证约束与单位

Run:

PGPASSWORD=123456 psql -h localhost -p 15432 -U postgres -d top-fans -tA -c "
SELECT
  (SELECT count(*)-count(DISTINCT (exhibition_id,cycle_start_time)) FROM exhibition_revenue_records) AS remaining_dups,
  (SELECT count(*) FROM exhibition_revenue_records WHERE created_at>0 AND created_at<100000000000) AS remaining_seconds,
  (SELECT count(*) FROM information_schema.constraint_column_usage WHERE constraint_name='uk_exhibition_revenue_cycle') AS constraint_cols,
  (SELECT count(*) FROM information_schema.columns WHERE table_name='exhibitions' AND column_name='settled_at') AS has_settled_at;"

Expected: remaining_dups=0、remaining_seconds=0、constraint_cols=2、has_settled_at=1。

• Step 5: Commit（用户批准后）

git add backend/migrations/2026_07_21_001_exhibition_revenue_idempotent.sql
git commit -m "feat(gallery): add exhibition revenue idempotency migration (batch 1.1/1.3)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Task 2: CreateRevenueRecord 幂等 + created_at 毫秒

Files:

• Modify: backend/services/taskService/repository/revenue_repo.go:30-37
• Test: backend/services/taskService/repository/revenue_repo_test.go（新建或追加）

Interfaces:

• Consumes: Task 1 的唯一约束 uk_exhibition_revenue_cycle。

• Produces: CreateRevenueRecord(record) (*model.ExhibitionRevenueRecord, error) —— 冲突时不报错，返回既有记录（ID != 0），保证调用方 revenue_service.go:394/528 的 createdRecord.ID 可用。

• Step 1: 写失败测试（依赖本地库或 sqlite；此处用本地 top-fans）

在 revenue_repo_test.go：

func TestCreateRevenueRecord_Idempotent(t *testing.T) {
    db := testDB(t) // 连接 top-fans 测试库；无则 t.Skip
    repo := NewRevenueRepository(db)
    rec := &model.ExhibitionRevenueRecord{
        UserID: 999001, StarID: 87, ExhibitionID: 999900001, AssetID: 1,
        SlotID: 1, SlotOwnerUID: 1, SlotType: "exhibition",
        CrystalAmount: 10, CycleStartTime: 1780000000000, CycleEndTime: 1780003600000,
        Status: "claimable",
    }
    r1, err := repo.CreateRevenueRecord(rec)
    if err != nil { t.Fatal(err) }
    rec2 := *rec // 同 (exhibition_id, cycle_start_time)
    r2, err := repo.CreateRevenueRecord(&rec2)
    if err != nil { t.Fatalf("second insert must not error: %v", err) }
    if r1.ID != r2.ID { t.Fatalf("want same id (idempotent), got %d vs %d", r1.ID, r2.ID) }
    // cleanup
    db.Exec("DELETE FROM exhibition_revenue_records WHERE exhibition_id=?", rec.ExhibitionID)
}

• Step 2: 跑测试确认失败

Run: cd backend/services/taskService && go test ./repository/ -run TestCreateRevenueRecord_Idempotent -v Expected: FAIL（当前无 ON CONFLICT，第二次插入报 duplicate key 或 id 不同）。

• Step 3: 实现幂等写入

替换 revenue_repo.go 的 CreateRevenueRecord：

func (r *revenueRepository) CreateRevenueRecord(record *model.ExhibitionRevenueRecord) (*model.ExhibitionRevenueRecord, error) {
	record.CreatedAt = time.Now().UnixMilli() // 统一毫秒（原为 Unix() 秒，见批次1.3）
	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 {
			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
}

并确保文件已 import "gorm.io/gorm/clause"。

• Step 4: 跑测试确认通过

Run: cd backend/services/taskService && go test ./repository/ -run TestCreateRevenueRecord_Idempotent -v Expected: PASS（两次返回同一 ID）。

• Step 5: 全模块编译

Run: cd backend/services/taskService && go build ./... Expected: 无错误。

• Step 6: Commit（用户批准后）

git add backend/services/taskService/repository/revenue_repo.go backend/services/taskService/repository/revenue_repo_test.go
git commit -m "fix(task): make CreateRevenueRecord idempotent via ON CONFLICT (batch 1.1)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Task 3: MQ isSettled/markSettled 切到 settled_at

Files:

• Modify: backend/services/galleryService/mq/consumer.go:204-221

Interfaces:

• Consumes: Task 1 的 exhibitions.settled_at 列。

• Produces: 结算幂等判定不再复用 is_processed，消除审计报告 §四 P2「一列两义」。

• Step 1: 改 isSettled

func isSettled(ctx context.Context, exhibitionID int64) bool {
	var settledAt *int64
	err := database.GetDB().Table("public.exhibitions").
		Select("settled_at").
		Where("id = ?", exhibitionID).
		Scan(&settledAt).Error
	if err != nil {
		return false // 查询失败视为未结算，允许 handler 继续（写入侧有唯一约束兜底）
	}
	return settledAt != nil && *settledAt > 0
}

• Step 2: 改 markSettled

func markSettled(ctx context.Context, exhibitionID int64) error {
	return database.GetDB().Table("public.exhibitions").
		Where("id = ?", exhibitionID).
		Update("settled_at", time.Now().UnixMilli()).Error
}

• Step 3: 删掉/更新 L202-203 的过时注释（"先简化用 is_processed / 待 migration 加 settled 列"——migration 已在 Task 1 加）。

• Step 4: 编译

Run: cd backend/services/galleryService && go build ./... Expected: 无错误（确认 time 已 import，consumer.go 顶部已有）。

• Step 5: Commit（用户批准后）

git add backend/services/galleryService/mq/consumer.go
git commit -m "refactor(gallery): use settled_at instead of is_processed for settle idempotency (batch 1.1)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Task 4: 端到端回归 — 结算重放不产生重复

Files: 无（验证任务）

• Step 1: 全 go.work 编译

Run: cd backend && go build ./... Expected: 无错误。

• Step 2: 重放幂等验证（SQL 级）

Run:

PGPASSWORD=123456 psql -h localhost -p 15432 -U postgres -d top-fans -tA -c "
-- 模拟同一展品同周期二次插入应被约束挡下
INSERT INTO exhibition_revenue_records (user_id,star_id,exhibition_id,asset_id,slot_id,slot_owner_uid,slot_type,crystal_amount,cycle_start_time,cycle_end_time,status,created_at)
VALUES (999001,87,999900002,1,1,1,'exhibition',10,1780000000000,1780003600000,'claimable',1780000000000)
ON CONFLICT (exhibition_id,cycle_start_time) DO NOTHING;
INSERT INTO exhibition_revenue_records (user_id,star_id,exhibition_id,asset_id,slot_id,slot_owner_uid,slot_type,crystal_amount,cycle_start_time,cycle_end_time,status,created_at)
VALUES (999001,87,999900002,1,1,1,'exhibition',10,1780000000000,1780003600000,'claimable',1780000000000)
ON CONFLICT (exhibition_id,cycle_start_time) DO NOTHING;
SELECT count(*) AS should_be_1 FROM exhibition_revenue_records WHERE exhibition_id=999900002;
DELETE FROM exhibition_revenue_records WHERE exhibition_id=999900002;"

Expected: should_be_1 = 1。

• Step 3: 回归清单（对照 CLAUDE.md）
  • query_graph callers_of CreateRevenueRecord 两个调用方（revenue_service.go:340/476）仍能拿到有效 createdRecord.ID。
  • like_bet 侧未受影响（其 BatchCreate 已幂等）。
  • MQ HandleExhibitionSettled 逻辑不依赖被删的 worker。

---

Task 5: 遗留项登记（不在本 plan 落地）

• display_status 兜底清理：随死 worker 删除的 cleanupInvalidDisplayStatus 需评估是否重新装配到 MQ 或独立小 worker。登记为独立任务，交由用户决定，不在本 plan 实现（避免夹带扩张）。

---

存量数据回收（超发水晶，批次 1.x 的收益侧）

Task 1 的 migration 已物理去重收益记录。已 claimed 的重复水晶回收口径需与运营确认；本库为测试数据可直接清账。

backend/scripts/fix_exhibition_revenue_dedup.sql（dry-run 版，先只 SELECT）：

-- 统计去重前已被领取的重复超发（供核对）
SELECT count(*) AS extra_claimed_records,
       COALESCE(sum(crystal_amount),0) AS extra_claimed_crystal
FROM (
  SELECT id, crystal_amount,
         row_number() OVER (PARTITION BY exhibition_id, cycle_start_time ORDER BY id) AS rn
  FROM exhibition_revenue_records WHERE status='claimed'
) t WHERE rn > 1;

执行回收（清账）需人工确认后另行编写，遵循备份+事务+序列同步。

---

Self-Review

• Spec 覆盖：批次1.1（唯一约束+ON CONFLICT+收敛入口+settled_at）→ Task 1/2/3；批次1.3（created_at 单位）→ Task 1 Step1 + Task 2 Step3；死 worker 删除 → 前置已完成。累计时长幂等（1.2）、mint（1.4-1.6）不在本 plan（各自独立）。
• Placeholder 扫描：无 TBD；migration/Go 代码均给出完整内容。
• 类型一致性：CreateRevenueRecord 签名与现有一致；冲突分支回查返回 *model.ExhibitionRevenueRecord，调用方 .ID 可用；settled_at 列名与 migration 一致。