package serving

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

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

const preparedEpochSelect = `
SELECT Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
       OrgId, AppInstallationId, AuthorizedWorkloadId,
       CatalogRevision, SchemaGeneration, EngineGeneration,
       SourceSnapshotDigest, BindingViewDigest,
       ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
       L2ConfigurationGenerationRef, L2ConfigurationDigest,
       L2VerificationReceipt, L2VerificationReceiptDigest
  FROM _System_ConnectorServingEpoch
 WHERE Id = ? AND BINARY Id = BINARY ?
 LIMIT 1`

const preparedBindingSelect = `
SELECT Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
       Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
       BusinessKey, DefinitionVersion, CredentialRefs,
       RuntimeCommandKind, RuntimePayloadFormatVersion,
       RuntimeDefinitionRef, RuntimeDefinitionVersion
  FROM _System_ConnectorServingEpochBinding
 WHERE ServingEpochId = ? AND BINARY ServingEpochId = BINARY ?
 ORDER BY BindingOrdinal, Id`

// Repository persists inert ServingEpoch candidates in the catalog-owned
// schema. It deliberately exposes no scope or digest lookup: exact binary
// epoch ID is the sole read path.
type Repository struct {
	db *sql.DB
}

// NewRepository constructs a ServingEpoch 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 fails closed unless the complete catalog-owned ServingEpoch
// schema is present with its required binary identities and constraints.
func (r *Repository) VerifySchema(ctx context.Context) error {
	if r == nil || r.db == nil {
		return fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	return verifyServingSchema(ctx, r.db)
}

// LoadPrepared loads one immutable candidate by its exact binary epoch ID.
// It never falls back to scope, digest, current-pack, or Catalog definition
// reads.
func (r *Repository) LoadPrepared(ctx context.Context, id string) (PreparedEpoch, bool, error) {
	if r == nil || r.db == nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validOpaqueIdentifier(id) {
		return PreparedEpoch{}, false, invalidCandidate("epoch ID is not canonical")
	}
	if err := r.VerifySchema(ctx); err != nil {
		return PreparedEpoch{}, false, err
	}
	return loadPrepared(ctx, r.db, id, false)
}

// Prepare atomically inserts a complete immutable candidate. Exact retries
// converge on the stored candidate; every other collision fails closed and no
// path updates or repairs existing rows.
func (r *Repository) Prepare(ctx context.Context, candidate PreparedEpoch) (PreparedEpoch, bool, error) {
	if r == nil || r.db == nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}

	// Detach all byte/slice fields before validation and persistence so the
	// repository never retains caller-owned mutable backing storage.
	candidate = candidate.copy()
	if err := candidate.validateStored(); err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: %v", ErrInvalidCandidate, err)
	}
	if err := r.VerifySchema(ctx); err != nil {
		return PreparedEpoch{}, false, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, r.db)
	if err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("begin serving epoch prepare: %w", err)
	}
	rollback := func(cause error) (PreparedEpoch, bool, error) {
		return PreparedEpoch{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}
	commit := func() error {
		return errors.Join(tx.Commit(), cleanup())
	}

	if err := insertPreparedEpoch(ctx, tx, candidate); err != nil {
		if !isMySQLDuplicateKey(err) {
			return rollback(fmt.Errorf("insert serving epoch: %w", err))
		}
		stored, found, loadErr := loadPrepared(ctx, tx, candidate.id, true)
		if loadErr != nil {
			if errors.Is(loadErr, errStoredProfiledServingEpoch) {
				return rollback(fmt.Errorf(
					"%w: epoch ID already names a profiled candidate",
					ErrIdentityConflict,
				))
			}
			return rollback(loadErr)
		}
		if !found || !candidate.equal(stored) {
			return rollback(fmt.Errorf("%w: epoch ID or digest already names different immutable bytes", ErrIdentityConflict))
		}
		if err := commit(); err != nil {
			return PreparedEpoch{}, false, fmt.Errorf("commit exact serving epoch duplicate: %w", err)
		}
		return stored.copy(), true, nil
	}

	for i := range candidate.bindings {
		if err := insertPreparedBinding(ctx, tx, candidate.bindings[i]); err != nil {
			if isMySQLDuplicateKey(err) {
				err = fmt.Errorf("%w: binding %d identity already names different immutable bytes: %v", ErrIdentityConflict, i, err)
			} else {
				err = fmt.Errorf("insert serving epoch binding %d: %w", i, err)
			}
			return rollback(err)
		}
	}

	stored, found, err := loadPrepared(ctx, tx, candidate.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found {
		return rollback(storedCandidateInvalid("inserted candidate cannot be reloaded by exact ID"))
	}
	if !candidate.equal(stored) {
		return rollback(storedCandidateInvalid("inserted candidate bytes changed during persistence"))
	}
	if err := commit(); err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("commit serving epoch prepare: %w", err)
	}
	return stored.copy(), false, nil
}

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

