Files
junhong_cmp_fiber/internal/task/phone_asset_unbind_import.go
break 70e680eb0a
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m2s
feat(手机号资产关联): AUG26-009 手机号—资产关联、十项上限与后台解绑
- 新增成对迁移 000223(tb_phone_asset_association,含有效关系部分唯一索引与 down 守卫)与 000224(解绑导入任务表),不回填历史
- H5:need_bind_phone 三支判定(开关关闭完全短路);已有主号幂等建联;十项上限按手机号 advisory 串行化(含换绑到全新号的并发场景);换绑原子迁移与冲突整单回滚;不写遗留列
- 后台:关联列表、单项/批量解绑、CSV 导入解绑(B1–B16),超管/平台 gate + 资产数据范围复核,三态统一文案
- 读侧:卡/设备列表与详情按页一次 IN 聚合;两类导出补「关联手机号」列并保留历史表头反解兼容
- 脱敏:关联审计走独立动作/资源只写脱敏手机号;访问日志手机号类字段脱敏
- 同步主 Spec openspec/specs/phone-asset-association 并归档 AUG26-009,补齐 requirement-evidence 与入口矩阵,context-health 通过
2026-09-15 11:54:56 +08:00

435 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package task
import (
"bytes"
"context"
"encoding/csv"
stderrors "errors"
"fmt"
"io"
"os"
"strconv"
"strings"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
"gorm.io/gorm"
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
"github.com/break/junhong_cmp_fiber/internal/model"
assetSvc "github.com/break/junhong_cmp_fiber/internal/service/asset"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
"github.com/break/junhong_cmp_fiber/pkg/constants"
pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/sanitizer"
"github.com/break/junhong_cmp_fiber/pkg/storage"
"github.com/break/junhong_cmp_fiber/pkg/utils"
)
// PhoneAssetUnbindImportPayload 手机号—资产关联 CSV 解绑导入任务载荷。
type PhoneAssetUnbindImportPayload struct {
TaskID uint `json:"task_id"`
}
// PhoneAssetUnbindImportHandler 手机号—资产关联 CSV 解绑导入任务处理器。
// 逐行独立事务:成功行提交、失败行不解除任何关系;任务级失败与行级失败分开记录。
// 行内只填资产标识与可选备注,解绑原因取任务级必填字段。
type PhoneAssetUnbindImportHandler struct {
db *gorm.DB
taskStore *postgres.PhoneAssetUnbindImportTaskStore
associationStore *postgres.PhoneAssetAssociationStore
assetIdentifierStore *postgres.AssetIdentifierStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
storageService *storage.Service
auditWriter *audit.Writer
logger *zap.Logger
}
// NewPhoneAssetUnbindImportHandler 创建解绑导入任务处理器。
func NewPhoneAssetUnbindImportHandler(
db *gorm.DB,
taskStore *postgres.PhoneAssetUnbindImportTaskStore,
associationStore *postgres.PhoneAssetAssociationStore,
assetIdentifierStore *postgres.AssetIdentifierStore,
iotCardStore *postgres.IotCardStore,
deviceStore *postgres.DeviceStore,
storageService *storage.Service,
logger *zap.Logger,
auditWriters ...*audit.Writer,
) *PhoneAssetUnbindImportHandler {
handler := &PhoneAssetUnbindImportHandler{
db: db, taskStore: taskStore, associationStore: associationStore,
assetIdentifierStore: assetIdentifierStore, iotCardStore: iotCardStore,
deviceStore: deviceStore, storageService: storageService, logger: logger,
}
if len(auditWriters) > 0 {
handler.auditWriter = auditWriters[0]
}
return handler
}
// Handle 处理手机号—资产关联 CSV 解绑导入任务。
func (h *PhoneAssetUnbindImportHandler) Handle(ctx context.Context, taskMessage *asynq.Task) error {
var payload PhoneAssetUnbindImportPayload
if err := sonic.Unmarshal(taskMessage.Payload(), &payload); err != nil {
h.logger.Error("解析手机号资产解绑导入任务载荷失败", zap.Error(err))
return asynq.SkipRetry
}
taskRecord, err := h.taskStore.GetByID(ctx, payload.TaskID)
if err != nil {
h.logger.Error("查询手机号资产解绑导入任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err))
return asynq.SkipRetry
}
if h.auditWriter == nil {
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "手机号资产解绑导入统一审计接缝未配置")
}
ctx = auditcontext.With(ctx, auditcontext.Context{
ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypePhoneAssetUnbindImport,
ActorName: "手机号资产解绑导入任务", Source: constants.AuditSourceWorker,
CorrelationID: taskRecord.TaskNo,
ParentEventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "completed"),
})
claimed, err := h.taskStore.ResetForProcessing(ctx, taskRecord.ID)
if err != nil {
return err
}
if !claimed {
h.logger.Info("手机号资产解绑导入任务已终结,跳过重复消费", zap.Uint("task_id", taskRecord.ID))
return nil
}
rows, err := h.downloadAndParse(ctx, taskRecord.StorageKey)
if err != nil {
h.logger.Warn("下载或解析手机号资产解绑导入CSV失败", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
if finishErr := h.finishTask(ctx, taskRecord, nil, 0, 0, model.ImportTaskStatusFailed, err.Error()); finishErr != nil {
return finishErr
}
return asynq.SkipRetry
}
items, successCount, err := h.processRows(ctx, taskRecord, rows)
if err != nil {
message := "导入执行中断:" + err.Error()
h.logger.Error("手机号资产解绑导入行执行中断", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
if finishErr := h.finishTask(ctx, taskRecord, nil, 0, 0, model.ImportTaskStatusFailed, message); finishErr != nil {
return finishErr
}
return asynq.SkipRetry
}
failCount := len(items) - successCount
if err := h.finishTask(ctx, taskRecord, items, successCount, failCount, model.ImportTaskStatusCompleted, ""); err != nil {
return err
}
h.logger.Info("手机号资产解绑导入任务完成",
zap.Uint("task_id", taskRecord.ID), zap.Int("success", successCount), zap.Int("fail", failCount))
return nil
}
// finishTask 在单事务内写任务终态、逐行明细与任务根审计事件。
// 任务级失败不产生行明细,与行级失败原因分开记录。
func (h *PhoneAssetUnbindImportHandler) finishTask(
ctx context.Context,
taskRecord *model.PhoneAssetUnbindImportTask,
items model.PhoneAssetUnbindImportResults,
successCount, failCount, status int,
errorMessage string,
) error {
return h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
store := h.taskStore.WithTx(tx)
if status == model.ImportTaskStatusFailed {
// 任务级失败:未命中非终态说明任务已被其他执行路径终结,按库内事实跳过重复收尾。
hit, err := store.MarkFailed(ctx, taskRecord.ID, errorMessage)
if err != nil {
return err
}
if !hit {
return nil
}
} else if err := store.Complete(ctx, taskRecord.ID, len(items), successCount, failCount, items); err != nil {
return err
}
result := batchAuditResult(successCount, failCount)
afterData := map[string]any{
"status": status, "total_count": len(items), "success_count": successCount, "fail_count": failCount,
}
if status == model.ImportTaskStatusFailed {
result = constants.AuditResultFailed
afterData["error_message"] = errorMessage
}
return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
EventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "completed"),
ActionCode: constants.AuditActionPhoneAssetUnbindImportTaskCompleted,
Summary: "完成手机号资产解绑导入任务", TaskID: taskRecord.ID, TaskNo: taskRecord.TaskNo,
Result: result, CorrelationID: taskRecord.TaskNo,
ParentEventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "created"),
BatchTotal: len(items), SuccessCount: successCount, FailCount: failCount,
IdentitySnapshot: map[string]any{
"id": taskRecord.ID, "task_no": taskRecord.TaskNo, "file_name": taskRecord.FileName,
},
BeforeData: map[string]any{"status": model.ImportTaskStatusProcessing},
AfterData: afterData,
ErrorSummary: errorMessage,
})
})
}
// downloadAndParse 下载并解析导入 CSV编码、表头或格式问题一律按任务级失败返回。
func (h *PhoneAssetUnbindImportHandler) downloadAndParse(ctx context.Context, key string) ([]phoneAssetUnbindImportRow, error) {
if h.storageService == nil {
return nil, phoneAssetUnbindImportError("对象存储服务未配置")
}
if key == "" {
return nil, phoneAssetUnbindImportError("导入文件Key不能为空")
}
localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, key)
if err != nil {
return nil, phoneAssetUnbindImportError("下载导入CSV失败")
}
defer cleanup()
// 不设行数与体积硬上限:体积沿用上传用途的既有校验,此处按文件实际大小读取。
data, err := os.ReadFile(localPath)
if err != nil {
return nil, phoneAssetUnbindImportError("读取导入CSV失败")
}
decoded, err := utils.DecodeTextToUTF8(data)
if err != nil {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorEncoding)
}
return parsePhoneAssetUnbindImportCSV(decoded)
}
// phoneAssetUnbindImportRow 是导入文件的单行业务事实;行号自数据首行起计,表头不计入。
type phoneAssetUnbindImportRow struct {
Line int
ColumnCountMatched bool
AssetIdentifier string
Remark string
}
// parsePhoneAssetUnbindImportCSV 解析固定列序的导入 CSV。
// 表头必须与固定列序完全一致,不一致即任务级失败且不进入逐行阶段;
// 数据行列数不符属行级「行格式错误」,因此必须关闭字段数一致性校验,
// 否则标准库在首条记录定型字段数后会让后续异常行直接返回 ErrFieldCount
// 把行级问题误升级为任务级失败且不产生行明细。
func parsePhoneAssetUnbindImportCSV(data []byte) ([]phoneAssetUnbindImportRow, error) {
reader := csv.NewReader(bytes.NewReader(data))
reader.TrimLeadingSpace = true
reader.FieldsPerRecord = -1
rows := make([]phoneAssetUnbindImportRow, 0)
line := 0
for {
record, err := reader.Read()
if err == io.EOF {
break
}
if err != nil {
// 关闭字段数校验后仍报错,说明是引号未闭合等真实 CSV 语法错误,属任务级失败。
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorFileFormat)
}
if line == 0 {
if !matchPhoneAssetUnbindImportHeader(record) {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorFileFormat)
}
line++
continue
}
line++
row := phoneAssetUnbindImportRow{Line: line - 1}
if len(record) != len(constants.PhoneAssetUnbindImportColumns) {
// 列数不符的行不参与业务校验,直接以行格式错误记录并保留原状。
rows = append(rows, row)
continue
}
row.ColumnCountMatched = true
row.AssetIdentifier = strings.TrimSpace(record[0])
row.Remark = strings.TrimSpace(record[1])
rows = append(rows, row)
}
if len(rows) == 0 {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorNoDataRow)
}
return rows, nil
}
// matchPhoneAssetUnbindImportHeader 逐列比较表头与固定列序,仅容忍列内两侧空白差异。
func matchPhoneAssetUnbindImportHeader(record []string) bool {
columns := constants.PhoneAssetUnbindImportColumns
if len(record) != len(columns) {
return false
}
for index, column := range columns {
if strings.TrimSpace(record[index]) != column {
return false
}
}
return true
}
// processRows 逐行独立执行并按批更新进度计数;进度写失败不回滚已提交行。
// 返回错误表示行执行遇到基础设施故障,由调用方按任务级失败收尾。
func (h *PhoneAssetUnbindImportHandler) processRows(ctx context.Context, taskRecord *model.PhoneAssetUnbindImportTask, rows []phoneAssetUnbindImportRow) (model.PhoneAssetUnbindImportResults, int, error) {
items := make(model.PhoneAssetUnbindImportResults, 0, len(rows))
successCount, failCount := 0, 0
for index, row := range rows {
item, err := h.processRow(ctx, taskRecord, row)
if err != nil {
return nil, 0, err
}
items = append(items, item)
if item.Status == constants.PhoneAssetUnbindImportItemStatusSuccess {
successCount++
} else {
failCount++
}
if (index+1)%constants.PhoneAssetUnbindImportProgressBatchSize == 0 {
if err := h.taskStore.UpdateProgress(ctx, taskRecord.ID, len(rows), successCount, failCount); err != nil {
h.logger.Warn("更新手机号资产解绑导入进度失败", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
}
}
}
return items, successCount, nil
}
// processRow 校验并执行单行;失败行只记录固定枚举原因,不解除任何关系。
func (h *PhoneAssetUnbindImportHandler) processRow(ctx context.Context, taskRecord *model.PhoneAssetUnbindImportTask, row phoneAssetUnbindImportRow) (model.PhoneAssetUnbindImportResultItem, error) {
item := model.PhoneAssetUnbindImportResultItem{Line: row.Line, AssetIdentifier: row.AssetIdentifier}
if !row.ColumnCountMatched {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorFormat), nil
}
if row.AssetIdentifier == "" {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorIdentifier), nil
}
if len(row.Remark) > constants.PhoneAssetUnbindImportRemarkMaxLength {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorRemarkTooLong), nil
}
assetType, assetID, err := h.resolveAsset(ctx, row.AssetIdentifier)
if err != nil {
return item, err
}
if assetType == "" {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorAssetMissing), nil
}
item.AssetType = assetType
item.AssetID = assetID
// 每行独立事务:成功行提交,失败行回滚并保留原状。
rowErr := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
locked, err := h.associationStore.WithTx(tx).LockValidByAsset(ctx, tx, assetType, assetID)
if err != nil {
return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "锁定资产关联失败")
}
if len(locked) == 0 {
return pkgerrors.New(pkgerrors.CodeNotFound, constants.PhoneAssetUnbindImportRowErrorNoAssociation)
}
ids := make([]uint, 0, len(locked))
phones := make([]string, 0, len(locked))
for _, association := range locked {
ids = append(ids, association.ID)
phones = append(phones, association.Phone)
}
now := time.Now()
affected, err := h.associationStore.WithTx(tx).InvalidateByIDs(ctx, tx, ids,
constants.PhoneAssetAssociationInvalidateMethodCSVImport, taskRecord.UnbindReason, taskRecord.Creator, now)
if err != nil {
return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "解除手机号资产关联失败")
}
if int(affected) != len(locked) {
return pkgerrors.New(pkgerrors.CodeNotFound, constants.PhoneAssetUnbindImportRowErrorNoAssociation)
}
item.UnboundCount = len(locked)
item.AssociatedPhones = phones
return h.appendRowAudit(ctx, tx, taskRecord, row, locked, now)
})
if rowErr != nil {
var appErr *pkgerrors.AppError
if stderrors.As(rowErr, &appErr) && appErr.Code == pkgerrors.CodeNotFound {
return failedPhoneAssetUnbindImportItem(item, appErr.Message), nil
}
return item, rowErr
}
item.Status = constants.PhoneAssetUnbindImportItemStatusSuccess
return item, nil
}
// resolveAsset 按资产标识定位资产,未命中返回空类型;解析口径与后台关联列表一致。
func (h *PhoneAssetUnbindImportHandler) resolveAsset(ctx context.Context, identifier string) (string, uint, error) {
return assetSvc.ResolveIdentifier(ctx, h.assetIdentifierStore, h.iotCardStore, h.deviceStore, identifier)
}
// appendRowAudit 在行事务内写单行解除审计,手机号一律脱敏,完整快照只进任务明细。
func (h *PhoneAssetUnbindImportHandler) appendRowAudit(
ctx context.Context,
tx *gorm.DB,
taskRecord *model.PhoneAssetUnbindImportTask,
row phoneAssetUnbindImportRow,
locked []*model.PhoneAssetAssociation,
now time.Time,
) error {
if h.auditWriter == nil {
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "手机号资产解绑导入统一审计接缝未配置")
}
metadata := map[string]any{
"import_task_id": taskRecord.ID, "import_task_no": taskRecord.TaskNo,
"line": row.Line, "unbound_count": len(locked),
}
if row.Remark != "" {
metadata["remark"] = row.Remark
}
resources := make([]audit.ResourceInput, 0, len(locked)+1)
for index, association := range locked {
relation := constants.AuditResourceRelationReference
if index == 0 {
relation = constants.AuditResourceRelationPrimary
}
associationID := strconv.FormatUint(uint64(association.ID), 10)
resources = append(resources, audit.ResourceInput{
Type: constants.AuditResourcePhoneAssetAssociation, ID: &associationID,
Key: associationID, DisplayName: sanitizer.MaskPhone(association.Phone),
Relation: relation, Role: constants.AuditResourceRolePhoneAssetAssociationTarget,
IdentitySnapshot: map[string]any{
"id": association.ID, "phone_masked": sanitizer.MaskPhone(association.Phone),
"asset_type": association.AssetType, "asset_id": association.AssetID,
"status": constants.PhoneAssetAssociationStatusInvalid, "source": association.Source,
"invalidated_at": now, "invalidation_method": constants.PhoneAssetAssociationInvalidateMethodCSVImport,
"invalidation_reason": taskRecord.UnbindReason,
},
BeforeData: map[string]any{"status": constants.PhoneAssetAssociationStatusValid},
AfterData: map[string]any{
"status": constants.PhoneAssetAssociationStatusInvalid, "invalidated_at": now,
"invalidation_method": constants.PhoneAssetAssociationInvalidateMethodCSVImport,
"invalidation_reason": taskRecord.UnbindReason,
},
SubjectVisibility: constants.AuditSubjectInternalOnly, SortOrder: index + 1,
})
}
assetID := strconv.FormatUint(uint64(locked[0].AssetID), 10)
resources = append(resources, audit.ResourceInput{
Type: locked[0].AssetType, ID: &assetID, Key: assetID, DisplayName: row.AssetIdentifier,
Relation: constants.AuditResourceRelationReference, Role: constants.AuditResourceRolePhoneAssetAssociationAsset,
IdentitySnapshot: map[string]any{"id": locked[0].AssetID},
SubjectVisibility: constants.AuditSubjectInternalOnly,
})
// 稳定事件 ID 由任务与行号决定,任务重复消费时同行为幂等重放。
return h.auditWriter.Append(ctx, tx, audit.AppendInput{
EventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, fmt.Sprintf("item:%d", row.Line)),
ActionCode: constants.AuditActionPhoneAssetAssociationImported,
Summary: "导入解除手机号资产关联",
Result: constants.AuditResultSuccess, Metadata: metadata,
ScopeType: constants.AuditScopePlatform, Resources: resources,
})
}
func failedPhoneAssetUnbindImportItem(item model.PhoneAssetUnbindImportResultItem, reason string) model.PhoneAssetUnbindImportResultItem {
item.Status, item.Reason = constants.PhoneAssetUnbindImportItemStatusFailed, reason
return item
}
// phoneAssetUnbindImportError 是任务级失败原因,与行级失败原因分开记录。
type phoneAssetUnbindImportError string
// Error 返回任务级失败原因原文。
func (e phoneAssetUnbindImportError) Error() string { return string(e) }