iotlcm/worker-svr/internal/client/upstream/client.go

package upstream

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/gclient"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
)

const inboundBasicInfoClientNameEnv = "UPSTREAM_INBOUND_BASIC_INFO_CLIENT_NAME"
const upstreamProxyMaxTimeoutConfigKey = "upstream.proxy.maxTimeout"

// apnAsyncProtocolVersionConfigKey 是异步APN提交的兼容开关。
// 默认值改为开启：常规发布不再需要人工逐环境开开关，兼容性由提交前的协议能力探测兜底。
// 只有需要临时回退到旧协议时，才显式配为 0。
const apnAsyncProtocolVersionConfigKey = "upstream.proxy.apnAsyncProtocolVersion"

// defaultApnAsyncProtocolVersion 默认开启异步APN提交。
const defaultApnAsyncProtocolVersion = 1

// defaultUpstreamProxyMaxTimeout 是单次批量转发的兜底上限。
// 必须大于 upstream-proxy 的 gateway.batchTimeout（默认 20 分钟），
// 让超时由更靠近通道的一层先判定，避免 worker 提前放弃而通道仍在办理，造成结果未知。
const defaultUpstreamProxyMaxTimeout = 25 * time.Minute

// defaultUpstreamProxyTimeout 是未配置单卡预算时的默认值。
const defaultUpstreamProxyTimeout = 10 * time.Second

type Client struct {
	currentUsageClientName     string
	currentUsageTimeout        time.Duration
	inboundBasicInfoClientName string
	imeiClientName             string
	imeiTimeout                time.Duration
}

type response[T any] struct {
	Code    int                   `json:"code"`
	Message string                `json:"message"`
	Data    []model.SimBaseRes[T] `json:"data"`
}

func New() *Client {
	return &Client{}
}

func (c *Client) QuerySimBasicInfo(ctx context.Context, input *model.SimBasicInfoInput) ([]model.SimBaseRes[model.SimBasicInfoItem], error) {
	address, err := c.inboundBasicInfoAddress(ctx)
	if err != nil {
		return nil, err
	}
	return requestAllWithAddress[model.SimBasicInfoItem](ctx, "/sim/query/sim-basic-info", input, len(input.SimKeys), address)
}

// NewWithCurrentUsageClient keeps the full-sync current-usage call independent
// from the configured upstream.proxy endpoint.
func NewWithCurrentUsageClient(address string, timeout ...time.Duration) *Client {
	var currentUsageTimeout time.Duration
	if len(timeout) > 0 {
		currentUsageTimeout = timeout[0]
	}
	return &Client{currentUsageClientName: address, currentUsageTimeout: currentUsageTimeout}
}

func NewWithInboundBasicInfoClient(address string) *Client {
	return &Client{inboundBasicInfoClientName: address}
}

// NewWithImeiClient pins the IMEI full-sync path to an explicit chan endpoint
// and timeout. An empty address falls back to the shared upstream.proxy
// configuration, which is only intended for local runs.
func NewWithImeiClient(address string, timeout ...time.Duration) *Client {
	var imeiTimeout time.Duration
	if len(timeout) > 0 {
		imTimeout := timeout[0]
		if imTimeout > 0 {
			imeiTimeout = imTimeout
		}
	}
	return &Client{imeiClientName: address, imeiTimeout: imeiTimeout}
}

func (c *Client) inboundBasicInfoAddress(ctx context.Context) (string, error) {
	if strings.TrimSpace(c.inboundBasicInfoClientName) != "" {
		return strings.TrimSpace(c.inboundBasicInfoClientName), nil
	}
	if address := strings.TrimSpace(os.Getenv(inboundBasicInfoClientNameEnv)); address != "" {
		return address, nil
	}
	config, ok, err := optionalUpstreamConfig(ctx, "upstream.gateway")
	if err != nil {
		return "", err
	}
	if !ok || strings.TrimSpace(config.Address) == "" {
		return "", gerror.Newf("upstream gateway address is empty, configure upstream.gateway.address or %s", inboundBasicInfoClientNameEnv)
	}
	return strings.TrimSpace(config.Address), nil
}

// QuerySimDataUsage calls Chan's current-month real-time usage endpoint. The
// response keeps one item per requested card so the caller can classify
// per-card errors without reissuing successful cards.
func (c *Client) QuerySimDataUsage(ctx context.Context, input *model.SimDataUsageInput) ([]model.SimBaseRes[model.SimDataUsageItem], error) {
	return requestAllWithAddressTimeout[model.SimDataUsageItem](ctx, "/sim/query/sim-datausage", input, len(input.SimKeys), c.currentUsageClientName, c.currentUsageTimeout)
}

