package bindingowner

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
)

func TestResolveExactUsesRetainedGenerationForEveryBindingKindAgainstTiDB(t *testing.T) {
	db, cleanup := openBindingOwnerTestTiDB(t)
	defer cleanup()
	createBindingOwnerTestTables(t, db)

	retainedCoordinate, retainedSet, retainedDefinitions := completeDefinitionSetFixture(t)
	insertBindingOwnerTestAuthority(t, db, retainedCoordinate, retainedSet, "retained")

	newerCoordinate := PackGenerationCoordinate{
		StorageID:          "Opaque-PackGeneration-Row/Active-02",
		OwnerPack:          retainedCoordinate.OwnerPack,
		PackInstallationID: "pack-installation-2",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("b", 64),
	}
	newerSet, _ := newBindingOwnerTestDefinitionSet(t, newerCoordinate, 1, "newer")
	insertBindingOwnerTestAuthority(t, db, newerCoordinate, newerSet, "active")
	insertBindingOwnerTestActivation(t, db, retainedCoordinate, newerCoordinate)

	repository := newBindingOwnerTiDBRepository(t, db)
	for _, expected := range retainedDefinitions {
		expected := expected
		t.Run(string(expected.Kind), func(t *testing.T) {
			actual, err := repository.ResolveExact(
				t.Context(),
				retainedCoordinate,
				bindingFromIdentity(expected.Identity),
			)
			if err != nil {
				t.Fatalf("ResolveExact() retained generation: %v", err)
			}
			if !reflect.DeepEqual(actual, expected) {
				t.Fatalf("ResolveExact() = %#v, want retained %#v", actual, expected)
			}
		})
	}
}

func TestResolveExactDistinguishesEmptyAndMissingAuthorityAgainstTiDB(t *testing.T) {
	db, cleanup := openBindingOwnerTestTiDB(t)
	defer cleanup()
	createBindingOwnerTestTables(t, db)

	emptyCoordinate, _, emptyDefinitions := completeDefinitionSetFixture(t)
	emptySet, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          emptyCoordinate.OwnerPack,
			PackInstallationID: emptyCoordinate.PackInstallationID,
			PackPayloadDigest:  emptyCoordinate.PackPayloadDigest,
		},
		nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	insertBindingOwnerTestAuthority(t, db, emptyCoordinate, emptySet, "retained")

	missingCoordinate := PackGenerationCoordinate{
		StorageID:          "Opaque-PackGeneration-Row/Missing-03",
		OwnerPack:          "missing-authority-pack",
		PackInstallationID: "missing-authority-installation",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("c", 64),
	}
	_, missingDefinitions := newBindingOwnerTestDefinitionSet(t, missingCoordinate, 1, "missing")
	insertBindingOwnerTestParent(t, db, missingCoordinate, "retained")

	repository := newBindingOwnerTiDBRepository(t, db)
	t.Run("explicit empty", func(t *testing.T) {
		_, err := repository.ResolveExact(
			t.Context(),
			emptyCoordinate,
			bindingFromIdentity(emptyDefinitions[0].Identity),
		)
		assertBindingOwnerTiDBError(t, err, ErrAuthorityConflict, "absent from its complete definition set")
	})
	t.Run("missing header", func(t *testing.T) {
		_, err := repository.ResolveExact(
			t.Context(),
			missingCoordinate,
			bindingFromIdentity(missingDefinitions[0].Identity),
		)
		assertBindingOwnerTiDBError(t, err, ErrAuthorityConflict, "missing its definition-set header")
	})
}

