package serving

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

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

// PrepareProfiled atomically persists one complete immutable V2 candidate and
// its complete reproducibility proof graph. Exact retries converge only after
// the failed write transaction has been rolled back and the stored graph has
// been lock-read in a new pessimistic transaction. No path updates, completes,
// or repairs an existing graph.
func (r *Repository) PrepareProfiled(
	ctx context.Context,
	candidate ProfiledPreparedEpoch,
) (ProfiledPreparedEpoch, bool, error) {
	if r == nil || r.db == nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}

	// Detach every caller-owned byte slice before validation or SQL. The same
	// immutable copy is used for persistence, duplicate comparison, and return.
	candidate = candidate.copy()
	if err := candidate.validateStored(); err != nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("%w: %v", ErrInvalidCandidate, err)
	}
	if err := r.VerifySchema(ctx); err != nil {
		return ProfiledPreparedEpoch{}, false, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, r.db)
	if err != nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("begin profiled serving epoch prepare: %w", err)
	}
	rollback := func(cause error) (ProfiledPreparedEpoch, bool, error) {
		return ProfiledPreparedEpoch{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}
	resolveParentDuplicate := func(duplicateErr error) (ProfiledPreparedEpoch, bool, error) {
		rollbackErr := errors.Join(tx.Rollback(), cleanup())
		if rollbackErr != nil {
			return ProfiledPreparedEpoch{}, false, fmt.Errorf(
				"rollback profiled serving epoch after parent identity collision: %w",
				errors.Join(duplicateErr, rollbackErr),
			)
		}
		return r.resolveProfiledDuplicate(ctx, candidate)
	}

	base := profiledPreparedBase(candidate)
	if err := insertPreparedEpoch(ctx, tx, base); err != nil {
		if isMySQLDuplicateKey(err) {
			return resolveParentDuplicate(err)
		}
		return rollback(fmt.Errorf("insert profiled serving epoch: %w", err))
	}
	if err := insertProfiledServingEpochHeader(ctx, tx, candidate); err != nil {
		return rollback(classifyProfiledChildInsertError(err, "reproducibility header"))
	}
	for index := range base.bindings {
		if err := insertPreparedBinding(ctx, tx, base.bindings[index]); err != nil {
			coordinate := fmt.Sprintf("binding %d", index)
			return rollback(classifyProfiledChildInsertError(err, coordinate))
		}
	}
	for index := range candidate.bindings {
		if err := insertProfiledBindingOwner(ctx, tx, candidate.bindings[index].owner); err != nil {
			coordinate := fmt.Sprintf("binding-owner proof %d", index)
			return rollback(classifyProfiledChildInsertError(err, coordinate))
		}
	}
	for index := range candidate.credentials {
		if err := insertProfiledCredential(ctx, tx, candidate.id, candidate.credentials[index]); err != nil {
			coordinate := fmt.Sprintf("credential-generation proof %d", index)
			return rollback(classifyProfiledChildInsertError(err, coordinate))
		}
	}

	stored, found, err := loadProfiledPrepared(ctx, tx, candidate.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found {
		return rollback(storedCandidateInvalid("inserted profiled candidate cannot be lock-read by exact ID"))
	}
	if !candidate.equal(stored) {
		return rollback(storedCandidateInvalid("inserted profiled candidate bytes changed during persistence"))
	}
	if err := errors.Join(tx.Commit(), cleanup()); err != nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("commit profiled serving epoch prepare: %w", err)
	}
	return stored.copy(), false, nil
}

