package data

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/gogf/gf/v2/database/gdb"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/batchresult"
	service "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/data"
)

func testSubBatchDB(t *testing.T) (*Repository, context.Context) {
	t.Helper()
	ctx := context.Background()
	var db gdb.DB
	var err error
	mysqlTest := os.Getenv("APN_WORKER_MYSQL_TEST") == "1"
	if mysqlTest {
		db = testSubBatchMySQL(t)
	} else {
		db, err = gdb.New(gdb.ConfigNode{Type: "sqlite", Link: "sqlite::@file(" + t.TempDir() + "/apn.db)"})
	}
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if err := db.Close(ctx); err != nil {
			t.Error(err)
		}
	})
	if !mysqlTest {
		if _, err := db.Exec(ctx, "PRAGMA journal_mode=WAL"); err != nil {
			t.Fatal(err)
		}
	}
	for _, statement := range []string{
		"CREATE TABLE batch_main (id INTEGER PRIMARY KEY, batch_no TEXT, batch_type INTEGER, batch_async_status INTEGER, tenant_id TEXT, start_time DATETIME, updated_at DATETIME, created_by INTEGER, created_dept INTEGER, valid_count INTEGER, success_count INTEGER, fail_count INTEGER, run_result INTEGER, fail_reason TEXT, batch_info TEXT, end_time DATETIME, updated_by INTEGER)",
		"CREATE TABLE data_operation_parent (batch_no TEXT PRIMARY KEY, ready INTEGER DEFAULT 0, finalize_token TEXT, finalize_until DATETIME, recovery_attempt_at DATETIME, last_reason TEXT)",
		"INSERT INTO data_operation_parent (batch_no,ready) VALUES ('S-1',1),('S-2',1)",
		"CREATE TABLE t_sim_card_info (id INTEGER PRIMARY KEY, tenant_id TEXT, iccid TEXT, msisdn TEXT, imsi TEXT, chan_key TEXT, data_service_status INTEGER, status_sync_time DATETIME, traffic_cutoff_triggered_at DATETIME, updated_by INTEGER)",
		"CREATE TABLE batch_detail (id INTEGER PRIMARY KEY, batch_no TEXT, batch_type INTEGER DEFAULT 110, row_no INTEGER, item_id TEXT, detail_status INTEGER, error_code TEXT, error_reason TEXT, snapshot TEXT, updated_by INTEGER)",
		"CREATE TABLE sys_oss (id INTEGER PRIMARY KEY AUTOINCREMENT,tenant_id TEXT,file_name TEXT,original_name TEXT,file_suffix TEXT,path TEXT,url TEXT,created_dept INTEGER,created_by INTEGER,updated_by INTEGER,service TEXT,md5 TEXT,file_size INTEGER)",
		"CREATE TABLE t_resource_batch_download (id INTEGER PRIMARY KEY AUTOINCREMENT,batch_no TEXT,task_name TEXT,resource_type INTEGER,export_scene INTEGER,total_count INTEGER,success_count INTEGER,fail_count INTEGER,download_status INTEGER,oss_id INTEGER,file_name TEXT,file_ext TEXT,file_url TEXT,file_path TEXT,file_size INTEGER,file_md5 TEXT,download_count INTEGER,max_download_count INTEGER,dept_id INTEGER,start_time DATETIME,finish_time DATETIME,tenant_id TEXT,created_dept INTEGER,created_by INTEGER,updated_by INTEGER)",
		"CREATE TABLE data_operation_sub_batch (id INTEGER PRIMARY KEY AUTOINCREMENT, batch_no TEXT, batch_type INTEGER, chan_key TEXT, operation_id TEXT UNIQUE, detail_ids TEXT, request_data TEXT, job_id TEXT, status TEXT, poll_count INTEGER DEFAULT 0, next_check_at DATETIME, submitted_at DATETIME, lease_token TEXT, lease_until DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_reason TEXT)",
		"CREATE TABLE data_operation_sub_detail (detail_id INTEGER PRIMARY KEY, operation_id TEXT, batch_no TEXT, iccid TEXT, UNIQUE(batch_no,iccid))",
		"CREATE TABLE data_operation_card_claim (tenant_id TEXT, iccid TEXT, batch_id INTEGER, operation_id TEXT, active_batch_id INTEGER DEFAULT 0, active_operation_id TEXT DEFAULT '', PRIMARY KEY(tenant_id,iccid))",
		"INSERT INTO batch_main (id,batch_no,batch_type,batch_async_status,tenant_id) VALUES (10,'S-1',110,2,'tenant'),(11,'S-2',109,2,'tenant')",
		"INSERT INTO batch_detail (id,batch_no,detail_status) VALUES (1,'S-1',1),(2,'S-1',1),(3,'S-2',1)",
	} {
		if mysqlTest {
			if strings.HasPrefix(statement, "CREATE TABLE data_operation_") {
				continue
			}
			statement = strings.ReplaceAll(statement, "INTEGER PRIMARY KEY AUTOINCREMENT", "BIGINT PRIMARY KEY AUTO_INCREMENT")
			statement = strings.ReplaceAll(statement, " TEXT", " VARCHAR(512)")
			statement = strings.ReplaceAll(statement, "snapshot VARCHAR(512)", "snapshot JSON")
			statement = strings.ReplaceAll(statement, "batch_info VARCHAR(512)", "batch_info JSON")
			if strings.HasPrefix(statement, "CREATE TABLE ") {
				statement += " DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci"
			}
		}
		if _, err := db.Exec(ctx, statement); err != nil {
			t.Fatalf("setup %q: %v", statement, err)
		}
	}
	return &Repository{db: db}, ctx
}