func TestResolveExactFailsClosedOnStoredCorruptionAgainstTiDB(t *testing.T) {
	tests := []struct {
		name       string
		mutate     func(*testing.T, *sql.DB, PackGenerationCoordinate, connectorbindingownerv1.DefinitionSet)
		coordinate func(PackGenerationCoordinate) PackGenerationCoordinate
		want       error
		message    string
	}{
		{
			name: "malformed header protocol",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, _ connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinitionSet
SET Protocol = 'nexuscrm.connector-binding-owner-definition-set/v2'
WHERE PackGenerationId = ?`, coordinate.StorageID)
			},
			want:    ErrAuthorityConflict,
			message: "Protocol",
		},
		{
			name: "header count mismatch",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, _ connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinitionSet
SET DefinitionCount = DefinitionCount + 1
WHERE PackGenerationId = ?`, coordinate.StorageID)
			},
			want:    ErrAuthorityConflict,
			message: "stored entry count",
		},
		{
			name: "child envelope digest mismatch",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, set connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinition
SET DefinitionEnvelopeDigest = ?
WHERE BINARY PackGenerationId = BINARY ? AND DefinitionOrdinal = ?`,
					"sha256:"+strings.Repeat("f", 64),
					coordinate.StorageID,
					strconv.FormatUint(set.Entries[0].Ordinal, 10),
				)
			},
			want:    ErrAuthorityConflict,
			message: "EnvelopeDigest disagrees",
		},
		{
			name: "child exceeds envelope budget",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, set connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinition
SET DefinitionEnvelope = ?
WHERE BINARY PackGenerationId = BINARY ? AND DefinitionOrdinal = ?`,
					bytes.Repeat(
						[]byte("x"),
						connectorbindingownerv1.MaxSerializedEnvelopeBytes+1,
					),
					coordinate.StorageID,
					strconv.FormatUint(set.Entries[0].Ordinal, 10),
				)
			},
			want:    ErrAuthorityConflict,
			message: "closed budget",
		},
		{
			name: "stored parent coordinate mismatch",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, _ connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGeneration
SET AppliedPayloadHash = ?
WHERE Id = ?`, "nexus-pack-sha256-v1:"+strings.Repeat("d", 64), coordinate.StorageID)
			},
			want:    ErrAuthorityConflict,
			message: "immutable parent disagrees",
		},
		{
			name: "collation-equivalent child parent alias",
			mutate: func(t *testing.T, db *sql.DB, coordinate PackGenerationCoordinate, set connectorbindingownerv1.DefinitionSet) {
				execBindingOwnerTestSQL(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinition
SET PackGenerationId = ?
WHERE BINARY PackGenerationId = BINARY ? AND DefinitionOrdinal = ?`,
					strings.ToUpper(coordinate.StorageID),
					coordinate.StorageID,
					strconv.FormatUint(set.Entries[0].Ordinal, 10),
				)
			},
			want:    ErrAuthorityConflict,
			message: "byte-distinct generation",
		},
		{
			name: "unresolved opaque input storage coordinate",
			coordinate: func(coordinate PackGenerationCoordinate) PackGenerationCoordinate {
				coordinate.StorageID = "Different-Owner-Assigned-Generation-ID"
				return coordinate
			},
			want:    ErrAuthorityConflict,
			message: "missing its immutable parent",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, cleanup := openBindingOwnerTestTiDB(t)
			defer cleanup()
			createBindingOwnerTestTables(t, db)
			coordinate, set, definitions := completeDefinitionSetFixture(t)
			insertBindingOwnerTestAuthority(t, db, coordinate, set, "retained")
			if test.mutate != nil {
				test.mutate(t, db, coordinate, set)
			}
			requestedCoordinate := coordinate
			if test.coordinate != nil {
				requestedCoordinate = test.coordinate(coordinate)
			}

			repository := newBindingOwnerTiDBRepository(t, db)
			actual, err := repository.ResolveExact(
				t.Context(),
				requestedCoordinate,
				bindingFromIdentity(definitions[0].Identity),
			)
			assertBindingOwnerTiDBError(t, err, test.want, test.message)
			if !reflect.DeepEqual(actual, connectorbindingownerv1.DefinitionEnvelope{}) {
				t.Fatalf("ResolveExact() returned authority after corruption: %#v", actual)
			}
		})
	}
}

