iotlcm/internal/logic/upstreamapi/apn_operation_mysql_test.go

package upstreamapi

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	_ "github.com/gogf/gf/contrib/drivers/mysql/v2"
	"github.com/gogf/gf/v2/database/gdb"
	"gopkg.in/yaml.v3"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model"
)

// 仅显式授权的loopback专用实例可写；身份核验始终先于DDL和DML。
func TestApnOperationMySQLIntegration(t *testing.T) {
	if os.Getenv("APN_MYSQL_INTEGRATION") != "1" {
		t.Skip("未启用本机专用MySQL集成；不会连接默认业务数据库")
	}
	const port = "3306"
	const name = "iotdb_apn_chan_test_20261015"
	const expectedUUID = "4fa545af-ab36-11f1-be02-040e3c9de361"
	const expectedDir = "D:/MySQL/MySQL Server 8.0/Data/"
	if value := os.Getenv("APN_MYSQL_PORT"); value != "" && value != port {
		t.Fatal("只允许已核实的127.0.0.1:3306")
	}
	if value := os.Getenv("APN_MYSQL_DATABASE"); value != "" && value != name {
		t.Fatal("只允许指定chan测试库")
	}
	configPath := filepath.Join("..", "..", "..", "..", "iotlcm-sms", "deploy", "config.yaml")
	configBody, err := os.ReadFile(configPath)
	if err != nil {
		t.Fatal("无法读取本机已有配置")
	}
	var config struct {
		SceneDbDsn string `yaml:"sceneDbDsn"`
	}
	if err := yaml.Unmarshal(configBody, &config); err != nil {
		t.Fatal("本机已有配置解析失败")
	}
	dsn, err := mysql.ParseDSN(config.SceneDbDsn)
	if err != nil || dsn.Net != "tcp" || (dsn.Addr != "127.0.0.1:3306" && dsn.Addr != "localhost:3306") {
		t.Fatal("已有配置不是已授权本机连接，禁止连接")
	}
	db, err := gdb.New(gdb.ConfigNode{Type: "mysql", Host: "127.0.0.1", Port: port, User: dsn.User, Pass: dsn.Passwd, Name: name})
	if err != nil {
		t.Fatal("建立本机测试连接失败")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	identity, err := db.GetOne(ctx, "SELECT @@hostname AS hostname,@@port AS port,@@datadir AS datadir,@@server_uuid AS server_uuid,DATABASE() AS dbname")
	if err != nil {
		t.Fatal("读取本机实例身份失败")
	}
	normalize := func(s string) string {
		return strings.ToLower(strings.TrimRight(filepath.ToSlash(strings.ReplaceAll(s, "\\", "/")), "/"))
	}
	if identity["port"].String() != port || identity["server_uuid"].String() != expectedUUID || identity["dbname"].String() != name || normalize(identity["datadir"].String()) != normalize(expectedDir) {
		t.Fatalf("隔离实例身份不符，禁止写入：%v", identity)
	}
	t.Logf("本机专用库核验通过 hostname=%s port=%s uuid=%s datadir=%s database=%s", identity["hostname"], identity["port"], identity["server_uuid"], identity["datadir"], identity["dbname"])
	exists, err := db.GetValue(ctx, "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='t_apn_operation'")
	if err != nil {
		t.Fatal(err)
	}
	if exists.Int() == 0 {
		_, err = db.Exec(ctx, "CREATE TABLE t_apn_operation (operation_id varchar(128) NOT NULL,chan_key varchar(128) NOT NULL,request_hash char(64) NOT NULL,response_json longtext NOT NULL,created_at datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),updated_at datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),PRIMARY KEY(operation_id,chan_key)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4")
		if err != nil {
			t.Fatal(err)
		}
	}
	migration, err := os.ReadFile(filepath.Join("..", "..", "..", "..", "iotlcm-database", "SQL", "20261015_apn_operation_chan.sql"))
	if err != nil {
		t.Fatal(err)
	}
	master, err := db.Master()
	if err != nil {
		t.Fatal(err)
	}
	connection, err := master.Conn(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	for repeat := 0; repeat < 2; repeat++ {
		for _, statement := range strings.Split(string(migration), ";") {
			if strings.TrimSpace(statement) == "" {
				continue
			}
			if _, err := connection.ExecContext(ctx, statement); err != nil {
				t.Fatalf("迁移执行失败：%v", err)
			}
		}
	}
	store := sqlApnOperationStore{db: db}
	id := fmt.Sprintf("integration-%d", time.Now().UnixNano())
	channel := "isolated-test-channel"
	hash := strings.Repeat("a", 64)
	claimed, err := store.Claim(ctx, id, channel, hash)
	if err != nil || !claimed {
		t.Fatalf("Claim失败：%v %v", claimed, err)
	}
	t.Cleanup(func() {
		_, _ = db.Exec(context.Background(), "DELETE FROM t_apn_operation WHERE operation_id=? AND chan_key=?", id, channel)
	})
	if claimed, err = store.Claim(ctx, id, channel, hash); err != nil || claimed {
		t.Fatalf("重复Claim未幂等：%v %v", claimed, err)
	}
	if err := store.Save(ctx, id+"-missing", channel, hash, "[]"); err == nil {
		t.Fatal("不存在的映射保存必须失败")
	}
	if err := store.Save(ctx, id, channel, strings.Repeat("b", 64), "[]"); err == nil {
		t.Fatal("不同摘要不得覆盖")
	}
	if claimed, err = store.Claim(ctx, strings.Repeat("x", 129), channel, hash); err == nil || claimed {
		t.Fatal("非法ID不得被INSERT忽略为成功")
	}
	var wg sync.WaitGroup
	errors := make(chan error, 10)
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			key := &model.SimKey{ChanKey: channel, ICCID: fmt.Sprintf("fake-%d", i)}
			body, _ := json.Marshal(model.SimCardOperateChangeApnOutput{&model.SimBaseRes[model.SimCardOperateChangeApnItem]{SimKey: key, ResData: model.SimCardOperateChangeApnItem{Pending: true, OrderNum: fmt.Sprintf("fake-job-%d", i)}}})
			errors <- store.Save(ctx, id, channel, hash, string(body))
		}(i)
	}
	wg.Wait()
	close(errors)
	for err := range errors {
		if err != nil {
			t.Fatal(err)
		}
	}
	_, saved, err := store.Load(ctx, id, channel)
	var output model.SimCardOperateChangeApnOutput
	if err != nil || json.Unmarshal([]byte(saved), &output) != nil || len(output) != 10 {
		t.Fatalf("并发保存丢失卡映射：%s %v", saved, err)
	}
	unknown := &model.SimCardOperateChangeApnInput{SimKeys: []*model.SimKey{output[0].SimKey}}
	body, _ := json.Marshal(unknownApnOperation(unknown, "模拟checkpoint后错误"))
	if err := store.Save(ctx, id, channel, hash, string(body)); err != nil {
		t.Fatal(err)
	}
	_, merged, err := store.Load(ctx, id, channel)
	if err != nil || merged != saved {
		t.Fatalf("已存jobId被unknown覆盖：%s %v", merged, err)
	}
	if err := store.Save(ctx, id, channel, hash, "[]"); err != nil {
		t.Fatal(err)
	}
	row, err := db.GetOne(ctx, "SELECT revision FROM t_apn_operation WHERE operation_id=? AND chan_key=?", id, channel)
	if err != nil || row["revision"].Int() != 12 {
		t.Fatalf("每次保存必须影响一行并递增revision：%v %v", row, err)
	}
}