diff --git a/.gitea/workflows/deploy.yaml b/.gitea/workflows/deploy.yaml index 738cf8b8..1737614c 100644 --- a/.gitea/workflows/deploy.yaml +++ b/.gitea/workflows/deploy.yaml @@ -50,8 +50,10 @@ jobs: - name: 部署到测试环境（仅八月迭代分支） if: github.ref == 'refs/heads/Iteration/8-11' run: | - # 确保部署目录存在（仅需日志目录，配置已嵌入二进制文件） - mkdir -p ${{ env.DEPLOY_DIR }}/logs + # 确保部署目录存在：api、worker 各自的日志目录与 worker 的采集记录归档目录（配置已嵌入二进制文件） + mkdir -p ${{ env.DEPLOY_DIR }}/api/logs ${{ env.DEPLOY_DIR }}/worker/logs ${{ env.DEPLOY_DIR }}/worker/archive + # 容器以 uid 1000（appuser）运行，新目录需可写；无权限修改属主时保持原样 + chown 1000:1000 ${{ env.DEPLOY_DIR }}/api/logs ${{ env.DEPLOY_DIR }}/worker/logs ${{ env.DEPLOY_DIR }}/worker/archive 2>/dev/null || true umask 077 { diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 35a5dfb6..24b60907 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -863,10 +863,10 @@ func startOutboxRelay(ctx context.Context, runtime *workerRuntime, instanceName appLogger.Info("Outbox Relay 已启动", zap.String("lease_owner", owner)) } -// startTrafficCollectionPartitionMaintainer 为 leader/all 维护卡流量采集记录月分区并扫描采集告警： -// 每 12 小时预建本月起 2 个月的分区、删除整月都早于保留期的分区；每天扫描一次前一天的采集告警。 +// startTrafficCollectionPartitionMaintainer 为 leader/all 维护卡流量采集记录（成功与失败）月分区并扫描采集告警： +// 每 12 小时预建本月起 2 个月的分区，到期分区导出为 csv.gz、校验后删除；每天扫描一次前一天的采集告警。 func startTrafficCollectionPartitionMaintainer(ctx context.Context, runtime *workerRuntime, appLogger *zap.Logger) { - store := postgres.NewCardTrafficCollectionStore(runtime.db) + archiver := cardObservationInfra.NewCollectionArchiver(runtime.db, constants.TrafficCollectionArchiveDir, constants.TrafficCollectionArchiveKeepFiles, appLogger) alerts := cardObservationInfra.NewTrafficAlertScanner(runtime.db, appLogger) go func() { ticker := time.NewTicker(time.Hour) @@ -876,7 +876,7 @@ func startTrafficCollectionPartitionMaintainer(ctx context.Context, runtime *wor for { now := time.Now() if now.Sub(lastMaintained) >= 12*time.Hour { - if err := store.MaintainPartitions(ctx, now, 2, constants.TrafficCollectionRetentionDays); err != nil && ctx.Err() == nil { + if err := archiver.Maintain(ctx, now); err != nil && ctx.Err() == nil { appLogger.Error("维护卡流量采集记录分区失败", zap.Error(err)) } else { lastMaintained = now diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 99b8325b..ff31d163 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -54,6 +54,9 @@ services: # 日志配置 - JUNHONG_LOGGING_LEVEL=info - JUNHONG_LOGGING_DEVELOPMENT=false + # 应用日志轮换：单文件 100MB（默认），保留 40 个压缩备份，api 与 worker 合计约 1GB + - JUNHONG_LOGGING_APP_LOG_MAX_BACKUPS=40 + - JUNHONG_LOGGING_APP_LOG_MAX_AGE=14 # 跨域配置 - JUNHONG_MIDDLEWARE_CORS_ENABLED=true - JUNHONG_MIDDLEWARE_CORS_ALLOW_ORIGINS=* @@ -88,7 +91,7 @@ services: - JUNHONG_MIDDLEWARE_CORS_ALLOW_ORIGINS=https://cmp-admin.boss160.cn,https://cmp-agent.boss160.cn,https://cmp-c.boss160.cn - JUNHONG_MIDDLEWARE_CORS_ALLOW_CREDENTIALS=true volumes: - - ./logs:/app/logs + - ./api/logs:/app/logs networks: - junhong-network healthcheck: @@ -134,6 +137,9 @@ services: # 日志配置 - JUNHONG_LOGGING_LEVEL=info - JUNHONG_LOGGING_DEVELOPMENT=false + # 应用日志轮换：单文件 100MB（默认），保留 40 个压缩备份，api 与 worker 合计约 1GB + - JUNHONG_LOGGING_APP_LOG_MAX_BACKUPS=40 + - JUNHONG_LOGGING_APP_LOG_MAX_AGE=14 # 对象存储配置 - JUNHONG_STORAGE_PROVIDER=s3 - JUNHONG_STORAGE_S3_ENDPOINT=https://obs-helf.cucloud.cn @@ -161,7 +167,9 @@ services: - JUNHONG_POLLING_CARRIER_NSQD_ADDRESS=${JUNHONG_POLLING_CARRIER_NSQD_ADDRESS:-nsqd:4150} - JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES=${JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES:-nsqlookupd:4161} volumes: - - ./logs:/app/logs + - ./worker/logs:/app/logs + # 采集记录到期归档压缩包（archive/traffic-collection），由同步服务器拉走 + - ./worker/archive:/app/archive networks: - junhong-network depends_on: diff --git a/internal/application/cardobservation/traffic.go b/internal/application/cardobservation/traffic.go index 2cca730c..d1950dcf 100644 --- a/internal/application/cardobservation/traffic.go +++ b/internal/application/cardobservation/traffic.go @@ -84,6 +84,7 @@ func (s *Service) ApplyTrafficObservation(ctx context.Context, observation domai return errors.Wrap(errors.CodeDatabaseError, err, "查询卡采集时刻有效套餐失败") } record.HasPackage = len(packages.Packages) > 0 + record.CarrierID = card.CarrierID if judge != cardflow.QueueWrite { if observation.Metadata.Scene == constants.CardObservationScenePreActivationSettlement { // 激活前结算没写入：用已有本月最大快照结算，不新写快照。 diff --git a/internal/application/cardobservation/traffic_collection.go b/internal/application/cardobservation/traffic_collection.go index 49e2f097..d7a531ae 100644 --- a/internal/application/cardobservation/traffic_collection.go +++ b/internal/application/cardobservation/traffic_collection.go @@ -4,8 +4,10 @@ import ( "context" "encoding/json" "time" + "unicode/utf8" "go.uber.org/zap" + "gorm.io/datatypes" cardflow "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" domain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation" @@ -17,10 +19,14 @@ import ( // collectionRecordTimeout 是事务提交后写采集记录的超时；采集记录只用于排查，写失败只告警。 const collectionRecordTimeout = 3 * time.Second -// collectionRecord 是一次采集的留痕内容；HasPackage 在观测事务内判断，读数无效的观测写记录前再查。 +// rawResponseMaxBytes 是原始响应体保存上限，超过时截断并标记。 +const rawResponseMaxBytes = 4096 + +// collectionRecord 是一次采集的留痕内容；HasPackage 在观测事务内判断（只对成功读数有意义）。 type collectionRecord struct { model.CardTrafficCollection HasPackage bool + failure *domain.CollectionFailure } func newCollectionRecord(observation domain.TrafficObservation) *collectionRecord { @@ -28,7 +34,8 @@ func newCollectionRecord(observation domain.TrafficObservation) *collectionRecor IotCardID: observation.CardID, ObservedAt: observation.Metadata.ObservedAt, ObservationID: observation.Metadata.ObservationID, Source: observation.Metadata.Source, Scene: observation.Metadata.Scene, Result: domain.ResolveReadingResult(observation), - }} + CollectionRaw: buildCollectionRaw(observation.Raw), + }, failure: observation.Failure} if record.Result == constants.TrafficCollectionResultOK { reading := observation.GatewayReadingMB record.VendorUsedMB = &reading @@ -51,20 +58,16 @@ func (r *collectionRecord) applyOutcome(outcome *flowOutcome) { r.MonthUsedMB, r.DeltaMB, r.StaleGuarded, r.EventID = &monthUsed, &delta, &stale, outcome.EventID } -// writeCollectionRecord 在观测事务之外为采集时刻有套餐的卡写采集记录；没套餐的卡不写。 -func (s *Service) writeCollectionRecord(ctx context.Context, record *collectionRecord, observedAt time.Time) { +// writeCollectionRecord 在观测事务之外写采集记录：成功读数只为采集时刻有套餐的卡写成功表； +// 失败读数不判断有无套餐，写失败表。 +func (s *Service) writeCollectionRecord(ctx context.Context, record *collectionRecord, _ time.Time) { writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), collectionRecordTimeout) defer cancel() - hasPackage := record.HasPackage if record.Result != constants.TrafficCollectionResultOK { - packages, err := postgres.FindEffectivePackagesAt(writeCtx, s.db, record.IotCardID, observedAt, false) - if err != nil { - s.logTrafficWarn("查询卡采集时刻有效套餐失败，跳过采集记录", zap.Uint("card_id", record.IotCardID), zap.Error(err)) - return - } - hasPackage = len(packages.Packages) > 0 + s.writeCollectionFailure(writeCtx, record) + return } - if !hasPackage { + if !record.HasPackage { return } if err := postgres.NewCardTrafficCollectionStore(s.db).Create(writeCtx, &record.CardTrafficCollection); err != nil { @@ -72,3 +75,72 @@ func (s *Service) writeCollectionRecord(ctx context.Context, record *collectionR zap.String("observation_id", record.ObservationID), zap.Error(err)) } } + +func (s *Service) writeCollectionFailure(ctx context.Context, record *collectionRecord) { + row := &model.CardTrafficCollectionFailure{ + IotCardID: record.IotCardID, ObservedAt: record.ObservedAt, ObservationID: record.ObservationID, + Source: record.Source, Scene: record.Scene, Result: record.Result, CollectionRaw: record.CollectionRaw, + } + if failure := record.failure; failure != nil { + row.ErrorCategory, row.ProviderCode = failure.ErrorCategory, failure.ProviderCode + row.ProviderMessage, row.DurationMS = truncateText(failure.ProviderMessage, 500), failure.DurationMS + } + var card model.IotCard + if err := s.db.WithContext(ctx).Unscoped().Select("iccid", "carrier_id").Where("id = ?", record.IotCardID).Take(&card).Error; err == nil { + row.ICCID, row.CarrierID = card.ICCID, card.CarrierID + } + if err := postgres.NewCardTrafficCollectionFailureStore(s.db).Create(ctx, row); err != nil { + s.logTrafficWarn("写卡流量采集失败记录失败", zap.Uint("card_id", record.IotCardID), + zap.String("observation_id", record.ObservationID), zap.Error(err)) + } +} + +// buildCollectionRaw 把网关原始数据转为留存字段：响应可解析为 JSON 且不超过 4KB 时存 JSON，否则存文本并按 4KB 截断。 +func buildCollectionRaw(raw *domain.GatewayRaw) model.CollectionRaw { + if raw == nil { + return model.CollectionRaw{} + } + out := model.CollectionRaw{ + GatewayCode: raw.GatewayCode, GatewayMsg: truncateText(raw.GatewayMsg, 500), + GatewayTraceID: truncateText(raw.TraceID, 128), GatewayAttempts: raw.Attempts, + } + if raw.HTTPStatus > 0 { + status := raw.HTTPStatus + out.HTTPStatus = &status + } + if len(raw.Request) > 0 && json.Valid(raw.Request) { + out.RawRequest = datatypes.JSON(raw.Request) + } + switch { + case len(raw.Response) == 0: + case len(raw.Response) > rawResponseMaxBytes: + out.RawResponseText, out.RawTruncated = validUTF8Prefix(raw.Response, rawResponseMaxBytes), true + case json.Valid(raw.Response): + out.RawResponse = datatypes.JSON(raw.Response) + default: + out.RawResponseText = validUTF8Prefix(raw.Response, rawResponseMaxBytes) + } + return out +} + +// validUTF8Prefix 返回不超过 limit 字节且不截断多字节字符的前缀，并替换非法 UTF-8（PostgreSQL text 不接受）。 +func validUTF8Prefix(data []byte, limit int) string { + if len(data) > limit { + data = data[:limit] + for len(data) > 0 && !utf8.Valid(data) { + data = data[:len(data)-1] + } + } + if !utf8.Valid(data) { + return string([]rune(string(data))) + } + return string(data) +} + +func truncateText(value string, limit int) string { + runes := []rune(value) + if len(runes) <= limit { + return value + } + return string(runes[:limit]) +} diff --git a/internal/domain/cardobservation/traffic.go b/internal/domain/cardobservation/traffic.go index bc39011c..03a6e6db 100644 --- a/internal/domain/cardobservation/traffic.go +++ b/internal/domain/cardobservation/traffic.go @@ -16,6 +16,29 @@ type TrafficObservation struct { ReadingResult string ResetDay int Metadata ObservationMetadata + // Raw 是本次网关调用的原始请求与响应，未收集时为 nil。 + Raw *GatewayRaw + // Failure 是读数结果非 ok 时的失败详情，未提供时为 nil。 + Failure *CollectionFailure +} + +// GatewayRaw 是一次网关调用的原始数据（最后一次 HTTP 尝试），只含明文业务请求与网关响应。 +type GatewayRaw struct { + Request []byte + HTTPStatus int + Response []byte + GatewayCode *int + GatewayMsg string + TraceID string + Attempts int +} + +// CollectionFailure 是一次采集失败的错误详情。 +type CollectionFailure struct { + ErrorCategory string + ProviderCode string + ProviderMessage string + DurationMS int } // TrafficDecision 描述一次流量观测的处理结果。 diff --git a/internal/gateway/client.go b/internal/gateway/client.go index c1fd11f9..da058d3d 100644 --- a/internal/gateway/client.go +++ b/internal/gateway/client.go @@ -114,6 +114,10 @@ func (c *Client) doRequest(ctx context.Context, path string, params interface{}) return nil, errors.Wrap(errors.CodeInternalError, err, "序列化业务数据失败") } + if capture := rawCaptureFrom(ctx); capture != nil { + capture.Request = append(json.RawMessage(nil), dataBytes...) + } + // 加密业务数据（加密结果不变，可在重试间复用） encryptedData, err := aesEncrypt(dataBytes, c.appSecret) if err != nil { @@ -184,8 +188,13 @@ func (c *Client) executeWithRetry(ctx context.Context, path, encryptedData strin return nil, err } } + capture := rawCaptureFrom(ctx) + if capture != nil { + capture.resetResponse() + capture.Attempts = attemptNumber + } attemptStartedAt := time.Now() - result, retryable, err := c.executeHTTPRequest(ctx, path, encryptedData) + result, retryable, err := c.executeHTTPRequest(ctx, path, encryptedData, capture != nil && logPayload) attemptDuration := time.Since(attemptStartedAt) if gwErr, ok := AsGatewayError(err); ok { gwErr.Duration = attemptDuration @@ -237,7 +246,12 @@ func (c *Client) executeWithRetry(ctx context.Context, path, encryptedData strin // executeHTTPRequest 执行单次 HTTP 请求（无重试逻辑） // 返回值：响应数据、是否可重试、错误 -func (c *Client) executeHTTPRequest(ctx context.Context, path string, encryptedData string) (json.RawMessage, bool, error) { +// captureRaw 为 true 时把状态码、响应体与外层字段写入上下文上的原始数据收集器。 +func (c *Client) executeHTTPRequest(ctx context.Context, path string, encryptedData string, captureRaw bool) (json.RawMessage, bool, error) { + var capture *RawCapture + if captureRaw { + capture = rawCaptureFrom(ctx) + } // 每次重试使用新的时间戳和签名 timestamp := time.Now().Unix() sign := generateSign(c.appID, encryptedData, timestamp, c.appSecret) @@ -277,6 +291,9 @@ func (c *Client) executeHTTPRequest(ctx context.Context, path string, encryptedD return nil, true, newGatewayError(constants.GatewayErrorNetwork, errors.Wrap(errors.CodeGatewayError, err, "发送 HTTP 请求失败")) } defer resp.Body.Close() + if capture != nil { + capture.HTTPStatus = resp.StatusCode + } // HTTP 状态码错误 — 不可重试 if resp.StatusCode != http.StatusOK { @@ -290,10 +307,18 @@ func (c *Client) executeHTTPRequest(ctx context.Context, path string, encryptedD return nil, false, newGatewayError(constants.GatewayErrorInvalidResponse, errors.Wrap(errors.CodeGatewayInvalidResp, err, "读取响应体失败")) } + if capture != nil { + capture.Response = body + } + var gatewayResp GatewayResponse if err := sonic.Unmarshal(body, &gatewayResp); err != nil { return nil, false, newGatewayError(constants.GatewayErrorInvalidResponse, errors.Wrap(errors.CodeGatewayInvalidResp, err, "解析 Gateway 响应失败")) } + if capture != nil { + code := gatewayResp.Code + capture.GatewayCode, capture.GatewayMsg, capture.TraceID = &code, gatewayResp.Msg, gatewayResp.TraceID + } // Gateway 业务错误 — 不可重试 if gatewayResp.Code != gatewaySuccessCode { diff --git a/internal/gateway/raw_capture.go b/internal/gateway/raw_capture.go new file mode 100644 index 00000000..fde576de --- /dev/null +++ b/internal/gateway/raw_capture.go @@ -0,0 +1,44 @@ +package gateway + +import ( + "context" + "encoding/json" +) + +// RawCapture 收集一次网关调用的原始数据（最后一次 HTTP 尝试），供业务记录留存。 +// 只保存加密前的明文业务请求与网关响应；不含加密报文、签名、应用标识、密钥与请求头。 +type RawCapture struct { + // Request 是加密前的明文业务请求（{"params": ...}）；不记录载荷的调用为空。 + Request json.RawMessage + // HTTPStatus 是 HTTP 状态码；没拿到响应（超时、连接失败）时为 0。 + HTTPStatus int + // Response 是网关原始响应体；没拿到响应体时为空。 + Response []byte + // GatewayCode、GatewayMsg、TraceID 是网关响应外层字段；响应无法解析时为空。 + GatewayCode *int + GatewayMsg string + TraceID string + // Attempts 是网关客户端内部 HTTP 尝试次数。 + Attempts int +} + +type rawCaptureKey struct{} + +// WithRawCapture 在请求上下文上挂载原始数据收集器；未挂载的调用不收集。 +func WithRawCapture(ctx context.Context) (context.Context, *RawCapture) { + capture := &RawCapture{} + return context.WithValue(ctx, rawCaptureKey{}, capture), capture +} + +func rawCaptureFrom(ctx context.Context) *RawCapture { + capture, _ := ctx.Value(rawCaptureKey{}).(*RawCapture) + return capture +} + +// resetResponse 在每次 HTTP 尝试前清空上一次的响应，保证留存的是最后一次尝试。 +func (c *RawCapture) resetResponse() { + if c == nil { + return + } + c.HTTPStatus, c.Response, c.GatewayCode, c.GatewayMsg, c.TraceID = 0, nil, nil, "", "" +} diff --git a/internal/handler/admin/package_usage.go b/internal/handler/admin/package_usage.go index 90743e9d..cb99eb26 100644 --- a/internal/handler/admin/package_usage.go +++ b/internal/handler/admin/package_usage.go @@ -37,7 +37,12 @@ func (h *PackageUsageHandler) GetDailyRecords(c *fiber.Ctx) error { endDate := c.Query("end_date", "") // 任务 16.4: 调用 DailyRecordService.GetDailyRecords 获取日记录 - records, err := h.dailyRecordService.GetDailyRecords(c.UserContext(), uint(id), startDate, endDate) + page := c.QueryInt("page", 0) + pageSize := c.QueryInt("page_size", 0) + if page < 0 || pageSize < 0 || pageSize > 100 { + return errors.New(errors.CodeInvalidParam, "分页参数不合法，page_size 最大 100") + } + records, err := h.dailyRecordService.GetDailyRecords(c.UserContext(), uint(id), startDate, endDate, page, pageSize) if err != nil { return err } diff --git a/internal/handler/admin/record_query.go b/internal/handler/admin/record_query.go new file mode 100644 index 00000000..5037105d --- /dev/null +++ b/internal/handler/admin/record_query.go @@ -0,0 +1,94 @@ +package admin + +import ( + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/handler/validation" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/response" +) + +// parseRecordQuery 解析并校验记录查询参数。 +func (h *AssetHandler) parseRecordQuery(c *fiber.Ctx, req any, label string) error { + if err := rejectEnterpriseAssetAdmin(c); err != nil { + return err + } + if err := c.QueryParser(req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数格式不正确") + } + if err := h.validator.Struct(req); err != nil { + return errors.New(errors.CodeInvalidParam, validation.Message(label, req, err)) + } + return nil +} + +// ListTrafficCollectionRecords 采集成功记录列表 +// GET /api/admin/traffic-collections +func (h *AssetHandler) ListTrafficCollectionRecords(c *fiber.Ctx) error { + var req dto.TrafficCollectionListRequest + if err := h.parseRecordQuery(c, &req, "采集成功记录查询参数不合法"); err != nil { + return err + } + result, err := h.assetService.ListTrafficCollectionRecords(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// ListTrafficCollectionFailures 采集失败记录列表 +// GET /api/admin/traffic-collection-failures +func (h *AssetHandler) ListTrafficCollectionFailures(c *fiber.Ctx) error { + var req dto.TrafficCollectionFailureListRequest + if err := h.parseRecordQuery(c, &req, "采集失败记录查询参数不合法"); err != nil { + return err + } + result, err := h.assetService.ListTrafficCollectionFailures(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// ListCardDailyUsages 卡日流量列表 +// GET /api/admin/card-daily-usages +func (h *AssetHandler) ListCardDailyUsages(c *fiber.Ctx) error { + var req dto.CardDailyUsageListRequest + if err := h.parseRecordQuery(c, &req, "卡日流量查询参数不合法"); err != nil { + return err + } + result, err := h.assetService.ListCardDailyUsages(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// ListPackageUsageDailyRecords 套餐日流量详单列表 +// GET /api/admin/package-usage-daily-records +func (h *AssetHandler) ListPackageUsageDailyRecords(c *fiber.Ctx) error { + var req dto.PackageUsageDailyRecordListRequest + if err := h.parseRecordQuery(c, &req, "套餐日流量详单查询参数不合法"); err != nil { + return err + } + result, err := h.assetService.ListPackageUsageDailyRecords(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// ListPackageTrafficAdjustmentRecords 套餐流量冲正流水列表 +// GET /api/admin/package-traffic-adjustments +func (h *AssetHandler) ListPackageTrafficAdjustmentRecords(c *fiber.Ctx) error { + var req dto.PackageTrafficAdjustmentQueryRequest + if err := h.parseRecordQuery(c, &req, "冲正流水查询参数不合法"); err != nil { + return err + } + result, err := h.assetService.ListPackageTrafficAdjustmentRecords(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} diff --git a/internal/handler/admin/refund.go b/internal/handler/admin/refund.go index fc0a8973..9802a805 100644 --- a/internal/handler/admin/refund.go +++ b/internal/handler/admin/refund.go @@ -111,6 +111,23 @@ func (h *RefundHandler) RetryAssetPostProcessing(c *fiber.Ctx) error { return response.Success(c, result) } +// ListAssetPostProcessAttempts 退款资产后处理尝试记录列表 +// GET /api/admin/refund-post-process-attempts +func (h *RefundHandler) ListAssetPostProcessAttempts(c *fiber.Ctx) error { + var req dto.RefundPostProcessAttemptListRequest + if err := c.QueryParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数格式不正确") + } + if req.StartTime == "" || req.EndTime == "" { + return errors.New(errors.CodeInvalidParam, "开始时间与结束时间必填") + } + result, err := h.service.ListAssetPostProcessAttempts(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + // Approve 审批通过退款申请 // POST /api/admin/refunds/:id/approve func (h *RefundHandler) Approve(c *fiber.Ctx) error { diff --git a/internal/infrastructure/cardobservation/collection_archive.go b/internal/infrastructure/cardobservation/collection_archive.go new file mode 100644 index 00000000..22fce6c1 --- /dev/null +++ b/internal/infrastructure/cardobservation/collection_archive.go @@ -0,0 +1,313 @@ +package cardobservation + +import ( + "compress/gzip" + "context" + "crypto/sha256" + "database/sql" + "encoding/csv" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/jackc/pgx/v5/stdlib" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// collectionArchiveTables 是按月分区、到期归档的采集记录父表。 +var collectionArchiveTables = []string{"tb_card_traffic_collection", "tb_card_traffic_collection_failure"} + +// CollectionArchiver 维护采集记录月分区：预建未来分区；到期分区按"摘出 → 导出 csv.gz → 校验 → 删除"归档。 +// 导出或校验失败时保留已摘出的表，下一轮重试，不会出现数据已删但没有压缩包的情况。 +type CollectionArchiver struct { + db *gorm.DB + dir string + keepFiles int + aheadMonth int + logger *zap.Logger +} + +// NewCollectionArchiver 创建采集记录归档器；dir 为压缩包目录，keepFiles 为每张表本地保留的压缩包个数。 +func NewCollectionArchiver(db *gorm.DB, dir string, keepFiles int, logger *zap.Logger) *CollectionArchiver { + if logger == nil { + logger = zap.NewNop() + } + if keepFiles <= 0 { + keepFiles = constants.TrafficCollectionArchiveKeepFiles + } + return &CollectionArchiver{db: db, dir: dir, keepFiles: keepFiles, aheadMonth: 2, logger: logger} +} + +// archiveManifest 是压缩包同名说明文件内容。 +type archiveManifest struct { + Table string `json:"table"` + Partition string `json:"partition"` + Month string `json:"month"` + Rows int64 `json:"rows"` + FirstObservedAt *string `json:"first_observed_at"` + LastObservedAt *string `json:"last_observed_at"` + File string `json:"file"` + FileBytes int64 `json:"file_bytes"` + SHA256 string `json:"sha256"` + ExportedAt time.Time `json:"exported_at"` +} + +// Maintain 对全部采集记录表执行一轮分区维护；单张表失败不影响其他表，返回第一个错误。 +func (a *CollectionArchiver) Maintain(ctx context.Context, now time.Time) error { + var firstErr error + for _, table := range collectionArchiveTables { + if err := a.maintainTable(ctx, table, now); err != nil { + a.logger.Error("维护采集记录分区失败", zap.String("table", table), zap.Error(err)) + if firstErr == nil { + firstErr = err + } + } + } + return firstErr +} + +func (a *CollectionArchiver) maintainTable(ctx context.Context, table string, now time.Time) error { + monthStart := cardflow.MonthStart(now) + for i := 0; i <= a.aheadMonth; i++ { + from := monthStart.AddDate(0, i, 0) + sqlText := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s_%s PARTITION OF %s FOR VALUES FROM ('%s') TO ('%s')`, + table, from.Format("200601"), table, from.Format(time.RFC3339), from.AddDate(0, 1, 0).Format(time.RFC3339)) + if err := a.db.WithContext(ctx).Exec(sqlText).Error; err != nil { + return fmt.Errorf("预建分区失败: %w", err) + } + } + // 在线保留当月加最近 N 个整月，更早的月分区到期。 + cutoff := monthStart.AddDate(0, -constants.TrafficCollectionRetentionMonths, 0) + partitions, err := a.monthPartitions(ctx, table) + if err != nil { + return err + } + for _, partition := range partitions { + if !partition.month.Before(cutoff) { + continue + } + if err := a.archivePartition(ctx, table, partition); err != nil { + return fmt.Errorf("归档分区 %s 失败: %w", partition.name, err) + } + } + a.cleanupLocalFiles(table) + return nil +} + +type monthPartition struct { + name string + month time.Time + attached bool +} + +// monthPartitions 返回该表全部按月命名的分区（含上一轮已摘出未删除的表），按月份升序。 +func (a *CollectionArchiver) monthPartitions(ctx context.Context, table string) ([]monthPartition, error) { + var rows []struct { + Relname string + Attached bool + } + err := a.db.WithContext(ctx).Raw(` +SELECT c.relname, EXISTS (SELECT 1 FROM pg_inherits i JOIN pg_class p ON p.oid = i.inhparent + WHERE i.inhrelid = c.oid AND p.relname = ?) AS attached +FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace +WHERE n.nspname = current_schema() AND c.relkind IN ('r', 'p') AND c.relname ~ ?`, + table, "^"+table+`_[0-9]{6}$`).Scan(&rows).Error + if err != nil { + return nil, fmt.Errorf("查询分区失败: %w", err) + } + partitions := make([]monthPartition, 0, len(rows)) + for _, row := range rows { + month, err := time.ParseInLocation("200601", row.Relname[len(table)+1:], cardflow.MonthStart(time.Now()).Location()) + if err != nil { + continue + } + partitions = append(partitions, monthPartition{name: row.Relname, month: month, attached: row.Attached}) + } + sort.Slice(partitions, func(i, j int) bool { return partitions[i].month.Before(partitions[j].month) }) + return partitions, nil +} + +func (a *CollectionArchiver) archivePartition(ctx context.Context, table string, partition monthPartition) error { + if partition.attached { + if err := a.db.WithContext(ctx).Exec(fmt.Sprintf("ALTER TABLE %s DETACH PARTITION %s", table, partition.name)).Error; err != nil { + return fmt.Errorf("摘出分区失败: %w", err) + } + } + if err := os.MkdirAll(a.dir, 0o755); err != nil { + return fmt.Errorf("创建归档目录失败: %w", err) + } + finalPath := filepath.Join(a.dir, partition.name+".csv.gz") + tmpPath := finalPath + ".tmp" + if err := a.exportCSV(ctx, partition.name, tmpPath); err != nil { + _ = os.Remove(tmpPath) + return err + } + manifest, err := a.verify(ctx, table, partition, tmpPath) + if err != nil { + _ = os.Remove(tmpPath) + return err + } + manifest.File = filepath.Base(finalPath) + if err := writeManifest(finalPath+".manifest.json", manifest); err != nil { + _ = os.Remove(tmpPath) + return err + } + if err := os.Rename(tmpPath, finalPath); err != nil { + return fmt.Errorf("压缩包改名失败: %w", err) + } + if err := a.db.WithContext(ctx).Exec("DROP TABLE IF EXISTS " + partition.name).Error; err != nil { + return fmt.Errorf("删除已归档分区失败: %w", err) + } + a.logger.Info("采集记录分区已归档", zap.String("partition", partition.name), zap.Int64("rows", manifest.Rows), + zap.String("file", finalPath), zap.Int64("file_bytes", manifest.FileBytes)) + return nil +} + +// exportCSV 用 COPY 把分区按采集时间与 ID 排序流式导出为带表头的 CSV，并边写边 gzip。 +func (a *CollectionArchiver) exportCSV(ctx context.Context, partition, path string) error { + file, err := os.Create(path) + if err != nil { + return fmt.Errorf("创建压缩包失败: %w", err) + } + defer file.Close() + gz := gzip.NewWriter(file) + sqlDB, err := a.db.DB() + if err != nil { + return fmt.Errorf("获取数据库连接失败: %w", err) + } + conn, err := sqlDB.Conn(ctx) + if err != nil { + return fmt.Errorf("获取数据库连接失败: %w", err) + } + defer conn.Close() + copySQL := fmt.Sprintf("COPY (SELECT * FROM %s ORDER BY observed_at, id) TO STDOUT WITH (FORMAT csv, HEADER)", partition) + if err := rawCopyTo(ctx, conn, gz, copySQL); err != nil { + return fmt.Errorf("导出分区失败: %w", err) + } + if err := gz.Close(); err != nil { + return fmt.Errorf("写压缩包失败: %w", err) + } + if err := file.Sync(); err != nil { + return fmt.Errorf("落盘压缩包失败: %w", err) + } + return nil +} + +func rawCopyTo(ctx context.Context, conn *sql.Conn, w io.Writer, copySQL string) error { + return conn.Raw(func(driverConn any) error { + stdConn, ok := driverConn.(*stdlib.Conn) + if !ok { + return fmt.Errorf("数据库驱动不支持 COPY 导出") + } + _, err := stdConn.Conn().PgConn().CopyTo(ctx, w, copySQL) + return err + }) +} + +// verify 重新读取压缩包统计行数并与分区行数比对，生成说明文件内容。 +func (a *CollectionArchiver) verify(ctx context.Context, table string, partition monthPartition, path string) (archiveManifest, error) { + var stats struct { + Rows int64 + First *time.Time + Last *time.Time + } + if err := a.db.WithContext(ctx).Raw(fmt.Sprintf( + "SELECT count(*) AS rows, min(observed_at) AS first, max(observed_at) AS last FROM %s", partition.name)).Scan(&stats).Error; err != nil { + return archiveManifest{}, fmt.Errorf("统计分区行数失败: %w", err) + } + fileRows, sum, size, err := countArchiveRows(path) + if err != nil { + return archiveManifest{}, err + } + if fileRows != stats.Rows { + return archiveManifest{}, fmt.Errorf("压缩包行数 %d 与分区行数 %d 不一致", fileRows, stats.Rows) + } + manifest := archiveManifest{ + Table: table, Partition: partition.name, Month: partition.month.Format("2006-01"), Rows: stats.Rows, + FileBytes: size, SHA256: sum, ExportedAt: time.Now(), + } + if stats.First != nil { + first, last := stats.First.Format(time.RFC3339), stats.Last.Format(time.RFC3339) + manifest.FirstObservedAt, manifest.LastObservedAt = &first, &last + } + return manifest, nil +} + +// countArchiveRows 解压并按 CSV 解析压缩包，返回数据行数（不含表头）、文件 SHA-256 与文件大小。 +func countArchiveRows(path string) (int64, string, int64, error) { + file, err := os.Open(path) + if err != nil { + return 0, "", 0, fmt.Errorf("打开压缩包失败: %w", err) + } + defer file.Close() + hash := sha256.New() + gz, err := gzip.NewReader(io.TeeReader(file, hash)) + if err != nil { + return 0, "", 0, fmt.Errorf("读取压缩包失败: %w", err) + } + reader := csv.NewReader(gz) + reader.FieldsPerRecord = -1 + reader.ReuseRecord = true + var records int64 + for { + if _, err := reader.Read(); err == io.EOF { + break + } else if err != nil { + return 0, "", 0, fmt.Errorf("解析压缩包失败: %w", err) + } + records++ + } + if _, err := io.Copy(io.Discard, gz); err != nil { + return 0, "", 0, fmt.Errorf("读取压缩包失败: %w", err) + } + info, err := file.Stat() + if err != nil { + return 0, "", 0, fmt.Errorf("读取压缩包大小失败: %w", err) + } + if records > 0 { + records-- // 表头 + } + return records, hex.EncodeToString(hash.Sum(nil)), info.Size(), nil +} + +func writeManifest(path string, manifest archiveManifest) error { + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return fmt.Errorf("生成说明文件失败: %w", err) + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, append(data, '\n'), 0o644); err != nil { + return fmt.Errorf("写说明文件失败: %w", err) + } + if err := os.Rename(tmp, path); err != nil { + return fmt.Errorf("说明文件改名失败: %w", err) + } + return nil +} + +// cleanupLocalFiles 每张表只保留最近 keepFiles 个月的压缩包与说明文件。 +func (a *CollectionArchiver) cleanupLocalFiles(table string) { + matches, err := filepath.Glob(filepath.Join(a.dir, table+"_[0-9][0-9][0-9][0-9][0-9][0-9].csv.gz")) + if err != nil || len(matches) <= a.keepFiles { + return + } + sort.Strings(matches) + for _, path := range matches[:len(matches)-a.keepFiles] { + if err := os.Remove(path); err != nil { + a.logger.Warn("删除过期采集记录压缩包失败", zap.String("file", path), zap.Error(err)) + continue + } + _ = os.Remove(path + ".manifest.json") + a.logger.Info("已删除过期采集记录压缩包", zap.String("file", strings.TrimPrefix(path, a.dir+"/"))) + } +} diff --git a/internal/model/card_traffic_collection.go b/internal/model/card_traffic_collection.go index 9223e487..17683b36 100644 --- a/internal/model/card_traffic_collection.go +++ b/internal/model/card_traffic_collection.go @@ -22,7 +22,9 @@ type CardTrafficCollection struct { StaleGuarded *bool `gorm:"column:stale_guarded" json:"stale_guarded"` DeltaMB *float64 `gorm:"column:delta_mb;type:numeric(14,3)" json:"delta_mb"` EventID string `gorm:"column:event_id;type:varchar(128);not null;default:''" json:"event_id"` - CreatedAt time.Time `gorm:"column:created_at;not null;default:now()" json:"created_at"` + CarrierID uint `gorm:"column:carrier_id;not null;default:0" json:"carrier_id"` + CollectionRaw + CreatedAt time.Time `gorm:"column:created_at;not null;default:now()" json:"created_at"` } // TableName 指定表名。 @@ -30,6 +32,43 @@ func (CardTrafficCollection) TableName() string { return "tb_card_traffic_collection" } +// CollectionRaw 是采集记录保存的网关原始数据（成功表与失败表共用）。 +type CollectionRaw struct { + RawRequest datatypes.JSON `gorm:"column:raw_request;type:jsonb" json:"raw_request"` + HTTPStatus *int `gorm:"column:http_status" json:"http_status"` + GatewayCode *int `gorm:"column:gateway_code" json:"gateway_code"` + GatewayMsg string `gorm:"column:gateway_msg;type:varchar(500);not null;default:''" json:"gateway_msg"` + GatewayTraceID string `gorm:"column:gateway_trace_id;type:varchar(128);not null;default:''" json:"gateway_trace_id"` + RawResponse datatypes.JSON `gorm:"column:raw_response;type:jsonb" json:"raw_response"` + RawResponseText string `gorm:"column:raw_response_text;type:text;not null;default:''" json:"raw_response_text"` + RawTruncated bool `gorm:"column:raw_truncated;not null;default:false" json:"raw_truncated"` + GatewayAttempts int `gorm:"column:gateway_attempts;not null;default:0" json:"gateway_attempts"` +} + +// CardTrafficCollectionFailure 卡流量采集失败记录：所有被采集的卡每次失败一条，按月分区，在线保留 2 个月。 +type CardTrafficCollectionFailure struct { + ID uint `gorm:"column:id;primaryKey;autoIncrement" json:"id"` + IotCardID uint `gorm:"column:iot_card_id;not null" json:"iot_card_id"` + ICCID string `gorm:"column:iccid;type:varchar(30);not null;default:''" json:"iccid"` + CarrierID uint `gorm:"column:carrier_id;not null;default:0" json:"carrier_id"` + ObservedAt time.Time `gorm:"column:observed_at;primaryKey;not null" json:"observed_at"` + ObservationID string `gorm:"column:observation_id;type:varchar(128);not null;default:''" json:"observation_id"` + Source string `gorm:"column:source;type:varchar(32);not null;default:''" json:"source"` + Scene string `gorm:"column:scene;type:varchar(64);not null;default:''" json:"scene"` + Result string `gorm:"column:result;type:varchar(32);not null" json:"result"` + ErrorCategory string `gorm:"column:error_category;type:varchar(30);not null;default:''" json:"error_category"` + ProviderCode string `gorm:"column:provider_code;type:varchar(30);not null;default:''" json:"provider_code"` + ProviderMessage string `gorm:"column:provider_message;type:varchar(500);not null;default:''" json:"provider_message"` + DurationMS int `gorm:"column:duration_ms;not null;default:0" json:"duration_ms"` + CollectionRaw + CreatedAt time.Time `gorm:"column:created_at;not null;default:now()" json:"created_at"` +} + +// TableName 指定表名。 +func (CardTrafficCollectionFailure) TableName() string { + return "tb_card_traffic_collection_failure" +} + // PackageUsageAdjustment 套餐流量人工冲正流水：只新增，不修改、不删除。 type PackageUsageAdjustment struct { ID uint `gorm:"column:id;primaryKey;autoIncrement" json:"id"` diff --git a/internal/model/dto/asset_dto.go b/internal/model/dto/asset_dto.go index b7f3999a..85bb375f 100644 --- a/internal/model/dto/asset_dto.go +++ b/internal/model/dto/asset_dto.go @@ -281,6 +281,8 @@ type PackageTrafficAdjustmentListRequest struct { PackageUsageID uint `path:"package_usage_id" description:"套餐使用记录ID" required:"true"` Page int `query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码，默认 1"` PageSize int `query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页条数，默认 20"` + StartTime string `query:"start_time" description:"开始时间（RFC3339 或 2006-01-02 15:04:05，上海时间），选填；与结束时间同时传，跨度不超过 31 天"` + EndTime string `query:"end_time" description:"结束时间（不含），选填"` } // PackageTrafficAdjustmentItem 冲正流水。 @@ -312,6 +314,11 @@ type CardTrafficCollectionListRequest struct { EndTime string `query:"end_time" validate:"required" required:"true" description:"结束时间（不含），最长跨度 31 天"` Page int `query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码，默认 1"` PageSize int `query:"page_size" validate:"omitempty,min=1,max=200" minimum:"1" maximum:"200" description:"每页条数，默认 50"` + Decision string `query:"decision" validate:"omitempty,max=32" description:"写入判断筛选（只作用于成功记录）"` + Source string `query:"source" validate:"omitempty,max=32" description:"来源筛选"` + Scene string `query:"scene" validate:"omitempty,max=64" description:"场景筛选"` + // IncludeFailures 为 true 时合并采集失败记录，按采集时间倒序返回。 + IncludeFailures bool `query:"include_failures" description:"是否包含采集失败记录，默认 false"` } // CardTrafficCollectionItem 卡流量采集记录。 @@ -329,6 +336,13 @@ type CardTrafficCollectionItem struct { StaleGuarded *bool `json:"stale_guarded" description:"写入时是否触发重置日防旧值"` DeltaMB *float64 `json:"delta_mb" description:"写入时本次增量(MB)"` EventID string `json:"event_id" description:"流量事件 ID"` + // RecordKind 区分成功表与失败表记录：success/failure。 + RecordKind string `json:"record_kind" description:"记录类型：success 采集成功，failure 采集失败"` + ErrorCategory string `json:"error_category,omitempty" description:"错误类别（失败记录）"` + ProviderCode string `json:"provider_code,omitempty" description:"上游返回码（失败记录）"` + ProviderMessage string `json:"provider_message,omitempty" description:"上游返回信息或错误信息（失败记录）"` + DurationMS int `json:"duration_ms,omitempty" description:"调用耗时毫秒（失败记录）"` + GatewayRawItem } // CardTrafficCollectionListResponse 卡流量采集记录分页结果。 diff --git a/internal/model/dto/gateway_rate_limit_dto.go b/internal/model/dto/gateway_rate_limit_dto.go index 062ad2c0..ad607528 100644 --- a/internal/model/dto/gateway_rate_limit_dto.go +++ b/internal/model/dto/gateway_rate_limit_dto.go @@ -140,6 +140,10 @@ type StopResumeRecordListRequest struct { Page int `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码"` PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"` ICCID string `json:"iccid" query:"iccid" validate:"omitempty,max=30" maxLength:"30" description:"ICCID"` + Identifier string `json:"identifier" query:"identifier" validate:"omitempty,max=50" maxLength:"50" description:"资产标识（虚拟号/ICCID/MSISDN，设备支持虚拟号/IMEI/SN，取其绑定卡）"` + Scene string `json:"scene" query:"scene" validate:"omitempty,max=50" maxLength:"50" description:"场景"` + Reason string `json:"reason" query:"reason" validate:"omitempty,max=50" maxLength:"50" description:"停机原因"` + CommandID string `json:"command_id" query:"command_id" validate:"omitempty,max=64" maxLength:"64" description:"指令号（同一次停复机命令的多次尝试共用）"` CarrierID *uint `json:"carrier_id" query:"carrier_id" validate:"omitempty" description:"运营商账户ID"` Action string `json:"action" query:"action" validate:"omitempty,oneof=stop start" description:"动作 (stop:停机, start:复机)"` Source string `json:"source" query:"source" validate:"omitempty,max=50" maxLength:"50" description:"来源"` @@ -156,6 +160,9 @@ type CardStopResumeRecordListRequest struct { PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"` StartTime time.Time `json:"start_time" query:"start_time" description:"开始时间（RFC3339），为空取最近 31 天"` EndTime time.Time `json:"end_time" query:"end_time" description:"结束时间（RFC3339），为空取当前"` + Action string `json:"action" query:"action" validate:"omitempty,oneof=stop start" description:"动作 (stop:停机, start:复机)"` + Result string `json:"result" query:"result" validate:"omitempty,oneof=success failed unknown already_target skipped" description:"结果 (success:成功, failed:失败, unknown:结果未知, already_target:已是目标状态, skipped:跳过)"` + Source string `json:"source" query:"source" validate:"omitempty,max=50" maxLength:"50" description:"来源"` } // StopResumeRecordResponse 停复机记录响应 diff --git a/internal/model/dto/package_dto.go b/internal/model/dto/package_dto.go index 75faad90..2dcd782a 100644 --- a/internal/model/dto/package_dto.go +++ b/internal/model/dto/package_dto.go @@ -194,5 +194,8 @@ type PackageUsageDetailResponse struct { PackageUsageID uint `json:"package_usage_id" description:"套餐使用记录ID"` PackageName string `json:"package_name" description:"套餐名称"` Records []PackageUsageDailyRecordResponse `json:"records" description:"流量日记录列表"` - TotalUsageMB int64 `json:"total_usage_mb" description:"总使用流量(MB)"` + TotalUsageMB int64 `json:"total_usage_mb" description:"总使用流量(MB)，为筛选范围内全部记录之和"` + Total int64 `json:"total" description:"筛选范围内记录总数"` + Page int `json:"page" description:"页码；未传 page_size 时为 0，返回全部记录"` + PageSize int `json:"page_size" description:"每页条数；未传时为 0，返回全部记录"` } diff --git a/internal/model/dto/record_query_dto.go b/internal/model/dto/record_query_dto.go new file mode 100644 index 00000000..f1f727d0 --- /dev/null +++ b/internal/model/dto/record_query_dto.go @@ -0,0 +1,215 @@ +package dto + +import ( + "encoding/json" + "time" +) + +// RecordRangeQuery 是记录查询的公共条件：资产标识选填（不传查全部），时间范围必填（跨度不超过 31 天），分页。 +type RecordRangeQuery struct { + Identifier string `query:"identifier" validate:"omitempty,max=50" maxLength:"50" description:"资产标识（虚拟号/ICCID/MSISDN，设备支持 IMEI/SN）；不传查全部"` + StartTime string `query:"start_time" validate:"required" required:"true" description:"开始时间（RFC3339 或 2006-01-02 15:04:05，上海时间）"` + EndTime string `query:"end_time" validate:"required" required:"true" description:"结束时间（不含），跨度不超过 31 天"` + Page int `query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码，默认 1"` + PageSize int `query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页条数，默认 20，最大 100"` +} + +// RecordDateRangeQuery 是按日期查询的公共条件：资产标识选填，日期范围必填（含首尾，跨度不超过 31 天），分页。 +type RecordDateRangeQuery struct { + Identifier string `query:"identifier" validate:"omitempty,max=50" maxLength:"50" description:"资产标识（虚拟号/ICCID/MSISDN，设备支持 IMEI/SN）；不传查全部"` + StartDate string `query:"start_date" validate:"required" required:"true" description:"开始日期（2006-01-02，含）"` + EndDate string `query:"end_date" validate:"required" required:"true" description:"结束日期（2006-01-02，含），跨度不超过 31 天"` + Page int `query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码，默认 1"` + PageSize int `query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页条数，默认 20，最大 100"` +} + +// GatewayRawItem 是采集记录保存的网关原始请求与响应。 +type GatewayRawItem struct { + RawRequest json.RawMessage `json:"raw_request" description:"网关明文业务请求（加密前），未收集时为 null"` + HTTPStatus *int `json:"http_status" description:"HTTP 状态码，未拿到响应时为 null"` + GatewayCode *int `json:"gateway_code" description:"网关响应 code"` + GatewayMsg string `json:"gateway_msg" description:"网关响应 msg"` + GatewayTraceID string `json:"gateway_trace_id" description:"网关响应 trace_id"` + RawResponse json.RawMessage `json:"raw_response" description:"网关原始响应体（JSON 且不超过 4KB 时）"` + RawResponseText string `json:"raw_response_text" description:"网关原始响应体文本（非 JSON 或超过 4KB 截断时）"` + RawTruncated bool `json:"raw_truncated" description:"原始响应是否被截断"` + GatewayAttempts int `json:"gateway_attempts" description:"网关客户端内部 HTTP 尝试次数（原始数据为最后一次）"` +} + +// TrafficCollectionListRequest 采集成功记录列表请求。 +type TrafficCollectionListRequest struct { + RecordRangeQuery + CarrierID *uint `query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID"` + Decision string `query:"decision" validate:"omitempty,max=32" description:"写入判断：written/baseline/queue_not_full/slope_negative/out_of_order/settled_existing"` + Source string `query:"source" validate:"omitempty,max=32" description:"来源：polling/manual_sync/business_event"` + Scene string `query:"scene" validate:"omitempty,max=64" description:"场景"` + HasDelta *bool `query:"has_delta" description:"是否产生增量：true 只看增量大于 0，false 只看没有增量"` + StaleGuarded *bool `query:"stale_guarded" description:"是否触发重置日防旧值"` +} + +// TrafficCollectionRecordItem 采集成功记录。 +type TrafficCollectionRecordItem struct { + CardTrafficCollectionItem + IotCardID uint `json:"iot_card_id" description:"卡ID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID（历史记录为 0）"` +} + +// TrafficCollectionRecordListResponse 采集成功记录分页结果。 +type TrafficCollectionRecordListResponse struct { + Items []TrafficCollectionRecordItem `json:"items" description:"采集成功记录"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} + +// TrafficCollectionFailureListRequest 采集失败记录列表请求。 +type TrafficCollectionFailureListRequest struct { + RecordRangeQuery + CarrierID *uint `query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID"` + Result string `query:"result" validate:"omitempty,oneof=failed rate_limited invalid_payload no_reading invalid_reading" description:"失败类型：failed/rate_limited/invalid_payload/no_reading/invalid_reading"` + ErrorCategory string `query:"error_category" validate:"omitempty,max=30" description:"错误类别：timeout/network/http/invalid_response/upstream 等"` + ProviderCode string `query:"provider_code" validate:"omitempty,max=30" description:"上游返回码"` + Source string `query:"source" validate:"omitempty,max=32" description:"来源"` + Scene string `query:"scene" validate:"omitempty,max=64" description:"场景"` +} + +// TrafficCollectionFailureItem 采集失败记录。 +type TrafficCollectionFailureItem struct { + ID uint `json:"id" description:"记录ID"` + IotCardID uint `json:"iot_card_id" description:"卡ID"` + ICCID string `json:"iccid" description:"ICCID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + ObservedAt time.Time `json:"observed_at" description:"采集时间"` + ObservationID string `json:"observation_id" description:"采集 ID（等于对接日志 integration_id）"` + Source string `json:"source" description:"来源"` + Scene string `json:"scene" description:"场景"` + Result string `json:"result" description:"失败类型"` + ErrorCategory string `json:"error_category" description:"错误类别"` + ProviderCode string `json:"provider_code" description:"上游返回码"` + ProviderMessage string `json:"provider_message" description:"上游返回信息或错误信息"` + DurationMS int `json:"duration_ms" description:"调用耗时（毫秒）"` + GatewayRawItem +} + +// TrafficCollectionFailureListResponse 采集失败记录分页结果。 +type TrafficCollectionFailureListResponse struct { + Items []TrafficCollectionFailureItem `json:"items" description:"采集失败记录"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} + +// CardDailyUsageListRequest 卡日流量列表请求。 +type CardDailyUsageListRequest struct { + RecordDateRangeQuery + CarrierID *uint `query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID"` + MinUsageMB *float64 `query:"min_usage_mb" validate:"omitempty,min=0" description:"单日用量下限(MB)，用于排查异常大流量"` +} + +// CardDailyUsageItem 卡日流量。 +type CardDailyUsageItem struct { + ID uint `json:"id" description:"记录ID"` + IotCardID uint `json:"iot_card_id" description:"卡ID"` + ICCID string `json:"iccid" description:"ICCID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + Date string `json:"date" description:"日期"` + UsageMB float64 `json:"usage_mb" description:"当日流量(MB)"` + UpdatedAt time.Time `json:"updated_at" description:"更新时间"` +} + +// CardDailyUsageListResponse 卡日流量分页结果。 +type CardDailyUsageListResponse struct { + Items []CardDailyUsageItem `json:"items" description:"卡日流量"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} + +// PackageUsageDailyRecordListRequest 套餐日流量详单列表请求。 +type PackageUsageDailyRecordListRequest struct { + RecordDateRangeQuery + PackageUsageID *uint `query:"package_usage_id" validate:"omitempty,min=1" description:"套餐使用记录ID"` + PackageID *uint `query:"package_id" validate:"omitempty,min=1" description:"套餐ID"` + CarrierID *uint `query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID（按卡所属账户）"` +} + +// PackageUsageDailyRecordItem 套餐日流量详单。 +type PackageUsageDailyRecordItem struct { + ID uint `json:"id" description:"记录ID"` + PackageUsageID uint `json:"package_usage_id" description:"套餐使用记录ID"` + PackageID uint `json:"package_id" description:"套餐ID"` + IotCardID uint `json:"iot_card_id" description:"卡ID，设备套餐为 0"` + DeviceID uint `json:"device_id" description:"设备ID，卡套餐为 0"` + Date string `json:"date" description:"日期"` + DailyUsageMB int `json:"daily_usage_mb" description:"当日流量(MB)"` + CumulativeUsageMB int64 `json:"cumulative_usage_mb" description:"截止当日累计(MB)"` + UpdatedAt time.Time `json:"updated_at" description:"更新时间"` +} + +// PackageUsageDailyRecordListResponse 套餐日流量详单分页结果。 +type PackageUsageDailyRecordListResponse struct { + Items []PackageUsageDailyRecordItem `json:"items" description:"套餐日流量详单"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} + +// PackageTrafficAdjustmentQueryRequest 套餐流量冲正流水列表请求。 +type PackageTrafficAdjustmentQueryRequest struct { + RecordRangeQuery + PackageUsageID *uint `query:"package_usage_id" validate:"omitempty,min=1" description:"套餐使用记录ID"` + OperatorID *uint `query:"operator_id" validate:"omitempty,min=1" description:"操作人账号ID"` + Direction string `query:"direction" validate:"omitempty,oneof=deduct refund" description:"调整方向：deduct 补扣（增量为正），refund 退回（增量为负）"` +} + +// PackageTrafficAdjustmentRecordItem 冲正流水（跨卡列表）。 +type PackageTrafficAdjustmentRecordItem struct { + PackageTrafficAdjustmentItem + IotCardID uint `json:"iot_card_id" description:"卡ID"` + DeviceID uint `json:"device_id" description:"设备ID"` +} + +// PackageTrafficAdjustmentRecordListResponse 冲正流水分页结果（跨卡列表）。 +type PackageTrafficAdjustmentRecordListResponse struct { + Items []PackageTrafficAdjustmentRecordItem `json:"items" description:"冲正流水"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} + +// RefundPostProcessAttemptListRequest 退款资产后处理尝试记录列表请求。 +type RefundPostProcessAttemptListRequest struct { + RecordRangeQuery + RefundID *uint `query:"refund_id" validate:"omitempty,min=1" description:"退款单ID"` + RefundNo string `query:"refund_no" validate:"omitempty,max=64" description:"退款单号"` + Step string `query:"step" validate:"omitempty,oneof=load invalidate activate check_main stop mark done" description:"处理步骤：load/invalidate/activate/check_main/stop/mark/done"` + Result string `query:"result" validate:"omitempty,oneof=success failed" description:"结果：success/failed"` + ErrorCode string `query:"error_code" validate:"omitempty,max=32" description:"错误码"` + Trigger string `query:"trigger" validate:"omitempty,oneof=outbox manual" description:"触发方式：outbox（含补偿扫描）/manual（人工重试）"` +} + +// RefundPostProcessAttemptItem 退款资产后处理尝试记录。 +type RefundPostProcessAttemptItem struct { + ID uint `json:"id" description:"记录ID"` + RefundID uint `json:"refund_id" description:"退款单ID"` + RefundNo string `json:"refund_no" description:"退款单号"` + Step string `json:"step" description:"处理步骤（失败时为失败步骤，成功为 done）"` + AssetType string `json:"asset_type" description:"资产类型"` + AssetID uint `json:"asset_id" description:"资产ID"` + ICCID string `json:"iccid" description:"ICCID（单卡资产）"` + Attempt int `json:"attempt" description:"尝试序号"` + Trigger string `json:"trigger" description:"触发方式"` + Result string `json:"result" description:"结果"` + ErrorCode string `json:"error_code" description:"错误码"` + ErrorMessage string `json:"error_message" description:"错误信息"` + DurationMS int `json:"duration_ms" description:"耗时（毫秒）"` + CreatedAt time.Time `json:"created_at" description:"记录时间"` +} + +// RefundPostProcessAttemptListResponse 退款资产后处理尝试记录分页结果。 +type RefundPostProcessAttemptListResponse struct { + Items []RefundPostProcessAttemptItem `json:"items" description:"尝试记录"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"页码"` + PageSize int `json:"page_size" description:"每页条数"` +} diff --git a/internal/model/dto/refund_dto.go b/internal/model/dto/refund_dto.go index dc5eb50f..768dd87d 100644 --- a/internal/model/dto/refund_dto.go +++ b/internal/model/dto/refund_dto.go @@ -1,5 +1,7 @@ package dto +import "time" + // CreateRefundRequest 创建退款申请请求 type CreateRefundRequest struct { OrderID uint `json:"order_id" validate:"required" required:"true" description:"关联订单ID"` @@ -60,6 +62,7 @@ type RefundListRequest struct { OrderID *uint `json:"order_id" query:"order_id" validate:"omitempty" description:"关联订单ID"` ShopID *uint `json:"shop_id" query:"shop_id" validate:"omitempty" description:"店铺ID"` AssetIdentifier string `json:"asset_identifier" query:"asset_identifier" validate:"omitempty,max=100" maxLength:"100" description:"资产标识精确检索（ICCID 或 设备虚拟号，非空时精确匹配）"` + AssetPostStatus *int `json:"asset_post_status" query:"asset_post_status" validate:"omitempty,min=0,max=3" minimum:"0" maximum:"3" description:"资产后处理状态 (0:未开始, 1:待重试, 2:已完成, 3:需人工处理)"` } // RefundResponse 退款申请详情响应 @@ -116,6 +119,10 @@ type RefundResponse struct { Remark string `json:"remark,omitempty" description:"审批备注"` CommissionDeducted bool `json:"commission_deducted" description:"佣金是否已回扣"` AssetReset bool `json:"asset_reset" description:"退款后资产处理是否完成"` + AssetPostStatus int `json:"asset_post_status" description:"资产后处理状态 (0:未开始, 1:待重试, 2:已完成, 3:需人工处理)"` + AssetPostAttempts int `json:"asset_post_attempts" description:"资产后处理已尝试次数（自动重试上限 10 次）"` + AssetPostLastError string `json:"asset_post_last_error" description:"资产后处理最后一次错误"` + AssetPostNextAt *time.Time `json:"asset_post_next_at,omitempty" description:"资产后处理下次自动重试时间"` SubmitterID uint `json:"submitter_id" description:"提交人账号ID"` SubmitterName string `json:"submitter_name" description:"提交人账号名称"` ApprovalInstanceID *uint `json:"approval_instance_id,omitempty" description:"通用审批实例ID"` diff --git a/internal/routes/admin.go b/internal/routes/admin.go index 28df5f49..f93f9678 100644 --- a/internal/routes/admin.go +++ b/internal/routes/admin.go @@ -188,6 +188,9 @@ func RegisterAdminRoutes(router fiber.Router, handlers *bootstrap.Handlers, midd if handlers.Refund != nil { registerRefundRoutes(authGroup, handlers.Refund, doc, basePath) } + if handlers.Asset != nil && handlers.Refund != nil { + registerRecordQueryRoutes(authGroup, handlers.Asset, handlers.Refund, doc, basePath) + } if handlers.OrderPackageInvalidate != nil { registerOrderPackageInvalidateRoutes(authGroup, handlers.OrderPackageInvalidate, doc, basePath) } diff --git a/internal/routes/package_usage.go b/internal/routes/package_usage.go index 3a9e7b26..d0ad7039 100644 --- a/internal/routes/package_usage.go +++ b/internal/routes/package_usage.go @@ -14,10 +14,11 @@ func registerPackageUsageRoutes(router fiber.Router, handler *admin.PackageUsage groupPath := basePath + "/package-usage" Register(packageUsage, doc, groupPath, "GET", "/:id/daily-records", handler.GetDailyRecords, RouteSpec{ - Summary: "获取套餐流量详单", - Tags: []string{"套餐使用记录"}, - Input: new(dto.IDReq), - Output: new(dto.PackageUsageDetailResponse), - Auth: true, + Summary: "获取套餐流量详单", + Description: "查询参数：start_date、end_date（2006-01-02，选填）；page、page_size（选填，page_size 最大 100）。传 page_size 时按日期倒序分页并返回 total/page/page_size，不传时按日期正序返回全部（兼容旧调用）；total_usage_mb 为筛选范围内全部记录之和。", + Tags: []string{"套餐使用记录"}, + Input: new(dto.IDReq), + Output: new(dto.PackageUsageDetailResponse), + Auth: true, }) } diff --git a/internal/routes/record_query.go b/internal/routes/record_query.go new file mode 100644 index 00000000..70727650 --- /dev/null +++ b/internal/routes/record_query.go @@ -0,0 +1,46 @@ +package routes + +import ( + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/handler/admin" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/openapi" +) + +// recordQueryDescription 是记录查询接口的公共说明。 +const recordQueryDescription = "identifier 选填：传入时只查该卡（设备取其绑定卡），解析不到资产返回空页；不传查全部，仅平台账号。" + + "时间范围必填，跨度不超过 31 天；按记录时间倒序分页，每页默认 20、最大 100。" + +func registerRecordQueryRoutes(router fiber.Router, asset *admin.AssetHandler, refund *admin.RefundHandler, doc *openapi.Generator, basePath string) { + Register(router, doc, basePath, "GET", "/traffic-collections", asset.ListTrafficCollectionRecords, RouteSpec{ + Summary: "运营商流量采集成功记录", + Description: recordQueryDescription + "只含读数有效的采集，带网关原始请求与响应；在线保留当月加最近 2 个整月。", + Tags: []string{"记录查询"}, Input: new(dto.TrafficCollectionListRequest), Output: new(dto.TrafficCollectionRecordListResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/traffic-collection-failures", asset.ListTrafficCollectionFailures, RouteSpec{ + Summary: "运营商流量采集失败记录", + Description: recordQueryDescription + "含失败、限流、响应无效、无读数、读数无效，带错误类别、上游返回与原始数据；observation_id 等于对接日志 integration_id。", + Tags: []string{"记录查询"}, Input: new(dto.TrafficCollectionFailureListRequest), Output: new(dto.TrafficCollectionFailureListResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/card-daily-usages", asset.ListCardDailyUsages, RouteSpec{ + Summary: "卡日流量记录", + Description: recordQueryDescription + "按日期范围（含首尾）查询；min_usage_mb 可筛异常大流量。", + Tags: []string{"记录查询"}, Input: new(dto.CardDailyUsageListRequest), Output: new(dto.CardDailyUsageListResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/package-usage-daily-records", asset.ListPackageUsageDailyRecords, RouteSpec{ + Summary: "套餐日流量详单记录", + Description: recordQueryDescription + "按日期范围（含首尾）查询；设备标识同时匹配设备套餐与其绑定卡的套餐。", + Tags: []string{"记录查询"}, Input: new(dto.PackageUsageDailyRecordListRequest), Output: new(dto.PackageUsageDailyRecordListResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/package-traffic-adjustments", asset.ListPackageTrafficAdjustmentRecords, RouteSpec{ + Summary: "套餐流量冲正流水记录", + Description: recordQueryDescription + "direction=deduct 为补扣（增量为正），refund 为退回（增量为负）。", + Tags: []string{"记录查询"}, Input: new(dto.PackageTrafficAdjustmentQueryRequest), Output: new(dto.PackageTrafficAdjustmentRecordListResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/refund-post-process-attempts", refund.ListAssetPostProcessAttempts, RouteSpec{ + Summary: "退款资产后处理尝试记录", + Description: "仅平台账号。可按退款单 ID 或单号、资产标识、步骤、结果、错误码、触发方式筛选，不传条件查全部；时间范围必填，跨度不超过 31 天；按时间倒序分页，每页默认 20、最大 100。", + Tags: []string{"记录查询"}, Input: new(dto.RefundPostProcessAttemptListRequest), Output: new(dto.RefundPostProcessAttemptListResponse), Auth: true, + }) +} diff --git a/internal/service/asset/record_query.go b/internal/service/asset/record_query.go new file mode 100644 index 00000000..54df7978 --- /dev/null +++ b/internal/service/asset/record_query.go @@ -0,0 +1,376 @@ +package asset + +import ( + "context" + "encoding/json" + stderrors "errors" + "time" + + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" +) + +const ( + recordDefaultPageSize = 20 + recordMaxPageSize = 100 + recordDateLayout = "2006-01-02" +) + +// recordScope 是资产标识解析后的查询范围：all 为 true 表示不按资产过滤。 +type recordScope struct { + all bool + cardIDs []uint + deviceID uint +} + +func (r recordScope) empty() bool { return !r.all && len(r.cardIDs) == 0 && r.deviceID == 0 } + +// resolveRecordScope 把选填的资产标识解析为卡 ID（设备取其绑定卡）；不传标识查全部时只允许平台账号。 +// 标识解析不到资产时返回空范围，调用方直接返回空页。 +func (s *Service) resolveRecordScope(ctx context.Context, identifier string) (recordScope, error) { + if identifier == "" { + userType := middleware.GetUserTypeFromContext(ctx) + if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform { + return recordScope{}, errors.New(errors.CodeForbidden, "只有平台账号可以不带资产标识查询全部记录") + } + return recordScope{all: true}, nil + } + asset, err := s.Resolve(ctx, identifier, false) + if err != nil { + var appErr *errors.AppError + if stderrors.As(err, &appErr) && appErr.Code == errors.CodeNotFound { + return recordScope{}, nil + } + return recordScope{}, err + } + if asset.AssetType == "card" { + return recordScope{cardIDs: []uint{asset.AssetID}}, nil + } + var cardIDs []uint + if err := s.db.WithContext(ctx).Model(&model.DeviceSimBinding{}). + Where("device_id = ? AND bind_status = ?", asset.AssetID, constants.BindStatusBound). + Pluck("iot_card_id", &cardIDs).Error; err != nil { + return recordScope{}, errors.Wrap(errors.CodeDatabaseError, err, "查询设备绑定卡失败") + } + return recordScope{cardIDs: cardIDs, deviceID: asset.AssetID}, nil +} + +// parseRecordTimeRange 解析必填时间范围（结束不含，跨度不超过 31 天）。 +func parseRecordTimeRange(start, end string) (time.Time, time.Time, error) { + from, err := parseCollectionTime(start) + if err != nil { + return time.Time{}, time.Time{}, err + } + to, err := parseCollectionTime(end) + if err != nil { + return time.Time{}, time.Time{}, err + } + if !to.After(from) || to.Sub(from) > maxCollectionQuerySpan { + return time.Time{}, time.Time{}, errors.New(errors.CodeInvalidParam, "结束时间必须晚于开始时间，且跨度不超过 31 天") + } + return from, to, nil +} + +// parseRecordDateRange 解析必填日期范围（含首尾，跨度不超过 31 天）。 +func parseRecordDateRange(start, end string) (time.Time, time.Time, error) { + location := cardflow.ShanghaiDate(time.Now()).Location() + from, err := time.ParseInLocation(recordDateLayout, start, location) + if err != nil { + return time.Time{}, time.Time{}, errors.New(errors.CodeInvalidParam, "开始日期格式无效，应为 2006-01-02") + } + to, err := time.ParseInLocation(recordDateLayout, end, location) + if err != nil { + return time.Time{}, time.Time{}, errors.New(errors.CodeInvalidParam, "结束日期格式无效，应为 2006-01-02") + } + if to.Before(from) || to.Sub(from) > maxCollectionQuerySpan { + return time.Time{}, time.Time{}, errors.New(errors.CodeInvalidParam, "结束日期不能早于开始日期，且跨度不超过 31 天") + } + return from, to, nil +} + +func recordPage(page, size, maxSize int) (int, int) { + if page <= 0 { + page = 1 + } + if size <= 0 { + size = recordDefaultPageSize + } + if size > maxSize { + size = maxSize + } + return page, size +} + +// pageQuery 统计总数后按排序取当前页。 +func pageQuery[T any](query *gorm.DB, order string, page, size int) ([]T, int64, error) { + var total int64 + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + rows := make([]T, 0, size) + if total == 0 { + return rows, 0, nil + } + err := query.Order(order).Offset((page - 1) * size).Limit(size).Find(&rows).Error + return rows, total, err +} + +func gatewayRawItem(raw model.CollectionRaw) dto.GatewayRawItem { + return dto.GatewayRawItem{ + RawRequest: json.RawMessage(raw.RawRequest), HTTPStatus: raw.HTTPStatus, GatewayCode: raw.GatewayCode, + GatewayMsg: raw.GatewayMsg, GatewayTraceID: raw.GatewayTraceID, RawResponse: json.RawMessage(raw.RawResponse), + RawResponseText: raw.RawResponseText, RawTruncated: raw.RawTruncated, GatewayAttempts: raw.GatewayAttempts, + } +} + +func collectionSuccessItem(row model.CardTrafficCollection) dto.CardTrafficCollectionItem { + return dto.CardTrafficCollectionItem{ + ID: row.ID, ObservedAt: row.ObservedAt, ObservationID: row.ObservationID, Source: row.Source, + Scene: row.Scene, Result: row.Result, VendorUsedMB: row.VendorUsedMB, QueuePoints: json.RawMessage(row.QueuePoints), + Decision: row.Decision, MonthUsedMB: row.MonthUsedMB, StaleGuarded: row.StaleGuarded, + DeltaMB: row.DeltaMB, EventID: row.EventID, RecordKind: "success", GatewayRawItem: gatewayRawItem(row.CollectionRaw), + } +} + +func collectionFailureItem(row model.CardTrafficCollectionFailure) dto.TrafficCollectionFailureItem { + return dto.TrafficCollectionFailureItem{ + ID: row.ID, IotCardID: row.IotCardID, ICCID: row.ICCID, CarrierID: row.CarrierID, ObservedAt: row.ObservedAt, + ObservationID: row.ObservationID, Source: row.Source, Scene: row.Scene, Result: row.Result, + ErrorCategory: row.ErrorCategory, ProviderCode: row.ProviderCode, ProviderMessage: row.ProviderMessage, + DurationMS: row.DurationMS, GatewayRawItem: gatewayRawItem(row.CollectionRaw), + } +} + +// ListTrafficCollectionRecords 查询采集成功记录（可查单个也可查全部）。 +func (s *Service) ListTrafficCollectionRecords(ctx context.Context, req *dto.TrafficCollectionListRequest) (*dto.TrafficCollectionRecordListResponse, error) { + from, to, err := parseRecordTimeRange(req.StartTime, req.EndTime) + if err != nil { + return nil, err + } + page, size := recordPage(req.Page, req.PageSize, recordMaxPageSize) + resp := &dto.TrafficCollectionRecordListResponse{Items: []dto.TrafficCollectionRecordItem{}, Page: page, PageSize: size} + scope, err := s.resolveRecordScope(ctx, req.Identifier) + if err != nil || scope.empty() { + return resp, err + } + query := s.db.WithContext(ctx).Model(&model.CardTrafficCollection{}). + Where("observed_at >= ? AND observed_at < ? AND result = ?", from, to, constants.TrafficCollectionResultOK) + if !scope.all { + query = query.Where("iot_card_id IN ?", scope.cardIDs) + } + if req.CarrierID != nil { + query = query.Where("carrier_id = ?", *req.CarrierID) + } + query = applyStringFilters(query, map[string]string{"decision": req.Decision, "source": req.Source, "scene": req.Scene}) + if req.HasDelta != nil { + if *req.HasDelta { + query = query.Where("delta_mb > 0") + } else { + query = query.Where("(delta_mb IS NULL OR delta_mb = 0)") + } + } + if req.StaleGuarded != nil { + query = query.Where("stale_guarded = ?", *req.StaleGuarded) + } + rows, total, err := pageQuery[model.CardTrafficCollection](query, "observed_at DESC, id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询采集成功记录失败") + } + for _, row := range rows { + resp.Items = append(resp.Items, dto.TrafficCollectionRecordItem{ + CardTrafficCollectionItem: collectionSuccessItem(row), IotCardID: row.IotCardID, CarrierID: row.CarrierID, + }) + } + resp.Total = total + return resp, nil +} + +// ListTrafficCollectionFailures 查询采集失败记录（可查单个也可查全部）。 +func (s *Service) ListTrafficCollectionFailures(ctx context.Context, req *dto.TrafficCollectionFailureListRequest) (*dto.TrafficCollectionFailureListResponse, error) { + from, to, err := parseRecordTimeRange(req.StartTime, req.EndTime) + if err != nil { + return nil, err + } + page, size := recordPage(req.Page, req.PageSize, recordMaxPageSize) + resp := &dto.TrafficCollectionFailureListResponse{Items: []dto.TrafficCollectionFailureItem{}, Page: page, PageSize: size} + scope, err := s.resolveRecordScope(ctx, req.Identifier) + if err != nil || scope.empty() { + return resp, err + } + query := s.db.WithContext(ctx).Model(&model.CardTrafficCollectionFailure{}).Where("observed_at >= ? AND observed_at < ?", from, to) + if !scope.all { + query = query.Where("iot_card_id IN ?", scope.cardIDs) + } + if req.CarrierID != nil { + query = query.Where("carrier_id = ?", *req.CarrierID) + } + query = applyStringFilters(query, map[string]string{ + "result": req.Result, "error_category": req.ErrorCategory, "provider_code": req.ProviderCode, "source": req.Source, "scene": req.Scene, + }) + rows, total, err := pageQuery[model.CardTrafficCollectionFailure](query, "observed_at DESC, id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询采集失败记录失败") + } + for _, row := range rows { + resp.Items = append(resp.Items, collectionFailureItem(row)) + } + resp.Total = total + return resp, nil +} + +// ListCardDailyUsages 查询卡日流量（可查单个也可查全部）。 +func (s *Service) ListCardDailyUsages(ctx context.Context, req *dto.CardDailyUsageListRequest) (*dto.CardDailyUsageListResponse, error) { + from, to, err := parseRecordDateRange(req.StartDate, req.EndDate) + if err != nil { + return nil, err + } + page, size := recordPage(req.Page, req.PageSize, recordMaxPageSize) + resp := &dto.CardDailyUsageListResponse{Items: []dto.CardDailyUsageItem{}, Page: page, PageSize: size} + scope, err := s.resolveRecordScope(ctx, req.Identifier) + if err != nil || scope.empty() { + return resp, err + } + type row struct { + model.CardDailyUsage + ICCID string + CarrierID uint + } + query := s.db.WithContext(ctx).Table("tb_card_daily_usage AS u"). + Select("u.*, c.iccid, c.carrier_id").Joins("LEFT JOIN tb_iot_card c ON c.id = u.iot_card_id"). + Where("u.date >= ? AND u.date <= ?", from.Format(recordDateLayout), to.Format(recordDateLayout)) + if !scope.all { + query = query.Where("u.iot_card_id IN ?", scope.cardIDs) + } + if req.CarrierID != nil { + query = query.Where("c.carrier_id = ?", *req.CarrierID) + } + if req.MinUsageMB != nil { + query = query.Where("u.usage_mb >= ?", *req.MinUsageMB) + } + rows, total, err := pageQuery[row](query, "u.date DESC, u.id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡日流量失败") + } + for _, item := range rows { + resp.Items = append(resp.Items, dto.CardDailyUsageItem{ + ID: item.ID, IotCardID: item.IotCardID, ICCID: item.ICCID, CarrierID: item.CarrierID, + Date: item.Date.Format(recordDateLayout), UsageMB: item.UsageMB, UpdatedAt: item.UpdatedAt, + }) + } + resp.Total = total + return resp, nil +} + +// ListPackageUsageDailyRecords 查询套餐日流量详单（可查单个也可查全部）。 +func (s *Service) ListPackageUsageDailyRecords(ctx context.Context, req *dto.PackageUsageDailyRecordListRequest) (*dto.PackageUsageDailyRecordListResponse, error) { + from, to, err := parseRecordDateRange(req.StartDate, req.EndDate) + if err != nil { + return nil, err + } + page, size := recordPage(req.Page, req.PageSize, recordMaxPageSize) + resp := &dto.PackageUsageDailyRecordListResponse{Items: []dto.PackageUsageDailyRecordItem{}, Page: page, PageSize: size} + scope, err := s.resolveRecordScope(ctx, req.Identifier) + if err != nil || scope.empty() { + return resp, err + } + type row struct { + model.PackageUsageDailyRecord + PackageID uint + IotCardID uint + DeviceID uint + } + query := s.db.WithContext(ctx).Table("tb_package_usage_daily_record AS r"). + Select("r.*, pu.package_id, COALESCE(pu.iot_card_id, 0) AS iot_card_id, COALESCE(pu.device_id, 0) AS device_id"). + Joins("JOIN tb_package_usage pu ON pu.id = r.package_usage_id"). + Where("r.date >= ? AND r.date <= ?", from.Format(recordDateLayout), to.Format(recordDateLayout)) + if !scope.all { + if scope.deviceID > 0 { + query = query.Where("(pu.device_id = ? OR pu.iot_card_id IN ?)", scope.deviceID, append([]uint{0}, scope.cardIDs...)) + } else { + query = query.Where("pu.iot_card_id IN ?", scope.cardIDs) + } + } + if req.PackageUsageID != nil { + query = query.Where("r.package_usage_id = ?", *req.PackageUsageID) + } + if req.PackageID != nil { + query = query.Where("pu.package_id = ?", *req.PackageID) + } + if req.CarrierID != nil { + query = query.Where("pu.iot_card_id IN (SELECT id FROM tb_iot_card WHERE carrier_id = ?)", *req.CarrierID) + } + rows, total, err := pageQuery[row](query, "r.date DESC, r.id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询套餐日流量详单失败") + } + for _, item := range rows { + resp.Items = append(resp.Items, dto.PackageUsageDailyRecordItem{ + ID: item.ID, PackageUsageID: item.PackageUsageID, PackageID: item.PackageID, IotCardID: item.IotCardID, + DeviceID: item.DeviceID, Date: item.Date.Format(recordDateLayout), DailyUsageMB: item.DailyUsageMB, + CumulativeUsageMB: item.CumulativeUsageMB, UpdatedAt: item.UpdatedAt, + }) + } + resp.Total = total + return resp, nil +} + +// ListPackageTrafficAdjustmentRecords 查询套餐流量冲正流水（可查单个也可查全部）。 +func (s *Service) ListPackageTrafficAdjustmentRecords(ctx context.Context, req *dto.PackageTrafficAdjustmentQueryRequest) (*dto.PackageTrafficAdjustmentRecordListResponse, error) { + from, to, err := parseRecordTimeRange(req.StartTime, req.EndTime) + if err != nil { + return nil, err + } + page, size := recordPage(req.Page, req.PageSize, recordMaxPageSize) + resp := &dto.PackageTrafficAdjustmentRecordListResponse{Items: []dto.PackageTrafficAdjustmentRecordItem{}, Page: page, PageSize: size} + scope, err := s.resolveRecordScope(ctx, req.Identifier) + if err != nil || scope.empty() { + return resp, err + } + query := s.db.WithContext(ctx).Model(&model.PackageUsageAdjustment{}).Where("created_at >= ? AND created_at < ?", from, to) + if !scope.all { + if scope.deviceID > 0 { + query = query.Where("(device_id = ? OR iot_card_id IN ?)", scope.deviceID, append([]uint{0}, scope.cardIDs...)) + } else { + query = query.Where("iot_card_id IN ?", scope.cardIDs) + } + } + if req.PackageUsageID != nil { + query = query.Where("package_usage_id = ?", *req.PackageUsageID) + } + if req.OperatorID != nil { + query = query.Where("operator_id = ?", *req.OperatorID) + } + switch req.Direction { + case "deduct": + query = query.Where("delta_mb > 0") + case "refund": + query = query.Where("delta_mb < 0") + } + rows, total, err := pageQuery[model.PackageUsageAdjustment](query, "created_at DESC, id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询套餐流量冲正流水失败") + } + for _, row := range rows { + resp.Items = append(resp.Items, dto.PackageTrafficAdjustmentRecordItem{ + PackageTrafficAdjustmentItem: toTrafficAdjustmentItem(row), IotCardID: row.IotCardID, DeviceID: row.DeviceID, + }) + } + resp.Total = total + return resp, nil +} + +// applyStringFilters 对非空字符串条件追加等值过滤（列名为内部常量，不来自请求）。 +func applyStringFilters(query *gorm.DB, filters map[string]string) *gorm.DB { + for column, value := range filters { + if value != "" { + query = query.Where(column+" = ?", value) + } + } + return query +} diff --git a/internal/service/asset/traffic_adjustment.go b/internal/service/asset/traffic_adjustment.go index a70ebdac..6ec18385 100644 --- a/internal/service/asset/traffic_adjustment.go +++ b/internal/service/asset/traffic_adjustment.go @@ -175,6 +175,13 @@ func (s *Service) ListPackageTrafficAdjustments(ctx context.Context, assetType s size = 20 } query := s.db.WithContext(ctx).Model(&model.PackageUsageAdjustment{}).Where("package_usage_id = ?", req.PackageUsageID) + if req.StartTime != "" || req.EndTime != "" { + from, to, err := parseRecordTimeRange(req.StartTime, req.EndTime) + if err != nil { + return nil, err + } + query = query.Where("created_at >= ? AND created_at < ?", from, to) + } var total int64 if err := query.Count(&total).Error; err != nil { return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询冲正流水失败") diff --git a/internal/service/asset/traffic_collection_query.go b/internal/service/asset/traffic_collection_query.go index 7f3401b6..6be9207c 100644 --- a/internal/service/asset/traffic_collection_query.go +++ b/internal/service/asset/traffic_collection_query.go @@ -2,12 +2,15 @@ package asset import ( "context" - "encoding/json" + "sort" "time" + "gorm.io/gorm" + "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" + "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/model/dto" - "github.com/break/junhong_cmp_fiber/internal/store/postgres" + "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" ) @@ -37,22 +40,72 @@ func (s *Service) ListCardTrafficCollections(ctx context.Context, cardID uint, r if size <= 0 { size = 50 } - rows, total, err := postgres.NewCardTrafficCollectionStore(s.db).List(ctx, postgres.CardTrafficCollectionQuery{ - CardID: cardID, From: from, To: to, Page: page, PageSize: size, - }) - if err != nil { + success := s.db.WithContext(ctx).Model(&model.CardTrafficCollection{}). + Where("iot_card_id = ? AND observed_at >= ? AND observed_at < ? AND result = ?", cardID, from, to, constants.TrafficCollectionResultOK) + success = applyStringFilters(success, map[string]string{"decision": req.Decision, "source": req.Source, "scene": req.Scene}) + if !req.IncludeFailures { + rows, total, err := pageQuery[model.CardTrafficCollection](success, "observed_at DESC, id DESC", page, size) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡流量采集记录失败") + } + items := make([]dto.CardTrafficCollectionItem, 0, len(rows)) + for _, row := range rows { + items = append(items, collectionSuccessItem(row)) + } + return &dto.CardTrafficCollectionListResponse{Items: items, Total: total}, nil + } + return s.listCardCollectionsWithFailures(ctx, success, cardID, from, to, req, page, size) +} + +// listCardCollectionsWithFailures 合并单卡的成功与失败记录，按采集时间倒序分页（单卡 31 天内数据量有限，取前 page*size 条归并）。 +func (s *Service) listCardCollectionsWithFailures(ctx context.Context, success *gorm.DB, cardID uint, from, to time.Time, + req *dto.CardTrafficCollectionListRequest, page, size int) (*dto.CardTrafficCollectionListResponse, error) { + failure := s.db.WithContext(ctx).Model(&model.CardTrafficCollectionFailure{}). + Where("iot_card_id = ? AND observed_at >= ? AND observed_at < ?", cardID, from, to) + failure = applyStringFilters(failure, map[string]string{"source": req.Source, "scene": req.Scene}) + var successTotal, failureTotal int64 + if err := success.Count(&successTotal).Error; err != nil { return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡流量采集记录失败") } - items := make([]dto.CardTrafficCollectionItem, 0, len(rows)) - for _, row := range rows { - items = append(items, dto.CardTrafficCollectionItem{ - ID: row.ID, ObservedAt: row.ObservedAt, ObservationID: row.ObservationID, Source: row.Source, - Scene: row.Scene, Result: row.Result, VendorUsedMB: row.VendorUsedMB, QueuePoints: json.RawMessage(row.QueuePoints), - Decision: row.Decision, MonthUsedMB: row.MonthUsedMB, StaleGuarded: row.StaleGuarded, - DeltaMB: row.DeltaMB, EventID: row.EventID, + if err := failure.Count(&failureTotal).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡流量采集失败记录失败") + } + window := page * size + var successRows []model.CardTrafficCollection + if err := success.Order("observed_at DESC, id DESC").Limit(window).Find(&successRows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡流量采集记录失败") + } + var failureRows []model.CardTrafficCollectionFailure + if err := failure.Order("observed_at DESC, id DESC").Limit(window).Find(&failureRows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡流量采集失败记录失败") + } + merged := make([]dto.CardTrafficCollectionItem, 0, len(successRows)+len(failureRows)) + for _, row := range successRows { + merged = append(merged, collectionSuccessItem(row)) + } + for _, row := range failureRows { + f := collectionFailureItem(row) + merged = append(merged, dto.CardTrafficCollectionItem{ + ID: f.ID, ObservedAt: f.ObservedAt, ObservationID: f.ObservationID, Source: f.Source, Scene: f.Scene, + Result: f.Result, RecordKind: "failure", ErrorCategory: f.ErrorCategory, ProviderCode: f.ProviderCode, + ProviderMessage: f.ProviderMessage, DurationMS: f.DurationMS, GatewayRawItem: f.GatewayRawItem, }) } - return &dto.CardTrafficCollectionListResponse{Items: items, Total: total}, nil + sort.SliceStable(merged, func(i, j int) bool { + if !merged[i].ObservedAt.Equal(merged[j].ObservedAt) { + return merged[i].ObservedAt.After(merged[j].ObservedAt) + } + return merged[i].ID > merged[j].ID + }) + start := (page - 1) * size + if start > len(merged) { + start = len(merged) + } + end := start + size + if end > len(merged) { + end = len(merged) + } + return &dto.CardTrafficCollectionListResponse{Items: merged[start:end], Total: successTotal + failureTotal}, nil } func parseCollectionTime(value string) (time.Time, error) { diff --git a/internal/service/package/daily_record_service.go b/internal/service/package/daily_record_service.go index f755713d..94b074df 100644 --- a/internal/service/package/daily_record_service.go +++ b/internal/service/package/daily_record_service.go @@ -36,7 +36,8 @@ func NewDailyRecordService( // GetDailyRecords 任务 13.2-13.5: 查询套餐流量详单 // 查询指定套餐使用记录的日流量明细 -func (s *DailyRecordService) GetDailyRecords(ctx context.Context, packageUsageID uint, startDate, endDate string) (*dto.PackageUsageDetailResponse, error) { +// GetDailyRecords 查询套餐流量详单；page_size 大于 0 时按日期倒序分页，否则按日期正序返回全部（兼容旧调用）。 +func (s *DailyRecordService) GetDailyRecords(ctx context.Context, packageUsageID uint, startDate, endDate string, page, pageSize int) (*dto.PackageUsageDetailResponse, error) { // 查询套餐使用记录 var usage model.PackageUsage if err := s.db.WithContext(ctx).First(&usage, packageUsageID).Error; err != nil { @@ -73,7 +74,27 @@ func (s *DailyRecordService) GetDailyRecords(ctx context.Context, packageUsageID query = query.Where("date <= ?", end) } - if err := query.Order("date ASC").Find(&records).Error; err != nil { + var summary struct { + Total int64 + Usage int64 + } + if err := query.Session(&gorm.Session{}).Model(&model.PackageUsageDailyRecord{}). + Select("count(*) AS total, COALESCE(sum(daily_usage_mb), 0) AS usage").Scan(&summary).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "统计日流量记录失败") + } + if pageSize > 0 { + if pageSize > 100 { + pageSize = 100 + } + if page <= 0 { + page = 1 + } + query = query.Order("date DESC").Offset((page - 1) * pageSize).Limit(pageSize) + } else { + page = 0 + query = query.Order("date ASC") + } + if err := query.Find(&records).Error; err != nil { return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询日流量记录失败") } @@ -87,14 +108,17 @@ func (s *DailyRecordService) GetDailyRecords(ctx context.Context, packageUsageID DailyUsageMB: record.DailyUsageMB, CumulativeUsageMB: record.CumulativeUsageMB, } - totalUsageMB += int64(record.DailyUsageMB) } + totalUsageMB = summary.Usage response := &dto.PackageUsageDetailResponse{ PackageUsageID: packageUsageID, PackageName: pkg.PackageName, Records: recordResponses, TotalUsageMB: totalUsageMB, + Total: summary.Total, + Page: page, + PageSize: pageSize, } return response, nil diff --git a/internal/service/polling/gateway_call_query.go b/internal/service/polling/gateway_call_query.go index 8e5d0dd5..d1faa53c 100644 --- a/internal/service/polling/gateway_call_query.go +++ b/internal/service/polling/gateway_call_query.go @@ -41,11 +41,19 @@ func (s *GatewayRateLimitService) ListStopResumeRecords(ctx context.Context, req return nil, errors.New(errors.CodeInvalidParam, "停复机记录查询参数不合法：时间跨度须大于 0 且不超过 31 天") } page, size := normalizePage(req.Page, req.PageSize) - return s.listRecords(ctx, postgres.StopResumeRecordFilter{ + filter := postgres.StopResumeRecordFilter{ ICCID: strings.TrimSpace(req.ICCID), CarrierID: req.CarrierID, Action: req.Action, Source: req.Source, Result: req.Result, ErrorCategory: req.ErrorCategory, StartTime: req.StartTime, EndTime: req.EndTime, - Page: page, PageSize: size, - }) + Scene: req.Scene, Reason: req.Reason, CommandID: strings.TrimSpace(req.CommandID), Page: page, PageSize: size, + } + if identifier := strings.TrimSpace(req.Identifier); identifier != "" { + cardIDs, err := s.cardIDsByIdentifier(ctx, identifier) + if err != nil { + return nil, err + } + filter.CardIDs = cardIDs + } + return s.listRecords(ctx, filter) } // ListCardStopResumeRecords 查询单卡停复机历史；未传时间默认最近 31 天。 @@ -73,7 +81,29 @@ func (s *GatewayRateLimitService) ListCardStopResumeRecords(ctx context.Context, } page, size := normalizePage(req.Page, req.PageSize) cardID := req.ID - return s.listRecords(ctx, postgres.StopResumeRecordFilter{CardID: &cardID, StartTime: start, EndTime: end, Page: page, PageSize: size}) + return s.listRecords(ctx, postgres.StopResumeRecordFilter{CardID: &cardID, StartTime: start, EndTime: end, Page: page, PageSize: size, + Action: req.Action, Result: req.Result, Source: req.Source}) +} + +// cardIDsByIdentifier 把资产标识解析为卡：卡按虚拟号、ICCID、MSISDN 匹配，设备按虚拟号、IMEI、SN 匹配后取其绑定卡。 +// 匹配不到时返回空切片（查询结果为空页）。 +func (s *GatewayRateLimitService) cardIDsByIdentifier(ctx context.Context, identifier string) ([]uint, error) { + cardIDs := []uint{} + if err := s.db.WithContext(ctx).Model(&model.IotCard{}). + Where("iccid = ? OR virtual_no = ? OR msisdn = ?", identifier, identifier, identifier). + Pluck("id", &cardIDs).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "按资产标识查询卡失败") + } + if len(cardIDs) > 0 { + return cardIDs, nil + } + if err := s.db.WithContext(ctx).Model(&model.DeviceSimBinding{}). + Joins("JOIN tb_device d ON d.id = tb_device_sim_binding.device_id AND d.deleted_at IS NULL"). + Where("(d.virtual_no = ? OR d.imei = ? OR d.sn = ?) AND tb_device_sim_binding.bind_status = ?", identifier, identifier, identifier, constants.BindStatusBound). + Pluck("tb_device_sim_binding.iot_card_id", &cardIDs).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "按资产标识查询设备绑定卡失败") + } + return cardIDs, nil } func (s *GatewayRateLimitService) listRecords(ctx context.Context, filter postgres.StopResumeRecordFilter) (*dto.StopResumeRecordPageResult, error) { diff --git a/internal/service/refund/asset_post_attempt_query.go b/internal/service/refund/asset_post_attempt_query.go new file mode 100644 index 00000000..9ea31065 --- /dev/null +++ b/internal/service/refund/asset_post_attempt_query.go @@ -0,0 +1,93 @@ +package refund + +import ( + "context" + "strings" + "time" + + "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" +) + +// attemptQueryMaxSpan 是尝试记录查询的最长时间跨度。 +const attemptQueryMaxSpan = 31 * 24 * time.Hour + +var attemptTimeLayouts = []string{time.RFC3339, "2006-01-02 15:04:05", "2006-01-02"} + +// ListAssetPostProcessAttempts 分页查询退款资产后处理尝试记录（仅平台账号；可查单张退款单或单个资产，也可查全部）。 +func (s *Service) ListAssetPostProcessAttempts(ctx context.Context, req *dto.RefundPostProcessAttemptListRequest) (*dto.RefundPostProcessAttemptListResponse, error) { + userType := middleware.GetUserTypeFromContext(ctx) + if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform { + return nil, errors.New(errors.CodeForbidden, "只有平台账号可以查询退款资产后处理尝试记录") + } + from, err := parseAttemptTime(req.StartTime) + if err != nil { + return nil, err + } + to, err := parseAttemptTime(req.EndTime) + if err != nil { + return nil, err + } + if !to.After(from) || to.Sub(from) > attemptQueryMaxSpan { + return nil, errors.New(errors.CodeInvalidParam, "结束时间必须晚于开始时间，且跨度不超过 31 天") + } + page, size := req.Page, req.PageSize + if page <= 0 { + page = 1 + } + if size <= 0 || size > 100 { + size = 20 + } + resp := &dto.RefundPostProcessAttemptListResponse{Items: []dto.RefundPostProcessAttemptItem{}, Page: page, PageSize: size} + query := s.db.WithContext(ctx).Model(&model.RefundPostProcessAttempt{}).Where("created_at >= ? AND created_at < ?", from, to) + if req.RefundID != nil { + query = query.Where("refund_id = ?", *req.RefundID) + } + if refundNo := strings.TrimSpace(req.RefundNo); refundNo != "" { + query = query.Where("refund_no = ?", refundNo) + } + if identifier := strings.TrimSpace(req.Identifier); identifier != "" { + query = query.Where(`(iccid = ? OR (asset_type = 'iot_card' AND asset_id IN (SELECT id FROM tb_iot_card WHERE virtual_no = ? OR msisdn = ?)) + OR (asset_type = 'device' AND asset_id IN (SELECT id FROM tb_device WHERE virtual_no = ? OR imei = ? OR sn = ?)))`, + identifier, identifier, identifier, identifier, identifier, identifier) + } + for column, value := range map[string]string{"step": req.Step, "result": req.Result, "error_code": req.ErrorCode, "trigger": req.Trigger} { + if value != "" { + query = query.Where(column+" = ?", value) + } + } + var total int64 + if err := query.Count(&total).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询退款资产后处理尝试记录失败") + } + resp.Total = total + if total == 0 { + return resp, nil + } + var rows []model.RefundPostProcessAttempt + if err := query.Order("created_at DESC, id DESC").Offset((page - 1) * size).Limit(size).Find(&rows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询退款资产后处理尝试记录失败") + } + for _, row := range rows { + resp.Items = append(resp.Items, dto.RefundPostProcessAttemptItem{ + ID: row.ID, RefundID: row.RefundID, RefundNo: row.RefundNo, Step: row.Step, AssetType: row.AssetType, + AssetID: row.AssetID, ICCID: row.ICCID, Attempt: row.Attempt, Trigger: row.Trigger, Result: row.Result, + ErrorCode: row.ErrorCode, ErrorMessage: row.ErrorMessage, DurationMS: row.DurationMS, CreatedAt: row.CreatedAt, + }) + } + return resp, nil +} + +func parseAttemptTime(value string) (time.Time, error) { + location := cardflow.ShanghaiDate(time.Now()).Location() + for _, layout := range attemptTimeLayouts { + if t, err := time.ParseInLocation(layout, value, location); err == nil { + return t, nil + } + } + return time.Time{}, errors.New(errors.CodeInvalidParam, "时间格式无效") +} diff --git a/internal/service/refund/service.go b/internal/service/refund/service.go index f5320a2f..53d98d6b 100644 --- a/internal/service/refund/service.go +++ b/internal/service/refund/service.go @@ -268,6 +268,7 @@ func (s *Service) List(ctx context.Context, req *dto.RefundListRequest) (*dto.Re OrderID: req.OrderID, ShopID: req.ShopID, AssetIdentifier: req.AssetIdentifier, + AssetPostStatus: req.AssetPostStatus, } requests, total, err := s.refundStore.List(ctx, opts, filters) @@ -1375,6 +1376,10 @@ func buildRefundResponse(r *model.RefundRequest) *dto.RefundResponse { Remark: r.Remark, CommissionDeducted: r.CommissionDeducted, AssetReset: r.AssetReset, + AssetPostStatus: r.AssetPostStatus, + AssetPostAttempts: r.AssetPostAttempts, + AssetPostLastError: r.AssetPostLastError, + AssetPostNextAt: r.AssetPostNextAt, SubmitterID: r.Creator, ApprovalInstanceID: r.ApprovalInstanceID, Creator: r.Creator, diff --git a/internal/store/postgres/card_traffic_collection_store.go b/internal/store/postgres/card_traffic_collection_store.go index 30c13d86..e3484a56 100644 --- a/internal/store/postgres/card_traffic_collection_store.go +++ b/internal/store/postgres/card_traffic_collection_store.go @@ -2,19 +2,12 @@ package postgres import ( "context" - "fmt" - "strings" - "time" "gorm.io/gorm" - "github.com/break/junhong_cmp_fiber/internal/domain/cardflow" "github.com/break/junhong_cmp_fiber/internal/model" ) -// collectionPartitionPrefix 是采集记录月分区表名前缀，后接 YYYYMM。 -const collectionPartitionPrefix = "tb_card_traffic_collection_" - // CardTrafficCollectionStore 卡流量采集记录存储。 type CardTrafficCollectionStore struct { db *gorm.DB @@ -30,68 +23,17 @@ func (s *CardTrafficCollectionStore) Create(ctx context.Context, record *model.C return s.db.WithContext(ctx).Create(record).Error } -// CardTrafficCollectionQuery 是采集记录查询条件。 -type CardTrafficCollectionQuery struct { - CardID uint - From, To time.Time - Page int - PageSize int +// CardTrafficCollectionFailureStore 卡流量采集失败记录存储。 +type CardTrafficCollectionFailureStore struct { + db *gorm.DB } -// List 按卡与采集时间范围分页查询采集记录，按采集时间倒序。 -func (s *CardTrafficCollectionStore) List(ctx context.Context, query CardTrafficCollectionQuery) ([]model.CardTrafficCollection, int64, error) { - db := s.db.WithContext(ctx).Model(&model.CardTrafficCollection{}). - Where("iot_card_id = ? AND observed_at >= ? AND observed_at < ?", query.CardID, query.From, query.To) - var total int64 - if err := db.Count(&total).Error; err != nil { - return nil, 0, err - } - var rows []model.CardTrafficCollection - err := db.Order("observed_at DESC, id DESC").Offset((query.Page - 1) * query.PageSize).Limit(query.PageSize).Find(&rows).Error - return rows, total, err +// NewCardTrafficCollectionFailureStore 创建采集失败记录存储。 +func NewCardTrafficCollectionFailureStore(db *gorm.DB) *CardTrafficCollectionFailureStore { + return &CardTrafficCollectionFailureStore{db: db} } -// FindByObservationID 按采集 ID 查询采集记录。 -func (s *CardTrafficCollectionStore) FindByObservationID(ctx context.Context, observationID string) ([]model.CardTrafficCollection, error) { - var rows []model.CardTrafficCollection - err := s.db.WithContext(ctx).Where("observation_id = ?", observationID).Order("observed_at DESC").Find(&rows).Error - return rows, err -} - -// MaintainPartitions 预建本月起 aheadMonths 个月的分区，并删除整月都早于保留期的分区。 -func (s *CardTrafficCollectionStore) MaintainPartitions(ctx context.Context, now time.Time, aheadMonths int, retentionDays int) error { - monthStart := cardflow.MonthStart(now) - for i := 0; i <= aheadMonths; i++ { - from := monthStart.AddDate(0, i, 0) - to := from.AddDate(0, 1, 0) - sql := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s%s PARTITION OF tb_card_traffic_collection FOR VALUES FROM ('%s') TO ('%s')`, - collectionPartitionPrefix, from.Format("200601"), from.Format(time.RFC3339), to.Format(time.RFC3339)) - if err := s.db.WithContext(ctx).Exec(sql).Error; err != nil { - return err - } - } - cutoff := cardflow.ShanghaiDate(now).AddDate(0, 0, -retentionDays) - var partitions []string - if err := s.db.WithContext(ctx).Raw(` -SELECT c.relname FROM pg_inherits i -JOIN pg_class c ON c.oid = i.inhrelid -JOIN pg_class p ON p.oid = i.inhparent -WHERE p.relname = 'tb_card_traffic_collection'`).Scan(&partitions).Error; err != nil { - return err - } - for _, name := range partitions { - if len(name) != len(collectionPartitionPrefix)+6 || !strings.HasPrefix(name, collectionPartitionPrefix) { - continue - } - start, err := time.ParseInLocation("200601", name[len(collectionPartitionPrefix):], cutoff.Location()) - if err != nil { - continue - } - if !start.AddDate(0, 1, 0).After(cutoff) { - if err := s.db.WithContext(ctx).Exec(fmt.Sprintf("DROP TABLE IF EXISTS %s", name)).Error; err != nil { - return err - } - } - } - return nil +// Create 写入一条采集失败记录。 +func (s *CardTrafficCollectionFailureStore) Create(ctx context.Context, record *model.CardTrafficCollectionFailure) error { + return s.db.WithContext(ctx).Create(record).Error } diff --git a/internal/store/postgres/gateway_rate_limit_store.go b/internal/store/postgres/gateway_rate_limit_store.go index 1edd1974..c2ad38b9 100644 --- a/internal/store/postgres/gateway_rate_limit_store.go +++ b/internal/store/postgres/gateway_rate_limit_store.go @@ -186,7 +186,11 @@ func NewCardStopResumeRecordStore(db *gorm.DB) *CardStopResumeRecordStore { // StopResumeRecordFilter 是停复机记录列表筛选条件；时间范围由调用方保证必填且有上限。 type StopResumeRecordFilter struct { CardID *uint + CardIDs []uint // 资产标识解析出的卡；非 nil 时按其过滤（空切片表示匹配不到任何卡） ICCID string + Scene string + Reason string + CommandID string CarrierID *uint Action string Source string @@ -210,9 +214,24 @@ func (s *CardStopResumeRecordStore) List(ctx context.Context, filter StopResumeR if filter.CardID != nil { query = query.Where("card_id = ?", *filter.CardID) } + if filter.CardIDs != nil { + if len(filter.CardIDs) == 0 { + return []model.CardStopResumeRecord{}, 0, nil + } + query = query.Where("card_id IN ?", filter.CardIDs) + } if filter.ICCID != "" { query = query.Where("iccid = ?", filter.ICCID) } + if filter.Scene != "" { + query = query.Where("scene = ?", filter.Scene) + } + if filter.Reason != "" { + query = query.Where("reason = ?", filter.Reason) + } + if filter.CommandID != "" { + query = query.Where("command_id = ?", filter.CommandID) + } if filter.CarrierID != nil { query = query.Where("carrier_id = ?", *filter.CarrierID) } diff --git a/internal/store/postgres/refund_store.go b/internal/store/postgres/refund_store.go index 63f3f66f..06d923d9 100644 --- a/internal/store/postgres/refund_store.go +++ b/internal/store/postgres/refund_store.go @@ -16,6 +16,7 @@ type RefundListFilters struct { OrderID *uint `json:"order_id"` ShopID *uint `json:"shop_id"` AssetIdentifier string `json:"asset_identifier"` + AssetPostStatus *int `json:"asset_post_status"` } // RefundStore 退款申请数据访问层 @@ -100,6 +101,9 @@ func (s *RefundStore) List(ctx context.Context, opts *store.QueryOptions, filter if filters.AssetIdentifier != "" { query = query.Where("tb_refund_request.asset_identifier = ?", filters.AssetIdentifier) } + if filters.AssetPostStatus != nil { + query = query.Where("tb_refund_request.asset_post_status = ?", *filters.AssetPostStatus) + } } if err := query.Count(&total).Error; err != nil { diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index c9c5b2c4..135b39fc 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -2,6 +2,7 @@ package task import ( "context" + "strconv" "strings" "time" @@ -90,20 +91,25 @@ func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutc } priority.MarkExecution(ctx) attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling) - result, err := h.gateway.QueryFlow(gateway.WithCarrier(ctx, card.CarrierID), &gateway.FlowQueryReq{CardNo: card.ICCID}) + queryCtx, rawCapture := gateway.WithRawCapture(gateway.WithCarrier(ctx, card.CarrierID)) + queryStartedAt := time.Now() + result, err := h.gateway.QueryFlow(queryCtx, &gateway.FlowQueryReq{CardNo: card.ICCID}) + unreadable := unreadableDetail{raw: domainGatewayRaw(rawCapture), failure: collectionFailure(err, time.Since(queryStartedAt))} if gateway.IsRateLimited(err) { - h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultRateLimited) + h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultRateLimited, unreadable) priority.FailRetryable(ctx, "运营商账户网关限流，稍后重试") return StepOutcome{Result: StepRateLimited} } if err != nil { - h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultFailed) + h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultFailed, unreadable) _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败", priority) priority.FailRetryable(ctx, "查询流量失败") return h.fail(ctx, cardID, startedAt, "查询流量失败", err) } if strings.TrimSpace(result.ICCID) == "" { - h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultInvalidPayload) + unreadable.failure = &carddomain.CollectionFailure{ErrorCategory: constants.GatewayErrorInvalidResponse, + ProviderMessage: "流量查询响应缺少 ICCID", DurationMS: int(time.Since(queryStartedAt).Milliseconds())} + h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultInvalidPayload, unreadable) _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID", priority) priority.FailRetryable(ctx, "流量查询响应缺少 ICCID") return h.fail(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil) @@ -116,7 +122,8 @@ func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutc now := time.Now() decision, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{ CardID: cardID, GatewayReadingMB: float64(result.Used), ReadingResult: gatewayReadingResult(result), - ResetDay: h.carrier.GetDataResetDay(ctx, card.CarrierID), + ResetDay: h.carrier.GetDataResetDay(ctx, card.CarrierID), Raw: unreadable.raw, + Failure: &carddomain.CollectionFailure{DurationMS: int(time.Since(queryStartedAt).Milliseconds())}, Metadata: carddomain.ObservationMetadata{ ObservationID: attempt.IntegrationID, Source: constants.CardObservationSourcePolling, Scene: constants.CardObservationSceneTrafficPolling, ObservedAt: now, CorrelationID: attempt.IntegrationID, @@ -159,12 +166,18 @@ func (h *PollingCarddataHandler) fail(ctx context.Context, cardID uint, startedA } // recordUnreadable 为没拿到有效读数的采集写采集记录（只记采集时刻有套餐的卡），失败只影响留痕。 -func (h *PollingCarddataHandler) recordUnreadable(ctx context.Context, cardID uint, observationID, result string) { +// unreadableDetail 是读数不可用时随采集失败记录保存的原始数据与错误详情。 +type unreadableDetail struct { + raw *carddomain.GatewayRaw + failure *carddomain.CollectionFailure +} + +func (h *PollingCarddataHandler) recordUnreadable(ctx context.Context, cardID uint, observationID, result string, detail unreadableDetail) { if h.observation == nil { return } if _, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{ - CardID: cardID, ReadingResult: result, ResetDay: 1, + CardID: cardID, ReadingResult: result, ResetDay: 1, Raw: detail.raw, Failure: detail.failure, Metadata: carddomain.ObservationMetadata{ ObservationID: observationID, Source: constants.CardObservationSourcePolling, Scene: constants.CardObservationSceneTrafficPolling, ObservedAt: time.Now(), CorrelationID: observationID, @@ -181,3 +194,35 @@ func gatewayReadingResult(result *gateway.FlowUsageResp) string { } return constants.TrafficCollectionResultOK } + +// domainGatewayRaw 把网关原始数据收集器转为领域层原始数据。 +func domainGatewayRaw(capture *gateway.RawCapture) *carddomain.GatewayRaw { + if capture == nil || (capture.Attempts == 0 && len(capture.Request) == 0) { + return nil + } + return &carddomain.GatewayRaw{ + Request: capture.Request, HTTPStatus: capture.HTTPStatus, Response: capture.Response, + GatewayCode: capture.GatewayCode, GatewayMsg: capture.GatewayMsg, TraceID: capture.TraceID, Attempts: capture.Attempts, + } +} + +// collectionFailure 从网关错误提取采集失败详情；无错误时只带耗时。 +func collectionFailure(err error, duration time.Duration) *carddomain.CollectionFailure { + failure := &carddomain.CollectionFailure{DurationMS: int(duration.Milliseconds())} + if err == nil { + return failure + } + failure.ProviderMessage = err.Error() + if gwErr, ok := gateway.AsGatewayError(err); ok { + failure.ErrorCategory = gwErr.Category + if gwErr.ProviderCode != 0 { + failure.ProviderCode = strconv.Itoa(gwErr.ProviderCode) + } + if gwErr.ProviderMessage != "" { + failure.ProviderMessage = gwErr.ProviderMessage + } + } else if gateway.IsRateLimited(err) { + failure.ErrorCategory = constants.TrafficCollectionResultRateLimited + } + return failure +} diff --git a/migrations/000272_split_card_traffic_collection_failure.down.sql b/migrations/000272_split_card_traffic_collection_failure.down.sql new file mode 100644 index 00000000..aa97f25c --- /dev/null +++ b/migrations/000272_split_card_traffic_collection_failure.down.sql @@ -0,0 +1,16 @@ +DROP TABLE IF EXISTS tb_card_traffic_collection_failure; +DROP INDEX IF EXISTS idx_card_traffic_collection_decision; +DROP INDEX IF EXISTS idx_card_traffic_collection_carrier; +DROP INDEX IF EXISTS idx_card_traffic_collection_time; +ALTER TABLE tb_card_traffic_collection + DROP COLUMN IF EXISTS gateway_attempts, + DROP COLUMN IF EXISTS raw_truncated, + DROP COLUMN IF EXISTS raw_response_text, + DROP COLUMN IF EXISTS raw_response, + DROP COLUMN IF EXISTS gateway_trace_id, + DROP COLUMN IF EXISTS gateway_msg, + DROP COLUMN IF EXISTS gateway_code, + DROP COLUMN IF EXISTS http_status, + DROP COLUMN IF EXISTS raw_request, + DROP COLUMN IF EXISTS carrier_id; +COMMENT ON TABLE tb_card_traffic_collection IS '卡流量采集记录：采集时刻有套餐的卡每次采集一条，按月分区保留 90 天'; diff --git a/migrations/000272_split_card_traffic_collection_failure.up.sql b/migrations/000272_split_card_traffic_collection_failure.up.sql new file mode 100644 index 00000000..0ea315b1 --- /dev/null +++ b/migrations/000272_split_card_traffic_collection_failure.up.sql @@ -0,0 +1,75 @@ +-- 采集记录成功与失败分表，并保存网关原始请求与响应（refactor-audit-log-persistence）。 +-- 1. tb_card_traffic_collection 只写成功读数；新增运营商账户冗余列与原始数据列（父表 ALTER 自动作用到各分区）。 +-- 历史行 carrier_id 为 0；历史上写在本表的失败记录不迁移，随分区到期归档。 +-- 2. 新增 tb_card_traffic_collection_failure：所有被采集的卡每次失败一条，按 observed_at 按月分区，带默认分区。 +-- 3. 两张表在线保留 2 个月，到期月分区由应用层导出为 csv.gz 并校验后删除。 +-- 注意：在已有大分区的父表上建索引会锁写入，执行时间与分区数据量相关，请在低峰执行。 + +ALTER TABLE tb_card_traffic_collection + ADD COLUMN IF NOT EXISTS carrier_id BIGINT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS raw_request JSONB, + ADD COLUMN IF NOT EXISTS http_status INT, + ADD COLUMN IF NOT EXISTS gateway_code INT, + ADD COLUMN IF NOT EXISTS gateway_msg VARCHAR(500) NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS gateway_trace_id VARCHAR(128) NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS raw_response JSONB, + ADD COLUMN IF NOT EXISTS raw_response_text TEXT NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS raw_truncated BOOLEAN NOT NULL DEFAULT FALSE, + ADD COLUMN IF NOT EXISTS gateway_attempts INT NOT NULL DEFAULT 0; + +COMMENT ON TABLE tb_card_traffic_collection IS '卡流量采集成功记录：采集时刻有套餐的卡每次成功读数一条，按月分区在线保留 2 个月，到期归档为 csv.gz'; +COMMENT ON COLUMN tb_card_traffic_collection.carrier_id IS '运营商账户ID（采集时写入的冗余列，历史行为 0）'; +COMMENT ON COLUMN tb_card_traffic_collection.raw_request IS '网关明文业务请求（加密前），不含签名与密钥'; +COMMENT ON COLUMN tb_card_traffic_collection.raw_response IS '网关原始响应体（可解析为 JSON 且不超过 4KB 时）'; +COMMENT ON COLUMN tb_card_traffic_collection.raw_response_text IS '网关原始响应体文本（无法解析为 JSON 或超过 4KB 截断时）'; +COMMENT ON COLUMN tb_card_traffic_collection.gateway_attempts IS '网关客户端内部 HTTP 尝试次数，原始数据为最后一次尝试'; + +CREATE INDEX IF NOT EXISTS idx_card_traffic_collection_time ON tb_card_traffic_collection (observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_collection_carrier ON tb_card_traffic_collection (carrier_id, observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_collection_decision ON tb_card_traffic_collection (decision, observed_at DESC); + +CREATE TABLE IF NOT EXISTS tb_card_traffic_collection_failure ( + id BIGSERIAL, + iot_card_id BIGINT NOT NULL, + iccid VARCHAR(30) NOT NULL DEFAULT '', + carrier_id BIGINT NOT NULL DEFAULT 0, + observed_at TIMESTAMPTZ NOT NULL, + observation_id VARCHAR(128) NOT NULL DEFAULT '', + source VARCHAR(32) NOT NULL DEFAULT '', + scene VARCHAR(64) NOT NULL DEFAULT '', + result VARCHAR(32) NOT NULL, + error_category VARCHAR(30) NOT NULL DEFAULT '', + provider_code VARCHAR(30) NOT NULL DEFAULT '', + provider_message VARCHAR(500) NOT NULL DEFAULT '', + duration_ms INT NOT NULL DEFAULT 0, + raw_request JSONB, + http_status INT, + gateway_code INT, + gateway_msg VARCHAR(500) NOT NULL DEFAULT '', + gateway_trace_id VARCHAR(128) NOT NULL DEFAULT '', + raw_response JSONB, + raw_response_text TEXT NOT NULL DEFAULT '', + raw_truncated BOOLEAN NOT NULL DEFAULT FALSE, + gateway_attempts INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (id, observed_at) +) PARTITION BY RANGE (observed_at); + +COMMENT ON TABLE tb_card_traffic_collection_failure IS '卡流量采集失败记录：所有被采集的卡每次失败一条，按月分区在线保留 2 个月，到期归档为 csv.gz'; +COMMENT ON COLUMN tb_card_traffic_collection_failure.observation_id IS '采集 ID，等于对接日志 integration_id'; +COMMENT ON COLUMN tb_card_traffic_collection_failure.result IS '失败类型：failed/rate_limited/invalid_payload/no_reading/invalid_reading'; +COMMENT ON COLUMN tb_card_traffic_collection_failure.error_category IS '网关错误类别：timeout/network/http/invalid_response/upstream 等'; + +CREATE TABLE IF NOT EXISTS tb_card_traffic_collection_failure_202611 PARTITION OF tb_card_traffic_collection_failure + FOR VALUES FROM ('2026-11-01 00:00:00+08') TO ('2026-12-01 00:00:00+08'); +CREATE TABLE IF NOT EXISTS tb_card_traffic_collection_failure_202612 PARTITION OF tb_card_traffic_collection_failure + FOR VALUES FROM ('2026-12-01 00:00:00+08') TO ('2027-01-01 00:00:00+08'); +CREATE TABLE IF NOT EXISTS tb_card_traffic_collection_failure_202701 PARTITION OF tb_card_traffic_collection_failure + FOR VALUES FROM ('2027-01-01 00:00:00+08') TO ('2027-02-01 00:00:00+08'); +CREATE TABLE IF NOT EXISTS tb_card_traffic_collection_failure_default PARTITION OF tb_card_traffic_collection_failure DEFAULT; + +CREATE INDEX IF NOT EXISTS idx_card_traffic_failure_card_time ON tb_card_traffic_collection_failure (iot_card_id, observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_failure_time ON tb_card_traffic_collection_failure (observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_failure_carrier ON tb_card_traffic_collection_failure (carrier_id, observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_failure_category ON tb_card_traffic_collection_failure (error_category, observed_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_traffic_failure_observation ON tb_card_traffic_collection_failure (observation_id); diff --git a/migrations/000273_add_record_query_indexes.down.sql b/migrations/000273_add_record_query_indexes.down.sql new file mode 100644 index 00000000..2d7d079d --- /dev/null +++ b/migrations/000273_add_record_query_indexes.down.sql @@ -0,0 +1,6 @@ +DROP INDEX IF EXISTS idx_package_usage_adjustment_time; +DROP INDEX IF EXISTS idx_package_usage_daily_record_day; +DROP INDEX IF EXISTS idx_card_daily_usage_date_usage; +DROP INDEX IF EXISTS idx_card_stop_resume_record_command; +DROP INDEX IF EXISTS idx_card_stop_resume_record_scene; +DROP INDEX IF EXISTS idx_card_stop_resume_record_time; diff --git a/migrations/000273_add_record_query_indexes.up.sql b/migrations/000273_add_record_query_indexes.up.sql new file mode 100644 index 00000000..2650424a --- /dev/null +++ b/migrations/000273_add_record_query_indexes.up.sql @@ -0,0 +1,13 @@ +-- 记录查询接口所需索引（refactor-audit-log-persistence）：查单个走"卡 + 时间"，查全部走"时间"，常用条件走"条件 + 时间"。 +-- B-tree 可反向扫描，已有的 (card_id, created_at) 等升序索引直接支持倒序分页，不重建。 +-- 分区表在父表上建索引会自动作用到各分区；在已有大分区的表上建索引会锁写入，请在低峰执行。 + +CREATE INDEX IF NOT EXISTS idx_card_stop_resume_record_time ON tb_card_stop_resume_record (created_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_stop_resume_record_scene ON tb_card_stop_resume_record (scene, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_card_stop_resume_record_command ON tb_card_stop_resume_record (command_id); + +CREATE INDEX IF NOT EXISTS idx_card_daily_usage_date_usage ON tb_card_daily_usage (date, usage_mb DESC); + +CREATE INDEX IF NOT EXISTS idx_package_usage_daily_record_day ON tb_package_usage_daily_record (date); + +CREATE INDEX IF NOT EXISTS idx_package_usage_adjustment_time ON tb_package_usage_adjustment (created_at DESC); diff --git a/openspec/changes/refactor-audit-log-persistence/design.md b/openspec/changes/refactor-audit-log-persistence/design.md index 7862bc1a..37ac182a 100644 --- a/openspec/changes/refactor-audit-log-persistence/design.md +++ b/openspec/changes/refactor-audit-log-persistence/design.md @@ -213,7 +213,7 @@ |---|---| | 采集成功表 | `(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(decision, observed_at DESC)`，保留现有 `(iot_card_id, observed_at DESC)` | | 采集失败表 | `(iot_card_id, observed_at DESC)`、`(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(error_category, observed_at DESC)`、`(observation_id)` | -| 停复机记录 | 现有三个索引重建为倒序；新增 `(created_at DESC)`、`(scene, created_at DESC)`、`(command_id)` | +| 停复机记录 | 现有三个升序索引保留（B-tree 可反向扫描，直接支持倒序分页）；新增 `(created_at DESC)`、`(scene, created_at DESC)`、`(command_id)` | | 卡日流量 | 新增 `(date, usage_mb DESC)` | | 套餐日流量详单 | 新增 `(package_usage_id, date DESC)`、`(date)` | | 冲正流水 | 新增 `(iot_card_id, created_at DESC)`、`(created_at DESC)` | diff --git a/openspec/changes/refactor-audit-log-persistence/tasks.md b/openspec/changes/refactor-audit-log-persistence/tasks.md index fe857720..cad97789 100644 --- a/openspec/changes/refactor-audit-log-persistence/tasks.md +++ b/openspec/changes/refactor-audit-log-persistence/tasks.md @@ -29,30 +29,30 @@ ## 3. 流量采集分表与原始数据（第 3 次提交） -- [ ] 3.1 `internal/gateway/client.go`：新增 `WithRawCapture`，`executeWithRetry` 按尝试写入最后一次的明文请求、HTTP 状态、响应体、`code`、`msg`、`trace_id` 和尝试次数；不记录加密报文、签名、`app_id`、密钥和请求头。 -- [ ] 3.2 迁移： +- [x] 3.1 `internal/gateway/client.go`：新增 `WithRawCapture`，`executeWithRetry` 按尝试写入最后一次的明文请求、HTTP 状态、响应体、`code`、`msg`、`trace_id` 和尝试次数；不记录加密报文、签名、`app_id`、密钥和请求头。 +- [x] 3.2 迁移： - 新建 `tb_card_traffic_collection_failure`，按月分区，并建立索引； - 采集成功表增加 `carrier_id` 和原始数据列（`raw_request`、`http_status`、`gateway_code`、`gateway_msg`、`gateway_trace_id`、`raw_response`、`raw_response_text`、`raw_truncated`、`gateway_attempts`），新增 D9 中列出的索引。 -- [ ] 3.3 `polling_carddata_handler`： +- [x] 3.3 `polling_carddata_handler`： - 调用 `QueryFlow` 前挂上收集器； - 网关失败时填入错误类别、上游码与信息、耗时； - 成功和失败都把原始数据带进 `TrafficObservation`。 -- [ ] 3.4 `internal/application/cardobservation`：`writeCollectionRecord` 按结果分流。成功写成功表（保持"有套餐才写"），失败直接写失败表（不查套餐）；原始响应超过 4 KB 时截断并标记。 -- [ ] 3.5 `MaintainPartitions` 改为成功表、失败表共用的四步归档（摘出、导出 `csv.gz.tmp`、校验并写 manifest 后改名、删除），并处理上一轮遗留的已摘出表、清理本地压缩包（默认保留 3 个月）；常量改为 `TrafficCollectionRetentionMonths=2`。 +- [x] 3.4 `internal/application/cardobservation`：`writeCollectionRecord` 按结果分流。成功写成功表（保持"有套餐才写"），失败直接写失败表（不查套餐）；原始响应超过 4 KB 时截断并标记。 +- [x] 3.5 `MaintainPartitions` 改为成功表、失败表共用的四步归档（摘出、导出 `csv.gz.tmp`、校验并写 manifest 后改名、删除），并处理上一轮遗留的已摘出表、清理本地压缩包（默认保留 3 个月）；常量改为 `TrafficCollectionRetentionMonths=2`。 ## 4. 记录查询接口（第 3 次提交） -- [ ] 4.1 在 `internal/store/postgres` 中为各记录表实现统一的 `List(ctx, Filter)`；资产标识复用现有资产解析，解析不到时返回空页。 -- [ ] 4.2 采集成功列表 `GET /traffic-collections`、采集失败列表 `GET /traffic-collection-failures`；单卡采集接口增加筛选条件和"包含失败"参数，成功表查询按 `result='ok'` 过滤。 -- [ ] 4.3 停复机记录列表：`iccid` 扩展为资产标识，增加场景、原因、指令号筛选；单卡历史增加动作、结果、来源筛选；新建或重建 D9 中列出的索引。 -- [ ] 4.4 新增卡日流量列表；新增套餐日流量详单列表，原单套餐接口改为分页；新增冲正流水列表，原单套餐接口增加时间范围筛选。 -- [ ] 4.5 新增退款后处理尝试列表；退款列表增加 `asset_post_status` 筛选，列表和详情返回后处理次数、最后错误、下次重试时间。 -- [ ] 4.6 统一校验规则：时间范围必填、跨度不超过 31 天、分页上限；所有接口在 `RouteSpec` 中登记输入输出 DTO，并生成接口文档。 +- [x] 4.1 在 `internal/store/postgres` 中为各记录表实现统一的 `List(ctx, Filter)`；资产标识复用现有资产解析，解析不到时返回空页。 +- [x] 4.2 采集成功列表 `GET /traffic-collections`、采集失败列表 `GET /traffic-collection-failures`；单卡采集接口增加筛选条件和"包含失败"参数，成功表查询按 `result='ok'` 过滤。 +- [x] 4.3 停复机记录列表：`iccid` 扩展为资产标识，增加场景、原因、指令号筛选；单卡历史增加动作、结果、来源筛选；新建或重建 D9 中列出的索引。 +- [x] 4.4 新增卡日流量列表；新增套餐日流量详单列表，原单套餐接口改为分页；新增冲正流水列表，原单套餐接口增加时间范围筛选。 +- [x] 4.5 新增退款后处理尝试列表；退款列表增加 `asset_post_status` 筛选，列表和详情返回后处理次数、最后错误、下次重试时间。 +- [x] 4.6 统一校验规则：时间范围必填、跨度不超过 31 天、分页上限；所有接口在 `RouteSpec` 中登记输入输出 DTO，并生成接口文档。 ## 5. 时间线与测试环境部署配置（第 3 次提交） - [ ] 5.1 先与前端或运营确认是否有页面依赖机器动作的审计，再在 `internal/query/audit` 中让卡和设备时间线合并停复机记录。 -- [ ] 5.2 `docker-compose.prod.yml`：api 挂载 `./api/logs`，worker 挂载 `./worker/logs` 和 `./worker/archive`；两者增加 `MAX_BACKUPS=40`、`MAX_AGE=14`。`.gitea/workflows/deploy.yaml` 改为创建这三个目录。 +- [x] 5.2 `docker-compose.prod.yml`：api 挂载 `./api/logs`，worker 挂载 `./worker/logs` 和 `./worker/archive`；两者增加 `MAX_BACKUPS=40`、`MAX_AGE=14`。`.gitea/workflows/deploy.yaml` 改为创建这三个目录。 - [ ] 5.3 构建、`openspec validate --all`、`./scripts/context-health.sh`；提交并推送 8-11。 ## 6. 测试环境验证（AGENTS.md 规定自动化测试为 N/A；按 ENG-TEST-001 在测试环境验证） diff --git a/pkg/constants/card_traffic_queue.go b/pkg/constants/card_traffic_queue.go index 314ed215..50298309 100644 --- a/pkg/constants/card_traffic_queue.go +++ b/pkg/constants/card_traffic_queue.go @@ -47,8 +47,15 @@ const ( TrafficDecisionSettledExisting = "settled_existing" ) -// TrafficCollectionRetentionDays 是采集记录保留天数，过期按整月分区删除。 -const TrafficCollectionRetentionDays = 90 +// TrafficCollectionRetentionMonths 是采集记录（成功与失败）在线保留的整月数：保留当月加最近 N 个整月， +// 更早的月分区导出为 csv.gz 并校验后删除。 +const TrafficCollectionRetentionMonths = 2 + +// TrafficCollectionArchiveDir 是采集记录压缩包目录（相对 worker 工作目录），由日志同步服务器拉走。 +const TrafficCollectionArchiveDir = "archive/traffic-collection" + +// TrafficCollectionArchiveKeepFiles 是每张采集记录表本地保留的压缩包个数（按月）。 +const TrafficCollectionArchiveKeepFiles = 3 // RedisCardTrafficQueueKey 返回卡流量采集队列 key。 func RedisCardTrafficQueueKey(cardID uint) string {