func TestResolveExactLocksParentHeaderAndWaitsForChildrenAgainstTiDB(t *testing.T) {
	db, cleanup := openBindingOwnerTestTiDB(t)
	defer cleanup()
	createBindingOwnerTestTables(t, db)
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	insertBindingOwnerTestAuthority(t, db, coordinate, set, "retained")

	holder, err := db.Conn(t.Context())
	if err != nil {
		t.Fatalf("acquire child-lock holder: %v", err)
	}
	defer holder.Close()
	if _, err := holder.ExecContext(t.Context(), "SET SESSION tidb_txn_mode = 'pessimistic'"); err != nil {
		t.Fatalf("enable pessimistic child-lock holder: %v", err)
	}
	holderTx, err := holder.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatalf("begin child-lock holder: %v", err)
	}
	defer holderTx.Rollback()
	last := set.Entries[len(set.Entries)-1]
	var lockedID string
	if err := holderTx.QueryRowContext(t.Context(), `SELECT Id
FROM _System_PackGenerationConnectorBindingDefinition
WHERE BINARY PackGenerationId = BINARY ? AND DefinitionOrdinal = ?
FOR UPDATE`, coordinate.StorageID, strconv.FormatUint(last.Ordinal, 10)).Scan(&lockedID); err != nil {
		t.Fatalf("lock last definition child: %v", err)
	}
	if lockedID == "" {
		t.Fatal("locked child has an empty owner-assigned identity")
	}

	readerContext, cancelReader := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelReader()
	type result struct {
		definition connectorbindingownerv1.DefinitionEnvelope
		err        error
	}
	resultChannel := make(chan result, 1)
	repository := newBindingOwnerTiDBRepository(t, db)
	go func() {
		definition, err := repository.ResolveExact(
			readerContext,
			coordinate,
			bindingFromIdentity(definitions[0].Identity),
		)
		resultChannel <- result{definition: definition, err: err}
	}()
	select {
	case early := <-resultChannel:
		t.Fatalf("ResolveExact() did not wait for locked final child: %v", early.err)
	case <-time.After(200 * time.Millisecond):
	}

	// Holding the final child proves ResolveExact does not bypass a locked child.
	// While that child read waits, the already-acquired parent and header locks
	// must keep independent mutations from slipping through the read transaction.
	waitForBindingOwnerReadLock(t, db, `UPDATE _System_PackGeneration
	SET Status = 'lock-probe' WHERE Id = ?`, coordinate.StorageID)
	waitForBindingOwnerReadLock(t, db, `UPDATE _System_PackGenerationConnectorBindingDefinitionSet
	SET Protocol = 'lock-probe' WHERE PackGenerationId = ?`, coordinate.StorageID)

	if err := holderTx.Commit(); err != nil {
		t.Fatalf("release final child lock: %v", err)
	}
	select {
	case resolved := <-resultChannel:
		if resolved.err != nil {
			t.Fatalf("ResolveExact() after lock release: %v", resolved.err)
		}
		if !reflect.DeepEqual(resolved.definition, definitions[0]) {
			t.Fatalf("ResolveExact() after lock release = %#v, want %#v", resolved.definition, definitions[0])
		}
	case <-time.After(5 * time.Second):
		t.Fatal("ResolveExact() remained blocked after final child lock release")
	}
}

func waitForBindingOwnerReadLock(
	t *testing.T,
	db *sql.DB,
	statement string,
	arguments ...any,
) {
	t.Helper()
	attemptContext, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
	defer cancel()
	_, err := db.ExecContext(attemptContext, statement, arguments...)
	if errors.Is(attemptContext.Err(), context.DeadlineExceeded) {
		return
	}
	if err != nil {
		t.Fatalf("probe binding-owner read lock: %v", err)
	}
	t.Fatal("concurrent mutation passed the binding-owner read lock")
}

func newBindingOwnerTiDBRepository(t *testing.T, db *sql.DB) *Repository {
	t.Helper()
	repository, err := NewRepository(db)
	if err != nil {
		t.Fatalf("NewRepository(): %v", err)
	}
	return repository
}

func newBindingOwnerTestDefinitionSet(
	t *testing.T,
	coordinate PackGenerationCoordinate,
	version uint64,
	marker string,
) (connectorbindingownerv1.DefinitionSet, []connectorbindingownerv1.DefinitionEnvelope) {
	t.Helper()
	_, _, template := completeDefinitionSetFixture(t)
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(template))
	for _, value := range template {
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
			connectorbindingownerv1.AuthoringInput{
				Identity: connectorbindingownerv1.Identity{
					Kind:               value.Kind,
					OwnerPack:          coordinate.OwnerPack,
					PackInstallationID: coordinate.PackInstallationID,
					PackPayloadDigest:  coordinate.PackPayloadDigest,
					BusinessKey:        value.BusinessKey,
					DefinitionVersion:  version,
				},
				CanonicalDefinition: []byte(`{"fixture":"` + marker + `"}`),
			},
		)
		if err != nil {
			t.Fatalf("NewDefinitionEnvelope(%s/%s): %v", value.Kind, value.BusinessKey, err)
		}
		definitions = append(definitions, definition)
	}
	set, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          coordinate.OwnerPack,
			PackInstallationID: coordinate.PackInstallationID,
			PackPayloadDigest:  coordinate.PackPayloadDigest,
		},
		definitions,
	)
	if err != nil {
		t.Fatalf("NewDefinitionSet(): %v", err)
	}
	validated, err := connectorbindingownerv1.ValidateDefinitionSet(set)
	if err != nil {
		t.Fatalf("ValidateDefinitionSet(): %v", err)
	}
	return set, validated
}