// resolveProfiledDuplicate lock-reads only after the colliding write
// transaction has been fully rolled back. This prevents a partial graph in the
// failed transaction from being mistaken for a durable exact duplicate.
func (r *Repository) resolveProfiledDuplicate(
	ctx context.Context,
	candidate ProfiledPreparedEpoch,
) (ProfiledPreparedEpoch, bool, error) {
	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, r.db)
	if err != nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf(
			"begin profiled serving epoch duplicate resolution: %w",
			err,
		)
	}
	rollback := func(cause error) (ProfiledPreparedEpoch, bool, error) {
		return ProfiledPreparedEpoch{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}

	stored, found, loadErr := loadProfiledPrepared(ctx, tx, candidate.id, true)
	if loadErr != nil {
		if errors.Is(loadErr, errStoredLegacyServingEpoch) {
			return rollback(fmt.Errorf(
				"%w: epoch identity collides with a legacy-unprofiled candidate",
				ErrIdentityConflict,
			))
		}
		if errors.Is(loadErr, errStoredDifferentProfileServingEpoch) {
			return rollback(errors.Join(
				fmt.Errorf("%w: epoch identity uses a different digest profile", ErrIdentityConflict),
				loadErr,
			))
		}
		if errors.Is(loadErr, errStoredTerminalProfiledServingEpoch) {
			return rollback(loadErr)
		}
		return rollback(loadErr)
	}
	if !found || !candidate.equal(stored) {
		return rollback(fmt.Errorf(
			"%w: epoch identity already names different immutable bytes",
			ErrIdentityConflict,
		))
	}
	if err := errors.Join(tx.Commit(), cleanup()); err != nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf(
			"commit exact profiled serving epoch duplicate: %w",
			err,
		)
	}
	return stored.copy(), true, nil
}

func classifyProfiledChildInsertError(err error, coordinate string) error {
	if isMySQLDuplicateKey(err) {
		return fmt.Errorf(
			"%w: %s already names different immutable bytes: %v",
			ErrIdentityConflict,
			coordinate,
			err,
		)
	}
	return fmt.Errorf("insert profiled serving epoch %s: %w", coordinate, err)
}

func profiledPreparedBase(epoch ProfiledPreparedEpoch) PreparedEpoch {
	bindings := make([]storedBinding, len(epoch.bindings))
	for index := range epoch.bindings {
		bindings[index] = epoch.bindings[index].binding.copy()
	}
	return PreparedEpoch{
		id:                 epoch.id,
		epochDigest:        epoch.epochDigest,
		servingScopeDigest: epoch.servingScopeDigest,
		state:              epoch.state,
		activeScopeDigest:  copyOptionalString(epoch.activeScopeDigest),
		catalog:            epoch.catalog,
		l2: L2Snapshot{
			configurationGenerationRef: epoch.serviceConfiguration.generationID,
			configurationDigest:        epoch.serviceConfiguration.generationDigest,
			verificationReceipt:        append([]byte(nil), epoch.serviceConfiguration.ownerReceipt...),
			verificationReceiptDigest:  epoch.serviceConfiguration.ownerReceiptDigest,
		},
		bindings: bindings,
	}
}

func insertProfiledServingEpochHeader(
	ctx context.Context,
	tx *sql.Tx,
	epoch ProfiledPreparedEpoch,
) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServingEpochReproducibility
       (ServingEpochId, ConnectorDigestProfileId,
        ServiceConfigurationGenerationProtocolVersion,
        ServiceConfigurationReceiptProtocolVersion)
VALUES (?, ?, ?, ?)`,
		epoch.id,
		string(epoch.profileID),
		string(epoch.serviceConfiguration.generationProtocolVersion),
		string(epoch.serviceConfiguration.receiptProtocolVersion),
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "serving epoch reproducibility header")
}

func insertProfiledBindingOwner(
	ctx context.Context,
	tx *sql.Tx,
	owner storedBindingOwnerProof,
) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServingEpochBindingOwner
       (ServingEpochBindingId, PackGenerationId,
        DefinitionEnvelopeVersion, DefinitionReceiptVersion,
        CanonicalDefinitionDigest, DefinitionOwnerReceipt,
        DefinitionOwnerReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
		owner.servingEpochBindingID,
		owner.packGenerationID,
		owner.definitionEnvelopeVersion,
		owner.definitionReceiptVersion,
		owner.canonicalDefinitionDigest,
		owner.definitionOwnerReceipt,
		owner.definitionOwnerReceiptDigest,
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "serving epoch binding-owner proof")
}

func insertProfiledCredential(
	ctx context.Context,
	tx *sql.Tx,
	epochID string,
	proof storedCredentialGenerationProof,
) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorServingEpochCredentialGeneration
       (ServingEpochId, CredentialOrdinal, CredentialGenerationId,
        ReceiptProtocolVersion, TransportConfigurationDigest,
        OwnerReceipt, OwnerReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
		epochID,
		proof.ordinal,
		proof.generationID,
		string(proof.receiptProtocolVersion),
		proof.transportConfigurationDigest,
		proof.ownerReceipt,
		proof.ownerReceiptDigest,
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "serving epoch credential-generation proof")
}