package credentialgeneration

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

	_ "github.com/go-sql-driver/mysql"
)

func TestCredentialGenerationRepositorySchemaReadinessAgainstTiDB(t *testing.T) {
	tests := []struct {
		name   string
		damage func(*testing.T, *sql.DB)
	}{
		{name: "complete schema"},
		{
			name: "non-binary identity collation",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  MODIFY COLUMN ReceiptProtocolVersion VARCHAR(32)
  CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL`)
			},
		},
		{
			name: "wrong timestamp default",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  MODIFY COLUMN CreatedDate DATETIME(6) NOT NULL
  DEFAULT '2000-01-01 00:00:00.000000'`)
			},
		},
		{
			name: "timestamp updates mutable row",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  MODIFY COLUMN LastModifiedDate DATETIME(6) NOT NULL
  DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)`)
			},
		},
		{
			name: "missing stable-identity history index",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  DROP INDEX idx_connector_credential_generation_identity`)
			},
		},
		{
			name: "unexpected unique current selector",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  ADD UNIQUE KEY ux_unreviewed_current (StableCredentialIdentity, ConnectorAccountId)`)
			},
		},
		{
			name: "unexpected column",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  ADD COLUMN IsCurrent BOOLEAN NULL`)
			},
		},
		{
			name: "unexpected mutable-parent foreign key",
			damage: func(t *testing.T, db *sql.DB) {
				execCredentialGenerationDDL(t, db, `CREATE TABLE _System_TestCredentialAccount (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PRIMARY KEY (Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`)
				execCredentialGenerationDDL(t, db, `ALTER TABLE _System_ConnectorCredentialGeneration
  ADD CONSTRAINT fk_unreviewed_credential_account
  FOREIGN KEY (ConnectorAccountId) REFERENCES _System_TestCredentialAccount (Id)
  ON UPDATE RESTRICT ON DELETE RESTRICT`)
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, cleanup := openCredentialGenerationTestTiDB(t)
			defer cleanup()
			createCredentialGenerationTestTable(t, db)
			repository := newCredentialGenerationTestRepository(t, db)

			if test.damage == nil {
				if err := repository.VerifySchema(t.Context()); err != nil {
					t.Fatalf("VerifySchema() error = %v", err)
				}
				return
			}

			test.damage(t, db)
			if err := repository.VerifySchema(t.Context()); !errors.Is(err, ErrSchemaUnavailable) {
				t.Fatalf("VerifySchema() error = %v, want ErrSchemaUnavailable", err)
			}
		})
	}
}

