package bindingowner

import (
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
)

func TestNewRepositoryRequiresDatabase(t *testing.T) {
	repository, err := NewRepository(nil)
	if repository != nil || !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("NewRepository(nil) = (%v, %v), want nil ErrDependencyMissing", repository, err)
	}
}

func TestResolveExactBindingsReadsCompleteSetOnceInCallerOrderAndResolveExactDelegates(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	repository, db, mock := bindingOwnerTestRepository(t)
	requestedIndexes := []int{3, 1, 0, 2}
	bindings := make([]catalogv1.IntegrationBinding, len(requestedIndexes))
	for index, definitionIndex := range requestedIndexes {
		bindings[index] = bindingFromIdentity(definitions[definitionIndex].Identity)
	}

	// One requested batch must lock and validate the complete definition set
	// exactly once, independent of the number or order of requested bindings.
	expectDefinitionSetTransaction(mock, coordinate, set)
	expectBindingOwnerCommit(mock)
	actual, err := repository.ResolveExactBindings(t.Context(), coordinate, bindings)
	if err != nil {
		t.Fatalf("ResolveExactBindings(): %v", err)
	}
	if len(actual) != len(bindings) {
		t.Fatalf("ResolveExactBindings() returned %d definitions, want %d", len(actual), len(bindings))
	}
	for index, definitionIndex := range requestedIndexes {
		if !reflect.DeepEqual(actual[index], definitions[definitionIndex]) {
			t.Fatalf(
				"ResolveExactBindings()[%d] = %#v, want caller-order definition %#v",
				index,
				actual[index],
				definitions[definitionIndex],
			)
		}
	}

	// Returned envelopes are detached from each other and from retained owner
	// facts. ResolveExact must delegate to the same singleton selection path and
	// recover the exact value after callers mutate a prior batch result.
	actual[0].CanonicalDefinition[0] ^= 0xff
	actual[0].OwnerReceipt[0] ^= 0xff
	if !reflect.DeepEqual(actual[1], definitions[requestedIndexes[1]]) {
		t.Fatal("mutating one returned envelope changed another batch position")
	}
	expectDefinitionSetTransaction(mock, coordinate, set)
	expectBindingOwnerCommit(mock)
	single, err := repository.ResolveExact(t.Context(), coordinate, bindings[0])
	if err != nil {
		t.Fatalf("ResolveExact(): %v", err)
	}
	if !reflect.DeepEqual(single, definitions[requestedIndexes[0]]) {
		t.Fatalf("ResolveExact() = %#v, want shared-path value %#v", single, definitions[requestedIndexes[0]])
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestResolveExactBindingSelectionsLocksGenerationsInByteOrderAndReturnsCallerOrder(t *testing.T) {
	coordinateA, setA, definitionsA := completeDefinitionSetFixture(t)
	coordinateA.StorageID = "Generation-A"
	coordinateB, setB, definitionsB := definitionSetFixtureForCoordinate(t, PackGenerationCoordinate{
		StorageID:          "generation-b",
		OwnerPack:          "neutral-connector-b",
		PackInstallationID: "pack-installation-b",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("b", 64),
	}, "generation-b")

	bindingsA := bindingsAtIndexes(definitionsA, 3, 1)
	bindingsB := bindingsAtIndexes(definitionsB, 2, 0)
	selectionA := mustExactBindingSelection(t, coordinateA, bindingsA)
	selectionB := mustExactBindingSelection(t, coordinateB, bindingsB)

	repository, db, mock := bindingOwnerTestRepository(t)
	expectBindingOwnerBegin(mock)
	// The caller deliberately requests the lowercase generation first. Exact
	// storage-ID byte order locks the uppercase generation first, while neither
	// the outer selection order nor either inner binding order is changed.
	expectDefinitionSetReads(mock, coordinateA, setA)
	expectDefinitionSetReads(mock, coordinateB, setB)
	expectBindingOwnerCommit(mock)

	actual, err := repository.ResolveExactBindingSelections(
		t.Context(),
		[]ExactBindingSelection{selectionB, selectionA},
	)
	if err != nil {
		t.Fatalf("ResolveExactBindingSelections(): %v", err)
	}
	want := [][]connectorbindingownerv1.DefinitionEnvelope{
		{definitionsB[2], definitionsB[0]},
		{definitionsA[3], definitionsA[1]},
	}
	if !reflect.DeepEqual(actual, want) {
		t.Fatalf("ResolveExactBindingSelections() = %#v, want caller-order %#v", actual, want)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestExactBindingSelectionAndMultiGenerationResultsAreDetached(t *testing.T) {
	coordinate, _, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	binding.CredentialRefs = []string{"credential-a"}
	binding.RuntimeCommand = &catalogv1.RuntimeCommandBinding{
		CommandKind:          "runtime.flow.execute",
		PayloadFormatVersion: 1,
		DefinitionRef:        "flow-a",
		DefinitionVersion:    "1",
	}
	source := []catalogv1.IntegrationBinding{binding}
	selection := mustExactBindingSelection(t, coordinate, source)

	source[0].BusinessKey = "mutated-after-construction"
	source[0].CredentialRefs[0] = "mutated-credential"
	source[0].RuntimeCommand.DefinitionRef = "mutated-flow"
	coordinate.StorageID = "mutated-storage-id"
	if selection.bindings[0].BusinessKey != definitions[0].BusinessKey ||
		selection.bindings[0].CredentialRefs[0] != "credential-a" ||
		selection.bindings[0].RuntimeCommand.DefinitionRef != "flow-a" {
		t.Fatal("NewExactBindingSelection retained caller-owned binding storage")
	}

	storedCoordinate, storedSet, _ := completeDefinitionSetFixture(t)
	repository, db, mock := bindingOwnerTestRepository(t)
	for attempt := 0; attempt < 2; attempt++ {
		expectDefinitionSetTransaction(mock, storedCoordinate, storedSet)
		expectBindingOwnerCommit(mock)
	}
	first, err := repository.ResolveExactBindingSelections(
		t.Context(),
		[]ExactBindingSelection{selection},
	)
	if err != nil || len(first) != 1 || len(first[0]) != 1 ||
		!reflect.DeepEqual(first[0][0], definitions[0]) {
		t.Fatalf("first ResolveExactBindingSelections() = %#v, %v", first, err)
	}
	first[0][0].CanonicalDefinition[0] ^= 0xff
	first[0][0].OwnerReceipt[0] ^= 0xff

	second, err := repository.ResolveExactBindingSelections(
		t.Context(),
		[]ExactBindingSelection{selection},
	)
	if err != nil || len(second) != 1 || len(second[0]) != 1 ||
		!reflect.DeepEqual(second[0][0], definitions[0]) {
		t.Fatalf("second ResolveExactBindingSelections() = %#v, %v", second, err)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestResolveExactBindingSelectionsPreflightsWholeRequestBeforeSQL(t *testing.T) {
	coordinate, _, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	selection := mustExactBindingSelection(
		t,
		coordinate,
		[]catalogv1.IntegrationBinding{binding},
	)

	t.Run("nil", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(t.Context(), nil)
		if !errors.Is(err, ErrInvalidRequest) || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(nil) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("explicit empty", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(
			t.Context(),
			[]ExactBindingSelection{},
		)
		if err != nil || actual == nil || len(actual) != 0 {
			t.Fatalf("ResolveExactBindingSelections(empty) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("explicit empty generation selection", func(t *testing.T) {
		empty := mustExactBindingSelection(t, coordinate, []catalogv1.IntegrationBinding{})
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(
			t.Context(),
			[]ExactBindingSelection{empty},
		)
		if err != nil || len(actual) != 1 || actual[0] == nil || len(actual[0]) != 0 {
			t.Fatalf("ResolveExactBindingSelections(empty generation) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("selection count", func(t *testing.T) {
		selections := make([]ExactBindingSelection, connectorbindingownerv1.MaxDefinitionCount+1)
		for index := range selections {
			candidateCoordinate := exactSelectionBudgetCoordinate(index)
			selections[index] = mustExactBindingSelection(
				t,
				candidateCoordinate,
				[]catalogv1.IntegrationBinding{},
			)
		}
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(t.Context(), selections)
		if !errors.Is(err, ErrInvalidRequest) || !strings.Contains(err.Error(), "selection count") || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(over-count) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("total binding count", func(t *testing.T) {
		firstCoordinate := exactSelectionBudgetCoordinate(0)
		secondCoordinate := exactSelectionBudgetCoordinate(1)
		firstCount := catalogv1.MaxBindingCount/2 + 1
		secondCount := catalogv1.MaxBindingCount - firstCount + 1
		first := mustExactBindingSelection(
			t,
			firstCoordinate,
			exactSelectionBudgetBindings(firstCoordinate, "first", firstCount),
		)
		second := mustExactBindingSelection(
			t,
			secondCoordinate,
			exactSelectionBudgetBindings(secondCoordinate, "second", secondCount),
		)
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(
			t.Context(),
			[]ExactBindingSelection{first, second},
		)
		if !errors.Is(err, ErrInvalidRequest) || !strings.Contains(err.Error(), "total bindings") || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(over-total) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("duplicate exact coordinate", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(
			t.Context(),
			[]ExactBindingSelection{selection, selection},
		)
		if !errors.Is(err, ErrInvalidRequest) || !strings.Contains(err.Error(), "repeat PackGeneration storage ID") || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(duplicate) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("same portable coordinate names different storage identities", func(t *testing.T) {
		aliasCoordinate := coordinate
		aliasCoordinate.StorageID = coordinate.StorageID + "-alias"
		alias := mustExactBindingSelection(
			t,
			aliasCoordinate,
			[]catalogv1.IntegrationBinding{binding},
		)
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindingSelections(
			t.Context(),
			[]ExactBindingSelection{selection, alias},
		)
		if !errors.Is(err, ErrInvalidRequest) || !strings.Contains(err.Error(), "one portable PackGeneration coordinate") || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(inverse disagreement) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactBindingSelectionsEnforcesAggregateDefinitionBudgetAtExactBoundary(t *testing.T) {
	coordinateA, setA, definitionsA := completeDefinitionSetFixture(t)
	coordinateA.StorageID = "Generation-A"
	coordinateB, setB, definitionsB := definitionSetFixtureForCoordinate(t, PackGenerationCoordinate{
		StorageID:          "generation-b",
		OwnerPack:          "neutral-connector-b",
		PackInstallationID: "pack-installation-b",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("b", 64),
	}, "generation-b")
	selectionA := mustExactBindingSelection(t, coordinateA, bindingsAtIndexes(definitionsA, 0))
	selectionB := mustExactBindingSelection(t, coordinateB, bindingsAtIndexes(definitionsB, 0))
	selections := []ExactBindingSelection{selectionB, selectionA}
	budget := definitionSetEnvelopeBytes(setA) + definitionSetEnvelopeBytes(setB)

	t.Run("exact", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		expectBindingOwnerBegin(mock)
		expectDefinitionSetReads(mock, coordinateA, setA)
		expectDefinitionSetReads(mock, coordinateB, setB)
		expectBindingOwnerCommit(mock)

		actual, err := repository.resolveExactBindingSelections(t.Context(), selections, budget)
		if err != nil || len(actual) != 2 || len(actual[0]) != 1 || len(actual[1]) != 1 {
			t.Fatalf("resolveExactBindingSelections(exact budget) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("one byte over", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		expectBindingOwnerBegin(mock)
		expectDefinitionSetReads(mock, coordinateA, setA)
		expectDefinitionSetReads(mock, coordinateB, setB)
		expectBindingOwnerRollback(mock)

		actual, err := repository.resolveExactBindingSelections(t.Context(), selections, budget-1)
		if !errors.Is(err, ErrExactReadBudgetExceeded) || actual != nil {
			t.Fatalf("resolveExactBindingSelections(over budget) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactBindingSelectionsReturnsNoPartialOutputOnLaterFailureOrCommitFailure(t *testing.T) {
	coordinateA, setA, definitionsA := completeDefinitionSetFixture(t)
	coordinateA.StorageID = "Generation-A"
	coordinateB, setB, definitionsB := definitionSetFixtureForCoordinate(t, PackGenerationCoordinate{
		StorageID:          "generation-b",
		OwnerPack:          "neutral-connector-b",
		PackInstallationID: "pack-installation-b",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("b", 64),
	}, "generation-b")
	selectionA := mustExactBindingSelection(t, coordinateA, bindingsAtIndexes(definitionsA, 0))
	selectionB := mustExactBindingSelection(t, coordinateB, bindingsAtIndexes(definitionsB, 0))
	selections := []ExactBindingSelection{selectionB, selectionA}

	t.Run("later generation read rolls back", func(t *testing.T) {
		databaseFailure := errors.New("second generation unavailable")
		repository, db, mock := bindingOwnerTestRepository(t)
		expectBindingOwnerBegin(mock)
		expectDefinitionSetReads(mock, coordinateA, setA)
		expectPackGenerationParent(mock, coordinateB, nil)
		mock.ExpectQuery(bindingOwnerHeaderQueryPattern()).
			WithArgs(coordinateB.StorageID, coordinateB.StorageID).
			WillReturnError(databaseFailure)
		expectBindingOwnerRollback(mock)

		actual, err := repository.ResolveExactBindingSelections(t.Context(), selections)
		if !errors.Is(err, databaseFailure) || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(later failure) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("commit failure discards complete result", func(t *testing.T) {
		commitFailure := errors.New("multi-generation commit failed")
		repository, db, mock := bindingOwnerTestRepository(t)
		expectBindingOwnerBegin(mock)
		expectDefinitionSetReads(mock, coordinateA, setA)
		expectDefinitionSetReads(mock, coordinateB, setB)
		mock.ExpectCommit().WillReturnError(commitFailure)
		expectBindingOwnerCleanup(mock)

		actual, err := repository.ResolveExactBindingSelections(t.Context(), selections)
		if !errors.Is(err, commitFailure) || !strings.Contains(err.Error(), "commit exact") || actual != nil {
			t.Fatalf("ResolveExactBindingSelections(commit failure) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactBindingsRejectsMissingOrDuplicateRequestedIdentity(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	first := bindingFromIdentity(definitions[0].Identity)
	second := bindingFromIdentity(definitions[1].Identity)

	t.Run("missing", func(t *testing.T) {
		missing := first
		missing.BusinessKey = "binding-that-is-not-published"
		repository, db, mock := bindingOwnerTestRepository(t)
		expectDefinitionSetTransaction(mock, coordinate, set)
		expectBindingOwnerRollback(mock)

		actual, err := repository.ResolveExactBindings(
			t.Context(),
			coordinate,
			[]catalogv1.IntegrationBinding{second, missing, first},
		)
		if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "absent") {
			t.Fatalf("ResolveExactBindings() error = %v, want missing authority conflict", err)
		}
		if actual != nil {
			t.Fatalf("ResolveExactBindings() returned partial definitions: %#v", actual)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	for _, test := range []struct {
		name      string
		duplicate catalogv1.IntegrationBinding
	}{
		{
			name: "same exact identity with different non-owner facts",
			duplicate: func() catalogv1.IntegrationBinding {
				value := first
				value.CredentialRefs = []string{"credential-owned-by-catalog"}
				return value
			}(),
		},
		{
			name: "same natural identity with disagreeing version",
			duplicate: func() catalogv1.IntegrationBinding {
				value := first
				value.DefinitionVersion++
				return value
			}(),
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			repository, db, mock := bindingOwnerTestRepository(t)
			actual, err := repository.ResolveExactBindings(
				t.Context(),
				coordinate,
				[]catalogv1.IntegrationBinding{first, test.duplicate},
			)
			if !errors.Is(err, ErrInvalidRequest) {
				t.Fatalf("ResolveExactBindings() error = %v, want duplicate invalid request", err)
			}
			if actual != nil {
				t.Fatalf("ResolveExactBindings() returned definitions for duplicate request: %#v", actual)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactBindingsRejectsInvalidSelectionBeforeSQL(t *testing.T) {
	coordinate, _, definitions := completeDefinitionSetFixture(t)
	valid := bindingFromIdentity(definitions[0].Identity)
	tooMany := make([]catalogv1.IntegrationBinding, connectorbindingownerv1.MaxDefinitionCount+1)
	for index := range tooMany {
		tooMany[index] = valid
	}
	differentGeneration := valid
	differentGeneration.OwnerPack = "byte-distinct-owner"

	for _, test := range []struct {
		name     string
		bindings []catalogv1.IntegrationBinding
		want     error
	}{
		{name: "nil", bindings: nil, want: ErrInvalidRequest},
		{name: "count", bindings: tooMany, want: ErrInvalidRequest},
		{name: "coordinate disagreement", bindings: []catalogv1.IntegrationBinding{valid, differentGeneration}, want: ErrAuthorityConflict},
	} {
		t.Run(test.name, func(t *testing.T) {
			repository, db, mock := bindingOwnerTestRepository(t)
			actual, err := repository.ResolveExactBindings(t.Context(), coordinate, test.bindings)
			if !errors.Is(err, test.want) {
				t.Fatalf("ResolveExactBindings() error = %v, want %v", err, test.want)
			}
			if actual != nil {
				t.Fatalf("ResolveExactBindings() returned definitions for invalid input: %#v", actual)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactBindingsAcceptsExplicitEmptySelectionWithoutSQL(t *testing.T) {
	coordinate, _, _ := completeDefinitionSetFixture(t)
	t.Run("valid coordinate", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindings(
			t.Context(),
			coordinate,
			[]catalogv1.IntegrationBinding{},
		)
		if err != nil {
			t.Fatalf("ResolveExactBindings(empty) error = %v", err)
		}
		if actual == nil || len(actual) != 0 {
			t.Fatalf("ResolveExactBindings(empty) = %#v, want non-nil empty selection", actual)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("coordinate still validated", func(t *testing.T) {
		coordinate.StorageID = ""
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindings(
			t.Context(),
			coordinate,
			[]catalogv1.IntegrationBinding{},
		)
		if !errors.Is(err, ErrInvalidRequest) || actual != nil {
			t.Fatalf("ResolveExactBindings(empty, malformed coordinate) = %#v, %v", actual, err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactBindingsTreatsStorageIDAsBoundedOpaqueBytes(t *testing.T) {
	baseCoordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)

	for _, test := range []struct {
		name      string
		storageID string
	}{
		{name: "exact byte limit", storageID: strings.Repeat("x", connectorbindingownerv1.MaxIdentifierBytes)},
		{name: "uninterpreted contents", storageID: " target-local/opaque:\x00\n "},
	} {
		t.Run(test.name, func(t *testing.T) {
			coordinate := baseCoordinate
			coordinate.StorageID = test.storageID
			repository, db, mock := bindingOwnerTestRepository(t)
			expectDefinitionSetTransaction(mock, coordinate, set)
			expectBindingOwnerCommit(mock)

			actual, err := repository.ResolveExactBindings(
				t.Context(),
				coordinate,
				[]catalogv1.IntegrationBinding{binding},
			)
			if err != nil || len(actual) != 1 || !reflect.DeepEqual(actual[0], definitions[0]) {
				t.Fatalf("ResolveExactBindings(storage ID %q) = %#v, %v", test.storageID, actual, err)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}

	t.Run("byte bound", func(t *testing.T) {
		coordinate := baseCoordinate
		coordinate.StorageID = strings.Repeat("x", connectorbindingownerv1.MaxIdentifierBytes+1)
		repository, db, mock := bindingOwnerTestRepository(t)
		actual, err := repository.ResolveExactBindings(
			t.Context(),
			coordinate,
			[]catalogv1.IntegrationBinding{binding},
		)
		if !errors.Is(err, ErrInvalidRequest) {
			t.Fatalf("ResolveExactBindings(oversized storage ID) error = %v, want ErrInvalidRequest", err)
		}
		if actual != nil {
			t.Fatalf("ResolveExactBindings(oversized storage ID) returned %#v", actual)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactBindingsValidatesWholeSetBeforeSelectingTargets(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	// Every requested target precedes the corrupt last row. The batch must still
	// validate the complete owner set rather than returning already-seen facts.
	set.Entries[len(set.Entries)-1].EnvelopeDigest = "sha256:" + strings.Repeat("0", 64)
	repository, db, mock := bindingOwnerTestRepository(t)
	expectDefinitionSetTransaction(mock, coordinate, set)
	expectBindingOwnerRollback(mock)

	actual, err := repository.ResolveExactBindings(
		t.Context(),
		coordinate,
		[]catalogv1.IntegrationBinding{
			bindingFromIdentity(definitions[0].Identity),
			bindingFromIdentity(definitions[1].Identity),
		},
	)
	if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "malformed") {
		t.Fatalf("ResolveExactBindings() error = %v, want whole-set authority conflict", err)
	}
	if actual != nil {
		t.Fatalf("ResolveExactBindings() returned definitions on corrupt later row: %#v", actual)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestResolveExactBindingsRejectsStoredAggregateEnvelopeBudget(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	const rowCount = int(connectorbindingownerv1.MaxDefinitionSetBytes/
		connectorbindingownerv1.MaxSerializedEnvelopeBytes) + 1
	set.Header.DefinitionCount = uint64(rowCount)

	repository, db, mock := bindingOwnerTestRepository(t)
	expectBindingOwnerBegin(mock)
	expectPackGenerationParent(mock, coordinate, nil)
	expectDefinitionSetHeader(mock, coordinate, set, nil)
	rows := bindingOwnerChildRows()
	maxEnvelope := make([]byte, connectorbindingownerv1.MaxSerializedEnvelopeBytes)
	for index := 0; index < rowCount; index++ {
		entry := set.Entries[0]
		entry.Ordinal = uint64(index)
		entry.BusinessKey = "aggregate-budget"
		entry.EnvelopeJSON = maxEnvelope
		addDefinitionChildRow(rows, coordinate, entry, nil)
	}
	mock.ExpectQuery(bindingOwnerChildrenQueryPattern()).
		WithArgs(coordinate.StorageID).
		WillReturnRows(rows)
	expectBindingOwnerRollback(mock)

	actual, err := repository.ResolveExactBindings(
		t.Context(),
		coordinate,
		[]catalogv1.IntegrationBinding{bindingFromIdentity(definitions[0].Identity)},
	)
	if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "closed budget") {
		t.Fatalf("ResolveExactBindings() error = %v, want aggregate budget authority conflict", err)
	}
	if actual != nil {
		t.Fatalf("ResolveExactBindings() returned definitions after aggregate budget failure: %#v", actual)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestResolveExactDistinguishesExplicitEmptySetFromMissingHeader(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	emptySet, err := connectorbindingownerv1.NewDefinitionSet(set.Header.DefinitionSetCoordinate, nil)
	if err != nil {
		t.Fatal(err)
	}

	t.Run("explicit empty", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		expectDefinitionSetTransaction(mock, coordinate, emptySet)
		expectBindingOwnerRollback(mock)

		_, err := repository.ResolveExact(t.Context(), coordinate, binding)
		if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "absent") {
			t.Fatalf("ResolveExact() explicit-empty error = %v, want absent authority conflict", err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("missing header", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		expectBindingOwnerBegin(mock)
		expectPackGenerationParent(mock, coordinate, nil)
		mock.ExpectQuery(bindingOwnerHeaderQueryPattern()).
			WithArgs(coordinate.StorageID, coordinate.StorageID).
			WillReturnRows(bindingOwnerHeaderRows())
		expectBindingOwnerRollback(mock)

		_, err := repository.ResolveExact(t.Context(), coordinate, binding)
		if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "missing its definition-set header") {
			t.Fatalf("ResolveExact() missing-header error = %v", err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactRejectsIndependentCoordinateDisagreementBeforeSQL(t *testing.T) {
	coordinate, _, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	tests := []struct {
		name       string
		coordinate PackGenerationCoordinate
		binding    catalogv1.IntegrationBinding
		want       error
	}{
		{
			name: "missing storage identity",
			coordinate: func() PackGenerationCoordinate {
				value := coordinate
				value.StorageID = ""
				return value
			}(),
			binding: binding,
			want:    ErrInvalidRequest,
		},
		{
			name:       "binding names another owner",
			coordinate: coordinate,
			binding: func() catalogv1.IntegrationBinding {
				value := binding
				value.OwnerPack = "other-owner"
				return value
			}(),
			want: ErrAuthorityConflict,
		},
		{
			name:       "binding names another installation",
			coordinate: coordinate,
			binding: func() catalogv1.IntegrationBinding {
				value := binding
				value.PackInstallationID = "other-installation"
				return value
			}(),
			want: ErrAuthorityConflict,
		},
		{
			name:       "binding names another payload",
			coordinate: coordinate,
			binding: func() catalogv1.IntegrationBinding {
				value := binding
				value.PackPayloadDigest = "nexus-pack-sha256-v1:" + strings.Repeat("b", 64)
				return value
			}(),
			want: ErrAuthorityConflict,
		},
		{
			name:       "malformed binding",
			coordinate: coordinate,
			binding: func() catalogv1.IntegrationBinding {
				value := binding
				value.DefinitionVersion = 0
				return value
			}(),
			want: ErrInvalidRequest,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			repository, db, mock := bindingOwnerTestRepository(t)
			_, err := repository.ResolveExact(t.Context(), test.coordinate, test.binding)
			if !errors.Is(err, test.want) {
				t.Fatalf("ResolveExact() error = %v, want %v", err, test.want)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactRejectsDefinitionVersionDisagreement(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	binding.DefinitionVersion++
	repository, db, mock := bindingOwnerTestRepository(t)
	expectDefinitionSetTransaction(mock, coordinate, set)
	expectBindingOwnerRollback(mock)

	_, err := repository.ResolveExact(t.Context(), coordinate, binding)
	if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), "disagrees") {
		t.Fatalf("ResolveExact() error = %v, want version authority conflict", err)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func TestResolveExactRejectsByteDistinctStorageAliases(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	tests := []struct {
		name         string
		mutateParent func([]driver.Value)
		mutateHeader func([]driver.Value)
		mutateChild  func([]driver.Value)
		want         string
	}{
		{
			name: "parent case alias",
			mutateParent: func(values []driver.Value) {
				values[0] = strings.ToUpper(coordinate.StorageID)
			},
			want: "immutable parent disagrees",
		},
		{
			name: "parent trailing-space alias",
			mutateParent: func(values []driver.Value) {
				values[1] = coordinate.OwnerPack + " "
			},
			want: "immutable parent disagrees",
		},
		{
			name: "header case alias",
			mutateHeader: func(values []driver.Value) {
				values[0] = strings.ToUpper(coordinate.StorageID)
			},
			want: "header names a different generation",
		},
		{
			name: "child trailing-space alias",
			mutateChild: func(values []driver.Value) {
				values[0] = coordinate.StorageID + " "
			},
			want: "byte-distinct generation",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			repository, db, mock := bindingOwnerTestRepository(t)
			expectBindingOwnerBegin(mock)
			expectPackGenerationParent(mock, coordinate, test.mutateParent)
			if test.mutateParent == nil {
				expectDefinitionSetHeader(mock, coordinate, set, test.mutateHeader)
				if test.mutateHeader == nil {
					expectDefinitionChildren(mock, coordinate, set, test.mutateChild)
				}
			}
			expectBindingOwnerRollback(mock)

			_, err := repository.ResolveExact(t.Context(), coordinate, binding)
			if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("ResolveExact() error = %v, want authority conflict containing %q", err, test.want)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactRejectsStoredDefinitionCorruption(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	tests := []struct {
		name   string
		mutate func(*connectorbindingownerv1.DefinitionSet)
		want   string
	}{
		{
			name: "count mismatch",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Header.DefinitionCount++
			},
			want: "stored entry count",
		},
		{
			name: "noncanonical ordinal",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Entries[0].Ordinal++
			},
			want: "Ordinal",
		},
		{
			name: "projected kind disagrees",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Entries[0].Kind = catalogv1.BindingConnectorPushContract
			},
			want: "projected identity disagrees",
		},
		{
			name: "envelope digest disagrees",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Entries[0].EnvelopeDigest = "sha256:" + strings.Repeat("f", 64)
			},
			want: "EnvelopeDigest disagrees",
		},
		{
			name: "ordered set digest disagrees",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Header.OrderedSetDigest = "sha256:" + strings.Repeat("e", 64)
			},
			want: "OrderedSetDigest disagrees",
		},
		{
			name: "malformed canonical version",
			mutate: func(value *connectorbindingownerv1.DefinitionSet) {
				value.Entries[0].DefinitionVersion = 0
			},
			want: "version is malformed",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			corrupt := cloneDefinitionSet(set)
			test.mutate(&corrupt)
			repository, db, mock := bindingOwnerTestRepository(t)
			expectDefinitionSetTransaction(mock, coordinate, corrupt)
			expectBindingOwnerRollback(mock)

			_, err := repository.ResolveExact(t.Context(), coordinate, binding)
			if !errors.Is(err, ErrAuthorityConflict) || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("ResolveExact() error = %v, want corruption containing %q", err, test.want)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactPropagatesDatabaseFailuresAndRollsBack(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	databaseFailure := errors.New("database unavailable")
	tests := []struct {
		name   string
		expect func(sqlmock.Sqlmock)
	}{
		{
			name: "parent read",
			expect: func(mock sqlmock.Sqlmock) {
				expectBindingOwnerBegin(mock)
				mock.ExpectQuery(bindingOwnerParentQueryPattern()).
					WithArgs(coordinate.StorageID, coordinate.StorageID).
					WillReturnError(databaseFailure)
				expectBindingOwnerRollback(mock)
			},
		},
		{
			name: "header read",
			expect: func(mock sqlmock.Sqlmock) {
				expectBindingOwnerBegin(mock)
				expectPackGenerationParent(mock, coordinate, nil)
				mock.ExpectQuery(bindingOwnerHeaderQueryPattern()).
					WithArgs(coordinate.StorageID, coordinate.StorageID).
					WillReturnError(databaseFailure)
				expectBindingOwnerRollback(mock)
			},
		},
		{
			name: "child read",
			expect: func(mock sqlmock.Sqlmock) {
				expectBindingOwnerBegin(mock)
				expectPackGenerationParent(mock, coordinate, nil)
				expectDefinitionSetHeader(mock, coordinate, set, nil)
				mock.ExpectQuery(bindingOwnerChildrenQueryPattern()).
					WithArgs(coordinate.StorageID).
					WillReturnError(databaseFailure)
				expectBindingOwnerRollback(mock)
			},
		},
		{
			name: "child iteration",
			expect: func(mock sqlmock.Sqlmock) {
				expectBindingOwnerBegin(mock)
				expectPackGenerationParent(mock, coordinate, nil)
				expectDefinitionSetHeader(mock, coordinate, set, nil)
				rows := bindingOwnerChildRows()
				addDefinitionChildRow(rows, coordinate, set.Entries[0], nil)
				rows.RowError(0, databaseFailure)
				mock.ExpectQuery(bindingOwnerChildrenQueryPattern()).
					WithArgs(coordinate.StorageID).
					WillReturnRows(rows)
				expectBindingOwnerRollback(mock)
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			repository, db, mock := bindingOwnerTestRepository(t)
			test.expect(mock)
			_, err := repository.ResolveExact(t.Context(), coordinate, binding)
			if !errors.Is(err, databaseFailure) {
				t.Fatalf("ResolveExact() error = %v, want database failure", err)
			}
			assertBindingOwnerExpectations(t, db, mock)
		})
	}
}

func TestResolveExactReportsBeginAndCommitFailures(t *testing.T) {
	coordinate, set, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)

	t.Run("begin", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		beginFailure := errors.New("set pessimistic mode failed")
		mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = 'pessimistic'")).
			WillReturnError(beginFailure)
		mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = DEFAULT")).
			WillReturnResult(sqlmock.NewResult(0, 0))

		_, err := repository.ResolveExact(t.Context(), coordinate, binding)
		if !errors.Is(err, beginFailure) || !strings.Contains(err.Error(), "begin exact") {
			t.Fatalf("ResolveExact() begin error = %v", err)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})

	t.Run("commit", func(t *testing.T) {
		repository, db, mock := bindingOwnerTestRepository(t)
		expectDefinitionSetTransaction(mock, coordinate, set)
		commitFailure := errors.New("commit failed")
		mock.ExpectCommit().WillReturnError(commitFailure)
		expectBindingOwnerCleanup(mock)

		actual, err := repository.ResolveExact(t.Context(), coordinate, binding)
		if !errors.Is(err, commitFailure) || !strings.Contains(err.Error(), "commit exact") {
			t.Fatalf("ResolveExact() commit error = %v", err)
		}
		if !reflect.DeepEqual(actual, connectorbindingownerv1.DefinitionEnvelope{}) {
			t.Fatalf("ResolveExact() returned definition after failed commit: %#v", actual)
		}
		assertBindingOwnerExpectations(t, db, mock)
	})
}

func TestResolveExactRejectsNilContextAndUnavailableRepository(t *testing.T) {
	coordinate, _, definitions := completeDefinitionSetFixture(t)
	binding := bindingFromIdentity(definitions[0].Identity)
	selection := mustExactBindingSelection(
		t,
		coordinate,
		[]catalogv1.IntegrationBinding{binding},
	)
	var repository *Repository
	if _, err := repository.ResolveExact(t.Context(), coordinate, binding); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("nil Repository.ResolveExact() error = %v", err)
	}
	if _, err := repository.ResolveExactBindings(
		t.Context(),
		coordinate,
		[]catalogv1.IntegrationBinding{binding},
	); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("nil Repository.ResolveExactBindings() error = %v", err)
	}
	if _, err := repository.ResolveExactBindingSelections(
		t.Context(),
		[]ExactBindingSelection{selection},
	); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("nil Repository.ResolveExactBindingSelections() error = %v", err)
	}

	repository, db, mock := bindingOwnerTestRepository(t)
	if _, err := repository.ResolveExact(nil, coordinate, binding); !errors.Is(err, ErrInvalidRequest) {
		t.Fatalf("ResolveExact(nil context) error = %v", err)
	}
	if _, err := repository.ResolveExactBindings(
		nil,
		coordinate,
		[]catalogv1.IntegrationBinding{binding},
	); !errors.Is(err, ErrInvalidRequest) {
		t.Fatalf("ResolveExactBindings(nil context) error = %v", err)
	}
	if _, err := repository.ResolveExactBindingSelections(
		nil,
		[]ExactBindingSelection{selection},
	); !errors.Is(err, ErrInvalidRequest) {
		t.Fatalf("ResolveExactBindingSelections(nil context) error = %v", err)
	}
	assertBindingOwnerExpectations(t, db, mock)
}

func completeDefinitionSetFixture(
	t *testing.T,
) (PackGenerationCoordinate, connectorbindingownerv1.DefinitionSet, []connectorbindingownerv1.DefinitionEnvelope) {
	t.Helper()
	coordinate := PackGenerationCoordinate{
		StorageID:          "Opaque-PackGeneration-Row/Retained-01",
		OwnerPack:          "neutral-connector",
		PackInstallationID: "pack-installation-1",
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("a", 64),
	}
	return definitionSetFixtureForCoordinate(t, coordinate, "")
}

func definitionSetFixtureForCoordinate(
	t *testing.T,
	coordinate PackGenerationCoordinate,
	businessKeyPrefix string,
) (PackGenerationCoordinate, connectorbindingownerv1.DefinitionSet, []connectorbindingownerv1.DefinitionEnvelope) {
	t.Helper()
	kinds := []catalogv1.IntegrationBindingKind{
		catalogv1.BindingConnectorInboundRoute,
		catalogv1.BindingConnectorCalloutProfile,
		catalogv1.BindingConnectorFieldMapping,
		catalogv1.BindingConnectorPushContract,
	}
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(kinds))
	for index, kind := range kinds {
		businessKey := "binding-" + string(rune('a'+index))
		if businessKeyPrefix != "" {
			businessKey = businessKeyPrefix + "/" + businessKey
		}
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(connectorbindingownerv1.AuthoringInput{
			Identity: connectorbindingownerv1.Identity{
				Kind:               kind,
				OwnerPack:          coordinate.OwnerPack,
				PackInstallationID: coordinate.PackInstallationID,
				PackPayloadDigest:  coordinate.PackPayloadDigest,
				BusinessKey:        businessKey,
				DefinitionVersion:  1,
			},
			CanonicalDefinition: []byte(`{"neutral":"definition-` + string(rune('a'+index)) + `"}`),
		})
		if err != nil {
			t.Fatalf("NewDefinitionEnvelope(%s): %v", kind, 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.Fatal(err)
	}
	validated, err := connectorbindingownerv1.ValidateDefinitionSet(set)
	if err != nil {
		t.Fatal(err)
	}
	return coordinate, set, validated
}

func bindingFromIdentity(identity connectorbindingownerv1.Identity) catalogv1.IntegrationBinding {
	return catalogv1.IntegrationBinding{
		Kind:               identity.Kind,
		OwnerPack:          identity.OwnerPack,
		PackInstallationID: identity.PackInstallationID,
		PackPayloadDigest:  identity.PackPayloadDigest,
		BusinessKey:        identity.BusinessKey,
		DefinitionVersion:  identity.DefinitionVersion,
	}
}

func bindingsAtIndexes(
	definitions []connectorbindingownerv1.DefinitionEnvelope,
	indexes ...int,
) []catalogv1.IntegrationBinding {
	bindings := make([]catalogv1.IntegrationBinding, len(indexes))
	for index, definitionIndex := range indexes {
		bindings[index] = bindingFromIdentity(definitions[definitionIndex].Identity)
	}
	return bindings
}

func mustExactBindingSelection(
	t *testing.T,
	coordinate PackGenerationCoordinate,
	bindings []catalogv1.IntegrationBinding,
) ExactBindingSelection {
	t.Helper()
	selection, err := NewExactBindingSelection(coordinate, bindings)
	if err != nil {
		t.Fatalf("NewExactBindingSelection(): %v", err)
	}
	return selection
}

func exactSelectionBudgetCoordinate(index int) PackGenerationCoordinate {
	return PackGenerationCoordinate{
		StorageID:          fmt.Sprintf("budget-generation-%04d", index),
		OwnerPack:          fmt.Sprintf("budget-owner-%04d", index),
		PackInstallationID: fmt.Sprintf("budget-installation-%04d", index),
		PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("d", 64),
	}
}

func exactSelectionBudgetBindings(
	coordinate PackGenerationCoordinate,
	prefix string,
	count int,
) []catalogv1.IntegrationBinding {
	bindings := make([]catalogv1.IntegrationBinding, count)
	for index := range bindings {
		bindings[index] = catalogv1.IntegrationBinding{
			Kind:               catalogv1.BindingConnectorInboundRoute,
			OwnerPack:          coordinate.OwnerPack,
			PackInstallationID: coordinate.PackInstallationID,
			PackPayloadDigest:  coordinate.PackPayloadDigest,
			BusinessKey:        fmt.Sprintf("%s/binding-%04d", prefix, index),
			DefinitionVersion:  1,
		}
	}
	return bindings
}

func definitionSetEnvelopeBytes(set connectorbindingownerv1.DefinitionSet) uint64 {
	var total uint64
	for _, entry := range set.Entries {
		total += uint64(len(entry.EnvelopeJSON))
	}
	return total
}

func bindingOwnerTestRepository(t *testing.T) (*Repository, *sql.DB, sqlmock.Sqlmock) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	repository, err := NewRepository(db)
	if err != nil {
		db.Close()
		t.Fatal(err)
	}
	return repository, db, mock
}

func expectDefinitionSetTransaction(
	mock sqlmock.Sqlmock,
	coordinate PackGenerationCoordinate,
	set connectorbindingownerv1.DefinitionSet,
) {
	expectBindingOwnerBegin(mock)
	expectDefinitionSetReads(mock, coordinate, set)
}

func expectDefinitionSetReads(
	mock sqlmock.Sqlmock,
	coordinate PackGenerationCoordinate,
	set connectorbindingownerv1.DefinitionSet,
) {
	expectPackGenerationParent(mock, coordinate, nil)
	expectDefinitionSetHeader(mock, coordinate, set, nil)
	expectDefinitionChildren(mock, coordinate, set, nil)
}

func expectBindingOwnerBegin(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = 'pessimistic'")).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectBindingOwnerCommit(mock sqlmock.Sqlmock) {
	mock.ExpectCommit()
	expectBindingOwnerCleanup(mock)
}

func expectBindingOwnerRollback(mock sqlmock.Sqlmock) {
	mock.ExpectRollback()
	expectBindingOwnerCleanup(mock)
}

func expectBindingOwnerCleanup(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = DEFAULT")).
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectPackGenerationParent(
	mock sqlmock.Sqlmock,
	coordinate PackGenerationCoordinate,
	mutate func([]driver.Value),
) {
	values := []driver.Value{
		coordinate.StorageID,
		coordinate.OwnerPack,
		coordinate.PackInstallationID,
		coordinate.PackPayloadDigest,
	}
	if mutate != nil {
		mutate(values)
	}
	mock.ExpectQuery(bindingOwnerParentQueryPattern()).
		WithArgs(coordinate.StorageID, coordinate.StorageID).
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "PackName", "PackInstallId", "AppliedPayloadHash",
		}).AddRow(values...))
}

func expectDefinitionSetHeader(
	mock sqlmock.Sqlmock,
	coordinate PackGenerationCoordinate,
	set connectorbindingownerv1.DefinitionSet,
	mutate func([]driver.Value),
) {
	values := []driver.Value{
		coordinate.StorageID,
		set.Header.Protocol,
		set.Header.DefinitionCount,
		set.Header.OrderedSetDigest,
	}
	if mutate != nil {
		mutate(values)
	}
	mock.ExpectQuery(bindingOwnerHeaderQueryPattern()).
		WithArgs(coordinate.StorageID, coordinate.StorageID).
		WillReturnRows(bindingOwnerHeaderRows().AddRow(values...))
}

func expectDefinitionChildren(
	mock sqlmock.Sqlmock,
	coordinate PackGenerationCoordinate,
	set connectorbindingownerv1.DefinitionSet,
	mutate func([]driver.Value),
) {
	rows := bindingOwnerChildRows()
	for index, entry := range set.Entries {
		rowMutator := func(values []driver.Value) {
			if mutate != nil && index == 0 {
				mutate(values)
			}
		}
		addDefinitionChildRow(rows, coordinate, entry, rowMutator)
	}
	mock.ExpectQuery(bindingOwnerChildrenQueryPattern()).
		WithArgs(coordinate.StorageID).
		WillReturnRows(rows)
}

func addDefinitionChildRow(
	rows *sqlmock.Rows,
	coordinate PackGenerationCoordinate,
	entry connectorbindingownerv1.DefinitionSetEntry,
	mutate func([]driver.Value),
) {
	values := []driver.Value{
		coordinate.StorageID,
		entry.PackInstallationID,
		entry.Ordinal,
		entry.Kind,
		entry.OwnerPack,
		entry.PackPayloadDigest,
		entry.BusinessKey,
		entry.DefinitionVersion,
		entry.EnvelopeDigest,
		entry.EnvelopeJSON,
	}
	if mutate != nil {
		mutate(values)
	}
	rows.AddRow(values...)
}

func bindingOwnerHeaderRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"PackGenerationId", "Protocol", "DefinitionCount", "DefinitionSetDigest",
	})
}

func bindingOwnerChildRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"PackGenerationId", "PackInstallId", "DefinitionOrdinal", "Kind",
		"OwnerPack", "PackPayloadDigest", "BusinessKey", "DefinitionVersion",
		"DefinitionEnvelopeDigest", "DefinitionEnvelope",
	})
}

func bindingOwnerParentQueryPattern() string {
	return `(?s)FROM _System_PackGeneration.*WHERE Id = \?\s+AND BINARY Id = BINARY \?.*LIMIT 1.*FOR UPDATE`
}

func bindingOwnerHeaderQueryPattern() string {
	return `(?s)FROM _System_PackGenerationConnectorBindingDefinitionSet.*WHERE PackGenerationId = \?\s+AND BINARY PackGenerationId = BINARY \?.*FOR UPDATE`
}

func bindingOwnerChildrenQueryPattern() string {
	return `(?s)FROM _System_PackGenerationConnectorBindingDefinition.*WHERE PackGenerationId = \?.*ORDER BY DefinitionOrdinal.*FOR UPDATE`
}

func cloneDefinitionSet(value connectorbindingownerv1.DefinitionSet) connectorbindingownerv1.DefinitionSet {
	value.Entries = append([]connectorbindingownerv1.DefinitionSetEntry(nil), value.Entries...)
	for index := range value.Entries {
		value.Entries[index].EnvelopeJSON = append([]byte(nil), value.Entries[index].EnvelopeJSON...)
	}
	return value
}

func assertBindingOwnerExpectations(t *testing.T, db *sql.DB, mock sqlmock.Sqlmock) {
	t.Helper()
	mock.ExpectClose()
	if err := db.Close(); err != nil {
		t.Fatalf("close database: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}