package persistence

import (
	"context"
	"github.com/nexuscrm/shared/pkg/constants"
	"testing"
	"time"
)

func TestRoleLifecycleLockFencesDeleteOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createCanonicalMetadataTables(t, db, constants.TableUserRole, constants.TableUser, constants.TableGroup, constants.TableGroupMember, constants.TableGroupMembershipClosure)
	metadataReadExec(t, db, `INSERT INTO UserRole (Id,Name,IsDeleted) VALUES ('role-target','Target',0)`)
	repo := NewPermissionRepository(db)
	if _, err := repo.LockRoleLifecycle(t.Context(), "role-target"); err == nil {
		t.Fatal("lock accepted without owning transaction")
	}
	tx, closeTx, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer closeTx()
	defer tx.Rollback()
	ctx := context.WithValue(t.Context(), txContextKey{}, tx)
	role, err := repo.LockRoleLifecycle(ctx, "ROLE-TARGET")
	if err != nil || role == nil {
		t.Fatalf("role=%v err=%v", role, err)
	}
	deleteCtx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
	defer cancel()
	deleted := make(chan error, 1)
	go func() { deleted <- repo.DeleteRole(deleteCtx, "role-target") }()
	// Observe the competing transaction in TiDB, rather than assuming a sleeping
	// goroutine reached the delete. The lifecycle row lock must be its blocker.
	deadline := time.Now().Add(5 * time.Second)
	for {
		select {
		case err := <-deleted:
			t.Fatalf("delete escaped lifecycle lock: %v", err)
		default:
		}
		var waiting int
		if err := db.QueryRowContext(deleteCtx, `SELECT COUNT(*) FROM information_schema.DATA_LOCK_WAITS`).Scan(&waiting); err != nil {
			t.Fatal(err)
		}
		if waiting > 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("delete did not reach TiDB lock wait")
		}
		time.Sleep(20 * time.Millisecond)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	if err := <-deleted; err != nil {
		t.Fatalf("delete after release: %v", err)
	}
	tx2, closeTx2, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer closeTx2()
	defer tx2.Rollback()
	role, err = repo.LockRoleLifecycle(context.WithValue(t.Context(), txContextKey{}, tx2), "role-target")
	if err != nil || role != nil {
		t.Fatalf("deleted role accepted: role=%v err=%v", role, err)
	}
}