func TestCredentialGenerationRepositoryConcurrentExactPublishAgainstTiDB(t *testing.T) {
	const workers = 12

	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	forceCredentialGenerationConstraintSessionsOff(t, db, workers)
	repository := newCredentialGenerationTestRepository(t, db)
	generation := credentialGenerationTestValue(t, "credential-generation-concurrent")

	type result struct {
		generation Generation
		duplicate  bool
		err        error
	}
	start := make(chan struct{})
	results := make(chan result, workers)
	var group sync.WaitGroup
	for index := 0; index < workers; index++ {
		group.Add(1)
		go func() {
			defer group.Done()
			<-start
			stored, duplicate, err := repository.Publish(t.Context(), generation)
			results <- result{generation: stored, duplicate: duplicate, err: err}
		}()
	}
	close(start)
	group.Wait()
	close(results)

	creators := 0
	duplicates := 0
	for result := range results {
		if result.err != nil {
			t.Errorf("Publish() error = %v", result.err)
			continue
		}
		if !credentialGenerationTestEqual(result.generation, generation) {
			t.Error("Publish() returned generation bytes that disagree with the exact candidate")
		}
		if result.duplicate {
			duplicates++
		} else {
			creators++
		}
	}
	if creators != 1 || duplicates != workers-1 {
		t.Fatalf("Publish() outcomes = %d creators/%d duplicates, want 1/%d", creators, duplicates, workers-1)
	}

	loaded, found, err := repository.ResolveExact(t.Context(), "credential-generation-concurrent")
	if err != nil || !found || !credentialGenerationTestEqual(loaded, generation) {
		t.Fatalf(
			"ResolveExact() = equal:%t found:%t err:%v",
			credentialGenerationTestEqual(loaded, generation),
			found,
			err,
		)
	}
	var count int
	if err := db.QueryRowContext(t.Context(),
		`SELECT COUNT(*) FROM _System_ConnectorCredentialGeneration`).Scan(&count); err != nil {
		t.Fatalf("count concurrently published generations: %v", err)
	}
	if count != 1 {
		t.Fatalf("concurrent exact Publish() stored %d rows, want 1", count)
	}
	var gateCount int
	var mode string
	var fenceToken int64
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*), MIN(Mode), MIN(FenceToken)
FROM _System_ConnectorCredentialGenerationDispatchGate
WHERE CredentialGenerationId = ?`, generation.ID()).Scan(&gateCount, &mode, &fenceToken); err != nil {
		t.Fatalf("read concurrently initialized dispatch gate: %v", err)
	}
	if gateCount != 1 || mode != string(DispatchModeEnabled) || fenceToken != 0 {
		t.Fatalf("concurrent dispatch gate = count:%d mode:%q fence:%d, want 1/enabled/0", gateCount, mode, fenceToken)
	}
}

func TestCredentialGenerationRepositoryRejectsIdentityConflictsAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	tests := []struct {
		name   string
		mutate func(*AuthoringInput)
	}{
		{
			name: "stable credential identity",
			mutate: func(input *AuthoringInput) {
				input.StableCredentialIdentity = "credential-stable-conflict"
			},
		},
		{
			name: "provider scope",
			mutate: func(input *AuthoringInput) {
				input.ProviderScopeID = "provider-scope-conflict"
			},
		},
		{
			name: "connector account",
			mutate: func(input *AuthoringInput) {
				input.ConnectorAccountID = "connector-account-conflict"
			},
		},
		{
			name: "secret backend protocol",
			mutate: func(input *AuthoringInput) {
				input.SecretBackendProtocol = "nexuscrm.secret-backend.v2"
			},
		},
		{
			name: "secret backend version",
			mutate: func(input *AuthoringInput) {
				input.SecretBackendVersionRef = "secret-version-conflict"
			},
		},
		{
			name: "transport configuration",
			mutate: func(input *AuthoringInput) {
				input.TransportConfiguration = []byte("transport-configuration-conflict\x00")
			},
		},
	}

	for index, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			id := fmt.Sprintf("credential-generation-conflict-%02d", index)
			base := credentialGenerationTestValue(t, id)
			publishCredentialGenerationTestValue(t, repository, base)

			input := credentialGenerationTestInput(id)
			test.mutate(&input)
			conflict := credentialGenerationFromInput(t, input)
			if _, _, err := repository.Publish(t.Context(), conflict); !errors.Is(err, ErrIdentityConflict) {
				t.Fatalf("Publish(conflict) error = %v, want ErrIdentityConflict", err)
			}
			loaded, found, err := repository.ResolveExact(t.Context(), id)
			if err != nil || !found || !credentialGenerationTestEqual(loaded, base) {
				t.Fatalf(
					"conflict changed durable generation: equal:%t found:%t err:%v",
					credentialGenerationTestEqual(loaded, base),
					found,
					err,
				)
			}
		})
	}
}

func TestCredentialGenerationRepositoryUsesExactBinaryIDsAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	upper := credentialGenerationTestValue(t, "Credential-Generation-Case-Exact")
	lower := credentialGenerationTestValue(t, "credential-generation-case-exact")
	publishCredentialGenerationTestValue(t, repository, upper)
	publishCredentialGenerationTestValue(t, repository, lower)

	for _, test := range []struct {
		id         string
		generation Generation
	}{
		{id: "Credential-Generation-Case-Exact", generation: upper},
		{id: "credential-generation-case-exact", generation: lower},
	} {
		loaded, found, err := repository.ResolveExact(t.Context(), test.id)
		if err != nil || !found || !credentialGenerationTestEqual(loaded, test.generation) {
			t.Fatalf(
				"ResolveExact(%q) = equal:%t found:%t err:%v",
				test.id,
				credentialGenerationTestEqual(loaded, test.generation),
				found,
				err,
			)
		}
	}
	if loaded, found, err := repository.ResolveExact(t.Context(), "CREDENTIAL-GENERATION-CASE-EXACT"); err != nil || found {
		t.Fatalf("case-folded ResolveExact() = %#v found:%t err:%v, want absent", loaded, found, err)
	}

	var count int
	if err := db.QueryRowContext(t.Context(),
		`SELECT COUNT(*) FROM _System_ConnectorCredentialGeneration`).Scan(&count); err != nil {
		t.Fatalf("count case-distinct generations: %v", err)
	}
	if count != 2 {
		t.Fatalf("case-distinct generation count = %d, want 2", count)
	}
}

func TestCredentialGenerationRepositoryResolveExactSetAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	upperInput := credentialGenerationTestInput("Credential-Generation-Exact-Set")
	upperInput.StableCredentialIdentity = "Credential-Stable-Exact-Set"
	upperInput.ProviderScopeID = "Provider-Scope-Exact-Set"
	upperInput.ConnectorAccountID = "Connector-Account-Exact-Set"
	upperInput.SecretBackendProtocol = "nexuscrm.secret-backend.exact-set.v1"
	upperInput.SecretBackendVersionRef = "secret-version-exact-set-upper"
	upperInput.TransportConfiguration = []byte{0x00, 0xff, 'u', 'p', 'p', 'e', 'r'}
	upper := credentialGenerationFromInput(t, upperInput)

	lowerInput := credentialGenerationTestInput("credential-generation-exact-set")
	lowerInput.StableCredentialIdentity = "credential-stable-exact-set"
	lowerInput.ProviderScopeID = "provider-scope-exact-set"
	lowerInput.ConnectorAccountID = "connector-account-exact-set"
	lowerInput.SecretBackendProtocol = "nexuscrm.secret-backend.exact-set.v2"
	lowerInput.SecretBackendVersionRef = "secret-version-exact-set-lower"
	lowerInput.TransportConfiguration = []byte{0xa5, 0x5a, 'l', 'o', 'w', 'e', 'r', 0x00}
	lower := credentialGenerationFromInput(t, lowerInput)

	publishCredentialGenerationTestValue(t, repository, upper)
	publishCredentialGenerationTestValue(t, repository, lower)

	missingID := "CREDENTIAL-GENERATION-EXACT-SET"
	ids := []string{lower.ID(), missingID, upper.ID()}
	loaded, found, err := repository.ResolveExactSet(t.Context(), ids)
	if err != nil {
		t.Fatalf("ResolveExactSet() error = %v", err)
	}
	if len(loaded) != len(ids) || len(found) != len(ids) {
		t.Fatalf("ResolveExactSet() lengths = %d, %d, want %d, %d", len(loaded), len(found), len(ids), len(ids))
	}
	if !found[0] || found[1] || !found[2] {
		t.Fatalf("ResolveExactSet() found = %#v, want [true false true]", found)
	}
	if !credentialGenerationTestEqual(loaded[0], lower) ||
		!credentialGenerationTestEqual(loaded[1], Generation{}) ||
		!credentialGenerationTestEqual(loaded[2], upper) {
		t.Fatal("ResolveExactSet() did not preserve caller order, missing position, and complete generation bytes")
	}

	reference := loaded[0].SecretBackendReference()
	if reference.Protocol() != lower.SecretBackendProtocol() ||
		reference.VersionRef() != lower.SecretBackendVersionRef() ||
		loaded[0].TransportConfigurationByteCount() != uint64(len(lower.TransportConfiguration())) ||
		loaded[0].OwnerReceiptByteCount() != uint64(len(lower.OwnerReceipt())) {
		t.Fatal("ResolveExactSet() public getters lost exact generation fields")
	}
	wantTransport := lower.TransportConfiguration()
	transport := loaded[0].TransportConfiguration()
	transport[0] ^= 0xff
	if !bytes.Equal(loaded[0].TransportConfiguration(), wantTransport) {
		t.Fatal("TransportConfiguration() exposed mutable stored bytes")
	}
	wantReceipt := lower.OwnerReceipt()
	receipt := loaded[0].OwnerReceipt()
	receipt[0] ^= 0xff
	if !bytes.Equal(loaded[0].OwnerReceipt(), wantReceipt) {
		t.Fatal("OwnerReceipt() exposed mutable stored bytes")
	}

	identities, identityFound, err := repository.ResolveExactIdentitySet(t.Context(), ids)
	if err != nil {
		t.Fatalf("ResolveExactIdentitySet() error = %v", err)
	}
	if len(identities) != len(loaded) || len(identityFound) != len(found) {
		t.Fatalf(
			"ResolveExactIdentitySet() lengths = %d, %d, want %d, %d",
			len(identities), len(identityFound), len(loaded), len(found),
		)
	}
	for index := range loaded {
		if identityFound[index] != found[index] {
			t.Fatalf("identity projection found[%d] = %t, want %t", index, identityFound[index], found[index])
		}
		if !found[index] {
			if identities[index] != (ExactIdentity{}) {
				t.Fatalf("missing identity projection[%d] = %#v, want zero value", index, identities[index])
			}
			continue
		}
		if identities[index].ID() != loaded[index].ID() ||
			identities[index].StableCredentialIdentity() != loaded[index].StableCredentialIdentity() ||
			identities[index].ProviderScopeID() != loaded[index].ProviderScopeID() ||
			identities[index].ConnectorAccountID() != loaded[index].ConnectorAccountID() {
			t.Fatalf("identity projection[%d] disagrees with the complete exact generation", index)
		}
	}

	var count int
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*)
FROM _System_ConnectorCredentialGeneration
WHERE Id IN (?, ?, ?)`, lower.ID(), missingID, upper.ID()).Scan(&count); err != nil {
		t.Fatalf("count exact-set case-distinct generations: %v", err)
	}
	if count != 2 {
		t.Fatalf("exact-set case-distinct generation count = %d, want 2", count)
	}
}

