iotlcm/worker-svr/internal/service/monthlyusage/handler.go

package monthlyusage

import (
	"context"
	"fmt"
	"strings"

	zsetconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/zset"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	queue "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
	monthlyusagerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/monthlyusage"
)

// errCodeTelecomChannelLimited is the carrier error code that means the whole
// channel is throttled or disabled and must stop for this run.
const errCodeTelecomChannelLimited = "S00004"

// The gateway uses code 9999 without an errCode for transport, HTTP, and
// response-decoding failures. These failures are safe to retry from the task
// lease; a populated errCode is treated as a business result.
const chanGatewayInternalErrorCode = 9999

// tokenExpiredCode is the carrier code for an expired/invalid upstream token.
// The gateway already invalidates and refreshes the token once on this code;
// when it still reaches the worker, retrying in a later round lets a refreshed
// token succeed instead of parking the card as a terminal business failure.
const tokenExpiredCode = 12021

const (
	mobileBusyFirstCode = 13000
	mobileBusyLastCode  = 13007
	mobileBusyClass     = "mobile_busy"
)

// Mobile usage templates are identified by the template returned with the
// channel metadata.  The channel name and error message are not stable
// carrier identifiers.
//
// The set is derived from the OneLink registration table in the upstream
// proxy (iotlcm-upstream-proxy/internal/logic/upstreamimpl/sim_yd.go
// init()): every key below is registered to the China Mobile OneLink
// sSimYD implementation, so numeric codes 13000-13007 emitted by any of
// them carry OneLink semantics and are safe to retry across rounds.
// Keep this list in sync with that registration table; templates that are
// not registered there (including unknown future keys) stay "generic" and
// are never classified as mobile.
var mobileUsageTemplates = map[string]struct{}{
	"sim_yd_v5":                    {},
	"sim_yd_mock_v5":               {},
	"sim_sdtywl_v1":                {},
	"sim_onelink_tengyu_v1":        {},
	"sim_onelink_xuzhou_mobile_v1": {},
	"sim_onelink_lelian_v1":        {},
	"sim_onelink_qianxun_v1":       {},
	"sim_onelink_henan_xinjing_v1": {},
	"sim_henan_xinjing_direct_v1":  {},
	"sim_onelink_zibo_v1":          {},
	"sim_tibiot_v1":                {},
	"sim_qianxun_v3":               {},
	"sim_yunda_v1":                 {},
	"sim_mingqian_v3":              {},
	"sim_daihe_v1":                 {},
	"sim_onelink_daihe_v1":         {},
	"sim_onelink_daliuliang_v1":    {},
	"sim__daliuliang_v1":           {},
	"sim_onelink_mobile_public_v1": {},
	"sim_onelink_kedang_v1":        {},
	"sim_onelink_beixin_v1":        {},
	"lanshang_8000":                {},
	"lanshang_8003":                {},
}

const telecomUsageTemplate = "sim_dx_v1"

func usageCarrierForTemplate(tempKey string) string {
	tempKey = strings.ToLower(strings.TrimSpace(tempKey))
	if model.IsTelecomTemplate(tempKey) {
		return "telecom"
	}
	if _, ok := mobileUsageTemplates[tempKey]; ok {
		return "mobile"
	}
	return "generic"
}

// classifyUsageFailure maps a gateway item to the queue semantics owned by
// this service.  Carrier identity is resolved from the template first; error
// text and channel names are deliberately excluded from the decision.
func classifyUsageFailure(tempKey string, code int, errCode string) (errorClass string, retryable, blockChannel bool) {
	errorClass = "business"
	carrier := usageCarrierForTemplate(tempKey)
	switch {
	case carrier == "telecom" && strings.EqualFold(strings.TrimSpace(errCode), errCodeTelecomChannelLimited):
		return "channel_limited", false, true
	case carrier == "mobile" && code >= mobileBusyFirstCode && code <= mobileBusyLastCode:
		return mobileBusyClass, true, false
	case code == chanGatewayInternalErrorCode && strings.TrimSpace(errCode) == "":
		return "network", true, false
	case code == tokenExpiredCode:
		return "token_expired", true, false
	default:
		return errorClass, false, false
	}
}

type UsageClient interface {
	QuerySimDataUsage(context.Context, *model.SimDataUsageInput) ([]model.SimBaseRes[model.SimDataUsageItem], error)
}

type Handler struct {
	client UsageClient
	repo   UsageRepository
}

type UsageRepository interface {
	ApplyBatch(context.Context, []monthlyusagerepository.UsageResult) error
}

func NewHandler(client UsageClient, repo UsageRepository) (*Handler, error) {
	if client == nil || repo == nil {
		return nil, fmt.Errorf("monthly usage handler dependencies are incomplete")
	}
	return &Handler{client: client, repo: repo}, nil
}

func (h *Handler) Query(ctx context.Context, task queue.Task) zsetconsumer.Result {
	result := zsetconsumer.Result{Task: task}
	output, err := h.client.QuerySimDataUsage(ctx, &model.SimDataUsageInput{
		SimKeys: []*model.SimKey{{
			ChanKey: task.ChanKey,
			ICCID:   task.ICCID,
			IMSI:    task.IMSI,
			MSISDN:  task.MSISDN,
		}},
	})
	if err != nil {
		result.ErrorClass = "network"
		result.Retryable = true
		result.Message = err.Error()
		return result
	}
	if len(output) == 0 || output[0].SimKey == nil {
		result.ErrorClass = "protocol"
		result.Retryable = true
		result.Message = "chan returned empty current usage result"
		return result
	}
	item := output[0]
	if task.ICCID != "" && item.SimKey.ICCID != task.ICCID {
		result.ErrorClass = "protocol"
		result.Retryable = true
		result.Message = "chan returned mismatched card identifier"
		return result
	}
	result.Code = item.Code
	result.Message = item.Message
	result.ErrCode = item.ErrCode
	if item.Code != 0 {
		result.ErrorClass, result.Retryable, result.BlockChannel = classifyUsageFailure(task.TempKey, item.Code, item.ErrCode)
		return result
	}
	result.Success = true
	result.UsageMB = item.ResData.DataUsage
	return result
}

func (h *Handler) Apply(ctx context.Context, results []zsetconsumer.Result) error {
	updates := make([]monthlyusagerepository.UsageResult, 0, len(results))
	for _, result := range results {
		if !result.Success {
			continue
		}
		updates = append(updates, monthlyusagerepository.UsageResult{
			CardID:  result.Task.CardID,
			UsageMB: result.UsageMB,
		})
	}
	return h.repo.ApplyBatch(ctx, updates)
}