// 运维/迁移型 SP 仓储(2 个 copy_* 存储过程的数据访问层)——AC-11。 // // 依据:unified/03-migration/academic-domain-design.md §3.6(迁移类 2 SP)/ §3.8 BUG#3 // / §10.3(事务模板)/ §10.4(游标 → 集合)。 // // 覆盖: // // copy_activity_discountLog_to_usedLog() → ListUsedLogRebuildRows + DeleteAllUsedLogs // + BatchInsertUsedLogs(SP30,dump 762745) // copy_activity_discount_to_signed() → ListSignedReplayRows + UpdateSignedFromReplay // (SP31,dump 762792) // // **本仓储有意不继承 BaseRepo**(设计 §1.7 例外的运维延伸,评审必看): // - SP30 的业务语义是"**全表重建**"(dump 用 `TRUNCATE t_courses_activity_used_log` 清空整表), // 按机构/校区裁剪会把其它租户的流水删掉/漏建 —— 作用域过滤在此**语义上不成立**; // - SP31 同理,dump 的游标全表扫描 `t_courses_activity_user_discount`。 // - 越权防线改为:端点 61~63 **仅 super_admin**(设计 §4.5)+ Service 层 `assertSuperAdmin()` 断言; // 且本仓储只被 MaintenanceService 调用,禁止被业务链路复用(需租户裁剪时请走 ActivityRepo)。 // // 事务纪律:删除 + 批量插入**必须同一 tx**(BUG#3 修正的核心:TRUNCATE 是 DDL,隐式提交, // 中途失败即数据全丢;`DELETE + INSERT` 在事务内失败会整体回滚,旧数据完好)。 package repository import ( "context" "strings" "time" "gorm.io/gorm" "yuansiwei/unified-backend/internal/academic/model" infraerrors "yuansiwei/unified-backend/internal/infra/errors" ) // MaintenanceRepo 迁移/修复型 SP 的专用仓储(**全局运维视角,不继承 BaseRepo**,见文件头)。 type MaintenanceRepo struct { // DB 数据库句柄(连接池由 infra/db 统一建立;nil 时任何查询返回 ErrInternal)。 DB *gorm.DB } // NewMaintenanceRepo 构造运维仓储。 // // 入参:db *gorm.DB——连接池句柄(必填)。出参:*MaintenanceRepo。 func NewMaintenanceRepo(db *gorm.DB) *MaintenanceRepo { return &MaintenanceRepo{DB: db} } // db 取查询句柄(ctx 内含 tx 时复用同一事务)。 // // 入参:ctx context.Context(可 nil)。出参:(*gorm.DB, error);未装配 DB → ErrInternal。 func (r *MaintenanceRepo) db(ctx context.Context) (*gorm.DB, error) { if r == nil || r.DB == nil { return nil, infraerrors.WithCause(infraerrors.ErrInternal, errString("repository: MaintenanceRepo.DB 为 nil")) } if tx := TxFromContext(ctx); tx != nil { return tx, nil } if ctx == nil { return r.DB, nil } return r.DB.WithContext(ctx), nil } // UsedLogRebuildRow 重建 used_log 的一行源数据(discount_log 左连 discount 池)。 // // 字段(**可空**:dump 是 LEFT JOIN,池行缺失时会取出 NULL): // - ActivityID *int64:活动 ID(来自池;LEFT JOIN 未命中 → nil); // - UserID *string:用户(来自池;未命中 → nil); // - CoursesStudentID *int64:报课记录 ID(新模型回填列;迁移期可能 NULL); // - CoursesStudentNo *string:报课记录旧单号(旧 UUID,= t_courses_student.biz_no); // - DiscountAmountUsed string:变动金额(元字符串;负=消费/正=退回); // - CreateTime *time.Time:**业务时间**(使用时间,§5.4 ① 类,逐值比对); // - Memo *string:备注。 type UsedLogRebuildRow struct { ActivityID *int64 `gorm:"column:activity_id"` UserID *string `gorm:"column:user_id"` CoursesStudentID *int64 `gorm:"column:courses_student_id"` CoursesStudentNo *string `gorm:"column:courses_student_no"` DiscountAmountUsed string `gorm:"column:discount_amount_used"` CreateTime *time.Time `gorm:"column:create_time"` Memo *string `gorm:"column:memo"` } // ListUsedLogRebuildRows 取 used_log 全量重建的源数据(SP30 游标集合化)。 // // 入参:ctx 请求上下文。 // 出参:[]UsedLogRebuildRow(**按 discount_log.id 升序** —— dump 游标 `ORDER BY t1.id`,顺序不可改);ErrDBError。 // // dump 原文(762754): // // SELECT t2.activityid,t2.userid,t1.courses_studentid,t1.discount_amount_used,t1.create_time,t1.memo // FROM t_courses_activity_user_discount_log t1 // LEFT JOIN t_courses_activity_user_discount t2 ON t1.activity_user_discountid=t2.id // ORDER BY t1.id; // // 1:N 说明:discount_log → 池 是 N:1(池 id 唯一),故不会出现同一流水被取多次。 func (r *MaintenanceRepo) ListUsedLogRebuildRows(ctx context.Context) ([]UsedLogRebuildRow, error) { db, err := r.db(ctx) if err != nil { return nil, err } rows := make([]UsedLogRebuildRow, 0, 256) q := db.Table("t_courses_activity_user_discount_log AS dlog"). Select("pool.activity_id AS activity_id, pool.user_id AS user_id, " + "dlog.courses_student_id AS courses_student_id, dlog.courses_student_no AS courses_student_no, " + "dlog.discount_amount_used AS discount_amount_used, dlog.create_time AS create_time, dlog.memo AS memo"). Joins("LEFT JOIN t_courses_activity_user_discount AS pool ON pool.id = dlog.activity_user_discount_id"). Order("dlog.id ASC") if err := q.Find(&rows).Error; err != nil { return nil, infraerrors.WithCause(infraerrors.ErrDBError, err) } return rows, nil } // DeleteAllUsedLogs 清空 used_log(**替代 dump 的 TRUNCATE**,BUG#3 修正)。 // // 入参:ctx;tx(**必传**:与后续批量插入同事务,失败整体回滚)。 // 出参:int64 删除行数;ErrDBError。 // // 为什么必须是 DELETE 而不是 TRUNCATE(设计 §3.8 BUG#3): // - TRUNCATE 是 DDL,MySQL 会**隐式提交**当前事务 → "清空成功、重建失败"时数据直接丢失, // 且事务回滚也救不回来; // - DELETE 走 MVCC/undo,与 INSERT 同事务 → 任一失败回滚后旧数据完好(可重跑)。 // // `Where("1 = 1")`:GORM 默认拒绝无条件的全表 Delete(安全护栏),语义上等价 `DELETE ... WHERE 1=1`。 func (r *MaintenanceRepo) DeleteAllUsedLogs(ctx context.Context, tx *gorm.DB) (int64, error) { if err := requireTx(tx, "DeleteAllUsedLogs"); err != nil { return 0, err } res := tx.WithContext(ctxOrBackground(ctx)).Where("1 = 1").Delete(&model.CoursesActivityUsedLog{}) if res.Error != nil { return 0, infraerrors.WithCause(infraerrors.ErrDBError, res.Error) } return res.RowsAffected, nil } // BatchInsertUsedLogs 批量写入 used_log(单条 `INSERT ... VALUES (...),(...)`,每批 200 行)。 // // 入参:ctx;tx(**必传**);rows 待插入行(调用方保证 ActivityID>0、UserID 非空)。 // 出参:int64 插入行数;ErrDBError。 // // 红线:`activity_money_used` **原值写入**(dump 的复制作业不做符号翻转;老带新结算才写负值), // 双跑时该列是重点比对项(§3.8 表格 ★)。 func (r *MaintenanceRepo) BatchInsertUsedLogs(ctx context.Context, tx *gorm.DB, rows []*model.CoursesActivityUsedLog) (int64, error) { if err := requireTx(tx, "BatchInsertUsedLogs"); err != nil { return 0, err } if len(rows) == 0 { return 0, nil } for _, row := range rows { if row == nil || row.ActivityID <= 0 || strings.TrimSpace(row.UserID) == "" { return 0, infraerrors.WithCause(infraerrors.ErrBadParam, errString("repository: BatchInsertUsedLogs 出现 activity_id/user_id 缺失的行")) } } if err := tx.WithContext(ctxOrBackground(ctx)).CreateInBatches(rows, 200).Error; err != nil { return 0, infraerrors.WithCause(infraerrors.ErrDBError, err) } return int64(len(rows)), nil } // SignedReplayRow signed 重放的一行源数据(discount 池内连 balance_request 审批通过行)。 // // 字段: // - PoolID int64:t_courses_activity_user_discount.id(游标排序键,**改动顺序会改变"后写覆盖前写"的终态**); // - ActivityID int64:活动 ID;UserID string:用户; // - Income string:充值金额;DiscountAmount string:赠送额;DiscountAmountUsed string:已用赠送额; // - RequestNo *string:工单业务单号(旧 requestid → t_user_balance_request.biz_no); // - ResponseUserID *string:审核人(ULID);ResponseTime *time.Time:审核时间;ResponseMsg *string:审核意见。 type SignedReplayRow struct { PoolID int64 `gorm:"column:pool_id"` ActivityID int64 `gorm:"column:activity_id"` UserID string `gorm:"column:user_id"` Income string `gorm:"column:income"` DiscountAmount string `gorm:"column:discount_amount"` DiscountAmountUsed string `gorm:"column:discount_amount_used"` RequestNo *string `gorm:"column:request_no"` ResponseUserID *string `gorm:"column:response_user_id"` ResponseTime *time.Time `gorm:"column:response_time"` ResponseMsg *string `gorm:"column:response_msg"` } // ListSignedReplayRows 取 signed 重放的源数据(SP31 游标集合化)。 // // 入参:ctx 请求上下文。 // 出参:[]SignedReplayRow(**按 pool.id 升序** —— dump 游标无 ORDER BY,MySQL 实际按 PK 序返回; // 该顺序决定"同一 (activity,user) 多次充值谁覆盖谁",因此**必须固定为 id 升序**);ErrDBError。 // // dump 原文(762800): // // SELECT t1.activityid,t1.userid,t1.income,t1.discount_amount,t1.discount_amount_used, // t1.balance_requestid,t2.response_userid,t2.response_time,t2.response_msg // FROM t_courses_activity_user_discount t1 // INNER JOIN t_user_balance_request t2 // ON t1.balance_requestid=t2.id AND t2.aproved=1 AND t2.type=100; // // 等价改写说明:`balance_requestid`(旧 varchar UUID)→ 新模型 `balance_request_id`(int64 FK), // 而 signed 表要落的是 `request_no`(业务单号)→ 取 `req.biz_no`(旧 UUID 原样,§2.2 拍板 11-10)。 func (r *MaintenanceRepo) ListSignedReplayRows(ctx context.Context) ([]SignedReplayRow, error) { db, err := r.db(ctx) if err != nil { return nil, err } rows := make([]SignedReplayRow, 0, 256) q := db.Table("t_courses_activity_user_discount AS pool"). Select("pool.id AS pool_id, pool.activity_id AS activity_id, pool.user_id AS user_id, "+ "pool.income AS income, pool.discount_amount AS discount_amount, "+ "pool.discount_amount_used AS discount_amount_used, req.biz_no AS request_no, "+ "req.response_user_id AS response_user_id, req.response_time AS response_time, "+ "req.response_msg AS response_msg"). Joins("INNER JOIN t_user_balance_request AS req ON req.id = pool.balance_request_id "+ "AND req.approved = ? AND req.type = ?", model.BalanceRequestApprovedPassed, model.BalanceRequestTypeRecharge). Order("pool.id ASC") if err := q.Find(&rows).Error; err != nil { return nil, infraerrors.WithCause(infraerrors.ErrDBError, err) } return rows, nil } // SignedReplayPatch 重放写入 signed 的列(dump 的 UPDATE SET 清单,7 列)。 // // 字段:RechargeAmount 充值金额;ActivityMoney 活动金额;ActivityMoneyUsed 已用活动金额; // RequestNo 申请单号;ResponseUserID 审核人;ResponseTime 审核时间;ResponseMsg 审核意见。 type SignedReplayPatch struct { RechargeAmount string ActivityMoney string ActivityMoneyUsed string RequestNo *string ResponseUserID *string ResponseTime *time.Time ResponseMsg *string } // UpdateSignedFromReplay 按(活动 + 用户)回写 signed 的 7 列(**仅 status > 0 的行**)。 // // 入参:ctx;tx(**必传**);activityID(>0);userID(非空);patch 列值。 // 出参:int64 受影响行数(0 = 该用户已取消报名 status=0 → dump 不更新,属正常);ErrBadParam / ErrDBError。 // // dump 原文(762812): // // UPDATE t_courses_activity_signed SET recharge_amount=?, activity_money=?, activity_money_used=?, // requestid=?, response_userid=?, response_time=?, response_msg=? // WHERE activityid=? AND userid=? AND status>0; // // 红线:`status > 0` 条件**不可省** —— 已取消(status=0)的报名不能被历史工单"复活", // 否则会把用户已撤销的赠送额度重新算回可用池(资金口径污染)。 func (r *MaintenanceRepo) UpdateSignedFromReplay(ctx context.Context, tx *gorm.DB, activityID int64, userID string, patch SignedReplayPatch) (int64, error) { if err := requireTx(tx, "UpdateSignedFromReplay"); err != nil { return 0, err } if activityID <= 0 || strings.TrimSpace(userID) == "" { return 0, infraerrors.WithCause(infraerrors.ErrBadParam, errString("repository: UpdateSignedFromReplay 需要 activity_id/user_id")) } fields := map[string]any{ "recharge_amount": patch.RechargeAmount, "activity_money": patch.ActivityMoney, "activity_money_used": patch.ActivityMoneyUsed, } // 可空列:NULL 也要能覆盖(dump 直接把 selectXxx 写进去,可能为 NULL) fields["request_no"] = patch.RequestNo fields["response_user_id"] = patch.ResponseUserID fields["response_time"] = patch.ResponseTime fields["response_msg"] = patch.ResponseMsg res := tx.WithContext(ctxOrBackground(ctx)).Model(&model.CoursesActivitySigned{}). Where("activity_id = ? AND user_id = ? AND status > 0", activityID, strings.TrimSpace(userID)). Updates(fields) if res.Error != nil { return 0, infraerrors.WithCause(infraerrors.ErrDBError, res.Error) } return res.RowsAffected, nil }