// QuerySimCardStatus queries the carrier status for the supplied cards. The
// status queue currently sends one card per request, while retaining the
// response shape needed for future batch support.
func (c *Client) QuerySimCardStatus(ctx context.Context, input *model.SimCardStatusInput) ([]model.SimBaseRes[model.SimCardStatusItem], error) {
	return requestAll[model.SimCardStatusItem](ctx, "/sim/query/sim-status", input, len(input.SimKeys))
}

// QuerySimImei queries the terminal IMEI for the supplied cards. The IMEI queue
// sends one card per request.
//
// It deliberately uses requestAll rather than request: a per-item carrier
// failure (unsupported template, missing value) must reach the handler as an
// item with a non-zero Code, not be silently dropped from the slice.
func (c *Client) QuerySimImei(ctx context.Context, input *model.SimImeiInput) ([]model.SimBaseRes[model.SimImeiItem], error) {
	return requestAllWithAddressTimeout[model.SimImeiItem](ctx, "/sim/query/sim-imei", input, len(input.SimKeys), c.imeiClientName, c.imeiTimeout)
}

func (c *Client) QuerySimDatausageDaily(ctx context.Context, input *model.SimDataUsageDailyInput) ([]model.SimBaseRes[model.SimDataUsageDailyOutputItem], error) {
	return requestAll[model.SimDataUsageDailyOutputItem](ctx, "/sim/query/sim-datausage-daily", input, len(input.SimKeys))
}

func (c *Client) QuerySimDatausageMonthly(ctx context.Context, input *model.SimDataUsageMonthlyInput) ([]model.SimBaseRes[model.SimDataUsageMonthlyOutputItem], error) {
	return requestAll[model.SimDataUsageMonthlyOutputItem](ctx, "/sim/query/sim-datausage-monthly", input, len(input.SimKeys))
}

// ChangeSimStatus returns one result for every upstream item, including failed items.
// Lifecycle batch execution needs those failures to be persisted to the matching detail.
func (c *Client) ChangeSimStatus(ctx context.Context, input *model.SimCardStatusChangeInput) ([]model.SimBaseRes[model.SimCardStatusChangeOutputItem], error) {
	return requestAll[model.SimCardStatusChangeOutputItem](ctx, "/sim/change/sim-status", input, len(input.SimKeys))
}

// ChangeSimApnFunction returns one result for every upstream item, including failed items.
// Data service batch execution needs those failures to be persisted to the matching detail.
func (c *Client) ChangeSimApnFunction(ctx context.Context, input *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	return requestAll[model.SimCardApnChangeOutputItem](ctx, "/sim/change/sim-apn", input, len(input.SimKeys))
}

// apnRequestNotSentError 仅描述任何APN写请求之前的能力拒绝。
// 提交开始后的通信错误绝不能使用此类型。
type apnRequestNotSentError struct{ cause error }

func (e *apnRequestNotSentError) Error() string           { return e.cause.Error() }
func (e *apnRequestNotSentError) Unwrap() error           { return e.cause }
func (e *apnRequestNotSentError) APNRequestNotSent() bool { return true }

func (c *Client) CheckSimApnAsyncCapability(ctx context.Context, input *model.SimCardApnChangeInput) (err error) {
	defer func() {
		if err != nil {
			err = &apnRequestNotSentError{cause: err}
		}
	}()
	if !apnAsyncSubmitEnabled(ctx) {
		return gerror.New("异步APN提交已被显式关闭（upstream.proxy.apnAsyncProtocolVersion=0）；清除该配置或置为1后重试")
	}
	if err := apnDependencyReady(ctx); err != nil {
		return gerror.Wrap(err, "异步APN依赖未就绪，未发送办理请求")
	}
	if input == nil || strings.TrimSpace(input.ReqData.OperationID) == "" {
		return gerror.New("异步APN提交能力探测缺少operationId")
	}
	return c.checkSimApnProtocol(ctx, input)
}

// apnDependencyReady 检查本地真实依赖（迁移表结构）；测试可替换以便固定结论。
var apnDependencyReady = func(ctx context.Context) error {
	return defaultAPNReadiness.check(ctx)
}

