package serving

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

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

func (repository *IngressReceiptRepository) PublishTransport(
	ctx context.Context,
	candidate TransportAdmission,
) (TransportAdmission, bool, error) {
	if repository == nil || repository.db == nil {
		return TransportAdmission{}, false, fmt.Errorf("%w: database is nil", ErrIngressReceiptDependencyMissing)
	}
	if ctx == nil {
		return TransportAdmission{}, false, fmt.Errorf("%w: context is nil", ErrIngressReceiptDependencyMissing)
	}
	candidate = candidate.copy()
	if candidate.key.disposition != ReceiptAdmissionDispositionV2Receipt || candidate.receipt == nil {
		return TransportAdmission{}, false, invalidIngressReceipt("only complete v2_receipt transport admission can be published")
	}
	if err := candidate.validateStored(); err != nil {
		return TransportAdmission{}, false, fmt.Errorf("%w: %v", ErrInvalidIngressReceipt, err)
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return TransportAdmission{}, false, err
	}
	// Immutable duplicates must remain resolvable after the account gate has
	// halted or advanced its fence. This source-free preflight uses the same
	// exact graph reader and cannot grant a new admission: only a complete
	// retained authority for the same transport identity and payload digest may
	// return here, even when its scope, epoch, or protocol is older.
	stored, found, err := loadTransportAdmissionByCandidate(ctx, repository.db, candidate, false)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if found {
		if !stored.authorizesTransportRequest(candidate) {
			return TransportAdmission{}, false, fmt.Errorf(
				"%w: transport identity already authorizes a different payload digest",
				ErrIngressReceiptIdentityConflict,
			)
		}
		return stored.copy(), true, nil
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return TransportAdmission{}, false, fmt.Errorf("begin ingress receipt publication: %w", err)
	}
	rollback := func(cause error) (TransportAdmission, bool, error) {
		return TransportAdmission{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}
	resolveDuplicate := func(duplicateErr error) (TransportAdmission, bool, error) {
		rollbackErr := errors.Join(tx.Rollback(), cleanup())
		if rollbackErr != nil {
			return TransportAdmission{}, false, fmt.Errorf(
				"rollback ingress receipt collision: %w",
				errors.Join(duplicateErr, rollbackErr),
			)
		}
		return repository.resolveTransportDuplicate(ctx, candidate)
	}

	receipt := *candidate.receipt
	gate, err := lockExpectedV2Serving(
		ctx,
		tx,
		receipt.servingScopeDigest,
		receipt.connectorAccountID,
		receipt.accountServingFenceToken,
		receipt.servingEpochID,
		receipt.servingEpochDigest,
	)
	if err != nil {
		admissionErr := classifyIngressAdmissionError(err)
		rollbackErr := errors.Join(tx.Rollback(), cleanup())
		if rollbackErr != nil {
			return TransportAdmission{}, false, errors.Join(admissionErr, rollbackErr)
		}
		if errors.Is(admissionErr, ErrIngressReceiptAdmissionConflict) {
			// A request can race behind a transaction that committed the exact
			// receipt and a later halt. Re-read only immutable authority after
			// releasing the gate transaction; never treat this as a new admission.
			stored, found, loadErr := loadTransportAdmissionByCandidate(ctx, repository.db, candidate, false)
			if loadErr != nil {
				return TransportAdmission{}, false, loadErr
			}
			if found && stored.authorizesTransportRequest(candidate) {
				return stored.copy(), true, nil
			}
		}
		return TransportAdmission{}, false, admissionErr
	}
	if gate.id != receipt.accountServingGateID || gate.id != candidate.key.accountServingGateID {
		return rollback(fmt.Errorf(
			"%w: locked account gate ID does not match the immutable candidate",
			ErrIngressReceiptAdmissionConflict,
		))
	}
	// A Receipt or v2_receipt AdmissionKey missing its exact counterpart is an
	// incomplete immutable authority graph. Before minting any new authority for
	// the account, fail closed on either direction. The locking form serializes
	// this check with direct row mutation while the account gate and epoch are
	// locked for publication.
	if err := rejectIncompleteTransportAuthorityForAccount(
		ctx,
		tx,
		candidate.key.identity.connectorAccountID,
		true,
	); err != nil {
		return rollback(err)
	}

	if err := insertV2Receipt(ctx, tx, receipt); err != nil {
		if isMySQLDuplicateKey(err) {
			return resolveDuplicate(err)
		}
		return rollback(fmt.Errorf("insert immutable V2Receipt: %w", err))
	}
	if err := insertReceiptAdmissionKey(ctx, tx, candidate.key); err != nil {
		if isMySQLDuplicateKey(err) {
			return resolveDuplicate(err)
		}
		return rollback(fmt.Errorf("insert immutable ReceiptAdmissionKey: %w", err))
	}

	stored, found, err = loadTransportAdmissionByCandidate(ctx, tx, candidate, true)
	if err != nil {
		return rollback(err)
	}
	if !found || !candidate.equal(stored) {
		return rollback(storedIngressReceiptInvalid("inserted authority cannot be read back byte-exact"))
	}
	if err := errors.Join(tx.Commit(), cleanup()); err != nil {
		return TransportAdmission{}, false, fmt.Errorf("commit ingress receipt publication: %w", err)
	}
	return stored.copy(), false, nil
}

func (repository *IngressReceiptRepository) resolveTransportDuplicate(
	ctx context.Context,
	candidate TransportAdmission,
) (TransportAdmission, bool, error) {
	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return TransportAdmission{}, false, fmt.Errorf("begin ingress receipt duplicate resolution: %w", err)
	}
	rollback := func(cause error) (TransportAdmission, bool, error) {
		return TransportAdmission{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}

	stored, found, err := loadTransportAdmissionByCandidate(ctx, tx, candidate, true)
	if err != nil {
		return rollback(err)
	}
	if !found || !stored.authorizesTransportRequest(candidate) {
		return rollback(fmt.Errorf(
			"%w: transport identity is absent or authorizes a different payload digest",
			ErrIngressReceiptIdentityConflict,
		))
	}
	if err := errors.Join(tx.Commit(), cleanup()); err != nil {
		return TransportAdmission{}, false, fmt.Errorf("commit exact ingress receipt duplicate: %w", err)
	}
	return stored.copy(), true, nil
}

func insertV2Receipt(ctx context.Context, tx *sql.Tx, receipt V2Receipt) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorV2Receipt
       (Id, ReceiptProtocolVersion, ServingScopeDigest, ConnectorAccountId,
        AccountServingGateId, AccountServingFenceToken,
        ServingEpochId, ServingEpochDigest, AdmissionKind,
        AdmissionAuthorityId, PayloadCiphertextProtocolVersion,
        PayloadCiphertext, PayloadDigest, PayloadSizeBytes,
        ReceiptComponentCount)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		receipt.id,
		string(receipt.receiptProtocolVersion),
		receipt.servingScopeDigest,
		receipt.connectorAccountID,
		receipt.accountServingGateID,
		receipt.accountServingFenceToken,
		receipt.servingEpochID,
		receipt.servingEpochDigest,
		string(receipt.admissionKind),
		receipt.admissionAuthorityID,
		string(receipt.payloadCiphertextProtocolVersion),
		receipt.payloadCiphertext,
		receipt.payloadDigest,
		receipt.payloadSizeBytes,
		receipt.receiptComponentCount,
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "immutable V2Receipt")
}

