package serviceconfiguration

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/credentialgeneration"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
)

const (
	serviceConfigurationInsertBatchSize = 256

	serviceConfigurationGenerationSelect = `
SELECT Id, GenerationProtocolVersion,
       OrgId, AppInstallationId, AuthorizedWorkloadId,
       CatalogRevision, SchemaGeneration, EngineGeneration,
       SourceSnapshotDigest, CatalogViewDigest,
       CatalogReceiptProtocolVersion, CatalogReceiptCapabilityVersion,
       CatalogSecurityEpoch,
       ConfigurationOwnerRevisionId, ConfigurationOwnerRevisionDigest,
       GenerationDigest, ReceiptProtocolVersion,
       OwnerReceipt, OwnerReceiptDigest,
       AccountCount, BindingCount, CredentialSelectionCount
  FROM _System_ConnectorServiceConfigurationGeneration
 WHERE Id = ? AND BINARY Id = BINARY ?
 LIMIT 1`

	serviceConfigurationAccountSelect = `
SELECT Id, AccountOrdinal, ConnectorAccountId, ProviderScopeId,
       AccountRevisionId, AccountRevisionDigest,
       ProviderRevisionId, ProviderRevisionDigest
  FROM _System_ConnectorServiceConfigurationAccount
 WHERE ServiceConfigurationGenerationId = ?
   AND BINARY ServiceConfigurationGenerationId = BINARY ?
 ORDER BY AccountOrdinal, Id`

	serviceConfigurationBindingSelect = `
SELECT binding_configuration.Id, binding_configuration.AccountConfigurationId,
       binding_configuration.CatalogBindingOrdinal,
       binding_configuration.ConfigurationRevisionId,
       binding_configuration.ConfigurationRevisionDigest,
       binding_configuration.ConfigurationFormatVersion,
       binding_configuration.CanonicalConfiguration
  FROM _System_ConnectorServiceConfigurationBinding AS binding_configuration
  JOIN _System_ConnectorServiceConfigurationAccount AS account_configuration
    ON account_configuration.Id = binding_configuration.AccountConfigurationId
   AND BINARY account_configuration.Id = BINARY binding_configuration.AccountConfigurationId
 WHERE account_configuration.ServiceConfigurationGenerationId = ?
   AND BINARY account_configuration.ServiceConfigurationGenerationId = BINARY ?
 ORDER BY account_configuration.AccountOrdinal,
          binding_configuration.CatalogBindingOrdinal,
          binding_configuration.Id`

	serviceConfigurationCredentialSelectionSelect = `
SELECT credential_selection.Id, credential_selection.BindingConfigurationId,
       credential_selection.CredentialOrdinal, credential_selection.CredentialRef,
       credential_selection.CredentialGenerationId
  FROM _System_ConnectorServiceConfigurationCredentialSelection AS credential_selection
  JOIN _System_ConnectorServiceConfigurationBinding AS binding_configuration
    ON binding_configuration.Id = credential_selection.BindingConfigurationId
   AND BINARY binding_configuration.Id = BINARY credential_selection.BindingConfigurationId
  JOIN _System_ConnectorServiceConfigurationAccount AS account_configuration
    ON account_configuration.Id = binding_configuration.AccountConfigurationId
   AND BINARY account_configuration.Id = BINARY binding_configuration.AccountConfigurationId
 WHERE account_configuration.ServiceConfigurationGenerationId = ?
   AND BINARY account_configuration.ServiceConfigurationGenerationId = BINARY ?
 ORDER BY account_configuration.AccountOrdinal,
          binding_configuration.CatalogBindingOrdinal,
          credential_selection.CredentialOrdinal,
          credential_selection.Id`
)

// Repository is the sole Connector persistence and exact-read authority for
// immutable whole-service configuration generations. It deliberately exposes
// no list, current/latest, update, delete, or mutable-owner lookup API.
type Repository struct {
	db                    *sql.DB
	credentialGenerations *credentialgeneration.Repository
}