func testSub(batchNo, operationID, chanKey string, detailIDs []int64, iccids []string) service.SubBatch {
	keys := make([]*model.SimKey, 0, len(detailIDs))
	for i, id := range detailIDs {
		_ = id
		keys = append(keys, &model.SimKey{ICCID: iccids[i], ChanKey: chanKey, MSISDN: fmt.Sprint(i + 1)})
	}
	return service.SubBatch{BatchNo: batchNo, BatchType: 110, ChanKey: chanKey, OperationID: operationID, DetailIDs: detailIDs, Request: model.SimCardApnChangeInput{SimKeys: keys, ReqData: model.SimCardApnChangeInputItem{OperType: "1", OperationID: operationID, SubmitOnly: true}}}
}

func TestSubBatchDurableClaimCASAndReview(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	b := testSub("S-1", "intent-1", "a", []int64{1, 2}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatalf("幂等准备失败: %v", err)
	}
	first, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner1", time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if first == nil || first.Status != "prepared" {
		t.Fatalf("claim=%+v", first)
	}
	second, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner2", time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if second != nil {
		t.Fatal("并发claim不应成功")
	}
	if err := repo.UpdateSubBatch(ctx, *first, "accepted", "job-1", "已受理", time.Now().Add(time.Minute), 0); err != nil {
		t.Fatal(err)
	}
	if err := repo.UpdateSubBatch(ctx, *first, "done", "", "stale", time.Time{}, 0); err == nil {
		t.Fatal("旧lease竟能覆盖")
	}
	row, err := repo.database().GetOne(ctx, "SELECT status,job_id FROM data_operation_sub_batch WHERE operation_id=?", b.OperationID)
	if err != nil {
		t.Fatal(err)
	}
	if row["status"].String() != "accepted" || row["job_id"].String() != "job-1" {
		t.Fatalf("jobId未立即落库: %+v", row)
	}
	if _, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner3", time.Now()); err != nil {
		t.Fatal(err)
	}
	if second, err = repo.ClaimSubBatch(ctx, b.OperationID, "owner3", time.Now().Add(2*time.Minute)); err != nil {
		t.Fatal(err)
	}
	if second == nil || second.JobID != "job-1" {
		t.Fatalf("无法恢复已受理jobId: %+v", second)
	}
	if err := repo.UpdateSubBatch(ctx, *second, "review", "job-1", "结果未确认", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	for _, id := range b.DetailIDs {
		row, err := repo.database().GetOne(ctx, "SELECT detail_status,error_reason FROM batch_detail WHERE id=?", id)
		if err != nil {
			t.Fatal(err)
		}
		if row["detail_status"].Int() != consts.BatchDetailStatusUnknown || row["error_reason"].String() != "结果未确认" {
			t.Fatalf("未知状态落库失败 id=%d row=%+v", id, row)
		}
	}
	if ids, err := repo.ListDueSubBatches(ctx, time.Now().Add(time.Hour), 10); err != nil {
		t.Fatal(err)
	} else if len(ids) != 0 {
		t.Fatalf("review不应自动重试: %+v", ids)
	}
}

func TestSubBatchPreparationRollbackAndReadyBarrier(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	b := testSub("S-1", "barrier", "a", []int64{1}, []string{"A"})
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_parent SET ready=0 WHERE batch_no='S-1'")
	interrupted := fmt.Errorf("模拟准备中断")
	err := repo.ResumeOrPrepare(ctx, "S-1", func(prep context.Context) error {
		if err := repo.PrepareSubBatch(prep, b); err != nil {
			return err
		}
		return interrupted
	})
	if err == nil {
		t.Fatal("中断应回滚")
	}
	n, _ := repo.database().Model(subTable).Ctx(ctx).Count()
	if n != 0 {
		t.Fatal("部分准备未回滚")
	}
	if err := repo.ResumeOrPrepare(ctx, "S-1", func(prep context.Context) error { return repo.PrepareSubBatch(prep, b) }); err != nil {
		t.Fatal(err)
	}
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_parent SET ready=0 WHERE batch_no='S-1'")
	if got, err := repo.ClaimSubBatch(ctx, b.OperationID, "premature", time.Now()); err != nil || got != nil {
		t.Fatalf("ready前可领取: %+v %v", got, err)
	}
	ids, err := repo.ListDueSubBatches(ctx, time.Now().Add(time.Minute), 10)
	if err != nil || len(ids) != 0 {
		t.Fatalf("ready前可扫描: %v %v", ids, err)
	}
}

