MiFaZhan
301 lines
11 KiB
Go
301 lines
11 KiB
Go
package upstreamapi
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"github.com/go-sql-driver/mysql"
|
|
"github.com/gogf/gf/v2/database/gdb"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model"
|
|
)
|
|
|
|
type apnOperationStore interface {
|
|
Claim(context.Context, string, string, string) (bool, error)
|
|
Load(context.Context, string, string) (string, string, error)
|
|
Save(context.Context, string, string, string, string) error
|
|
}
|
|
|
|
type sqlApnOperationStore struct{ db gdb.DB }
|
|
|
|
func (s sqlApnOperationStore) database() gdb.DB {
|
|
if s.db != nil {
|
|
return s.db
|
|
}
|
|
return g.DB()
|
|
}
|
|
|
|
func validateApnOperationIdentity(id, channel, hash string) error {
|
|
for _, value := range []string{id, channel} {
|
|
if len(value) == 0 || len(value) > 128 || !utf8.ValidString(value) || strings.TrimSpace(value) != value || strings.ContainsRune(value, 0) {
|
|
return fmt.Errorf("operationId和chanKey必须为1至128字节且不含首尾空白或空字符")
|
|
}
|
|
}
|
|
decoded, err := hex.DecodeString(hash)
|
|
if err != nil || len(decoded) != sha256.Size || hash != strings.ToLower(hash) {
|
|
return fmt.Errorf("APN请求摘要必须为小写SHA-256")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s sqlApnOperationStore) Claim(ctx context.Context, id, channel, hash string) (bool, error) {
|
|
if err := validateApnOperationIdentity(id, channel, hash); err != nil {
|
|
return false, err
|
|
}
|
|
result, err := s.database().Exec(ctx, "INSERT INTO t_apn_operation (operation_id,chan_key,request_hash,response_json,created_at,updated_at) VALUES (?,?,?,'',NOW(),NOW())", id, channel, hash)
|
|
if err != nil {
|
|
var duplicate *mysql.MySQLError
|
|
if errors.As(err, &duplicate) && duplicate.Number == 1062 {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if count != 1 {
|
|
return false, fmt.Errorf("APN提交意图插入行数异常：%d", count)
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (s sqlApnOperationStore) Load(ctx context.Context, id, channel string) (string, string, error) {
|
|
row, err := s.database().GetOne(ctx, "SELECT request_hash,response_json FROM t_apn_operation WHERE operation_id=? AND chan_key=? AND BINARY operation_id=BINARY ? AND BINARY chan_key=BINARY ?", id, channel, id, channel)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
if len(row) == 0 {
|
|
return "", "", nil
|
|
}
|
|
return row["request_hash"].String(), row["response_json"].String(), nil
|
|
}
|
|
|
|
func (s sqlApnOperationStore) Save(ctx context.Context, id, channel, hash, response string) error {
|
|
if err := validateApnOperationIdentity(id, channel, hash); err != nil {
|
|
return err
|
|
}
|
|
return s.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
row, err := tx.GetOne("SELECT request_hash,response_json FROM t_apn_operation WHERE operation_id=? AND chan_key=? AND BINARY operation_id=BINARY ? AND BINARY chan_key=BINARY ? FOR UPDATE", id, channel, id, channel)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(row) == 0 {
|
|
return fmt.Errorf("APN受理映射不存在，禁止无意图保存")
|
|
}
|
|
if row["request_hash"].String() != hash {
|
|
return fmt.Errorf("operationId已绑定不同的办理请求")
|
|
}
|
|
merged, err := mergeApnOperationJSON(row["response_json"].String(), response)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result, err := tx.Exec("UPDATE t_apn_operation SET response_json=?,revision=revision+1,updated_at=NOW(3) WHERE operation_id=? AND chan_key=? AND request_hash=?", merged, id, channel, hash)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count != 1 {
|
|
return fmt.Errorf("APN受理映射保存行数异常：%d", count)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// 已确认逐卡结果不回退未知，已持久化jobId不能被后续空响应或错误抹掉。
|
|
func mergeApnOperationJSON(saved, incoming string) (string, error) {
|
|
var output, updates model.SimCardOperateChangeApnOutput
|
|
if saved != "" {
|
|
if err := json.Unmarshal([]byte(saved), &output); err != nil {
|
|
return "", fmt.Errorf("解析已存APN受理映射失败：%w", err)
|
|
}
|
|
}
|
|
if incoming != "" {
|
|
if err := json.Unmarshal([]byte(incoming), &updates); err != nil {
|
|
return "", fmt.Errorf("解析待存APN受理映射失败：%w", err)
|
|
}
|
|
}
|
|
positions := make(map[model.SimKey]int)
|
|
for i, item := range output {
|
|
if item == nil || item.SimKey == nil {
|
|
return "", fmt.Errorf("已存APN受理映射缺少卡关联")
|
|
}
|
|
positions[*item.SimKey] = i
|
|
}
|
|
for _, item := range updates {
|
|
if item == nil || item.SimKey == nil {
|
|
return "", fmt.Errorf("待存APN受理映射缺少卡关联")
|
|
}
|
|
i, exists := positions[*item.SimKey]
|
|
if !exists {
|
|
positions[*item.SimKey] = len(output)
|
|
output = append(output, item)
|
|
continue
|
|
}
|
|
old := output[i]
|
|
if old.ResData.OrderNum != "" && item.ResData.OrderNum != "" && old.ResData.OrderNum != item.ResData.OrderNum {
|
|
return "", fmt.Errorf("同卡APN受理映射出现不同jobId，禁止覆盖")
|
|
}
|
|
oldTerminal := !old.ResData.Pending && !old.ResData.Uncertain
|
|
newTerminal := !item.ResData.Pending && !item.ResData.Uncertain
|
|
if oldTerminal || (!newTerminal && old.ResData.OrderNum != "") {
|
|
continue
|
|
}
|
|
if item.ResData.OrderNum == "" {
|
|
item.ResData.OrderNum = old.ResData.OrderNum
|
|
}
|
|
output[i] = item
|
|
}
|
|
body, err := json.Marshal(output)
|
|
return string(body), err
|
|
}
|
|
|
|
func unknownApnOperation(input *model.SimCardOperateChangeApnInput, reason string) model.SimCardOperateChangeApnOutput {
|
|
output := make(model.SimCardOperateChangeApnOutput, 0, len(input.SimKeys))
|
|
for _, key := range input.SimKeys {
|
|
output = append(output, &model.SimBaseRes[model.SimCardOperateChangeApnItem]{SimKey: key, Code: consts.ApiStatusError, Message: reason, ResData: model.SimCardOperateChangeApnItem{Uncertain: true}})
|
|
}
|
|
return output
|
|
}
|
|
|
|
// durableApnOperation 先落提交意图，重复投递与重启恢复均不重新办理。
|
|
func durableApnOperation(ctx context.Context, store apnOperationStore, channel string, input *model.SimCardOperateChangeApnInput, submit func(context.Context) (model.SimCardOperateChangeApnOutput, error)) (model.SimCardOperateChangeApnOutput, error) {
|
|
id := input.ReqData.OperationID
|
|
if len(input.SimKeys) == 0 || len(input.SimKeys) > 100 {
|
|
return nil, fmt.Errorf("异步APN子任务必须包含1至100张卡")
|
|
}
|
|
normalized := *input
|
|
normalized.SimKeys = append([]*model.SimKey(nil), input.SimKeys...)
|
|
for _, key := range normalized.SimKeys {
|
|
if key == nil || key.ChanKey != channel || key.ICCID == "" {
|
|
return nil, fmt.Errorf("异步APN子任务卡标识或通道归属无效")
|
|
}
|
|
}
|
|
sort.Slice(normalized.SimKeys, func(i, j int) bool {
|
|
a, b := normalized.SimKeys[i], normalized.SimKeys[j]
|
|
return a.ICCID+"|"+a.MSISDN+"|"+a.IMSI < b.ICCID+"|"+b.MSISDN+"|"+b.IMSI
|
|
})
|
|
normalized.ReqData.QueryJobID = ""
|
|
normalized.ReqData.RecoverOnly = false
|
|
normalized.ReqData.SubmitOnly = true
|
|
payload, err := json.Marshal(normalized)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
digest := sha256.Sum256(payload)
|
|
hash := hex.EncodeToString(digest[:])
|
|
if err := validateApnOperationIdentity(id, channel, hash); err != nil {
|
|
return nil, err
|
|
}
|
|
claimed := false
|
|
if !input.ReqData.RecoverOnly {
|
|
claimed, err = store.Claim(ctx, id, channel, hash)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("保存APN提交意图失败，未调用运营商：%w", err)
|
|
}
|
|
}
|
|
if !claimed {
|
|
savedHash, saved, err := store.Load(ctx, id, channel)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("读取APN受理映射失败：%w", err)
|
|
}
|
|
if savedHash == "" {
|
|
return unknownApnOperation(input, "未找到提交意图，禁止自动重新办理，请核对"), nil
|
|
}
|
|
if savedHash != hash {
|
|
return nil, fmt.Errorf("operationId已绑定不同的办理请求")
|
|
}
|
|
if saved == "" {
|
|
return unknownApnOperation(input, "已记录提交意图但受理结果未确认，禁止重复办理，请核对"), nil
|
|
}
|
|
return completeApnOperation(input, saved)
|
|
}
|
|
var checkpointMu sync.Mutex
|
|
latest := ""
|
|
checkpoint := func(partial model.SimCardOperateChangeApnOutput) error {
|
|
checkpointMu.Lock()
|
|
defer checkpointMu.Unlock()
|
|
snapshot, err := json.Marshal(partial)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
merged, err := mergeApnOperationJSON(latest, string(snapshot))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
latest = merged
|
|
saveCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
|
defer cancel()
|
|
return store.Save(saveCtx, id, channel, hash, latest)
|
|
}
|
|
submitCtx := context.WithValue(ctx, model.ApnOperationCheckpointKey{}, checkpoint)
|
|
output, submitErr := submit(submitCtx)
|
|
if submitErr != nil {
|
|
output = unknownApnOperation(input, "提交结果未确认："+submitErr.Error())
|
|
} else if len(output) == 0 {
|
|
output = unknownApnOperation(input, "提交未返回逐卡受理结果，请核对")
|
|
}
|
|
if err := checkpoint(output); err != nil {
|
|
g.Log().Errorf(context.WithoutCancel(ctx), "APN受理结果持久化失败 operationId=%s chanKey=%s error=%v", id, channel, err)
|
|
return nil, fmt.Errorf("上游受理结果持久化失败，禁止重新办理：%w", err)
|
|
}
|
|
return completeApnOperation(input, latest)
|
|
}
|
|
|
|
func (u *sUpstreamAPI) QuerySimApnCapability(ctx context.Context, input *model.SimCardOperateChangeApnInput) (model.SimApnCapabilityOutput, error) {
|
|
if len(input.SimKeys) == 0 || len(input.SimKeys) > 100 {
|
|
return nil, fmt.Errorf("APN能力探测必须包含1至100张卡")
|
|
}
|
|
output := make(model.SimApnCapabilityOutput, 0, len(input.SimKeys))
|
|
for _, key := range input.SimKeys {
|
|
if key == nil || key.ChanKey == "" || key.ICCID == "" {
|
|
return nil, fmt.Errorf("APN能力探测缺少冻结卡标识")
|
|
}
|
|
output = append(output, &model.SimBaseRes[model.SimApnCapabilityItem]{SimKey: key, ResData: model.SimApnCapabilityItem{ProtocolVersion: 1}})
|
|
}
|
|
// 已有 jobId 只读查询运营商，不依赖本地提交意图表；提交和 operationId 恢复仍需该表。
|
|
if strings.TrimSpace(input.ReqData.QueryJobID) == "" || input.ReqData.SubmitOnly || input.ReqData.RecoverOnly {
|
|
probeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
|
defer cancel()
|
|
if _, err := g.DB().GetAll(probeCtx, "SELECT revision FROM t_apn_operation LIMIT 0"); err != nil {
|
|
return nil, fmt.Errorf("APN持久化迁移未就绪：%w", err)
|
|
}
|
|
}
|
|
return output, nil
|
|
}
|
|
|
|
func completeApnOperation(input *model.SimCardOperateChangeApnInput, saved string) (model.SimCardOperateChangeApnOutput, error) {
|
|
var output model.SimCardOperateChangeApnOutput
|
|
if err := json.Unmarshal([]byte(saved), &output); err != nil {
|
|
return nil, fmt.Errorf("解析已持久化受理映射失败：%w", err)
|
|
}
|
|
for _, key := range input.SimKeys {
|
|
found := false
|
|
for _, item := range output {
|
|
if item != nil && item.SimKey != nil && *item.SimKey == *key {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
missing := *input
|
|
missing.SimKeys = []*model.SimKey{key}
|
|
output = append(output, unknownApnOperation(&missing, "该卡未持久化受理结果，禁止重新办理，请核对")...)
|
|
}
|
|
}
|
|
return output, nil
|
|
}
|