// apnAsyncSubmitEnabled 判断应急总闸是否被显式关闭。
// 默认放行：是否可提交由程序自行检查真实依赖决定——
//   - 本地依赖：apn_readiness 核对迁移表结构是否就绪；
//   - 远端兼容：每次提交前的协议能力探测，旧版 chan/proxy 会明确拒绝而不是按同步语义执行。
// 因此常规发布不再需要人工逐环境开开关；仅回退到旧协议时把配置显式设为 0。
func apnAsyncSubmitEnabled(ctx context.Context) bool {
	return g.Cfg().MustGet(ctx, apnAsyncProtocolVersionConfigKey, defaultApnAsyncProtocolVersion).Int() == 1
}

// 仅验证远端协议：已有受理映射的只读补查不受新增提交门禁控制。
// 旧版代理/通道缺少能力接口时必须拒绝，不能把查询当作普通办理转发。
func (c *Client) checkSimApnProtocol(ctx context.Context, input *model.SimCardApnChangeInput) error {
	if len(input.SimKeys) == 0 || len(input.SimKeys) > 100 || (strings.TrimSpace(input.ReqData.OperationID) == "" && strings.TrimSpace(input.ReqData.QueryJobID) == "") {
		return gerror.New("异步APN能力探测需要operationId和1至100张卡快照")
	}
	type capability struct {
		ProtocolVersion int `json:"protocolVersion"`
	}
	probeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	items, err := requestAll[capability](probeCtx, "/sim/query/sim-apn-capability", input, len(input.SimKeys))
	if err != nil {
		return gerror.Wrap(err, "异步APN协议能力探测失败，未发送办理或恢复请求")
	}
	if len(items) != len(input.SimKeys) {
		return gerror.New("异步APN协议能力响应卡数不匹配，禁止发送办理请求")
	}
	seen := make(map[model.SimKey]bool)
	for _, item := range items {
		if item.SimKey == nil || item.Code != 0 || item.ResData.ProtocolVersion != 1 || seen[*item.SimKey] {
			return gerror.New("通道未就绪或不支持异步APN协议，禁止发送办理请求")
		}
		seen[*item.SimKey] = true
	}
	for _, key := range input.SimKeys {
		if key == nil || !seen[*key] {
			return gerror.New("异步APN协议能力响应卡关联不匹配")
		}
	}
	return nil
}

// SubmitSimApnFunction 仅提交异步批量任务；受理状态不能按办理成功回写。
func (c *Client) SubmitSimApnFunction(ctx context.Context, input *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	if err := c.CheckSimApnAsyncCapability(ctx, input); err != nil {
		return nil, err
	}
	request := *input
	request.ReqData.SubmitOnly = true
	request.ReqData.QueryJobID = ""
	request.ReqData.RecoverOnly = false
	return c.ChangeSimApnFunction(ctx, &request)
}

// QuerySimApnFunctionResult 根据已持久化的jobId只查一次，不重新提交。
func (c *Client) QuerySimApnFunctionResult(ctx context.Context, input *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	if strings.TrimSpace(input.ReqData.QueryJobID) == "" && strings.TrimSpace(input.ReqData.OperationID) == "" {
		return nil, gerror.New("异步APN结果查询缺少jobId和operationId")
	}
	request := *input
	request.ReqData.QueryJobID = strings.TrimSpace(request.ReqData.QueryJobID)
	request.ReqData.SubmitOnly = false
	request.ReqData.RecoverOnly = request.ReqData.QueryJobID == ""
	if err := c.checkSimApnProtocol(ctx, &request); err != nil {
		return nil, gerror.Wrap(err, "异步APN只读查询协议未就绪，未发送查询请求")
	}
	return c.ChangeSimApnFunction(ctx, &request)
}

func request[T any](ctx context.Context, path string, input any, itemCount int) ([]model.SimBaseRes[T], error) {
	items, err := requestAll[T](ctx, path, input, itemCount)
	if err != nil {
		return nil, err
	}

	successItems := make([]model.SimBaseRes[T], 0, len(items))
	for _, item := range items {
		if item.Code == 0 {
			successItems = append(successItems, item)
		}
	}
	return successItems, nil
}

func requestAll[T any](ctx context.Context, path string, input any, itemCount int) ([]model.SimBaseRes[T], error) {
	return requestAllWithAddress[T](ctx, path, input, itemCount, "")
}

func requestAllWithAddress[T any](ctx context.Context, path string, input any, itemCount int, address string) ([]model.SimBaseRes[T], error) {
	return requestAllWithAddressTimeout[T](ctx, path, input, itemCount, address, 0)
}

