package data

import (
	"context"
	"fmt"
	"os"
	"sync"
	"sync/atomic"

	"github.com/gogf/gf/v2/database/gdb"
	"testing"
	"time"

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

func TestSubBatchPrecheckDelayCASNeverOverwritesClaimOrCompletion(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	b := testSub("S-1", "precheck-cas", "a", []int64{1}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	snapshot, err := repo.GetSubBatch(ctx, b.OperationID)
	if err != nil {
		t.Fatal(err)
	}
	next := time.Now().Add(30 * time.Second)
	if err := repo.DelaySubBatchPrecheck(ctx, *snapshot, next, "能力拒绝"); err != nil {
		t.Fatal(err)
	}
	stored, err := repo.GetSubBatch(ctx, b.OperationID)
	if err != nil || stored.NextCheckAt.Sub(next).Abs() > time.Millisecond {
		t.Fatalf("退避未保存: %+v %v", stored, err)
	}
	if stored.Status != "prepared" || stored.PollCount != 0 || !stored.SubmittedAt.IsZero() || stored.LeaseToken != "" {
		t.Fatalf("退避改变状态: %+v", stored)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET next_check_at=NULL WHERE operation_id=?", b.OperationID); err != nil {
		t.Fatal(err)
	}
	snapshot, _ = repo.GetSubBatch(ctx, b.OperationID)
	claimed, err := repo.ClaimSubBatch(ctx, b.OperationID, "owner", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("领取: %+v %v", claimed, err)
	}
	if err := repo.DelaySubBatchPrecheck(ctx, *snapshot, time.Now().Add(time.Hour), "陈旧拒绝"); err != nil {
		t.Fatal(err)
	}
	stored, _ = repo.GetSubBatch(ctx, b.OperationID)
	if stored.Status != "submitting" || stored.LeaseToken != "owner" || stored.NextCheckAt.After(time.Now().Add(6*time.Minute)) {
		t.Fatalf("陈旧预检覆盖领取: %+v", stored)
	}
	if _, err := repo.database().Exec(ctx, "UPDATE batch_detail SET item_id='A' WHERE id=1"); err != nil {
		t.Fatal(err)
	}
	if err := repo.CompleteSubBatch(ctx, *claimed, []service.Outcome{{Code: 9999, Message: "明确失败"}}); err != nil {
		t.Fatal(err)
	}
	if err := repo.DelaySubBatchPrecheck(ctx, *snapshot, time.Now().Add(time.Hour), "更晚拒绝"); err != nil {
		t.Fatal(err)
	}
	stored, _ = repo.GetSubBatch(ctx, b.OperationID)
	if stored.Status != "done" || !stored.NextCheckAt.IsZero() {
		t.Fatalf("陈旧预检覆盖完成: %+v", stored)
	}
}

func TestSubBatchConcurrentPredecessorCompletionAndNewClaim(t *testing.T) {
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 20*time.Second)
	defer cancel()
	if _, err := repo.database().Exec(ctx, "UPDATE batch_detail SET item_id='A',batch_type=CASE WHEN batch_no='S-2' THEN 109 ELSE 110 END"); err != nil {
		t.Fatal(err)
	}
	older := testSub("S-1", "concurrent-old", "a", []int64{1}, []string{"A"})
	newer := testSub("S-2", "concurrent-new", "a", []int64{3}, []string{"A"})
	newer.BatchType = 109
	for _, b := range []service.SubBatch{older, newer} {
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	first, err := repo.ClaimSubBatch(ctx, older.OperationID, "old", time.Now())
	if err != nil || first == nil {
		t.Fatalf("旧批领取: %+v %v", first, err)
	}
	start := make(chan struct{})
	errs := make(chan error, 2)
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		<-start
		errs <- repo.CompleteSubBatch(ctx, *first, []service.Outcome{{Success: true}})
	}()
	go func() {
		defer wg.Done()
		<-start
		_, err := repo.ClaimSubBatch(ctx, newer.OperationID, "new", time.Now())
		errs <- err
	}()
	close(start)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatal(err)
		}
	}
	stored, _ := repo.GetSubBatch(ctx, newer.OperationID)
	if stored.LeaseToken == "" {
		if _, err := repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET next_check_at=NULL WHERE operation_id=?", newer.OperationID); err != nil {
			t.Fatal(err)
		}
		got, err := repo.ClaimSubBatch(ctx, newer.OperationID, "after", time.Now())
		if err != nil || got == nil {
			t.Fatal(fmt.Errorf("旧批完成后新批不可领取: %+v %v", got, err))
		}
	}
	row, err := repo.database().GetOne(ctx, "SELECT active_operation_id FROM data_operation_card_claim WHERE iccid='A'")
	if err != nil || row["active_operation_id"].String() != newer.OperationID {
		t.Fatalf("顺序屏障结果: %v %v", row, err)
	}
}