func loadPrepared(
	ctx context.Context,
	queryer preparedEpochQueryer,
	id string,
	forUpdate bool,
) (PreparedEpoch, bool, error) {
	epoch, found, err := loadPreparedBase(ctx, queryer, id, forUpdate)
	if err != nil || !found {
		return PreparedEpoch{}, found, err
	}
	proofs, err := loadProfiledProofGraph(ctx, queryer, id, forUpdate)
	if err != nil {
		return PreparedEpoch{}, false, err
	}
	if proofs.headerFound || len(proofs.owners) != 0 || len(proofs.credentials) != 0 {
		if !proofs.headerFound {
			return PreparedEpoch{}, false, storedCandidateInvalid(
				"reproducibility proof rows exist without a header",
			)
		}
		if _, profileErr := profiledPreActivationFromStored(epoch, proofs, id); profileErr != nil {
			return PreparedEpoch{}, false, profileErr
		}
		return PreparedEpoch{}, false, errors.Join(
			storedCandidateInvalid("epoch is a profiled candidate, not a legacy candidate"),
			errStoredProfiledServingEpoch,
		)
	}
	if err := epoch.validateStored(); err != nil {
		return PreparedEpoch{}, false, err
	}
	return epoch.copy(), true, nil
}

// loadPreparedBase reads the storage shared by legacy and profiled candidates
// without assigning either semantic interpretation to the four legacy-named
// configuration columns. Callers must inspect the reproducibility proof graph
// before validating the result as exactly one candidate type.
func loadPreparedBase(
	ctx context.Context,
	queryer preparedEpochQueryer,
	id string,
	forUpdate bool,
) (PreparedEpoch, bool, error) {
	parentQuery := preparedEpochSelect
	bindingQuery := preparedBindingSelect
	if forUpdate {
		parentQuery += " FOR UPDATE"
		bindingQuery += " FOR UPDATE"
	}

	var (
		epoch                             PreparedEpoch
		activeScope                       sql.NullString
		catalogRevision, schemaGeneration uint64
		engineGeneration, securityEpoch   uint64
	)
	parentRow := queryer.QueryRowContext(ctx, parentQuery, id, id)
	if err := parentRow.Err(); err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("load serving epoch row: %w", err)
	}
	err := parentRow.Scan(
		&epoch.id,
		&epoch.epochDigest,
		&epoch.servingScopeDigest,
		&activeScope,
		&epoch.state,
		&epoch.catalog.orgID,
		&epoch.catalog.appInstallationID,
		&epoch.catalog.authorizedWorkloadID,
		&catalogRevision,
		&schemaGeneration,
		&engineGeneration,
		&epoch.catalog.lineage.SourceSnapshotDigest,
		&epoch.catalog.lineage.ViewDigest,
		&epoch.catalog.receipts.ProtocolVersion,
		&epoch.catalog.receipts.CapabilityVersion,
		&securityEpoch,
		&epoch.l2.configurationGenerationRef,
		&epoch.l2.configurationDigest,
		&epoch.l2.verificationReceipt,
		&epoch.l2.verificationReceiptDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return PreparedEpoch{}, false, nil
	}
	if err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("%w: scan epoch row: %w", ErrStoredCandidateInvalid, err)
	}
	if activeScope.Valid {
		epoch.activeScopeDigest = stringPointer(activeScope.String)
	}
	epoch.catalog.lineage.CatalogRevision = catalogRevision
	epoch.catalog.lineage.SchemaGeneration = schemaGeneration
	epoch.catalog.lineage.EngineGeneration = engineGeneration
	// The schema normalizes the receipt's repeated revision tuple into the
	// lineage columns. Reconstruct it exactly before canonical validation.
	epoch.catalog.receipts.CatalogRevision = catalogRevision
	epoch.catalog.receipts.SchemaGeneration = schemaGeneration
	epoch.catalog.receipts.EngineGeneration = engineGeneration
	epoch.catalog.receipts.SecurityEpoch = securityEpoch
	if len(epoch.l2.verificationReceipt) > maxPersistedL2VerificationReceiptBytes {
		return PreparedEpoch{}, false, storedCandidateInvalid(
			"persisted L2 or ServiceConfiguration receipt exceeds the shared storage admission bound",
		)
	}
	epoch.l2.verificationReceipt = append([]byte(nil), epoch.l2.verificationReceipt...)

	rows, err := queryer.QueryContext(ctx, bindingQuery, id, id)
	if err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("load serving epoch bindings: %w", err)
	}
	defer rows.Close()

	epoch.bindings = make([]storedBinding, 0)
	for rows.Next() {
		if len(epoch.bindings) >= catalogv1.MaxBindingCount {
			return PreparedEpoch{}, false, storedCandidateInvalid("binding count exceeds Catalog maximum")
		}
		var (
			binding              storedBinding
			kind                 string
			runtimeCommandKind   sql.NullString
			runtimePayloadFormat sql.NullInt64
			runtimeDefinitionRef sql.NullString
			runtimeDefinitionVer sql.NullString
		)
		if err := rows.Scan(
			&binding.id,
			&binding.servingEpochID,
			&binding.ordinal,
			&binding.identityDigest,
			&kind,
			&binding.ownerPack,
			&binding.packInstallationID,
			&binding.packPayloadDigest,
			&binding.businessKey,
			&binding.definitionVersion,
			&binding.credentialRefs,
			&runtimeCommandKind,
			&runtimePayloadFormat,
			&runtimeDefinitionRef,
			&runtimeDefinitionVer,
		); err != nil {
			return PreparedEpoch{}, false, fmt.Errorf("%w: scan binding row: %w", ErrStoredCandidateInvalid, err)
		}
		binding.kind = catalogv1.IntegrationBindingKind(kind)
		binding.credentialRefs = append([]byte(nil), binding.credentialRefs...)
		if runtimeCommandKind.Valid {
			binding.runtimeCommandKind = stringPointer(runtimeCommandKind.String)
		}
		if runtimePayloadFormat.Valid {
			if runtimePayloadFormat.Int64 < 0 || uint64(runtimePayloadFormat.Int64) > math.MaxUint32 {
				return PreparedEpoch{}, false, storedCandidateInvalid("runtime payload format version exceeds uint32")
			}
			binding.runtimePayloadFormatVersion = uint32Pointer(uint32(runtimePayloadFormat.Int64))
		}
		if runtimeDefinitionRef.Valid {
			binding.runtimeDefinitionRef = stringPointer(runtimeDefinitionRef.String)
		}
		if runtimeDefinitionVer.Valid {
			binding.runtimeDefinitionVersion = stringPointer(runtimeDefinitionVer.String)
		}
		epoch.bindings = append(epoch.bindings, binding)
	}
	if err := rows.Err(); err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("load serving epoch bindings: %w", err)
	}
	if err := rows.Close(); err != nil {
		return PreparedEpoch{}, false, fmt.Errorf("close serving epoch binding rows: %w", err)
	}
	return epoch, true, nil
}