func TestCredentialGenerationRepositoryPersistsEmptyTransportConfigurationAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	input := credentialGenerationTestInput("credential-generation-empty-transport")
	input.TransportConfiguration = nil
	generation := credentialGenerationFromInput(t, input)
	if configuration := generation.TransportConfiguration(); configuration == nil || len(configuration) != 0 {
		t.Fatalf("TransportConfiguration() = %#v, want non-nil empty bytes", configuration)
	}
	publishCredentialGenerationTestValue(t, repository, generation)

	var isNull, byteLength int
	if err := db.QueryRowContext(t.Context(), `SELECT TransportConfiguration IS NULL,
       OCTET_LENGTH(TransportConfiguration)
  FROM _System_ConnectorCredentialGeneration
 WHERE Id = ?`, generation.ID()).Scan(&isNull, &byteLength); err != nil {
		t.Fatalf("read persisted empty transport configuration: %v", err)
	}
	if isNull != 0 || byteLength != 0 {
		t.Fatalf("persisted transport configuration isNull=%d bytes=%d, want 0/0", isNull, byteLength)
	}

	loaded, found, err := repository.ResolveExact(t.Context(), generation.ID())
	if err != nil || !found || !credentialGenerationTestEqual(loaded, generation) {
		t.Fatalf(
			"ResolveExact(empty transport) = equal:%t found:%t err:%v",
			credentialGenerationTestEqual(loaded, generation),
			found,
			err,
		)
	}
	if configuration := loaded.TransportConfiguration(); configuration == nil || len(configuration) != 0 {
		t.Fatalf("loaded TransportConfiguration() = %#v, want non-nil empty bytes", configuration)
	}
}