func TestSubBatchPrecheckBackoffMakes129thDatabaseTaskDue(t *testing.T) {
	repo, ctx := testSubBatchDB(t)
	base := time.Now().Add(-time.Minute)
	for i := 0; i < 129; i++ {
		id := int64(1000 + i)
		if _, err := repo.database().Exec(ctx, "INSERT INTO batch_detail(id,batch_no,detail_status)VALUES(?,'S-1',1)", id); err != nil {
			t.Fatal(err)
		}
		channel := "removed"
		if i == 128 {
			channel = "healthy"
		}
		b := testSub("S-1", fmt.Sprintf("db-fair-%03d", i), channel, []int64{id}, []string{fmt.Sprintf("fake-%03d", i)})
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
		if _, err := repo.database().Exec(ctx, "UPDATE data_operation_sub_batch SET next_check_at=? WHERE operation_id=?", base.Add(time.Duration(i)*time.Millisecond), b.OperationID); err != nil {
			t.Fatal(err)
		}
	}
	ids, err := repo.ListDueSubBatches(ctx, time.Now(), 128)
	if err != nil || len(ids) != 128 {
		t.Fatalf("首窗口: %d %v", len(ids), err)
	}
	for _, id := range ids {
		snapshot, err := repo.GetSubBatch(ctx, id)
		if err != nil {
			t.Fatal(err)
		}
		if snapshot.ChanKey != "removed" {
			t.Fatal("健康任务不应处于首128")
		}
		if err := repo.DelaySubBatchPrecheck(ctx, *snapshot, time.Now().Add(30*time.Second), "通道已移除"); err != nil {
			t.Fatal(err)
		}
	}
	ids, err = repo.ListDueSubBatches(ctx, time.Now(), 128)
	if err != nil || len(ids) != 1 || ids[0] != "db-fair-128" {
		t.Fatalf("健康第129仍被饿死: %v %v", ids, err)
	}
	if got, err := repo.ClaimSubBatch(ctx, ids[0], "healthy", time.Now()); err != nil || got == nil {
		t.Fatalf("健康任务无法领取: %+v %v", got, err)
	}
	n, err := repo.database().Model(subTable).Ctx(ctx).Where("chan_key", "removed").Where("status", "prepared").Where("poll_count", 0).WhereNull("submitted_at").WhereNull("lease_token").Count()
	if err != nil || n != 128 {
		t.Fatalf("预检失败意图被消耗: %d %v", n, err)
	}
	n, err = repo.database().Model(claimTable).Ctx(ctx).Where("active_operation_id != ''").Where("active_operation_id != ?", "db-fair-128").Count()
	if err != nil || n != 0 {
		t.Fatalf("预检退避错误占active: %d %v", n, err)
	}
}