func insertPreparedEpoch(ctx context.Context, tx *sql.Tx, epoch PreparedEpoch) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServingEpoch
       (Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
        OrgId, AppInstallationId, AuthorizedWorkloadId,
        CatalogRevision, SchemaGeneration, EngineGeneration,
        SourceSnapshotDigest, BindingViewDigest,
        ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
        L2ConfigurationGenerationRef, L2ConfigurationDigest,
        L2VerificationReceipt, L2VerificationReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		epoch.id,
		epoch.epochDigest,
		epoch.servingScopeDigest,
		nil,
		epoch.state,
		epoch.catalog.orgID,
		epoch.catalog.appInstallationID,
		epoch.catalog.authorizedWorkloadID,
		epoch.catalog.lineage.CatalogRevision,
		epoch.catalog.lineage.SchemaGeneration,
		epoch.catalog.lineage.EngineGeneration,
		epoch.catalog.lineage.SourceSnapshotDigest,
		epoch.catalog.lineage.ViewDigest,
		epoch.catalog.receipts.ProtocolVersion,
		epoch.catalog.receipts.CapabilityVersion,
		epoch.catalog.receipts.SecurityEpoch,
		epoch.l2.configurationGenerationRef,
		epoch.l2.configurationDigest,
		epoch.l2.verificationReceipt,
		epoch.l2.verificationReceiptDigest,
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "serving epoch")
}

func insertPreparedBinding(ctx context.Context, tx *sql.Tx, binding storedBinding) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServingEpochBinding
       (Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
        Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
        BusinessKey, DefinitionVersion, CredentialRefs,
        RuntimeCommandKind, RuntimePayloadFormatVersion,
        RuntimeDefinitionRef, RuntimeDefinitionVersion)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		binding.id,
		binding.servingEpochID,
		binding.ordinal,
		binding.identityDigest,
		string(binding.kind),
		binding.ownerPack,
		binding.packInstallationID,
		binding.packPayloadDigest,
		binding.businessKey,
		binding.definitionVersion,
		binding.credentialRefs,
		nullableString(binding.runtimeCommandKind),
		nullableUint32(binding.runtimePayloadFormatVersion),
		nullableString(binding.runtimeDefinitionRef),
		nullableString(binding.runtimeDefinitionVersion),
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "serving epoch binding")
}

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

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

func nullableString(value *string) any {
	if value == nil {
		return nil
	}
	return *value
}

func nullableUint32(value *uint32) any {
	if value == nil {
		return nil
	}
	return int64(*value)
}