// NewRepository constructs the service-configuration authority over the
// supplied TiDB pool and binds it to the credential-generation authority over
// that same pool.
func NewRepository(db *sql.DB) (*Repository, error) {
	if db == nil {
		return nil, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	credentialGenerations, err := credentialgeneration.NewRepository(db)
	if err != nil {
		return nil, fmt.Errorf("%w: credential-generation repository: %v", ErrDependencyMissing, err)
	}
	return &Repository{db: db, credentialGenerations: credentialGenerations}, nil
}

// VerifySchema checks the exact NexusCRM-owned four-table persistence shape.
// It never creates, alters, or repairs schema.
func (repository *Repository) VerifySchema(ctx context.Context) error {
	if repository == nil || repository.db == nil || repository.credentialGenerations == nil {
		return fmt.Errorf("%w: repository dependencies are unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	return verifyServiceConfigurationSchema(ctx, repository.db)
}

// ResolveExact loads one complete immutable generation by exact binary ID.
// It never falls back to a mutable source revision, Catalog view, account,
// current generation, or latest generation.
func (repository *Repository) ResolveExact(
	ctx context.Context,
	id string,
) (Generation, bool, error) {
	if repository == nil || repository.db == nil || repository.credentialGenerations == nil {
		return Generation{}, false, fmt.Errorf(
			"%w: repository dependencies are unavailable",
			ErrDependencyMissing,
		)
	}
	if ctx == nil {
		return Generation{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validIdentifier(id) {
		return Generation{}, false, invalidGeneration("generation ID is not canonical")
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Generation{}, false, err
	}
	return loadServiceConfigurationExact(ctx, repository.db, id, false)
}

// Publish resolves every selected credential through its exact authority,
// derives one closed generation, and inserts the complete four-level graph.
// An exact retry converges; a same-ID material difference or partial stored
// graph fails closed and is never repaired.
func (repository *Repository) Publish(
	ctx context.Context,
	input AuthoringInput,
	view catalogv1.IntegrationBindingView,
) (Generation, bool, error) {
	if repository == nil || repository.db == nil || repository.credentialGenerations == nil {
		return Generation{}, false, fmt.Errorf(
			"%w: repository dependencies are unavailable",
			ErrDependencyMissing,
		)
	}
	if ctx == nil {
		return Generation{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}

	credentialGenerationIDs, err := preflightCredentialGenerationIDs(input, view)
	if err != nil {
		return Generation{}, false, err
	}
	credentialGenerations, found, err := repository.credentialGenerations.ResolveExactIdentitySet(
		ctx,
		credentialGenerationIDs,
	)
	if err != nil {
		return Generation{}, false, fmt.Errorf("resolve exact credential generations: %w", err)
	}
	for index, exists := range found {
		if !exists {
			return Generation{}, false, fmt.Errorf(
				"%w: %q",
				ErrCredentialGenerationUnavailable,
				credentialGenerationIDs[index],
			)
		}
	}

	generation, err := deriveGenerationFromExactIdentities(input, view, credentialGenerations)
	if err != nil {
		return Generation{}, false, err
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Generation{}, false, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return Generation{}, false, fmt.Errorf("begin service configuration publish: %w", err)
	}
	rollback := func(cause error) (Generation, bool, error) {
		return Generation{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}
	commit := func() error {
		return errors.Join(tx.Commit(), cleanup())
	}

	if err := insertServiceConfigurationGeneration(ctx, tx, generation); err != nil {
		if !isServiceConfigurationDuplicateKey(err) {
			return rollback(fmt.Errorf("insert service configuration generation: %w", err))
		}
		stored, exists, loadErr := loadServiceConfigurationExact(ctx, tx, generation.id, true)
		if loadErr != nil {
			return rollback(loadErr)
		}
		if !exists || !generation.equal(stored) {
			return rollback(fmt.Errorf(
				"%w: generation ID already names different or incomplete immutable bytes",
				ErrIdentityConflict,
			))
		}
		if err := commit(); err != nil {
			return Generation{}, false, fmt.Errorf(
				"commit exact service configuration duplicate: %w",
				err,
			)
		}
		return stored.copy(), true, nil
	}

	if err := insertServiceConfigurationAccounts(ctx, tx, generation); err != nil {
		return rollback(serviceConfigurationChildInsertError("account", err))
	}
	if err := insertServiceConfigurationBindings(ctx, tx, generation); err != nil {
		return rollback(serviceConfigurationChildInsertError("binding", err))
	}
	if err := insertServiceConfigurationCredentialSelections(ctx, tx, generation); err != nil {
		return rollback(serviceConfigurationChildInsertError("credential selection", err))
	}

	stored, exists, err := loadServiceConfigurationExact(ctx, tx, generation.id, true)
	if err != nil {
		return rollback(err)
	}
	if !exists {
		return rollback(storedGenerationInvalid(
			"inserted generation cannot be reloaded by exact ID",
		))
	}
	if !generation.equal(stored) {
		return rollback(storedGenerationInvalid(
			"inserted generation bytes changed during persistence",
		))
	}
	if err := commit(); err != nil {
		return Generation{}, false, fmt.Errorf("commit service configuration publish: %w", err)
	}
	return stored.copy(), false, nil
}

func preflightCredentialGenerationIDs(
	input AuthoringInput,
	view catalogv1.IntegrationBindingView,
) ([]string, error) {
	if _, err := ResolveGenerationProtocolForAuthoring(input.GenerationProtocolVersion); err != nil {
		return nil, err
	}
	if _, err := ResolveReceiptProtocolForAuthoring(input.ReceiptProtocolVersion); err != nil {
		return nil, err
	}
	if err := validateCatalogIntegrationBindingView(view); err != nil {
		return nil, err
	}
	catalog := catalogReferenceFromView(view)
	if err := validateCatalogReference(catalog); err != nil {
		return nil, invalidGeneration("Catalog reference: %v", err)
	}
	if err := validateAuthoringInputShape(input, len(view.Bindings)); err != nil {
		return nil, err
	}
	if size, ok := authoringCanonicalSize(input, catalog); !ok || size > MaxCanonicalGenerationBytes {
		return nil, invalidGeneration(
			"canonical generation exceeds %d bytes",
			MaxCanonicalGenerationBytes,
		)
	}

	uniqueIDs := make(map[string]struct{})
	for accountIndex, account := range input.Accounts {
		for bindingIndex, binding := range account.Bindings {
			if err := validateSelectionsAgainstBinding(
				binding,
				view.Bindings[binding.CatalogBindingOrdinal],
			); err != nil {
				return nil, invalidGeneration(
					"accounts[%d].bindings[%d]: %v",
					accountIndex,
					bindingIndex,
					err,
				)
			}
			for _, selection := range binding.CredentialSelections {
				uniqueIDs[selection.CredentialGenerationID] = struct{}{}
			}
		}
	}
	ids := make([]string, 0, len(uniqueIDs))
	for id := range uniqueIDs {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	return ids, nil
}

type serviceConfigurationQueryer interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

func loadServiceConfigurationExact(
	ctx context.Context,
	queryer serviceConfigurationQueryer,
	id string,
	forUpdate bool,
) (Generation, bool, error) {
	parentQuery := serviceConfigurationGenerationSelect
	if forUpdate {
		parentQuery += " FOR UPDATE"
	}

	var (
		generation         Generation
		generationProtocol string
		receiptProtocol    string
	)
	parentRows, err := queryer.QueryContext(ctx, parentQuery, id, id)
	if err != nil {
		return Generation{}, false, fmt.Errorf("load service configuration generation row: %w", err)
	}
	defer parentRows.Close()
	if !parentRows.Next() {
		if err := parentRows.Err(); err != nil {
			return Generation{}, false, fmt.Errorf(
				"iterate service configuration generation row: %w",
				err,
			)
		}
		if err := parentRows.Close(); err != nil {
			return Generation{}, false, fmt.Errorf("close service configuration generation row: %w", err)
		}
		return Generation{}, false, nil
	}
	err = parentRows.Scan(
		&generation.id,
		&generationProtocol,
		&generation.catalog.orgID,
		&generation.catalog.appInstallationID,
		&generation.catalog.authorizedWorkloadID,
		&generation.catalog.lineage.CatalogRevision,
		&generation.catalog.lineage.SchemaGeneration,
		&generation.catalog.lineage.EngineGeneration,
		&generation.catalog.lineage.SourceSnapshotDigest,
		&generation.catalog.lineage.ViewDigest,
		&generation.catalog.receipts.ProtocolVersion,
		&generation.catalog.receipts.CapabilityVersion,
		&generation.catalog.receipts.SecurityEpoch,
		&generation.ownerRevision.RevisionID,
		&generation.ownerRevision.ValueDigest,
		&generation.generationDigest,
		&receiptProtocol,
		&generation.ownerReceipt,
		&generation.ownerReceiptDigest,
		&generation.accountCount,
		&generation.bindingCount,
		&generation.credentialSelectionCount,
	)
	if err != nil {
		return Generation{}, false, storedGenerationInvalid(
			"scan service configuration generation row: %v",
			err,
		)
	}
	if parentRows.Next() {
		return Generation{}, false, storedGenerationInvalid(
			"exact generation query returned more than one parent row",
		)
	}
	if err := parentRows.Err(); err != nil {
		return Generation{}, false, fmt.Errorf(
			"iterate service configuration generation row: %w",
			err,
		)
	}
	if err := parentRows.Close(); err != nil {
		return Generation{}, false, fmt.Errorf("close service configuration generation row: %w", err)
	}
	generation.generationProtocolVersion = GenerationProtocolVersion(generationProtocol)
	generation.receiptProtocolVersion = ReceiptProtocolVersion(receiptProtocol)
	generation.catalog.receipts.CatalogRevision = generation.catalog.lineage.CatalogRevision
	generation.catalog.receipts.SchemaGeneration = generation.catalog.lineage.SchemaGeneration
	generation.catalog.receipts.EngineGeneration = generation.catalog.lineage.EngineGeneration
	generation.ownerReceipt = cloneBinary(generation.ownerReceipt)
	if generation.accountCount > MaxAccounts ||
		generation.bindingCount > MaxBindingConfigurations ||
		generation.credentialSelectionCount > MaxCredentialSelections {
		return Generation{}, false, storedGenerationInvalid(
			"declared child counts exceed the retained protocol limits",
		)
	}

	accounts, accountByRowID, err := loadServiceConfigurationAccounts(
		ctx,
		queryer,
		id,
		forUpdate,
	)
	if err != nil {
		return Generation{}, false, err
	}
	generation.accounts = accounts
	bindingByRowID, err := loadServiceConfigurationBindings(
		ctx,
		queryer,
		id,
		forUpdate,
		generation.accounts,
		accountByRowID,
	)
	if err != nil {
		return Generation{}, false, err
	}
	if err := loadServiceConfigurationCredentialSelections(
		ctx,
		queryer,
		id,
		forUpdate,
		generation.accounts,
		bindingByRowID,
	); err != nil {
		return Generation{}, false, err
	}
	if err := generation.validateStored(); err != nil {
		return Generation{}, false, err
	}
	return generation.copy(), true, nil
}

func loadServiceConfigurationAccounts(
	ctx context.Context,
	queryer serviceConfigurationQueryer,
	generationID string,
	forUpdate bool,
) ([]AccountConfiguration, map[string]int, error) {
	query := serviceConfigurationAccountSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	rows, err := queryer.QueryContext(ctx, query, generationID, generationID)
	if err != nil {
		return nil, nil, fmt.Errorf("load service configuration account rows: %w", err)
	}
	defer rows.Close()

	accounts := make([]AccountConfiguration, 0)
	byRowID := make(map[string]int)
	for rows.Next() {
		if len(accounts) >= MaxAccounts {
			return nil, nil, storedGenerationInvalid("account rows exceed the retained protocol limit")
		}
		var (
			account AccountConfiguration
			ordinal int64
		)
		if err := rows.Scan(
			&account.rowID,
			&ordinal,
			&account.connectorAccountID,
			&account.providerScopeID,
			&account.accountRevision.RevisionID,
			&account.accountRevision.ValueDigest,
			&account.providerRevision.RevisionID,
			&account.providerRevision.ValueDigest,
		); err != nil {
			return nil, nil, storedGenerationInvalid(
				"scan service configuration account row: %v",
				err,
			)
		}
		if ordinal < 0 || ordinal > math.MaxInt32 {
			return nil, nil, storedGenerationInvalid("account ordinal is outside signed INT range")
		}
		account.ordinal = uint64(ordinal)
		account.bindings = make([]BindingConfiguration, 0)
		if _, duplicate := byRowID[account.rowID]; duplicate {
			return nil, nil, storedGenerationInvalid("duplicate account row ID")
		}
		byRowID[account.rowID] = len(accounts)
		accounts = append(accounts, account)
	}
	if err := rows.Err(); err != nil {
		return nil, nil, fmt.Errorf("iterate service configuration account rows: %w", err)
	}
	if err := rows.Close(); err != nil {
		return nil, nil, fmt.Errorf("close service configuration account rows: %w", err)
	}
	return accounts, byRowID, nil
}

type serviceConfigurationBindingLocation struct {
	accountIndex int
	bindingIndex int
}

func loadServiceConfigurationBindings(
	ctx context.Context,
	queryer serviceConfigurationQueryer,
	generationID string,
	forUpdate bool,
	accounts []AccountConfiguration,
	accountByRowID map[string]int,
) (map[string]serviceConfigurationBindingLocation, error) {
	query := serviceConfigurationBindingSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	rows, err := queryer.QueryContext(ctx, query, generationID, generationID)
	if err != nil {
		return nil, fmt.Errorf("load service configuration binding rows: %w", err)
	}
	defer rows.Close()

	total := 0
	byRowID := make(map[string]serviceConfigurationBindingLocation)
	for rows.Next() {
		if total >= MaxBindingConfigurations {
			return nil, storedGenerationInvalid("binding rows exceed the retained protocol limit")
		}
		var (
			binding BindingConfiguration
			ordinal int64
		)
		if err := rows.Scan(
			&binding.rowID,
			&binding.accountRowID,
			&ordinal,
			&binding.configurationRevision.RevisionID,
			&binding.configurationRevision.ValueDigest,
			&binding.configurationFormatVersion,
			&binding.canonicalConfiguration,
		); err != nil {
			return nil, storedGenerationInvalid(
				"scan service configuration binding row: %v",
				err,
			)
		}
		if ordinal < 0 || ordinal > math.MaxUint32 {
			return nil, storedGenerationInvalid("Catalog binding ordinal is outside uint32 range")
		}
		binding.catalogBindingOrdinal = uint32(ordinal)
		binding.canonicalConfiguration = cloneBinary(binding.canonicalConfiguration)
		binding.credentialSelections = make([]CredentialSelection, 0)
		accountIndex, exists := accountByRowID[binding.accountRowID]
		if !exists {
			return nil, storedGenerationInvalid("binding row references an unknown account row")
		}
		if len(accounts[accountIndex].bindings) >= MaxBindingsPerAccount {
			return nil, storedGenerationInvalid(
				"one account's binding rows exceed the retained protocol limit",
			)
		}
		if _, duplicate := byRowID[binding.rowID]; duplicate {
			return nil, storedGenerationInvalid("duplicate binding row ID")
		}
		location := serviceConfigurationBindingLocation{
			accountIndex: accountIndex,
			bindingIndex: len(accounts[accountIndex].bindings),
		}
		accounts[accountIndex].bindings = append(accounts[accountIndex].bindings, binding)
		byRowID[binding.rowID] = location
		total++
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service configuration binding rows: %w", err)
	}
	if err := rows.Close(); err != nil {
		return nil, fmt.Errorf("close service configuration binding rows: %w", err)
	}
	return byRowID, nil
}

func loadServiceConfigurationCredentialSelections(
	ctx context.Context,
	queryer serviceConfigurationQueryer,
	generationID string,
	forUpdate bool,
	accounts []AccountConfiguration,
	bindingByRowID map[string]serviceConfigurationBindingLocation,
) error {
	query := serviceConfigurationCredentialSelectionSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	rows, err := queryer.QueryContext(ctx, query, generationID, generationID)
	if err != nil {
		return fmt.Errorf("load service configuration credential selection rows: %w", err)
	}
	defer rows.Close()

	total := 0
	seenRowIDs := make(map[string]struct{})
	for rows.Next() {
		if total >= MaxCredentialSelections {
			return storedGenerationInvalid(
				"credential selection rows exceed the retained protocol limit",
			)
		}
		var (
			selection CredentialSelection
			ordinal   int64
		)
		if err := rows.Scan(
			&selection.rowID,
			&selection.bindingRowID,
			&ordinal,
			&selection.credentialRef,
			&selection.credentialGenerationID,
		); err != nil {
			return storedGenerationInvalid(
				"scan service configuration credential selection row: %v",
				err,
			)
		}
		if ordinal < 0 || ordinal > math.MaxInt32 {
			return storedGenerationInvalid("credential ordinal is outside signed INT range")
		}
		selection.ordinal = uint64(ordinal)
		location, exists := bindingByRowID[selection.bindingRowID]
		if !exists {
			return storedGenerationInvalid("credential selection references an unknown binding row")
		}
		binding := &accounts[location.accountIndex].bindings[location.bindingIndex]
		if len(binding.credentialSelections) >= catalogv1.MaxCredentialRefCount {
			return storedGenerationInvalid(
				"one binding's credential selections exceed the retained protocol limit",
			)
		}
		if _, duplicate := seenRowIDs[selection.rowID]; duplicate {
			return storedGenerationInvalid("duplicate credential selection row ID")
		}
		seenRowIDs[selection.rowID] = struct{}{}
		binding.credentialSelections = append(binding.credentialSelections, selection)
		total++
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate service configuration credential selection rows: %w", err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("close service configuration credential selection rows: %w", err)
	}
	return nil
}

func insertServiceConfigurationGeneration(
	ctx context.Context,
	tx *sql.Tx,
	generation Generation,
) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServiceConfigurationGeneration
       (Id, GenerationProtocolVersion,
        OrgId, AppInstallationId, AuthorizedWorkloadId,
        CatalogRevision, SchemaGeneration, EngineGeneration,
        SourceSnapshotDigest, CatalogViewDigest,
        CatalogReceiptProtocolVersion, CatalogReceiptCapabilityVersion,
        CatalogSecurityEpoch,
        ConfigurationOwnerRevisionId, ConfigurationOwnerRevisionDigest,
        GenerationDigest, ReceiptProtocolVersion,
        OwnerReceipt, OwnerReceiptDigest,
        AccountCount, BindingCount, CredentialSelectionCount)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		generation.id,
		string(generation.generationProtocolVersion),
		generation.catalog.orgID,
		generation.catalog.appInstallationID,
		generation.catalog.authorizedWorkloadID,
		generation.catalog.lineage.CatalogRevision,
		generation.catalog.lineage.SchemaGeneration,
		generation.catalog.lineage.EngineGeneration,
		generation.catalog.lineage.SourceSnapshotDigest,
		generation.catalog.lineage.ViewDigest,
		generation.catalog.receipts.ProtocolVersion,
		generation.catalog.receipts.CapabilityVersion,
		generation.catalog.receipts.SecurityEpoch,
		generation.ownerRevision.RevisionID,
		generation.ownerRevision.ValueDigest,
		generation.generationDigest,
		string(generation.receiptProtocolVersion),
		generation.ownerReceipt,
		generation.ownerReceiptDigest,
		generation.accountCount,
		generation.bindingCount,
		generation.credentialSelectionCount,
	)
	if err != nil {
		return err
	}
	return requireServiceConfigurationInsertedRows(result, 1)
}

func insertServiceConfigurationAccounts(
	ctx context.Context,
	tx *sql.Tx,
	generation Generation,
) error {
	for start := 0; start < len(generation.accounts); start += serviceConfigurationInsertBatchSize {
		end := min(start+serviceConfigurationInsertBatchSize, len(generation.accounts))
		accounts := generation.accounts[start:end]
		query := `INSERT INTO _System_ConnectorServiceConfigurationAccount
       (Id, ServiceConfigurationGenerationId, AccountOrdinal,
        ConnectorAccountId, ProviderScopeId,
        AccountRevisionId, AccountRevisionDigest,
        ProviderRevisionId, ProviderRevisionDigest)
VALUES ` + serviceConfigurationValueTuples(len(accounts), 9)
		arguments := make([]any, 0, len(accounts)*9)
		for _, account := range accounts {
			arguments = append(arguments,
				account.rowID,
				generation.id,
				account.ordinal,
				account.connectorAccountID,
				account.providerScopeID,
				account.accountRevision.RevisionID,
				account.accountRevision.ValueDigest,
				account.providerRevision.RevisionID,
				account.providerRevision.ValueDigest,
			)
		}
		result, err := tx.ExecContext(ctx, query, arguments...)
		if err != nil {
			return err
		}
		if err := requireServiceConfigurationInsertedRows(result, int64(len(accounts))); err != nil {
			return err
		}
	}
	return nil
}

func insertServiceConfigurationBindings(
	ctx context.Context,
	tx *sql.Tx,
	generation Generation,
) error {
	bindings := make([]BindingConfiguration, 0, generation.bindingCount)
	for _, account := range generation.accounts {
		bindings = append(bindings, account.bindings...)
	}
	for start := 0; start < len(bindings); start += serviceConfigurationInsertBatchSize {
		end := min(start+serviceConfigurationInsertBatchSize, len(bindings))
		batch := bindings[start:end]
		query := `INSERT INTO _System_ConnectorServiceConfigurationBinding
       (Id, AccountConfigurationId, CatalogBindingOrdinal,
        ConfigurationRevisionId, ConfigurationRevisionDigest,
        ConfigurationFormatVersion, CanonicalConfiguration)
VALUES ` + serviceConfigurationValueTuples(len(batch), 7)
		arguments := make([]any, 0, len(batch)*7)
		for _, binding := range batch {
			arguments = append(arguments,
				binding.rowID,
				binding.accountRowID,
				binding.catalogBindingOrdinal,
				binding.configurationRevision.RevisionID,
				binding.configurationRevision.ValueDigest,
				binding.configurationFormatVersion,
				binding.canonicalConfiguration,
			)
		}
		result, err := tx.ExecContext(ctx, query, arguments...)
		if err != nil {
			return err
		}
		if err := requireServiceConfigurationInsertedRows(result, int64(len(batch))); err != nil {
			return err
		}
	}
	return nil
}

func insertServiceConfigurationCredentialSelections(
	ctx context.Context,
	tx *sql.Tx,
	generation Generation,
) error {
	selections := make([]CredentialSelection, 0, generation.credentialSelectionCount)
	for _, account := range generation.accounts {
		for _, binding := range account.bindings {
			selections = append(selections, binding.credentialSelections...)
		}
	}
	for start := 0; start < len(selections); start += serviceConfigurationInsertBatchSize {
		end := min(start+serviceConfigurationInsertBatchSize, len(selections))
		batch := selections[start:end]
		query := `INSERT INTO _System_ConnectorServiceConfigurationCredentialSelection
       (Id, BindingConfigurationId, CredentialOrdinal,
        CredentialRef, CredentialGenerationId)
VALUES ` + serviceConfigurationValueTuples(len(batch), 5)
		arguments := make([]any, 0, len(batch)*5)
		for _, selection := range batch {
			arguments = append(arguments,
				selection.rowID,
				selection.bindingRowID,
				selection.ordinal,
				selection.credentialRef,
				selection.credentialGenerationID,
			)
		}
		result, err := tx.ExecContext(ctx, query, arguments...)
		if err != nil {
			return err
		}
		if err := requireServiceConfigurationInsertedRows(result, int64(len(batch))); err != nil {
			return err
		}
	}
	return nil
}

func serviceConfigurationValueTuples(rows, columns int) string {
	if rows <= 0 || columns <= 0 {
		return ""
	}
	tuple := "(" + strings.TrimSuffix(strings.Repeat("?,", columns), ",") + ")"
	return strings.TrimSuffix(strings.Repeat(tuple+",", rows), ",")
}

func requireServiceConfigurationInsertedRows(result sql.Result, want int64) error {
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("service configuration rows affected: %w", err)
	}
	if rows != want {
		return fmt.Errorf("service configuration rows affected = %d, want %d", rows, want)
	}
	return nil
}

func isServiceConfigurationDuplicateKey(err error) bool {
	var mysqlError *mysql.MySQLError
	return errors.As(err, &mysqlError) && mysqlError.Number == 1062
}

func serviceConfigurationChildInsertError(kind string, err error) error {
	if isServiceConfigurationDuplicateKey(err) {
		return fmt.Errorf(
			"%w: derived %s rows conflict with existing immutable graph rows: %v",
			ErrIdentityConflict,
			kind,
			err,
		)
	}
	return fmt.Errorf("insert service configuration %s rows: %w", kind, err)
}