func TestSubBatchMySQLCardWaitThenConcurrentReleaseAndReverseClaim(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证真实锁等待和两卡反向顺序")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 20*time.Second)
	defer cancel()
	if _, err := repo.database().Exec(ctx, "INSERT INTO batch_detail(id,batch_no,detail_status)VALUES(4,'S-2',1)"); err != nil {
		t.Fatal(err)
	}
	older := testSub("S-1", "wait-old-BA", "a", []int64{1, 2}, []string{"B", "A"})
	newer := testSub("S-2", "wait-new-AB", "a", []int64{3, 4}, []string{"A", "B"})
	for _, b := range []service.SubBatch{older, newer} {
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	claimed, err := repo.ClaimSubBatch(ctx, older.OperationID, "old-owner", time.Now())
	if err != nil || claimed == nil {
		t.Fatalf("旧批领取: %+v %v", claimed, err)
	}
	holder, err := repo.database().Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer holder.Rollback()
	if _, err := holder.GetAll("SELECT iccid FROM data_operation_card_claim WHERE tenant_id='tenant' AND iccid='A' FOR UPDATE"); err != nil {
		t.Fatal(err)
	}
	start := make(chan struct{})
	errs := make(chan error, 2)
	go func() {
		<-start
		errs <- repo.ReleaseUnsentSubBatch(ctx, *claimed, "确定未发送")
	}()
	go func() {
		<-start
		_, err := repo.ClaimSubBatch(ctx, newer.OperationID, "new-owner", time.Now())
		errs <- err
	}()
	close(start)
	// 实际观察InnoDB锁等待，而非只靠两goroutine恰好同时运行。
	waiting := false
	deadline := time.After(5 * time.Second)
	tick := time.NewTicker(10 * time.Millisecond)
	defer tick.Stop()
observe:
	for {
		select {
		case <-deadline:
			break observe
		case <-tick.C:
			value, err := repo.database().GetValue(ctx, "SELECT COUNT(*) FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='data_operation_card_claim'")
			if err != nil {
				t.Fatal(err)
			}
			if value.Int() >= 2 {
				waiting = true
				break observe
			}
		}
	}
	if err := holder.Commit(); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 2; i++ {
		if err := <-errs; err != nil {
			t.Fatal(err)
		}
	}
	if !waiting {
		t.Fatal("未实际观察到两条同卡交易的InnoDB锁等待")
	}
	stored, err := repo.GetSubBatch(ctx, older.OperationID)
	if err != nil || stored.Status != "prepared" || !stored.SubmittedAt.IsZero() {
		t.Fatalf("确定未发送回退状态: %+v %v", stored, err)
	}
	n, err := repo.database().Model(claimTable).Ctx(ctx).Where("active_operation_id != ''").Count()
	if err != nil || n != 0 {
		t.Fatalf("旧prepared未完成而新批抢先或残留部分active: %d %v", n, err)
	}
	t.Log("已观察两条事务等待同卡锁，释放后均提交成功，两卡输入BA/AB未形成反向环")
}

func TestSubBatchMySQLClaimDoesNotLockBusinessParent(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证前序JOIN不锁业务父批")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 10*time.Second)
	defer cancel()
	older := testSub("S-1", "main-lock-old", "a", []int64{1}, []string{"A"})
	newer := testSub("S-2", "main-lock-new", "a", []int64{3}, []string{"A"})
	for _, b := range []service.SubBatch{older, newer} {
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	holder, err := repo.database().Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer holder.Rollback()
	if _, err := holder.GetAll("SELECT id FROM batch_main WHERE batch_no='S-1' FOR UPDATE"); err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() {
		got, err := repo.ClaimSubBatch(ctx, newer.OperationID, "new-owner", time.Now())
		if err == nil && got != nil {
			err = fmt.Errorf("新批越过旧批")
		}
		done <- err
	}()
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-ctx.Done():
		t.Fatal("Claim持card等待业务main，存在main→card反向环")
	}
	t.Log("旧业务main保持FOR UPDATE期间，新批前序当前读完成且未越序")
}

func TestSubBatchMySQLCrossTenantSameCardsDoNotLockOtherSub(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证跨租户同ICCID双卡并发")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 15*time.Second)
	defer cancel()
	for _, q := range []string{
		"UPDATE batch_main SET tenant_id='other-tenant' WHERE batch_no='S-2'",
		"INSERT INTO batch_detail(id,batch_no,detail_status)VALUES(4,'S-2',1)",
	} {
		if _, err := repo.database().Exec(ctx, q); err != nil {
			t.Fatal(err)
		}
	}
	first := testSub("S-1", "tenant-one-BA", "a", []int64{1, 2}, []string{"B", "A"})
	second := testSub("S-2", "tenant-two-AB", "a", []int64{3, 4}, []string{"A", "B"})
	for _, b := range []service.SubBatch{first, second} {
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	start := make(chan struct{})
	errs := make(chan error, 2)
	for _, b := range []service.SubBatch{first, second} {
		go func(b service.SubBatch) {
			<-start
			got, err := repo.ClaimSubBatch(ctx, b.OperationID, b.OperationID, time.Now())
			if err == nil && got == nil {
				err = fmt.Errorf("跨租户意图不应互相阻断: %s", b.OperationID)
			}
			errs <- err
		}(b)
	}
	close(start)
	for i := 0; i < 2; i++ {
		if err := <-errs; err != nil {
			t.Fatal(err)
		}
	}
	n, err := repo.database().Model(claimTable).Ctx(ctx).Where("active_operation_id != ''").Count()
	if err != nil || n != 4 {
		t.Fatalf("跨tenant同ICCID active归属丢失: %d %v", n, err)
	}
}

func TestSubBatchMySQLCrossTenantForeignDetailLockDoesNotBlock(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证异租户相同ICCID不锁候选范围")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 15*time.Second)
	defer cancel()
	if _, err := repo.database().Exec(ctx, "UPDATE batch_main SET tenant_id='other-tenant' WHERE batch_no='S-2'"); err != nil {
		t.Fatal(err)
	}
	foreign := testSub("S-1", "foreign-A", "a", []int64{1}, []string{"A"})
	local := testSub("S-2", "local-A", "a", []int64{3}, []string{"A"})
	for _, b := range []service.SubBatch{foreign, local} {
		if err := repo.PrepareSubBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	holder, err := repo.database().Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer holder.Rollback()
	if _, err := holder.GetAll("SELECT detail_id FROM data_operation_sub_detail WHERE operation_id=? FOR UPDATE", foreign.OperationID); err != nil {
		t.Fatal(err)
	}
	result := make(chan error, 1)
	go func() {
		got, err := repo.ClaimSubBatch(ctx, local.OperationID, "other-tenant-owner", time.Now())
		if err == nil && got == nil {
			err = fmt.Errorf("异租户独立意图未领取")
		}
		result <- err
	}()
	blocked := false
	var claimErr error
	ticker := time.NewTicker(20 * time.Millisecond)
	defer ticker.Stop()
	wait := time.After(2 * time.Second)
loop:
	for {
		select {
		case claimErr = <-result:
			break loop
		case <-wait:
			blocked = true
			break loop
		case <-ticker.C:
			value, err := repo.database().GetValue(ctx, "SELECT COUNT(*) FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='data_operation_sub_detail'")
			if err != nil {
				t.Fatal(err)
			}
			if value.Int() > 0 {
				blocked = true
				break loop
			}
		}
	}
	if err := holder.Commit(); err != nil {
		t.Fatal(err)
	}
	if blocked {
		claimErr = <-result
	}
	if claimErr != nil {
		t.Fatal(claimErr)
	}
	if blocked {
		t.Fatal("异租户旧明细持锁无谓阻塞新租户同ICCID Claim")
	}
}

func TestSubBatchMySQLRepeatableReadStillSeesCommittedPredecessor(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证RR快照之后的前序当前读")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 15*time.Second)
	defer cancel()
	older := testSub("S-1", "rr-old-stop", "a", []int64{1}, []string{"A"})
	newer := testSub("S-2", "rr-new-open", "a", []int64{3}, []string{"A"})
	if err := repo.PrepareSubBatch(ctx, newer); err != nil {
		t.Fatal(err)
	}
	// 外层事务先建立无旧意图的RR快照，再由独立连接提交旧父批的冻结意图。
	tx, err := repo.database().Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback()
	if _, err := tx.GetAll("SELECT operation_id FROM data_operation_sub_batch"); err != nil {
		t.Fatal(err)
	}
	prepared := make(chan error, 1)
	go func() { prepared <- repo.PrepareSubBatch(base, older) }()
	if err := <-prepared; err != nil {
		t.Fatal(err)
	}
	rows, err := tx.GetAll("SELECT operation_id FROM data_operation_sub_batch WHERE operation_id=?", older.OperationID)
	if err != nil || len(rows) != 0 {
		t.Fatalf("未实际建立看不到新提交前序的RR快照: %v %v", rows, err)
	}
	if gdb.TXFromCtx(tx.GetCtx(), repo.database().GetGroup()) == nil {
		t.Fatal("外层RR事务未绑定上下文")
	}
	got, err := repo.ClaimSubBatch(tx.GetCtx(), newer.OperationID, "rr-owner", time.Now())
	if err != nil || got != nil {
		t.Fatalf("当前读遗漏已提交前序: %+v %v", got, err)
	}
	t.Log("普通RR快照仍看不到旧意图，Claim的锁定当前读已阻止新批越序")
}

func TestSubBatchMySQLDeadlockRetriesWholeTransaction(t *testing.T) {
	if os.Getenv("APN_WORKER_MYSQL_TEST") != "1" {
		t.Skip("仅隔离MySQL验证真实1213回滚重试")
	}
	repo, base := testSubBatchDB(t)
	ctx, cancel := context.WithTimeout(base, 15*time.Second)
	defer cancel()
	if _, err := repo.database().Exec(ctx, "INSERT INTO data_operation_card_claim(tenant_id,iccid,batch_id,operation_id)VALUES('tenant','dead-A',0,''),('tenant','dead-B',0,'')"); err != nil {
		t.Fatal(err)
	}
	locked := make(chan struct{}, 2)
	proceed := make(chan struct{})
	errs := make(chan error, 2)
	var total atomic.Int32
	for _, keys := range [][2]string{{"dead-A", "dead-B"}, {"dead-B", "dead-A"}} {
		go func(keys [2]string) {
			attempt := 0
			errs <- repo.durableTx(ctx, func(ctx context.Context, tx gdb.TX) error {
				attempt++
				total.Add(1)
				if _, err := tx.Exec("UPDATE data_operation_card_claim SET batch_id=batch_id+1 WHERE tenant_id='tenant' AND iccid=?", keys[0]); err != nil {
					return err
				}
				if attempt == 1 {
					locked <- struct{}{}
					select {
					case <-proceed:
					case <-ctx.Done():
						return ctx.Err()
					}
				}
				_, err := tx.Exec("UPDATE data_operation_card_claim SET batch_id=batch_id+1 WHERE tenant_id='tenant' AND iccid=?", keys[1])
				return err
			})
		}(keys)
	}
	for i := 0; i < 2; i++ {
		select {
		case <-locked:
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		}
	}
	close(proceed)
	for i := 0; i < 2; i++ {
		if err := <-errs; err != nil {
			t.Fatal(err)
		}
	}
	if total.Load() < 3 {
		t.Fatal("未触发真实死锁重试")
	}
	rows, err := repo.database().GetAll(ctx, "SELECT batch_id FROM data_operation_card_claim WHERE iccid IN ('dead-A','dead-B')")
	if err != nil || len(rows) != 2 {
		t.Fatalf("结果: %v %v", rows, err)
	}
	for _, row := range rows {
		if row["batch_id"].Int() != 2 {
			t.Fatalf("回滚后重复或遗漏更新: %v", rows)
		}
	}
	t.Logf("两事务成功提交，实际执行轮数=%d", total.Load())
}