func TestCredentialGenerationRepositoryRejectsCorruptStoredRowsAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	tests := []struct {
		name   string
		mutate func(*testing.T, *sql.DB, string)
	}{
		{
			name: "stable identity",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET StableCredentialIdentity = ? WHERE Id = ?`,
					"credential-stable-corrupt", id)
			},
		},
		{
			name: "provider scope",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET ProviderScopeId = ? WHERE Id = ?`,
					"provider-scope-corrupt", id)
			},
		},
		{
			name: "connector account",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET ConnectorAccountId = ? WHERE Id = ?`,
					"connector-account-corrupt", id)
			},
		},
		{
			name: "secret backend protocol",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET SecretBackendProtocol = ? WHERE Id = ?`,
					"nexuscrm.secret-backend.corrupt", id)
			},
		},
		{
			name: "secret backend version reference",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET SecretBackendVersionRef = ? WHERE Id = ?`,
					"secret-version-corrupt", id)
			},
		},
		{
			name: "transport configuration",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET TransportConfiguration = ? WHERE Id = ?`,
					[]byte{0x00, 0xff, 0x5a}, id)
			},
		},
		{
			name: "transport configuration digest",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET TransportConfigurationDigest = ? WHERE Id = ?`,
					"sha256:"+strings.Repeat("0", 64), id)
			},
		},
		{
			name: "receipt protocol",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET ReceiptProtocolVersion = ? WHERE Id = ?`,
					"unsupported.receipt.v9", id)
			},
		},
		{
			name: "owner receipt",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET OwnerReceipt = ? WHERE Id = ?`,
					[]byte("not-the-canonical-owner-receipt"), id)
			},
		},
		{
			name: "owner receipt digest",
			mutate: func(t *testing.T, db *sql.DB, id string) {
				execCredentialGenerationCorruption(t, db,
					`UPDATE _System_ConnectorCredentialGeneration SET OwnerReceiptDigest = ? WHERE Id = ?`,
					"sha256:"+strings.Repeat("0", 64), id)
			},
		},
	}

	for index, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			id := fmt.Sprintf("credential-generation-corrupt-%02d", index)
			generation := credentialGenerationTestValue(t, id)
			publishCredentialGenerationTestValue(t, repository, generation)
			test.mutate(t, db, id)

			if _, _, err := repository.ResolveExact(t.Context(), id); !errors.Is(err, ErrStoredGenerationInvalid) {
				t.Fatalf("ResolveExact(corrupt row) error = %v, want ErrStoredGenerationInvalid", err)
			}
		})
	}
}

func TestCredentialGenerationRepositoryRetainsRotationsAgainstTiDB(t *testing.T) {
	db, cleanup := openCredentialGenerationTestTiDB(t)
	defer cleanup()
	createCredentialGenerationTestTable(t, db)
	repository := newCredentialGenerationTestRepository(t, db)

	firstInput := credentialGenerationTestInput("credential-generation-rotation-a")
	firstInput.SecretBackendVersionRef = "secret-version-rotation-a"
	secondInput := credentialGenerationTestInput("credential-generation-rotation-b")
	secondInput.SecretBackendVersionRef = "secret-version-rotation-b"
	first := credentialGenerationFromInput(t, firstInput)
	second := credentialGenerationFromInput(t, secondInput)

	publishCredentialGenerationTestValue(t, repository, first)
	publishCredentialGenerationTestValue(t, repository, second)

	for _, test := range []struct {
		id         string
		generation Generation
	}{
		{id: firstInput.ID, generation: first},
		{id: secondInput.ID, generation: second},
	} {
		loaded, found, err := repository.ResolveExact(t.Context(), test.id)
		if err != nil || !found || !credentialGenerationTestEqual(loaded, test.generation) {
			t.Fatalf(
				"ResolveExact(%q) after rotation = equal:%t found:%t err:%v",
				test.id,
				credentialGenerationTestEqual(loaded, test.generation),
				found,
				err,
			)
		}
	}

	stored, duplicate, err := repository.Publish(t.Context(), first)
	if err != nil || !duplicate || !credentialGenerationTestEqual(stored, first) {
		t.Fatalf(
			"Publish(first rotation again) = equal:%t duplicate:%t err:%v",
			credentialGenerationTestEqual(stored, first),
			duplicate,
			err,
		)
	}
	var count int
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*)
FROM _System_ConnectorCredentialGeneration
WHERE StableCredentialIdentity = ? AND ProviderScopeId = ? AND ConnectorAccountId = ?`,
		firstInput.StableCredentialIdentity,
		firstInput.ProviderScopeID,
		firstInput.ConnectorAccountID,
	).Scan(&count); err != nil {
		t.Fatalf("count retained credential rotations: %v", err)
	}
	if count != 2 {
		t.Fatalf("retained credential rotation count = %d, want 2", count)
	}
}

