feat(员工代收款): 新增员工代收款账单闭环

- 新增 6 张表与成对迁移 000212，扩展企业微信审批场景业务类型白名单 - 后台线下套餐订单与两条代理线下充值入账路径在来源成功事务内建账，来源唯一键幂等 - 核销申请、审批尝试记录、账单分摊预占与驳回重提，审批业务类型 employee_collection_approval - 企业微信终态消费幂等：通过转已核销、驳回释放预占、通过后撤销不回滚并转异常终态 - 退款成功事务内按 bill_id+refund_id 幂等冲销账单或仅写退款关联提示 - 线下收款方式字典、账单查询/统计/关闭、申请查询与代办权限，均写入事务内审计 OpenSpec Change: add-employee-collection-bills
2026-09-10 18:24:05 +08:00
parent dc4e0d4103
commit ce24d5612e
54 changed files with 5975 additions and 465 deletions
--- a/internal/application/employeecollection/approval_decision.go
+++ b/internal/application/employeecollection/approval_decision.go
@@ -0,0 +1,501 @@
+package employeecollection
+
+import (
+	"context"
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	approvalapp "github.com/break/junhong_cmp_fiber/internal/application/approval"
+	"github.com/break/junhong_cmp_fiber/internal/model"
+	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+	"github.com/break/junhong_cmp_fiber/pkg/errors"
+)
+
+// ApprovalDecisionHandler 将渠道无关企业微信审批终态应用到员工代收款核销申请。
+// 通过才增加账单已核销金额，驳回才释放预占；重复、乱序或延迟回调都不重复入账。
+type ApprovalDecisionHandler struct {
+	db    *gorm.DB
+	audit ApplicationAuditWriter
+}
+
+// NewApprovalDecisionHandler 创建员工代收款核销审批终态消费者。
+func NewApprovalDecisionHandler(db *gorm.DB, audit ApplicationAuditWriter) *ApprovalDecisionHandler {
+	return &ApprovalDecisionHandler{db: db, audit: audit}
+}
+
+// Handle 幂等消费标准审批终态。
+// 业务标识为审批尝试记录主键：先锁定尝试记录并校验审批实例一致，再按申请与账单 ID 升序加锁。
+func (h *ApprovalDecisionHandler) Handle(ctx context.Context, event approvalapp.TerminalDecisionEvent) error {
+	if h == nil || h.db == nil || h.audit == nil {
+		return errors.New(errors.CodeInternalError, "员工代收款核销审批终态能力未配置")
+	}
+	if event.BusinessType != constants.ApprovalBusinessTypeEmployeeCollection || event.BusinessID == 0 || event.InstanceID == 0 {
+		return errors.New(errors.CodeInvalidParam, "员工代收款核销审批终态参数无效")
+	}
+	ctx = auditcontext.With(ctx, auditcontext.Context{CorrelationID: event.CorrelationID, ParentEventID: event.EventID})
+	return h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var attempt model.EmployeeCollectionApplicationAttempt
+		if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).
+			First(&attempt, event.BusinessID).Error; err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return errors.New(errors.CodeNotFound, "核销审批尝试记录不存在")
+			}
+			return errors.Wrap(errors.CodeDatabaseError, err, "锁定核销审批尝试记录失败")
+		}
+		if attempt.ApprovalInstanceID == nil || *attempt.ApprovalInstanceID != event.InstanceID {
+			return errors.New(errors.CodeConflict, "核销审批尝试记录关联的审批实例不一致")
+		}
+		application, err := lockApplication(ctx, tx, attempt.ApplicationID)
+		if err != nil {
+			return err
+		}
+		if application.LatestAttemptID != attempt.ID {
+			// 已被更新尝试取代的历史尝试终态不再改变申请事实。
+			return nil
+		}
+		switch event.Decision {
+		case constants.ApprovalDecisionApproved:
+			return h.applyApproved(ctx, tx, application, &attempt, event)
+		case constants.ApprovalDecisionRejected:
+			return h.applyClosed(ctx, tx, application, &attempt, event,
+				constants.EmployeeCollectionApplicationStatusRejected, "企业微信审批已驳回")
+		case constants.ApprovalDecisionCancelled:
+			return h.applyClosed(ctx, tx, application, &attempt, event,
+				constants.EmployeeCollectionApplicationStatusRevoked, "企业微信审批已撤销")
+		case constants.ApprovalDecisionDeleted:
+			return h.applyClosed(ctx, tx, application, &attempt, event,
+				constants.EmployeeCollectionApplicationStatusRevoked, "企业微信审批已删除")
+		case constants.ApprovalDecisionRevokedAfterApproved:
+			return h.applyRevoked(ctx, tx, application, &attempt, event)
+		default:
+			return errors.New(errors.CodeInvalidParam, "不支持的核销申请审批终态")
+		}
+	})
+}
+
+// applyApproved 将本次尝试的全部预占分摊转入已核销并重算账单状态。
+// 仅当申请仍处于审批中时推进，重复或乱序回调不重复增加已核销金额。
+func (h *ApprovalDecisionHandler) applyApproved(
+	ctx context.Context,
+	tx *gorm.DB,
+	application *model.EmployeeCollectionApplication,
+	attempt *model.EmployeeCollectionApplicationAttempt,
+	event approvalapp.TerminalDecisionEvent,
+) error {
+	if application.Status != constants.EmployeeCollectionApplicationStatusPending {
+		return nil
+	}
+	allocations, err := loadAttemptAllocations(ctx, tx, attempt.ID)
+	if err != nil {
+		return err
+	}
+	if _, err := lockBillsInAscendingOrder(ctx, tx, allocationCommandsOf(allocations)); err != nil {
+		return err
+	}
+	now := time.Now().UTC()
+	for _, allocation := range allocations {
+		if err := approveAllocation(ctx, tx, allocation, now); err != nil {
+			return err
+		}
+		if err := settleBillReservation(ctx, tx, allocation); err != nil {
+			return err
+		}
+	}
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplication{}).
+		Where("id = ? AND status = ?", application.ID, constants.EmployeeCollectionApplicationStatusPending).
+		Updates(map[string]any{
+			"status":     constants.EmployeeCollectionApplicationStatusApproved,
+			"decided_at": now, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "标记核销申请已通过失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销申请状态已变化")
+	}
+	before := application.Status
+	application.Status = constants.EmployeeCollectionApplicationStatusApproved
+	application.DecidedAt = &now
+	application.Updater = 0
+	bills, err := reloadBills(ctx, tx, allocationCommandsOf(allocations))
+	if err != nil {
+		return err
+	}
+	approvedEventID, err := decisionEventID(application.ID, attempt.AttemptNo, "approved")
+	if err != nil {
+		return err
+	}
+	return h.audit.WriteEmployeeCollectionApplication(ctx, tx, ApplicationAudit{
+		EventID:     approvedEventID,
+		ActionCode:  constants.AuditActionEmployeeCollectionApplicationApproved,
+		Summary:     "企业微信审批通过，核销分摊转入已核销",
+		Application: application, Attempt: attempt, Allocations: allocations, Bills: bills,
+		BeforeData: map[string]any{"status": before},
+		AfterData:  applicationAuditSnapshot(application), CorrelationID: event.CorrelationID,
+	})
+}
+
+// applyClosed 处理最终驳回与渠道撤销、删除：释放全部预占并把申请置为对应终态。
+func (h *ApprovalDecisionHandler) applyClosed(
+	ctx context.Context,
+	tx *gorm.DB,
+	application *model.EmployeeCollectionApplication,
+	attempt *model.EmployeeCollectionApplicationAttempt,
+	event approvalapp.TerminalDecisionEvent,
+	targetStatus int,
+	reason string,
+) error {
+	if application.Status != constants.EmployeeCollectionApplicationStatusPending {
+		return nil
+	}
+	allocations, err := loadAttemptAllocations(ctx, tx, attempt.ID)
+	if err != nil {
+		return err
+	}
+	if _, err := lockBillsInAscendingOrder(ctx, tx, allocationCommandsOf(allocations)); err != nil {
+		return err
+	}
+	now := time.Now().UTC()
+	for _, allocation := range allocations {
+		if err := releaseAllocation(ctx, tx, allocation, now); err != nil {
+			return err
+		}
+		if err := releaseBillReservation(ctx, tx, allocation); err != nil {
+			return err
+		}
+	}
+	terminalReason := ""
+	if targetStatus == constants.EmployeeCollectionApplicationStatusRevoked {
+		terminalReason = reason
+	}
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplication{}).
+		Where("id = ? AND status = ?", application.ID, constants.EmployeeCollectionApplicationStatusPending).
+		Updates(map[string]any{
+			"status": targetStatus, "decided_at": now,
+			"terminal_reason": terminalReason, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新核销申请终态失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销申请状态已变化")
+	}
+	before := application.Status
+	application.Status = targetStatus
+	application.DecidedAt = &now
+	application.TerminalReason = terminalReason
+	application.Updater = 0
+	bills, err := reloadBills(ctx, tx, allocationCommandsOf(allocations))
+	if err != nil {
+		return err
+	}
+	closedEventID, err := decisionEventID(application.ID, attempt.AttemptNo, decisionEventSuffix(event.Decision))
+	if err != nil {
+		return err
+	}
+	return h.audit.WriteEmployeeCollectionApplication(ctx, tx, ApplicationAudit{
+		EventID:     closedEventID,
+		ActionCode:  constants.AuditActionEmployeeCollectionApplicationRejected,
+		Summary:     "企业微信审批未通过，核销申请预占已释放",
+		Application: application, Attempt: attempt, Allocations: allocations, Bills: bills,
+		BeforeData: map[string]any{"status": before},
+		AfterData:  applicationAuditSnapshot(application), CorrelationID: event.CorrelationID,
+	})
+}
+
+// applyRevoked 处理通过后撤销。
+// 申请已通过：不回滚已核销金额，只转异常终态并禁止自动重提。
+// 申请仍在审批中（渠道乱序投递）：释放全部审批中预占并转异常终态，避免预占永久占用账单。
+func (h *ApprovalDecisionHandler) applyRevoked(
+	ctx context.Context,
+	tx *gorm.DB,
+	application *model.EmployeeCollectionApplication,
+	attempt *model.EmployeeCollectionApplicationAttempt,
+	event approvalapp.TerminalDecisionEvent,
+) error {
+	switch application.Status {
+	case constants.EmployeeCollectionApplicationStatusApproved:
+		return h.revokeApproved(ctx, tx, application, attempt, event)
+	case constants.EmployeeCollectionApplicationStatusPending:
+		return h.revokePending(ctx, tx, application, attempt, event)
+	default:
+		// 已驳回、已撤销等终态不再改变事实。
+		return nil
+	}
+}
+
+// revokeApproved 在已通过态撤销：保留已核销金额，仅转异常终态。
+func (h *ApprovalDecisionHandler) revokeApproved(
+	ctx context.Context,
+	tx *gorm.DB,
+	application *model.EmployeeCollectionApplication,
+	attempt *model.EmployeeCollectionApplicationAttempt,
+	event approvalapp.TerminalDecisionEvent,
+) error {
+	now := time.Now().UTC()
+	terminalReason := "企业微信通过后撤销"
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplication{}).
+		Where("id = ? AND status = ?", application.ID, constants.EmployeeCollectionApplicationStatusApproved).
+		Updates(map[string]any{
+			"status":     constants.EmployeeCollectionApplicationStatusRevoked,
+			"decided_at": now, "terminal_reason": terminalReason, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "标记核销申请通过后撤销失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销申请状态已变化")
+	}
+	before := application.Status
+	application.Status = constants.EmployeeCollectionApplicationStatusRevoked
+	application.DecidedAt = &now
+	application.TerminalReason = terminalReason
+	application.Updater = 0
+	allocations, err := loadAttemptAllocations(ctx, tx, attempt.ID)
+	if err != nil {
+		return err
+	}
+	bills, err := reloadBills(ctx, tx, allocationCommandsOf(allocations))
+	if err != nil {
+		return err
+	}
+	revokedEventID, err := decisionEventID(application.ID, attempt.AttemptNo, "revoked")
+	if err != nil {
+		return err
+	}
+	return h.audit.WriteEmployeeCollectionApplication(ctx, tx, ApplicationAudit{
+		EventID:     revokedEventID,
+		ActionCode:  constants.AuditActionEmployeeCollectionApplicationRevoked,
+		Summary:     "企业微信通过后撤销，已核销金额不回滚",
+		Application: application, Attempt: attempt, Allocations: allocations, Bills: bills,
+		BeforeData: map[string]any{"status": before},
+		AfterData:  applicationAuditSnapshot(application), CorrelationID: event.CorrelationID,
+	})
+}
+
+// revokePending 在审批中态撤销：释放全部审批中预占并转异常终态。
+func (h *ApprovalDecisionHandler) revokePending(
+	ctx context.Context,
+	tx *gorm.DB,
+	application *model.EmployeeCollectionApplication,
+	attempt *model.EmployeeCollectionApplicationAttempt,
+	event approvalapp.TerminalDecisionEvent,
+) error {
+	allocations, err := loadAttemptAllocations(ctx, tx, attempt.ID)
+	if err != nil {
+		return err
+	}
+	if _, err := lockBillsInAscendingOrder(ctx, tx, allocationCommandsOf(allocations)); err != nil {
+		return err
+	}
+	now := time.Now().UTC()
+	for _, allocation := range allocations {
+		if err := releaseAllocation(ctx, tx, allocation, now); err != nil {
+			return err
+		}
+		if err := releaseBillReservation(ctx, tx, allocation); err != nil {
+			return err
+		}
+	}
+	terminalReason := "企业微信通过后在本地审批中状态被撤销"
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplication{}).
+		Where("id = ? AND status = ?", application.ID, constants.EmployeeCollectionApplicationStatusPending).
+		Updates(map[string]any{
+			"status":     constants.EmployeeCollectionApplicationStatusRevoked,
+			"decided_at": now, "terminal_reason": terminalReason, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "标记核销申请撤销失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销申请状态已变化")
+	}
+	before := application.Status
+	application.Status = constants.EmployeeCollectionApplicationStatusRevoked
+	application.DecidedAt = &now
+	application.TerminalReason = terminalReason
+	application.Updater = 0
+	bills, err := reloadBills(ctx, tx, allocationCommandsOf(allocations))
+	if err != nil {
+		return err
+	}
+	revokedEventID, err := decisionEventID(application.ID, attempt.AttemptNo, "revoked")
+	if err != nil {
+		return err
+	}
+	return h.audit.WriteEmployeeCollectionApplication(ctx, tx, ApplicationAudit{
+		EventID:     revokedEventID,
+		ActionCode:  constants.AuditActionEmployeeCollectionApplicationRevoked,
+		Summary:     "企业微信通过后撤销，申请仍在审批中，已释放预占",
+		Application: application, Attempt: attempt, Allocations: allocations, Bills: bills,
+		BeforeData: map[string]any{"status": before},
+		AfterData:  applicationAuditSnapshot(application), CorrelationID: event.CorrelationID,
+	})
+}
+
+// loadAttemptAllocations 按账单 ID 升序读取本次尝试的分摊，保证后续加锁与写入次序唯一。
+func loadAttemptAllocations(ctx context.Context, tx *gorm.DB, attemptID uint) ([]*model.EmployeeCollectionApplicationAllocation, error) {
+	var allocations []model.EmployeeCollectionApplicationAllocation
+	if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).
+		Where("attempt_id = ?", attemptID).Order("bill_id ASC, id ASC").Find(&allocations).Error; err != nil {
+		return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询核销审批尝试分摊失败")
+	}
+	if len(allocations) == 0 {
+		return nil, errors.New(errors.CodeConflict, "核销审批尝试记录缺少分摊事实")
+	}
+	result := make([]*model.EmployeeCollectionApplicationAllocation, 0, len(allocations))
+	for index := range allocations {
+		result = append(result, &allocations[index])
+	}
+	return result, nil
+}
+
+// allocationCommandsOf 提取分摊涉及的账单与金额，用于复用升序加锁函数。
+func allocationCommandsOf(allocations []*model.EmployeeCollectionApplicationAllocation) []ApplicationAllocationCommand {
+	items := make([]ApplicationAllocationCommand, 0, len(allocations))
+	for _, allocation := range allocations {
+		items = append(items, ApplicationAllocationCommand{BillID: allocation.BillID, Amount: allocation.Amount})
+	}
+	return items
+}
+
+// approveAllocation 把分摊从审批中预占条件更新为已通过；重复处理时返回冲突。
+func approveAllocation(
+	ctx context.Context,
+	tx *gorm.DB,
+	allocation *model.EmployeeCollectionApplicationAllocation,
+	now time.Time,
+) error {
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplicationAllocation{}).
+		Where("id = ? AND status = ?", allocation.ID, constants.EmployeeCollectionAllocationStatusPending).
+		Updates(map[string]any{
+			"status": constants.EmployeeCollectionAllocationStatusApproved, "released_at": now, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "标记核销分摊已通过失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销分摊状态已变化")
+	}
+	allocation.Status = constants.EmployeeCollectionAllocationStatusApproved
+	allocation.ReleasedAt = &now
+	return nil
+}
+
+// releaseAllocation 把分摊从审批中预占条件更新为已释放；重复处理时返回冲突。
+func releaseAllocation(
+	ctx context.Context,
+	tx *gorm.DB,
+	allocation *model.EmployeeCollectionApplicationAllocation,
+	now time.Time,
+) error {
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionApplicationAllocation{}).
+		Where("id = ? AND status = ?", allocation.ID, constants.EmployeeCollectionAllocationStatusPending).
+		Updates(map[string]any{
+			"status": constants.EmployeeCollectionAllocationStatusReleased, "released_at": now, "updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "释放核销分摊预占失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "核销分摊状态已变化")
+	}
+	allocation.Status = constants.EmployeeCollectionAllocationStatusReleased
+	allocation.ReleasedAt = &now
+	return nil
+}
+
+// settleBillReservation 将账单预占转为已核销并重算账单状态。
+// 条件更新要求账单预占不小于分摊金额，并检查 RowsAffected，避免并发下重复入账。
+func settleBillReservation(
+	ctx context.Context,
+	tx *gorm.DB,
+	allocation *model.EmployeeCollectionApplicationAllocation,
+) error {
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionBill{}).
+		Where("id = ? AND reserved_amount >= ?", allocation.BillID, allocation.Amount).
+		Updates(map[string]any{
+			"received_amount": gorm.Expr("received_amount + ?", allocation.Amount),
+			"reserved_amount": gorm.Expr("reserved_amount - ?", allocation.Amount),
+			"status": gorm.Expr(
+				"CASE WHEN received_amount + ? >= receivable_amount THEN ?::smallint WHEN received_amount + ? > 0 THEN ?::smallint ELSE ?::smallint END",
+				allocation.Amount, constants.EmployeeCollectionBillStatusSettled,
+				allocation.Amount, constants.EmployeeCollectionBillStatusPartial,
+				constants.EmployeeCollectionBillStatusPending,
+			),
+			"updater": 0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "账单预占转入已核销失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "账单预占已变化，核销未入账")
+	}
+	return nil
+}
+
+// releaseBillReservation 释放账单预占金额，条件更新并检查 RowsAffected。
+func releaseBillReservation(
+	ctx context.Context,
+	tx *gorm.DB,
+	allocation *model.EmployeeCollectionApplicationAllocation,
+) error {
+	result := tx.WithContext(ctx).Model(&model.EmployeeCollectionBill{}).
+		Where("id = ? AND reserved_amount >= ?", allocation.BillID, allocation.Amount).
+		Updates(map[string]any{
+			"reserved_amount": gorm.Expr("reserved_amount - ?", allocation.Amount),
+			"updater":         0,
+		})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "释放账单预占失败")
+	}
+	if result.RowsAffected != 1 {
+		return errors.New(errors.CodeConflict, "账单预占已变化")
+	}
+	return nil
+}
+
+// reloadBills 重新读取受影响账单，保证审计快照反映终态金额。
+// 账单行已在同一事务内持有排他锁，这里只做一次按 ID 升序的普通读取。
+func reloadBills(
+	ctx context.Context,
+	tx *gorm.DB,
+	items []ApplicationAllocationCommand,
+) ([]*model.EmployeeCollectionBill, error) {
+	billIDs := make([]uint, 0, len(items))
+	for _, item := range items {
+		billIDs = append(billIDs, item.BillID)
+	}
+	var bills []model.EmployeeCollectionBill
+	if err := tx.WithContext(ctx).Where("id IN ?", billIDs).Order("id ASC").Find(&bills).Error; err != nil {
+		return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询受影响员工代收款账单失败")
+	}
+	result := make([]*model.EmployeeCollectionBill, 0, len(bills))
+	for index := range bills {
+		result = append(result, &bills[index])
+	}
+	return result, nil
+}
+
+// decisionEventID 生成审批终态审计事件的稳定标识，并约束在审计列宽内。
+func decisionEventID(applicationID uint, attemptNo int, suffix string) (string, error) {
+	return composeAuditEventID(
+		"employee_collection", "application", uintText(applicationID), "attempt", intText(attemptNo), suffix)
+}
+
+// decisionEventSuffix 把渠道决策映射为审计事件后缀。
+func decisionEventSuffix(decision string) string {
+	switch decision {
+	case constants.ApprovalDecisionRejected:
+		return "rejected"
+	case constants.ApprovalDecisionCancelled:
+		return "cancelled"
+	case constants.ApprovalDecisionDeleted:
+		return "deleted"
+	default:
+		return "closed"
+	}
+}