func insertBindingOwnerTestAuthority(
	t *testing.T,
	db *sql.DB,
	coordinate PackGenerationCoordinate,
	set connectorbindingownerv1.DefinitionSet,
	status string,
) {
	t.Helper()
	insertBindingOwnerTestParent(t, db, coordinate, status)
	now := time.Now().UTC()
	execBindingOwnerTestSQL(t, db, `INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
(PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?)`, coordinate.StorageID, set.Header.Protocol,
		strconv.FormatUint(set.Header.DefinitionCount, 10), set.Header.OrderedSetDigest, now, now)
	for _, entry := range set.Entries {
		execBindingOwnerTestSQL(t, db, `INSERT INTO _System_PackGenerationConnectorBindingDefinition
(Id, PackGenerationId, PackInstallId, DefinitionOrdinal, Kind, OwnerPack,
 PackPayloadDigest, BusinessKey, DefinitionVersion, DefinitionEnvelopeDigest,
 DefinitionEnvelope, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			fmt.Sprintf("Opaque-Definition-Row/%s/%d", coordinate.StorageID, entry.Ordinal),
			coordinate.StorageID,
			entry.PackInstallationID,
			strconv.FormatUint(entry.Ordinal, 10),
			string(entry.Kind),
			entry.OwnerPack,
			entry.PackPayloadDigest,
			entry.BusinessKey,
			strconv.FormatUint(entry.DefinitionVersion, 10),
			entry.EnvelopeDigest,
			entry.EnvelopeJSON,
			now,
			now,
		)
	}
}

func insertBindingOwnerTestParent(
	t *testing.T,
	db *sql.DB,
	coordinate PackGenerationCoordinate,
	status string,
) {
	t.Helper()
	execBindingOwnerTestSQL(t, db, `INSERT INTO _System_PackGeneration
(Id, PackName, PackInstallId, Status, DesiredPayloadHash, AppliedPayloadHash,
 RuntimeStartupShallowDeclarations, IsDeleted)
VALUES (?, ?, ?, ?, ?, ?, NULL, 0)`, coordinate.StorageID, coordinate.OwnerPack,
		coordinate.PackInstallationID, status, coordinate.PackPayloadDigest, coordinate.PackPayloadDigest)
}

func insertBindingOwnerTestActivation(
	t *testing.T,
	db *sql.DB,
	previous PackGenerationCoordinate,
	active PackGenerationCoordinate,
) {
	t.Helper()
	now := time.Now().UTC()
	execBindingOwnerTestSQL(t, db, `INSERT INTO _System_PackActivation
(Id, PackName, ActivePackInstallId, PreviousPackInstallId, ActivationEpoch,
 ActivatedDate, CreatedDate, LastModifiedDate, IsDeleted)
VALUES (?, ?, ?, ?, 2, ?, ?, ?, 0)`, "activation-fixture", active.OwnerPack,
		active.PackInstallationID, previous.PackInstallationID, now, now, now)
}

func createBindingOwnerTestTables(t *testing.T, db *sql.DB) {
	t.Helper()
	statements := []string{
		`CREATE TABLE _System_PackGeneration (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PackName VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PackInstallId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  Status VARCHAR(32) NOT NULL,
  DesiredPayloadHash VARCHAR(128) NOT NULL,
  AppliedPayloadHash VARCHAR(128) NOT NULL,
  RuntimeStartupShallowDeclarations JSON NULL,
  IsDeleted TINYINT(1) NOT NULL,
  PRIMARY KEY (Id),
  UNIQUE KEY uq_pack_generation_install (PackInstallId)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`,
		`CREATE TABLE _System_PackGenerationConnectorBindingDefinitionSet (
  PackGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  Protocol VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionCount DECIMAL(20,0) NOT NULL,
  DefinitionSetDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL,
  LastModifiedDate DATETIME(6) NOT NULL,
  PRIMARY KEY (PackGenerationId),
  CONSTRAINT fk_binding_owner_set_generation FOREIGN KEY (PackGenerationId)
    REFERENCES _System_PackGeneration (Id) ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`,
		`CREATE TABLE _System_PackGenerationConnectorBindingDefinition (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PackInstallId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionOrdinal DECIMAL(20,0) NOT NULL,
  Kind VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  OwnerPack VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackPayloadDigest CHAR(85) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  BusinessKey VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionVersion DECIMAL(20,0) NOT NULL,
  DefinitionEnvelopeDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionEnvelope MEDIUMBLOB NOT NULL,
  CreatedDate DATETIME(6) NOT NULL,
  LastModifiedDate DATETIME(6) NOT NULL,
  PRIMARY KEY (Id),
  UNIQUE KEY uq_binding_owner_ordinal (PackGenerationId, DefinitionOrdinal),
  UNIQUE KEY uq_binding_owner_identity (PackGenerationId, Kind, BusinessKey),
  CONSTRAINT fk_binding_owner_definition_set FOREIGN KEY (PackGenerationId)
    REFERENCES _System_PackGenerationConnectorBindingDefinitionSet (PackGenerationId)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`,
		`CREATE TABLE _System_PackActivation (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PackName VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  ActivePackInstallId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PreviousPackInstallId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  ActivationEpoch DECIMAL(20,0) NOT NULL,
  ActivatedDate DATETIME(6) NOT NULL,
  CreatedDate DATETIME(6) NOT NULL,
  LastModifiedDate DATETIME(6) NOT NULL,
  IsDeleted TINYINT(1) NOT NULL,
  PRIMARY KEY (Id),
  UNIQUE KEY uq_pack_activation_pack (PackName)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`,
	}
	for _, statement := range statements {
		execBindingOwnerTestSQL(t, db, statement)
	}
}

func execBindingOwnerTestSQL(t *testing.T, db *sql.DB, statement string, arguments ...any) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), statement, arguments...); err != nil {
		t.Fatalf("execute binding-owner TiDB fixture SQL: %v", err)
	}
}

func assertBindingOwnerTiDBError(t *testing.T, err, want error, message string) {
	t.Helper()
	if !errors.Is(err, want) || !strings.Contains(err.Error(), message) {
		t.Fatalf("ResolveExact() error = %v, want %v containing %q", err, want, message)
	}
}

func openBindingOwnerTestTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := bindingOwnerTestEnvironmentOr("TIDB_HOST", "127.0.0.1")
	port := bindingOwnerTestEnvironmentOr("TIDB_PORT", "4001")
	user := bindingOwnerTestEnvironmentOr("TIDB_USER", "root")
	password := os.Getenv("TIDB_PASSWORD")
	tlsParameter := ""
	switch strings.ToLower(strings.TrimSpace(os.Getenv("TIDB_TLS"))) {
	case "1", "true", "yes", "require", "verify-full":
		tlsParameter = "&tls=true"
	}
	admin, err := sql.Open("mysql", fmt.Sprintf(
		"%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC&timeout=2s%s",
		user, password, host, port, tlsParameter,
	))
	if err != nil {
		unavailableBindingOwnerTestTiDB(t, "open admin connection: %v", err)
	}
	if err := admin.PingContext(context.Background()); err != nil {
		_ = admin.Close()
		unavailableBindingOwnerTestTiDB(t, "TiDB not reachable at %s:%s: %v", host, port, err)
	}

	databaseName := fmt.Sprintf("nexuscrm_connector_binding_owner_%d", time.Now().UnixNano())
	if _, err := admin.ExecContext(t.Context(), "CREATE DATABASE `"+databaseName+
		"` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); err != nil {
		_ = admin.Close()
		unavailableBindingOwnerTestTiDB(t, "create temporary database: %v", err)
	}
	db, err := sql.Open("mysql", fmt.Sprintf(
		"%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC&timeout=2s%s",
		user, password, host, port, databaseName, tlsParameter,
	))
	if err != nil {
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
		t.Fatalf("open binding-owner test database: %v", err)
	}
	db.SetMaxOpenConns(16)
	if err := db.PingContext(t.Context()); err != nil {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
		t.Fatalf("ping binding-owner test database: %v", err)
	}
	cleanup := func() {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
	}
	return db, cleanup
}

func unavailableBindingOwnerTestTiDB(t *testing.T, format string, arguments ...any) {
	t.Helper()
	if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
		t.Fatalf("required TiDB integration unavailable: "+format, arguments...)
	}
	t.Skipf("skipping TiDB integration: "+format, arguments...)
}

func bindingOwnerTestEnvironmentOr(name, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(name)); value != "" {
		return value
	}
	return fallback
}