func credentialGenerationTestInput(id string) AuthoringInput {
	return AuthoringInput{
		ReceiptProtocolVersion:   ReceiptProtocolV1,
		ID:                       id,
		StableCredentialIdentity: "credential-stable-1",
		ProviderScopeID:          "provider-scope-1",
		ConnectorAccountID:       "connector-account-1",
		SecretBackendProtocol:    "nexuscrm.secret-backend.v1",
		SecretBackendVersionRef:  "secret-version-1",
		TransportConfiguration: []byte{
			0x00, 0xff, 0x5a, 0xa5,
			'c', 'a', 'n', 'o', 'n', 'i', 'c', 'a', 'l',
		},
	}
}

func credentialGenerationTestValue(t *testing.T, id string) Generation {
	t.Helper()
	return credentialGenerationFromInput(t, credentialGenerationTestInput(id))
}

func credentialGenerationFromInput(t *testing.T, input AuthoringInput) Generation {
	t.Helper()
	generation, err := NewGeneration(input)
	if err != nil {
		t.Fatalf("NewGeneration(%q): %v", input.ID, err)
	}
	return generation
}

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

func publishCredentialGenerationTestValue(t *testing.T, repository *Repository, generation Generation) {
	t.Helper()
	stored, duplicate, err := repository.Publish(t.Context(), generation)
	if err != nil || duplicate || !credentialGenerationTestEqual(stored, generation) {
		t.Fatalf(
			"Publish() = equal:%t duplicate:%t err:%v",
			credentialGenerationTestEqual(stored, generation),
			duplicate,
			err,
		)
	}
}

