package credentialgeneration

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

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

const credentialGenerationSelect = `
SELECT Id, StableCredentialIdentity, ProviderScopeId, ConnectorAccountId,
       SecretBackendProtocol, SecretBackendVersionRef,
       TransportConfiguration, TransportConfigurationDigest,
       ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest
  FROM _System_ConnectorCredentialGeneration
 WHERE Id = ? AND BINARY Id = BINARY ?
 LIMIT 1`

const (
	credentialGenerationExactSetBatchSize = 256
	credentialGenerationExactSetColumns   = `Id, StableCredentialIdentity, ProviderScopeId, ConnectorAccountId,
       SecretBackendProtocol, SecretBackendVersionRef,
       TransportConfiguration, TransportConfigurationDigest,
       ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest`
)

// Repository is the sole Connector persistence and exact-read authority for
// immutable credential-generation metadata. It deliberately exposes no list,
// mutable-name, current-generation, update, or delete API.
type Repository struct {
	db *sql.DB
}

// NewRepository constructs a credential-generation repository over the
// provided TiDB connection pool.
func NewRepository(db *sql.DB) (*Repository, error) {
	if db == nil {
		return nil, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	return &Repository{db: db}, nil
}

// VerifySchema checks only immutable generation metadata so retained exact
// generations remain readable independently from dispatch-gate readiness.
func (repository *Repository) VerifySchema(ctx context.Context) error {
	if repository == nil || repository.db == nil {
		return fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	return verifyCredentialGenerationSchema(ctx, repository.db)
}

// ResolveExact loads one immutable generation by exact binary ID. It never
// falls back to a mutable name, current generation, provider scope, account,
// or newer backend version.
func (repository *Repository) ResolveExact(
	ctx context.Context,
	id string,
) (Generation, bool, error) {
	if repository == nil || repository.db == nil {
		return Generation{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return Generation{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validOpaqueIdentifier(id) {
		return Generation{}, false, invalidGeneration("generation ID is not canonical")
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Generation{}, false, err
	}
	return loadExact(ctx, repository.db, id, false)
}

// ResolveExactSet validates and returns complete immutable generations by exact
// binary ID in caller order. It first rejects oversized or duplicate input,
// then verifies aggregate stored payload size without loading BLOBs, and
// finally reads complete rows in bounded batches. Any failure discards the
// whole result. A missing ID retains its caller-order position as a zero
// Generation with a false found bit.
//
// This is not a list, current-generation, scope, or account lookup. The input
// slice must be non-nil; a non-nil empty set is valid and still verifies schema
// readiness.
func (repository *Repository) ResolveExactSet(
	ctx context.Context,
	ids []string,
) ([]Generation, []bool, error) {
	return repository.resolveExactSet(ctx, ids)
}

// ResolveExactIdentitySet returns the bounded identity projections of complete
// generations loaded through ResolveExactSet's sole batched exact-read path.
// It preserves caller order and missing-entry positions, and never issues a
// weaker identity-only storage read.
func (repository *Repository) ResolveExactIdentitySet(
	ctx context.Context,
	ids []string,
) ([]ExactIdentity, []bool, error) {
	generations, found, err := repository.resolveExactSet(ctx, ids)
	if err != nil {
		return nil, nil, err
	}
	identities := make([]ExactIdentity, len(generations))
	for index := range generations {
		if found[index] {
			identities[index] = generations[index].exactIdentity()
		}
	}
	return identities, found, nil
}

func (repository *Repository) resolveExactSet(
	ctx context.Context,
	ids []string,
) ([]Generation, []bool, error) {
	if repository == nil || repository.db == nil {
		return nil, nil, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return nil, nil, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if err := validateExactIdentitySetInput(ids); err != nil {
		return nil, nil, err
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return nil, nil, err
	}
	generations := make([]Generation, len(ids))
	found := make([]bool, len(ids))
	if len(ids) == 0 {
		return generations, found, nil
	}
	if err := repository.verifyExactIdentitySetPayloadBudget(ctx, ids); err != nil {
		return nil, nil, err
	}

	byID, err := repository.loadExactSet(ctx, ids)
	if err != nil {
		return nil, nil, err
	}
	for index, id := range ids {
		generation, exists := byID[id]
		if exists {
			generations[index] = generation
			found[index] = true
		}
	}
	return generations, found, nil
}

func validateExactIdentitySetInput(ids []string) error {
	if ids == nil {
		return invalidGeneration("generation ID set is nil")
	}
	if len(ids) > MaxExactIdentitySetEntries {
		return exactSetBudgetExceeded(
			"generation ID count %d exceeds %d",
			len(ids),
			MaxExactIdentitySetEntries,
		)
	}
	seen := make(map[string]struct{}, len(ids))
	totalIDBytes := 0
	for _, id := range ids {
		if !validOpaqueIdentifier(id) {
			return invalidGeneration("generation ID is not canonical")
		}
		if _, duplicate := seen[id]; duplicate {
			return invalidGeneration("generation ID set contains a duplicate exact ID")
		}
		seen[id] = struct{}{}
		if len(id) > MaxExactIdentitySetIDBytes-totalIDBytes {
			return exactSetBudgetExceeded(
				"generation ID bytes exceed %d",
				MaxExactIdentitySetIDBytes,
			)
		}
		totalIDBytes += len(id)
	}
	return nil
}

func (repository *Repository) verifyExactIdentitySetPayloadBudget(
	ctx context.Context,
	ids []string,
) error {
	totalPayloadBytes := uint64(0)
	for start := 0; start < len(ids); start += credentialGenerationExactSetBatchSize {
		end := min(start+credentialGenerationExactSetBatchSize, len(ids))
		batch := ids[start:end]
		query := credentialGenerationExactSetPayloadQuery(len(batch))
		arguments := credentialGenerationExactSetArguments(batch)
		var batchPayloadBytes uint64
		if err := repository.db.QueryRowContext(ctx, query, arguments...).Scan(&batchPayloadBytes); err != nil {
			return fmt.Errorf("measure exact credential generation payload: %w", err)
		}
		if batchPayloadBytes > MaxExactIdentitySetPayloadBytes-totalPayloadBytes {
			return exactSetBudgetExceeded(
				"credential generation payload exceeds %d bytes",
				MaxExactIdentitySetPayloadBytes,
			)
		}
		totalPayloadBytes += batchPayloadBytes
	}
	return nil
}

func (repository *Repository) loadExactSet(
	ctx context.Context,
	ids []string,
) (map[string]Generation, error) {
	generations := make(map[string]Generation, len(ids))
	requested := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		requested[id] = struct{}{}
	}
	totalIdentityBytes := 0
	totalPayloadBytes := 0
	for start := 0; start < len(ids); start += credentialGenerationExactSetBatchSize {
		end := min(start+credentialGenerationExactSetBatchSize, len(ids))
		batch := ids[start:end]
		query := credentialGenerationExactSetLoadQuery(len(batch))
		rows, err := repository.db.QueryContext(
			ctx,
			query,
			credentialGenerationExactSetArguments(batch)...,
		)
		if err != nil {
			return nil, fmt.Errorf("load exact credential generation set: %w", err)
		}
		for rows.Next() {
			var generation Generation
			if err := rows.Scan(
				&generation.id,
				&generation.stableCredentialIdentity,
				&generation.providerScopeID,
				&generation.connectorAccountID,
				&generation.secretBackendProtocol,
				&generation.secretBackendVersionRef,
				&generation.transportConfiguration,
				&generation.transportConfigurationDigest,
				&generation.receiptProtocolVersion,
				&generation.ownerReceipt,
				&generation.ownerReceiptDigest,
			); err != nil {
				_ = rows.Close()
				return nil, storedGenerationInvalid("scan exact-set row: %v", err)
			}
			if _, expected := requested[generation.id]; !expected {
				_ = rows.Close()
				return nil, storedGenerationInvalid(
					"exact-set query returned unrequested generation %q",
					generation.id,
				)
			}
			if _, duplicate := generations[generation.id]; duplicate {
				_ = rows.Close()
				return nil, storedGenerationInvalid(
					"exact-set query returned generation %q more than once",
					generation.id,
				)
			}
			payloadBytes := len(generation.transportConfiguration) + len(generation.ownerReceipt)
			if payloadBytes > MaxExactIdentitySetPayloadBytes-totalPayloadBytes {
				_ = rows.Close()
				return nil, exactSetBudgetExceeded(
					"credential generation payload changed beyond %d bytes",
					MaxExactIdentitySetPayloadBytes,
				)
			}
			totalPayloadBytes += payloadBytes
			generation = generation.copy()
			if err := generation.validateStored(); err != nil {
				_ = rows.Close()
				return nil, err
			}
			identityBytes := len(generation.id) + len(generation.stableCredentialIdentity) +
				len(generation.providerScopeID) + len(generation.connectorAccountID)
			if identityBytes > MaxExactIdentitySetIdentityBytes-totalIdentityBytes {
				_ = rows.Close()
				return nil, exactSetBudgetExceeded(
					"credential generation identity bytes exceed %d",
					MaxExactIdentitySetIdentityBytes,
				)
			}
			totalIdentityBytes += identityBytes
			generations[generation.id] = generation
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return nil, fmt.Errorf("iterate exact credential generation set: %w", err)
		}
		if err := rows.Close(); err != nil {
			return nil, fmt.Errorf("close exact credential generation set: %w", err)
		}
	}
	return generations, nil
}

func credentialGenerationExactSetPredicate(count int) string {
	return strings.TrimSuffix(
		strings.Repeat("(Id = ? AND BINARY Id = BINARY ?) OR ", count),
		" OR ",
	)
}

func credentialGenerationExactSetPayloadQuery(count int) string {
	return `SELECT COALESCE(SUM(OCTET_LENGTH(TransportConfiguration) + OCTET_LENGTH(OwnerReceipt)), 0)
  FROM _System_ConnectorCredentialGeneration
 WHERE ` + credentialGenerationExactSetPredicate(count)
}

func credentialGenerationExactSetLoadQuery(count int) string {
	return `SELECT ` + credentialGenerationExactSetColumns + `
  FROM _System_ConnectorCredentialGeneration
 WHERE ` + credentialGenerationExactSetPredicate(count) + `
 ORDER BY BINARY Id`
}

func credentialGenerationExactSetArguments(ids []string) []any {
	arguments := make([]any, 0, len(ids)*2)
	for _, id := range ids {
		arguments = append(arguments, id, id)
	}
	return arguments
}

func exactSetBudgetExceeded(format string, args ...any) error {
	return fmt.Errorf(
		"%w: %s",
		ErrExactSetBudgetExceeded,
		fmt.Sprintf(format, args...),
	)
}

// Publish inserts one complete immutable generation. Exact retries converge on
// the stored row; any same-ID byte difference fails closed. No path updates or
// repairs an existing generation.
func (repository *Repository) Publish(
	ctx context.Context,
	generation Generation,
) (Generation, bool, error) {
	if repository == nil || repository.db == nil {
		return Generation{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return Generation{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}

	generation = generation.copy()
	if err := generation.validateStored(); err != nil {
		return Generation{}, false, fmt.Errorf("%w: %v", ErrInvalidGeneration, err)
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Generation{}, false, err
	}
	if err := verifyDispatchGateSchema(ctx, repository.db); err != nil {
		return Generation{}, false, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return Generation{}, false, fmt.Errorf("begin credential generation 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 := insertGeneration(ctx, tx, generation); err != nil {
		if !isMySQLDuplicateKey(err) {
			return rollback(fmt.Errorf("insert credential generation: %w", err))
		}
		stored, found, loadErr := loadExact(ctx, tx, generation.id, true)
		if loadErr != nil {
			return rollback(loadErr)
		}
		if !found || !generation.equal(stored) {
			return rollback(fmt.Errorf(
				"%w: generation ID already names different immutable bytes",
				ErrIdentityConflict,
			))
		}
		// An exact duplicate from before the gate schema landed may establish
		// its absent enabled/fence-zero baseline. A valid existing gate is only
		// loaded and validated; publication retry never rewrites its mode/fence.
		if err := initializeDispatchGateIfAbsent(ctx, tx, generation.id); err != nil {
			return rollback(fmt.Errorf("initialize exact credential generation dispatch gate: %w", err))
		}
		if err := commit(); err != nil {
			return Generation{}, false, fmt.Errorf("commit exact credential generation duplicate: %w", err)
		}
		return stored.copy(), true, nil
	}
	if err := insertInitialDispatchGate(ctx, tx, generation.id); err != nil {
		return rollback(fmt.Errorf("insert credential generation dispatch gate: %w", err))
	}

	stored, found, err := loadExact(ctx, tx, generation.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found {
		return rollback(storedGenerationInvalid("inserted generation cannot be reloaded by exact ID"))
	}
	if !generation.equal(stored) {
		return rollback(storedGenerationInvalid("inserted generation bytes changed during persistence"))
	}
	gate, found, err := loadDispatchGateExact(ctx, tx, generation.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found || gate.mode != DispatchModeEnabled || gate.fenceToken != 0 {
		return rollback(storedDispatchGateInvalid("new generation dispatch gate is not enabled at fence zero"))
	}
	if err := commit(); err != nil {
		return Generation{}, false, fmt.Errorf("commit credential generation publish: %w", err)
	}
	return stored.copy(), false, nil
}

type credentialGenerationQueryer interface {
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

func loadExact(
	ctx context.Context,
	queryer credentialGenerationQueryer,
	id string,
	forUpdate bool,
) (Generation, bool, error) {
	query := credentialGenerationSelect
	if forUpdate {
		query += " FOR UPDATE"
	}

	var generation Generation
	err := queryer.QueryRowContext(ctx, query, id, id).Scan(
		&generation.id,
		&generation.stableCredentialIdentity,
		&generation.providerScopeID,
		&generation.connectorAccountID,
		&generation.secretBackendProtocol,
		&generation.secretBackendVersionRef,
		&generation.transportConfiguration,
		&generation.transportConfigurationDigest,
		&generation.receiptProtocolVersion,
		&generation.ownerReceipt,
		&generation.ownerReceiptDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return Generation{}, false, nil
	}
	if err != nil {
		return Generation{}, false, fmt.Errorf("load credential generation row: %w", err)
	}
	generation = generation.copy()
	if err := generation.validateStored(); err != nil {
		return Generation{}, false, err
	}
	return generation, true, nil
}

func insertGeneration(ctx context.Context, tx *sql.Tx, generation Generation) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorCredentialGeneration
       (Id, StableCredentialIdentity, ProviderScopeId, ConnectorAccountId,
        SecretBackendProtocol, SecretBackendVersionRef,
        TransportConfiguration, TransportConfigurationDigest,
        ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		generation.id,
		generation.stableCredentialIdentity,
		generation.providerScopeID,
		generation.connectorAccountID,
		generation.secretBackendProtocol,
		generation.secretBackendVersionRef,
		generation.transportConfiguration,
		generation.transportConfigurationDigest,
		string(generation.receiptProtocolVersion),
		generation.ownerReceipt,
		generation.ownerReceiptDigest,
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result)
}

func requireOneInsertedRow(result sql.Result) error {
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("credential generation rows affected: %w", err)
	}
	if rows != 1 {
		return fmt.Errorf("credential generation rows affected = %d, want 1", rows)
	}
	return nil
}

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