func insertReceiptAdmissionKey(ctx context.Context, tx *sql.Tx, key receiptAdmissionKey) error {
	result, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorV2ReceiptAdmissionKey
       (Id, AdmissionKeyProtocolVersion, ServingScopeDigest,
        ConnectorAccountId, AccountServingGateId, IngressChannel,
        TransportIdempotencyKey, AuthorizedPayloadDigest, Disposition,
        ReceiptId, PredecessorRefProtocolVersion, PredecessorRef,
        PredecessorRefDigest)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		key.id,
		string(key.protocolVersion),
		key.servingScopeDigest,
		key.identity.connectorAccountID,
		key.accountServingGateID,
		string(key.identity.ingressChannel),
		key.identity.transportIdempotencyKey,
		key.authorizedPayloadDigest,
		string(key.disposition),
		nullableString(key.receiptID),
		nullablePredecessorProtocol(key.predecessorProtocolVersion),
		nullableBytes(key.predecessorRef),
		nullableString(key.predecessorRefDigest),
	)
	if err != nil {
		return err
	}
	return requireOneInsertedRow(result, "immutable ReceiptAdmissionKey")
}

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

func nullableBytes(value []byte) any {
	if value == nil {
		return nil
	}
	return value
}

func classifyIngressAdmissionError(err error) error {
	if errors.Is(err, errAccountServingGateNotFound) ||
		errors.Is(err, errAccountServingGateFenceConflict) ||
		errors.Is(err, errAccountServingGateModeConflict) ||
		errors.Is(err, errAccountServingGateEpochConflict) {
		return errors.Join(ErrIngressReceiptAdmissionConflict, err)
	}
	return err
}