func credentialGenerationTestEqual(first, second Generation) bool {
	return first.ReceiptProtocolVersion() == second.ReceiptProtocolVersion() &&
		first.ID() == second.ID() &&
		first.StableCredentialIdentity() == second.StableCredentialIdentity() &&
		first.ProviderScopeID() == second.ProviderScopeID() &&
		first.ConnectorAccountID() == second.ConnectorAccountID() &&
		first.SecretBackendProtocol() == second.SecretBackendProtocol() &&
		first.SecretBackendVersionRef() == second.SecretBackendVersionRef() &&
		bytes.Equal(first.TransportConfiguration(), second.TransportConfiguration()) &&
		first.TransportConfigurationDigest() == second.TransportConfigurationDigest() &&
		bytes.Equal(first.OwnerReceipt(), second.OwnerReceipt()) &&
		first.OwnerReceiptDigest() == second.OwnerReceiptDigest()
}

func execCredentialGenerationCorruption(t *testing.T, db *sql.DB, statement string, args ...any) {
	t.Helper()
	result, err := db.ExecContext(t.Context(), statement, args...)
	if err != nil {
		t.Fatalf("corrupt CredentialGeneration fixture: %v", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		t.Fatalf("corrupt CredentialGeneration fixture rows: %v", err)
	}
	if rows != 1 {
		t.Fatalf("corrupt CredentialGeneration fixture affected %d rows, want 1", rows)
	}
}

func createCredentialGenerationTestTable(t *testing.T, db *sql.DB) {
	t.Helper()
	execCredentialGenerationDDL(t, db, credentialGenerationTestDDL)
	execCredentialGenerationDDL(t, db, credentialGenerationDispatchGateTestDDL)
}

func execCredentialGenerationDDL(t *testing.T, db *sql.DB, statement string) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), statement); err != nil {
		t.Fatalf("execute CredentialGeneration fixture DDL: %v", err)
	}
}