func TestSubBatchActiveBlocksReverseAndReviewTerminatesUnsubmitted(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	older := testSub("S-1", "active-old", "a", []int64{1}, []string{"A"})
	newer := testSub("S-2", "queued-new", "a", []int64{3}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, older); err != nil {
		t.Fatal(err)
	}
	first, err := repo.ClaimSubBatch(ctx, older.OperationID, "owner1", time.Now())
	if err != nil || first == nil {
		t.Fatalf("%+v %v", first, err)
	}
	if err := repo.PrepareSubBatch(ctx, newer); err != nil {
		t.Fatal(err)
	}
	if got, err := repo.ClaimSubBatch(ctx, newer.OperationID, "reverse", time.Now()); err != nil || got != nil {
		t.Fatalf("在途前序未阻断反向: %+v %v", got, err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT active_operation_id,operation_id FROM data_operation_card_claim WHERE iccid='A'")
	if row["active_operation_id"].String() != older.OperationID || row["operation_id"].String() != newer.OperationID {
		t.Fatalf("queued抢占active: %v", row)
	}
	if err := repo.UpdateSubBatch(ctx, *first, "review", "", "前序未知", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET next_check_at=NULL WHERE operation_id=?", newer.OperationID)
	if got, err := repo.ClaimSubBatch(ctx, newer.OperationID, "reverse", time.Now()); err != nil || got != nil {
		t.Fatalf("未知前序未阻断: %+v %v", got, err)
	}
	row, _ = repo.database().GetOne(ctx, "SELECT status,submitted_at,poll_count FROM data_operation_sub_batch WHERE operation_id=?", newer.OperationID)
	if row["status"].String() != "review" || !row["submitted_at"].IsNil() || row["poll_count"].Int() != 0 {
		t.Fatalf("未提交后继应有限收尾: %v", row)
	}
	row, _ = repo.database().GetOne(ctx, "SELECT active_operation_id FROM data_operation_card_claim WHERE iccid='A'")
	if row["active_operation_id"].String() != older.OperationID {
		t.Fatal("未知active被释放")
	}
}

func TestSubBatchBlockedGroupNeverAcquiresPartialActive(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	old := testSub("S-1", "old-B", "a", []int64{1}, []string{"B"})
	if err := repo.PrepareSubBatch(ctx, old); err != nil {
		t.Fatal(err)
	}
	if _, err := repo.ClaimSubBatch(ctx, old.OperationID, "old", time.Now()); err != nil {
		t.Fatal(err)
	}
	_, _ = repo.database().Exec(ctx, "INSERT INTO batch_detail(id,batch_no,detail_status)VALUES(4,'S-2',1)")
	newer := testSub("S-2", "AB", "a", []int64{3, 4}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, newer); err != nil {
		t.Fatal(err)
	}
	if got, err := repo.ClaimSubBatch(ctx, newer.OperationID, "new", time.Now()); err != nil || got != nil {
		t.Fatalf("阻塞组竟领取: %+v %v", got, err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT active_operation_id FROM data_operation_card_claim WHERE iccid='A'")
	if row["active_operation_id"].String() != "" {
		t.Fatal("部分卡占active产生死锁")
	}
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET created_at=?,next_check_at=NULL WHERE operation_id=?", time.Now().Add(-16*time.Minute), newer.OperationID)
	if _, err := repo.ClaimSubBatch(ctx, newer.OperationID, "deadline", time.Now()); err != nil {
		t.Fatal(err)
	}
	row, _ = repo.database().GetOne(ctx, "SELECT status FROM data_operation_sub_batch WHERE operation_id=?", newer.OperationID)
	if row["status"].String() != "review" {
		t.Fatal("在途阻塞超过窗口仍永久Running")
	}
}

func TestSubBatchRecoveryWindowIsFair(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	for i := 0; i < 35; i++ {
		no := fmt.Sprintf("R-%02d", i)
		if _, err := repo.database().Exec(ctx, "INSERT INTO batch_main(id,batch_no,batch_type,batch_async_status,tenant_id)VALUES(?,?,109,2,'tenant')", 100+i, no); err != nil {
			t.Fatal(err)
		}
		if _, err := repo.database().Exec(ctx, "INSERT INTO data_operation_parent(batch_no,ready)VALUES(?,0)", no); err != nil {
			t.Fatal(err)
		}
	}
	seen := map[string]bool{}
	for i := 0; i < 3; i++ {
		ids, err := repo.ListRecoverableBatches(ctx, 16)
		if err != nil {
			t.Fatal(err)
		}
		for _, id := range ids {
			seen[id] = true
		}
	}
	for i := 0; i < 35; i++ {
		if !seen[fmt.Sprintf("R-%02d", i)] {
			t.Fatalf("固定恢复窗口饥饿: %d", i)
		}
	}
}

type durableGenerator struct {
	calls   atomic.Int32
	create  func(context.Context, batchresult.Request) (batchresult.Artifact, error)
	request batchresult.Request
}

func (f *durableGenerator) Create(ctx context.Context, request batchresult.Request) (batchresult.Artifact, error) {
	n := f.calls.Add(1)
	f.request = request
	if f.create != nil {
		return f.create(ctx, request)
	}
	return batchresult.Artifact{FileName: fmt.Sprintf("%s-%d.xlsx", request.BatchNo, n), OriginalName: request.BatchNo + ".xlsx"}, nil
}

func TestSubBatchProgressIsMonotonicAndMissingCardSafe(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET item_id=CASE id WHEN 1 THEN 'A' ELSE 'B' END,snapshot='{\"authorized\":true,\"statusBefore\":1}' WHERE batch_no='S-1'")
	_, _ = repo.database().Exec(ctx, "INSERT INTO t_sim_card_info(id,tenant_id,iccid,data_service_status)VALUES(1,'tenant','A',1)")
	b := testSub("S-1", "partial", "a", []int64{1, 2}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.ClaimSubBatch(ctx, b.OperationID, "first", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("claim: %v %v", claimed, err)
	}
	claimed.JobID = "job-partial"
	if err := repo.SaveSubBatchProgress(ctx, *claimed, []service.Outcome{{Success: true, OrderNum: "order-A"}, {Pending: true}}); err != nil {
		t.Fatal(err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT job_id FROM data_operation_sub_batch WHERE operation_id=?", b.OperationID)
	if row["job_id"].String() != "job-partial" {
		t.Fatal("jobId未随部分结果落库")
	}
	// A已明确成功，后续缺失不得覆盖；B无卡实体也不能panic丢掉整组结果。
	if err := repo.SaveSubBatchProgress(ctx, *claimed, []service.Outcome{{Pending: true}, {Success: false, Code: 42, Message: "明确拒绝"}}); err != nil {
		t.Fatal(err)
	}
	if err := repo.UpdateSubBatch(ctx, *claimed, "review", claimed.JobID, "窗口结束", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	rows, err := repo.database().GetAll(ctx, "SELECT detail_status,error_reason FROM batch_detail WHERE batch_no='S-1' ORDER BY id")
	if err != nil {
		t.Fatal(err)
	}
	if rows[0]["detail_status"].Int() != consts.BatchDetailStatusSuccess || rows[1]["detail_status"].Int() != consts.BatchDetailStatusFailed || rows[1]["error_reason"].String() != "明确拒绝" {
		t.Fatalf("review覆盖逐卡终态: %v", rows)
	}
	gen := &durableGenerator{}
	repo.resultFiles = gen
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err != nil {
		t.Fatal(err)
	}
	if len(gen.request.Rows) != 2 || !gen.request.Rows[0].Success || gen.request.Rows[1].Success {
		t.Fatalf("真实DB导出接线错误: %+v", gen.request)
	}
}

func TestSubBatchUnknownOnlyChangesUnconfirmedRows(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET item_id=CASE id WHEN 1 THEN 'A' ELSE 'B' END,snapshot='{\"authorized\":true}' WHERE batch_no='S-1'")
	b := testSub("S-1", "unknown", "a", []int64{1, 2}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner", time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if err := repo.SaveSubBatchProgress(ctx, *claimed, []service.Outcome{{Success: true, OrderNum: "confirmed"}, {Pending: true}}); err != nil {
		t.Fatal(err)
	}
	if err := repo.UpdateSubBatch(ctx, *claimed, "review", "job", "具体超时原因", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	gen := &durableGenerator{}
	repo.resultFiles = gen
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err != nil {
		t.Fatal(err)
	}
	main, _ := repo.database().GetOne(ctx, "SELECT batch_async_status,success_count,fail_count,valid_count,batch_info FROM batch_main WHERE batch_no='S-1'")
	if main["batch_async_status"].Int() != consts.BatchAsyncStatusReview || main["success_count"].Int() != 1 || main["fail_count"].Int() != 0 || main["valid_count"].Int() != 2 {
		t.Fatalf("未知误计失败或清零有效数: %v", main)
	}
	if len(gen.request.Rows) != 2 || gen.request.Rows[1].DetailStatus != consts.BatchDetailStatusUnknown || gen.request.Rows[1].Reason != "具体超时原因" {
		t.Fatalf("待核查导出原因丢失: %+v", gen.request)
	}
}

func TestSubBatchZeroChildrenFinalizeLeaseRecoveryAndConcurrency(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET detail_status=3,error_reason='校验不通过' WHERE batch_no='S-1'")
	started := make(chan struct{})
	release := make(chan struct{})
	gen := &durableGenerator{create: func(context.Context, batchresult.Request) (batchresult.Artifact, error) {
		close(started)
		<-release
		return batchresult.Artifact{FileName: "unique.xlsx"}, nil
	}}
	repo.resultFiles = gen
	var wg sync.WaitGroup
	wg.Add(1)
	var firstErr error
	go func() { defer wg.Done(); firstErr = repo.FinalizeReadyBatch(ctx, "S-1") }()
	<-started
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err != nil {
		t.Fatal(err)
	}
	close(release)
	wg.Wait()
	if firstErr != nil {
		t.Fatal(firstErr)
	}
	if gen.calls.Load() != 1 {
		t.Fatal("并发最终化重复生成")
	}
	n, _ := repo.database().Model("sys_oss").Ctx(ctx).Count()
	if n != 1 {
		t.Fatal("重复发布OSS")
	}
	n, _ = repo.database().Model("t_resource_batch_download").Ctx(ctx).Count()
	if n != 1 {
		t.Fatal("重复下载记录")
	}
	// 真正零明细父批，以及前任已过期租约应能恢复。
	_, _ = repo.database().Exec(ctx, "DELETE FROM batch_detail WHERE batch_no='S-2'")
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_parent SET finalize_token='crashed',finalize_until=? WHERE batch_no='S-2'", time.Now().Add(-time.Minute))
	repo.resultFiles = &durableGenerator{}
	if err := repo.FinalizeReadyBatch(ctx, "S-2"); err != nil {
		t.Fatalf("零明细/过期租约未恢复: %v", err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT batch_async_status FROM batch_main WHERE batch_no='S-2'")
	if row["batch_async_status"].Int() == consts.BatchAsyncStatusProcessing {
		t.Fatal("零子批不收尾")
	}
}

func TestSubBatchFinalizeFailureReleasesLease(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET detail_status=3 WHERE batch_no='S-1'")
	repo.resultFiles = &durableGenerator{create: func(context.Context, batchresult.Request) (batchresult.Artifact, error) {
		return batchresult.Artifact{}, errors.New("模拟文件生成失败")
	}}
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err == nil {
		t.Fatal("应暴露生成失败")
	}
	repo.resultFiles = &durableGenerator{}
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err != nil {
		t.Fatal(err)
	}
}

func TestSubBatchExplicitFailureReleasesActiveForReverse(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET item_id='A' WHERE id=1")
	old := testSub("S-1", "failed-old", "a", []int64{1}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, old); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.ClaimSubBatch(ctx, old.OperationID, "owner", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("claim: %+v %v", claimed, err)
	}
	logMySQLLease(t, repo, claimed.ID, claimed.LeaseToken)
	if err := repo.CompleteSubBatch(ctx, *claimed, []service.Outcome{{Code: 9999, Message: "任务状态4，已完成且明确失败"}}); err != nil {
		t.Fatal(err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT detail_status FROM batch_detail WHERE id=1")
	if row["detail_status"].Int() != consts.BatchDetailStatusFailed {
		t.Fatal("明确9999未持久终态")
	}
	row, _ = repo.database().GetOne(ctx, "SELECT active_operation_id FROM data_operation_card_claim WHERE iccid='A'")
	if row["active_operation_id"].String() != "" {
		t.Fatal("明确失败仍保留active")
	}
	reverse := testSub("S-2", "reverse-after-failed", "a", []int64{3}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, reverse); err != nil {
		t.Fatal(err)
	}
	if got, err := repo.ClaimSubBatch(ctx, reverse.OperationID, "reverse", time.Now()); err != nil || got == nil {
		t.Fatalf("明确失败后反向被阻塞: %+v %v", got, err)
	}
}

func TestSubBatchTypedUnsentReleasesOnlyOwnActiveAndRestoresPrepared(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	b := testSub("S-1", "unsent", "a", []int64{1, 2}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.ClaimSubBatch(ctx, b.OperationID, "first", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("%+v %v", claimed, err)
	}
	if err := repo.ReleaseUnsentSubBatch(ctx, *claimed, "第二次能力探测拒绝"); err != nil {
		t.Fatal(err)
	}
	row, _ := repo.database().GetOne(ctx, "SELECT status,submitted_at,poll_count,lease_token FROM data_operation_sub_batch WHERE operation_id=?", b.OperationID)
	if row["status"].String() != "prepared" || !row["submitted_at"].IsNil() || row["poll_count"].Int() != 0 || row["lease_token"].String() != "" {
		t.Fatalf("回退prepared失败: %v", row)
	}
	n, _ := repo.database().Model(claimTable).Ctx(ctx).Where("active_operation_id", b.OperationID).Count()
	if n != 0 {
		t.Fatal("未发送仍占active")
	}
	if err := repo.ReleaseUnsentSubBatch(ctx, *claimed, "陈旧回退"); err == nil {
		t.Fatal("陈旧lease可重放回退")
	}
	_, _ = repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET next_check_at=NULL WHERE operation_id=?", b.OperationID)
	retry, err := repo.ClaimSubBatch(ctx, b.OperationID, "retry", time.Now())
	if err != nil || retry == nil || retry.Status != "prepared" {
		t.Fatalf("回退后无法唯一首发: %+v %v", retry, err)
	}
	if err := repo.UpdateSubBatch(ctx, *retry, "accepted", "job", "已受理", time.Now(), 0); err != nil {
		t.Fatal(err)
	}
	query, err := repo.ClaimSubBatch(ctx, b.OperationID, "query", time.Now().Add(time.Second))
	if err != nil || query == nil {
		t.Fatalf("%+v %v", query, err)
	}
	submitted := query.SubmittedAt
	if err := repo.ReleaseUnsentSubBatch(ctx, *query, "查询探测拒绝"); err != nil {
		t.Fatal(err)
	}
	stored, err := repo.GetSubBatch(ctx, b.OperationID)
	if err != nil || stored.PollCount != 0 || stored.Status != "accepted" || !stored.SubmittedAt.Equal(submitted) {
		t.Fatalf("未发查询消耗轮次或丢首发时点: %+v %v", stored, err)
	}
	n, _ = repo.database().Model(claimTable).Ctx(ctx).Where("active_operation_id", b.OperationID).Count()
	if n != 2 {
		t.Fatal("已真实提交的active被查询门禁释放")
	}
}

type prepareOnlyAPN struct{}

func (prepareOnlyAPN) ChangeSimApnFunction(context.Context, *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	return nil, errors.New("准备阶段不得调用上游")
}
func (prepareOnlyAPN) SubmitSimApnFunction(context.Context, *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	return nil, errors.New("准备阶段不得首发")
}
func (prepareOnlyAPN) QuerySimApnFunctionResult(context.Context, *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
	return nil, errors.New("准备阶段不得补查")
}

func TestSubBatchCutoffDoesNotSkipDisabledCardWithActiveOpen(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	if _, err := repo.database().Exec(ctx, "UPDATE batch_main SET batch_type=109 WHERE batch_no='S-1'"); err != nil {
		t.Fatal(err)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE batch_main SET batch_type=107 WHERE batch_no='S-2'"); err != nil {
		t.Fatal(err)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE batch_detail SET batch_type=107,item_id='A',snapshot='{\"authorized\":true}' WHERE id=3"); err != nil {
		t.Fatal(err)
	}
	if _, err := repo.database().Exec(ctx, "INSERT INTO t_sim_card_info(id,tenant_id,iccid,chan_key,data_service_status)VALUES(1,'tenant','A','a',?)", consts.SimDataServiceStatusDisabled); err != nil {
		t.Fatal(err)
	}
	old := testSub("S-1", "opening", "a", []int64{1}, []string{"A"})
	old.BatchType = 109
	old.Request.ReqData.OperType = consts.SimApnOperTypeOpen
	if err := repo.PrepareSubBatch(ctx, old); err != nil {
		t.Fatal(err)
	}
	if got, err := repo.ClaimSubBatch(ctx, old.OperationID, "open-owner", time.Now()); err != nil || got == nil {
		t.Fatalf("开通领取: %+v %v", got, err)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE data_operation_parent SET ready=0 WHERE batch_no='S-2'"); err != nil {
		t.Fatal(err)
	}
	if err := service.NewExecutor(repo, prepareOnlyAPN{}).Execute(ctx, "S-2"); err != nil {
		t.Fatal(err)
	}
	row, err := repo.database().GetOne(ctx, "SELECT operation_id,status FROM data_operation_sub_batch WHERE batch_no='S-2'")
	if err != nil || len(row) == 0 {
		t.Fatalf("本地disabled绕过冻结: %v %v", row, err)
	}
	detail, _ := repo.database().GetOne(ctx, "SELECT detail_status FROM batch_detail WHERE id=3")
	if detail["detail_status"].Int() != consts.BatchDetailStatusRunning {
		t.Fatal("开在途时达量停被提前成功")
	}
	if got, err := repo.ClaimSubBatch(ctx, row["operation_id"].String(), "cutoff", time.Now()); err != nil || got != nil {
		t.Fatalf("达量停绕过前序active: %+v %v", got, err)
	}
}

func TestSubBatchLegacyProcessingWithoutParentNeverCreatesIntent(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	if _, err := repo.database().Exec(ctx, "DELETE FROM data_operation_parent WHERE batch_no='S-1'"); err != nil {
		t.Fatal(err)
	}
	ids, err := repo.ListRecoverableBatches(ctx, 128)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range ids {
		if id == "S-1" {
			t.Fatal("恢复扫描捡起无凭据旧批")
		}
	}
	called := false
	if err := repo.ResumeOrPrepare(ctx, "S-1", func(context.Context) error { called = true; return nil }); err == nil || called {
		t.Fatalf("旧批进入准备/提交: called=%t err=%v", called, err)
	}
	if err := repo.ProtectExecutionFailure(ctx, "S-1", "旧MQ重投失败"); err != nil {
		t.Fatal(err)
	}
	if exists, err := repo.HasDurableParent(ctx, "S-1"); err != nil || exists {
		t.Fatalf("失败保护或扫描凭空补凭据: exists=%t err=%v", exists, err)
	}
	n, err := repo.database().Model(subTable).Ctx(ctx).Where("batch_no", "S-1").Count()
	if err != nil || n != 0 {
		t.Fatalf("旧批产生新提交意图: %d %v", n, err)
	}
}

func TestSubBatchDetailAndCardClaimProtectNewerOperation(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	older := testSub("S-1", "intent-old", "a", []int64{1}, []string{"A"})
	newer := testSub("S-2", "intent-new", "a", []int64{3}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, older); err != nil {
		t.Fatal(err)
	}
	if err := repo.PrepareSubBatch(ctx, newer); err != nil {
		t.Fatal(err)
	}
	claim, err := repo.database().GetOne(ctx, "SELECT batch_id,operation_id FROM data_operation_card_claim WHERE tenant_id='tenant' AND iccid='A'")
	if err != nil {
		t.Fatal(err)
	}
	if claim["batch_id"].Int() != 11 || claim["operation_id"].String() != newer.OperationID {
		t.Fatalf("后续反向操作claim未生效: %+v", claim)
	}
	// Even with a differently named intent, the same detail cannot join another submission.
	duplicate := testSub("S-1", "intent-duplicate", "a", []int64{1}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, duplicate); err == nil {
		t.Fatal("同一明细重新归属另一提交意图应失败")
	}
	count, err := repo.database().GetOne(ctx, "SELECT COUNT(*) AS n FROM data_operation_sub_batch WHERE batch_no='S-1'")
	if err != nil {
		t.Fatal(err)
	}
	if count["n"].Int() != 1 {
		t.Fatalf("失败准备未回滚: %+v", count)
	}
}

// 并发的两次手动补查（不同 operationId）不得互相阻塞退回：
// 在途 review_query 子批不是“未终结”，否则两边都会以为对方还没完成而同时放弃退回。
func TestSubBatchReviewParentReturnsDespiteInFlightManualQuery(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	inflight := testSub("S-1", "inflight", "a", []int64{1}, []string{"A"})
	settle := testSub("S-1", "settle", "a", []int64{2}, []string{"B"})
	for _, sub := range []service.SubBatch{inflight, settle} {
		if err := repo.PrepareSubBatch(ctx, sub); err != nil {
			t.Fatal(err)
		}
	}
	first, err := repo.ClaimSubBatch(ctx, inflight.OperationID, "owner-a", time.Now())
	if err != nil || first == nil {
		t.Fatalf("claim A: %+v %v", first, err)
	}
	if err := repo.UpdateSubBatch(ctx, *first, "review", "job-a", "补查窗口耗尽", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	second, err := repo.ClaimSubBatch(ctx, settle.OperationID, "owner-b", time.Now())
	if err != nil || second == nil {
		t.Fatalf("claim B: %+v %v", second, err)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE batch_main SET batch_async_status=? WHERE batch_no='S-1'", consts.BatchAsyncStatusReview); err != nil {
		t.Fatal(err)
	}
	// A 处于已提交的在途补查状态；B 此时写回终态，仍必须能退回处理中。
	if manual, err := repo.ClaimReviewSubBatch(ctx, inflight.OperationID, "manual", time.Now()); err != nil || manual == nil {
		t.Fatalf("A 在途补查领取失败: %+v %v", manual, err)
	}
	if err := repo.UpdateSubBatch(ctx, *second, "done", "job-b", "", time.Time{}, second.PollCount); err != nil {
		t.Fatal(err)
	}
	main, _ := repo.database().GetOne(ctx, "SELECT batch_async_status FROM batch_main WHERE batch_no='S-1'")
	if main["batch_async_status"].Int() != consts.BatchAsyncStatusProcessing {
		t.Fatalf("在途补查阻塞了退回，父批将永久挂待核查: %v", main)
	}
}

// 子批全部终结后父批必须能回到恢复扫描重新汇总：
// 修复“手动补查已把子批写成终态，但汇总失败/进程退出后父批永久挂待核查”的死角，
// 同时保证合法停查的批次不会因此被反复捞起空转。
func TestSubBatchReviewParentReentersRecoveryOnceSettled(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	_, _ = repo.database().Exec(ctx, "UPDATE batch_detail SET item_id=CASE id WHEN 1 THEN 'A' ELSE 'B' END,snapshot='{\"authorized\":true}' WHERE batch_no='S-1'")
	b := testSub("S-1", "settle", "a", []int64{1, 2}, []string{"A", "B"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("claim: %+v %v", claimed, err)
	}
	if err := repo.UpdateSubBatch(ctx, *claimed, "review", "job-1", "补查窗口耗尽", time.Time{}, 4); err != nil {
		t.Fatal(err)
	}
	// 模拟此前一次汇总已把父批停在待核查。
	if _, err := repo.database().Exec(ctx, "UPDATE batch_main SET batch_async_status=? WHERE batch_no='S-1'", consts.BatchAsyncStatusReview); err != nil {
		t.Fatal(err)
	}
	// 手动同步领取并写回终态：这正是汇总失败/进程退出前最后一步。
	manual, err := repo.ClaimReviewSubBatch(ctx, b.OperationID, "manual", time.Now())
	if err != nil || manual == nil {
		t.Fatalf("手动补查领取失败: %+v %v", manual, err)
	}
	if err := repo.UpdateSubBatch(ctx, *manual, "done", "job-1", "", time.Time{}, manual.PollCount); err != nil {
		t.Fatal(err)
	}
	main, _ := repo.database().GetOne(ctx, "SELECT batch_async_status FROM batch_main WHERE batch_no='S-1'")
	if main["batch_async_status"].Int() != consts.BatchAsyncStatusProcessing {
		t.Fatalf("终态后未退回处理中，父批将永久挂待核查: %v", main)
	}
	if !recoverableContains(t, repo, ctx, "S-1") {
		t.Fatal("退回处理中后仍未被恢复扫描领取，无法自动汇总")
	}
	// 汇总成功后若仍有未知明细，父批回到待核查且不再被捞起，避免每 5 秒空转生成结果文件。
	repo.resultFiles = &durableGenerator{}
	if err := repo.FinalizeReadyBatch(ctx, "S-1"); err != nil {
		t.Fatal(err)
	}
	main, _ = repo.database().GetOne(ctx, "SELECT batch_async_status FROM batch_main WHERE batch_no='S-1'")
	if main["batch_async_status"].Int() != consts.BatchAsyncStatusReview {
		t.Fatalf("待核查明细应仍停在待核查: %v", main)
	}
	if recoverableContains(t, repo, ctx, "S-1") {
		t.Fatal("合法待核查批次被反复纳入恢复扫描，将导致空转汇总")
	}
}

func recoverableContains(t *testing.T, repo *Repository, ctx context.Context, batchNo string) bool {
	t.Helper()
	ids, err := repo.ListRecoverableBatches(ctx, 128)
	if err != nil {
		t.Fatal(err)
	}
	for _, id := range ids {
		if id == batchNo {
			return true
		}
	}
	return false
}

// 手动同步只能看到该父批下仍待核查的子批，且不会因为已完成的子批而误触发。
func TestSubBatchListReviewSubBatchesForManualSync(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	review := testSub("S-1", "intent-review", "a", []int64{1}, []string{"A"})
	done := testSub("S-1", "intent-done", "a", []int64{2}, []string{"B"})
	other := testSub("S-2", "intent-other", "a", []int64{3}, []string{"C"})
	for _, sub := range []service.SubBatch{review, done, other} {
		if err := repo.PrepareSubBatch(ctx, sub); err != nil {
			t.Fatal(err)
		}
		claimed, err := repo.ClaimSubBatch(ctx, sub.OperationID, "owner-"+sub.OperationID, time.Now())
		if err != nil || claimed == nil {
			t.Fatalf("claim失败 operationId=%s err=%v", sub.OperationID, err)
		}
		status, jobID, reason := "review", "", "补查窗口耗尽，结果未知"
		if sub.OperationID == done.OperationID {
			status, jobID, reason = "done", "job-done", ""
		}
		if err := repo.UpdateSubBatch(ctx, *claimed, status, jobID, reason, time.Time{}, 4); err != nil {
			t.Fatal(err)
		}
	}

	ids, err := repo.ListReviewSubBatches(ctx, "S-1")
	if err != nil {
		t.Fatal(err)
	}
	if len(ids) != 1 || ids[0] != review.OperationID {
		t.Fatalf("手动同步候选错误: %+v", ids)
	}
	// 租约未过期的在途补查不能被第二个请求抢走。
	if _, err := repo.database().Exec(ctx,
		"UPDATE data_operation_sub_batch SET status='review_query', lease_token='owner', lease_until=? WHERE operation_id=?",
		time.Now().Add(time.Minute), review.OperationID,
	); err != nil {
		t.Fatal(err)
	}
	if ids, err = repo.ListReviewSubBatches(ctx, "S-1"); err != nil || len(ids) != 0 {
		t.Fatalf("在途补查不应进入手动同步: %+v err=%v", ids, err)
	}
	// 租约过期后允许恢复，避免卡在 review_query 永不收敛。
	if _, err := repo.database().Exec(ctx,
		"UPDATE data_operation_sub_batch SET lease_until=? WHERE operation_id=?",
		time.Now().Add(-time.Minute), review.OperationID,
	); err != nil {
		t.Fatal(err)
	}
	if ids, err = repo.ListReviewSubBatches(ctx, "S-1"); err != nil || len(ids) != 1 || ids[0] != review.OperationID {
		t.Fatalf("过期租约应可恢复: %+v err=%v", ids, err)
	}
}