func requestAllWithAddressTimeout[T any](ctx context.Context, path string, input any, itemCount int, address string, timeoutOverride time.Duration) ([]model.SimBaseRes[T], error) {
	config, err := upstreamConfig(ctx, "upstream.proxy")
	if err != nil {
		return nil, err
	}
	if address == "" {
		address = config.Address
	}
	if address == "" {
		return nil, gerror.New("upstream proxy address is empty")
	}
	if itemCount < 1 {
		itemCount = 1
	}

	client := g.Client()
	if path == "/sim/change/sim-apn" || path == "/sim/query/sim-apn-capability" {
		client.SetContentType("application/json")
	}
	timeout := resolveUpstreamTimeout(path, itemCount, config.Timeout,
		timeoutOverride,
		configDuration(ctx, upstreamProxyMaxTimeoutConfigKey, defaultUpstreamProxyMaxTimeout),
		configDuration(ctx, "upstream.proxy.dataServiceTimeout", 240*time.Second))
	setClientTimeout(client, timeout)
	clientResponse, err := client.Post(ctx, fmt.Sprintf("http://%s%s", address, path), input)
	if err != nil {
		return nil, gerror.Wrap(err, "request upstream failed")
	}
	defer clientResponse.Close()

	return decodeResponse[T](clientResponse.ReadAll())
}

func optionalUpstreamConfig(ctx context.Context, key string) (model.UpstreamChanConfig, bool, error) {
	configValue, err := g.Cfg().Get(ctx, key)
	if err != nil {
		return model.UpstreamChanConfig{}, false, gerror.Wrapf(err, "read upstream configuration failed: %s", key)
	}
	if configValue == nil || configValue.IsNil() {
		return model.UpstreamChanConfig{}, false, nil
	}
	config := model.UpstreamChanConfig{}
	if err = configValue.Scan(&config); err != nil {
		return model.UpstreamChanConfig{}, false, gerror.Wrapf(err, "parse upstream configuration failed: %s", key)
	}
	return config, true, nil
}

func upstreamConfig(ctx context.Context, key string) (model.UpstreamChanConfig, error) {
	config, ok, err := optionalUpstreamConfig(ctx, key)
	if err != nil {
		return model.UpstreamChanConfig{}, err
	}
	if !ok {
		return model.UpstreamChanConfig{}, gerror.Newf("upstream configuration is missing: %s", key)
	}
	return config, nil
}

func decodeResponse[T any](body []byte) ([]model.SimBaseRes[T], error) {
	output := response[T]{}
	if err := json.Unmarshal(body, &output); err != nil {
		return nil, gerror.Wrap(err, "parse upstream response failed")
	}
	if output.Code != 0 {
		return nil, gerror.New(output.Message)
	}
	return output.Data, nil
}

func setClientTimeout(client *gclient.Client, timeout time.Duration) {
	client.SetTimeout(timeout)
	if transport, ok := client.Transport.(*http.Transport); ok {
		transport.ResponseHeaderTimeout = timeout
	}
}

// resolveUpstreamTimeout 计算单次转发预算：
// 普通批量按「单卡预算 × 卡数」放大，并以 maxTimeout 兜底，避免批量操作被默认 2 分钟上限提前判失败；
// 数据开停沿用独立预算，优先于按卡计时。
func resolveUpstreamTimeout(path string, itemCount int, perCard, timeoutOverride, maxTimeout, dataServiceTimeout time.Duration) time.Duration {
	if itemCount < 1 {
		itemCount = 1
	}
	timeout := timeoutOverride
	if timeout <= 0 {
		timeout = perCard
	}
	if timeout <= 0 {
		timeout = defaultUpstreamProxyTimeout
	}
	timeout *= time.Duration(itemCount)
	if path == "/sim/change/sim-apn" {
		if dataServiceTimeout > 0 {
			return dataServiceTimeout
		}
		return 240 * time.Second
	}
	if maxTimeout > 0 && timeout > maxTimeout {
		timeout = maxTimeout
	}
	return timeout
}

// configDuration 读取时长配置，兼容写成 "20m" 字符串或秒数整数两种形式。
func configDuration(ctx context.Context, key string, fallback time.Duration) time.Duration {
	value := g.Cfg().MustGet(ctx, key)
	if value == nil || value.IsNil() {
		return fallback
	}
	if duration := value.Duration(); duration >= time.Second {
		return duration
	}
	if seconds := value.Int(); seconds > 0 {
		return time.Duration(seconds) * time.Second
	}
	return fallback
}