const credentialGenerationTestDDL = `CREATE TABLE _System_ConnectorCredentialGeneration (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  StableCredentialIdentity VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ProviderScopeId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ConnectorAccountId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  SecretBackendProtocol VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  SecretBackendVersionRef VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  TransportConfiguration MEDIUMBLOB NOT NULL,
  TransportConfigurationDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ReceiptProtocolVersion VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  OwnerReceipt MEDIUMBLOB NOT NULL,
  OwnerReceiptDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  KEY idx_connector_credential_generation_identity (StableCredentialIdentity, Id),
  KEY idx_connector_credential_generation_scope (ProviderScopeId, ConnectorAccountId)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const credentialGenerationDispatchGateTestDDL = `CREATE TABLE _System_ConnectorCredentialGenerationDispatchGate (
  CredentialGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  Mode VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT 'enabled',
  FenceToken BIGINT NOT NULL DEFAULT 0,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (CredentialGenerationId),
  CONSTRAINT fk_connector_credential_generation_dispatch_gate
    FOREIGN KEY (CredentialGenerationId)
    REFERENCES _System_ConnectorCredentialGeneration (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

func forceCredentialGenerationConstraintSessionsOff(t *testing.T, db *sql.DB, sessionCount int) {
	t.Helper()
	const variable = "tidb_constraint_check_in_place_pessimistic"
	if sessionCount < 1 {
		t.Fatalf("session count = %d, want at least 1", sessionCount)
	}

	db.SetMaxOpenConns(sessionCount)
	db.SetMaxIdleConns(0)
	held := make([]*sql.Conn, 0, sessionCount)
	defer func() {
		for _, conn := range held {
			if conn != nil {
				_ = conn.Close()
			}
		}
	}()

	for index := 0; index < sessionCount; index++ {
		conn, err := db.Conn(t.Context())
		if err != nil {
			t.Fatalf("acquire TiDB session %d: %v", index, err)
		}
		held = append(held, conn)
		if _, err := conn.ExecContext(t.Context(), "SET SESSION "+variable+" = OFF"); err != nil {
			t.Fatalf("set SESSION %s=OFF on connection %d: %v", variable, index, err)
		}
		var raw string
		if err := conn.QueryRowContext(t.Context(), "SELECT @@SESSION."+variable).Scan(&raw); err != nil {
			t.Fatalf("read SESSION %s on connection %d: %v", variable, index, err)
		}
		if canonicalCredentialGenerationTiDBBoolean(t, variable, raw) != "OFF" {
			t.Fatalf("SESSION %s on connection %d = %s, want OFF", variable, index, raw)
		}
	}

	db.SetMaxIdleConns(sessionCount)
	for index := range held {
		if err := held[index].Close(); err != nil {
			t.Fatalf("return TiDB session %d to bounded idle pool: %v", index, err)
		}
		held[index] = nil
	}
}

func canonicalCredentialGenerationTiDBBoolean(t *testing.T, name, value string) string {
	t.Helper()
	switch strings.ToUpper(strings.TrimSpace(value)) {
	case "ON", "1":
		return "ON"
	case "OFF", "0":
		return "OFF"
	default:
		t.Fatalf("unexpected %s value %q", name, value)
		return ""
	}
}

func openCredentialGenerationTestTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := credentialGenerationEnvironmentOr("TIDB_HOST", "127.0.0.1")
	port := credentialGenerationEnvironmentOr("TIDB_PORT", "4001")
	user := credentialGenerationEnvironmentOr("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 {
		unavailableCredentialGenerationTiDB(t, "open admin connection: %v", err)
	}
	if err := admin.PingContext(context.Background()); err != nil {
		_ = admin.Close()
		unavailableCredentialGenerationTiDB(t, "TiDB not reachable at %s:%s: %v", host, port, err)
	}

	databaseName := fmt.Sprintf("nexuscrm_connector_credential_generation_%d", time.Now().UnixNano())
	if _, err := admin.ExecContext(t.Context(), "CREATE DATABASE `"+databaseName+"`"); err != nil {
		_ = admin.Close()
		unavailableCredentialGenerationTiDB(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 CredentialGeneration test database: %v", err)
	}
	db.SetMaxOpenConns(32)
	if err := db.PingContext(t.Context()); err != nil {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
		t.Fatalf("ping CredentialGeneration test database: %v", err)
	}

	cleanup := func() {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
	}
	return db, cleanup
}

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

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