package catalogpublication

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

	"github.com/go-sql-driver/mysql"
	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
)

// Begin durably reserves one publication command and its complete immutable
// candidate. An exact retry returns the existing command without inspecting
// the current active pointer. Reusing CommandID for a changed request fails.
func (a *Authority) Begin(ctx context.Context, request BeginRequest) (Command, error) {
	candidateDigest, requestDigest, err := validateBeginRequest(request)
	if err != nil {
		return Command{}, err
	}
	wanted := commandFromRequest(request, candidateDigest, requestDigest)

	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Command{}, err
	}
	defer rollbackTx(tx)

	existing, err := readCommandDB(ctx, tx, request.Candidate.OrgID, request.CommandID, true)
	if err == nil {
		if err := rejectCandidateSourceJournalLocked(
			ctx, tx, request.Candidate.OrgID, request.CommandID, true,
		); err != nil {
			return Command{}, err
		}
		return commitExistingCommand(tx, existing, wanted)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return Command{}, err
	}
	if err := rejectCandidateSourceJournalLocked(
		ctx, tx, request.Candidate.OrgID, request.CommandID, false,
	); err != nil {
		return Command{}, err
	}

	writer, err := catalogwriter.LockOrCreate(ctx, tx, request.Candidate.OrgID, request.RuntimeAudience)
	if err != nil {
		return Command{}, mapWriterError(err)
	}
	// The first absent-row check can precede a competing source command that
	// reserves and releases the writer while Begin waits. Re-check under the
	// acquired writer fence before creating any publication rows.
	if err := rejectCandidateSourceJournalLocked(
		ctx, tx, request.Candidate.OrgID, request.CommandID, false,
	); err != nil {
		return Command{}, err
	}
	if writer.ActiveCommandID != "" {
		if writer.ActiveCommandID != request.CommandID {
			return Command{}, ErrConflict
		}
		// A concurrent exact Begin may have committed while this transaction
		// waited for the org writer. Re-read its journal under the same fence.
		existing, err = readCommandDB(ctx, tx, request.Candidate.OrgID, request.CommandID, true)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return Command{}, fmt.Errorf("%w: writer fence has no publication command", ErrCorrupt)
			}
			return Command{}, err
		}
		return commitExistingCommand(tx, existing, wanted)
	}

	state, err := lockOrCreateActiveState(ctx, tx, request.Candidate.OrgID)
	if err != nil {
		return Command{}, err
	}
	activeBinding, err := lockedActiveBinding(ctx, tx, request.Candidate.OrgID, state)
	if err != nil {
		return Command{}, err
	}
	if err := validateWriterCoherence(writer, activeBinding); err != nil {
		return Command{}, err
	}

	if request.ExpectedPrior == state.Active && state.Active.Digest != candidateDigest {
		if err := validateSelectionTransition(
			writer.CatalogRevision,
			writer.SecurityEpoch,
			activeBinding,
			request.Candidate,
		); err != nil {
			return Command{}, fmt.Errorf("%w: invalid publication transition: %v", ErrConflict, err)
		}
	}
	if err := persistPublication(ctx, tx, Publication{Digest: candidateDigest, Binding: request.Candidate}); err != nil {
		return Command{}, err
	}
	if err := insertCommand(ctx, tx, wanted); err != nil {
		return Command{}, err
	}
	if err := insertCandidate(ctx, tx, wanted); err != nil {
		return Command{}, err
	}
	if err := catalogwriter.Reserve(ctx, tx, writer, request.CommandID); err != nil {
		return Command{}, mapWriterError(err)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return wanted, nil
}

// Commit evaluates the command's exact expected-prior CAS and atomically
// persists its immutable outcome, org history entry, pointer decision, and
// writer-fence release. A completed retry returns its stored outcome before
// reading the current pointer, which may already have advanced again.
func (a *Authority) Commit(ctx context.Context, orgID, commandID string) (Outcome, error) {
	if err := validateOpaque("OrgID", orgID, 255); err != nil {
		return Outcome{}, err
	}
	if !validCommandID(commandID) {
		return Outcome{}, fmt.Errorf("catalog publication: CommandID must be a lowercase UUIDv7")
	}

	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Outcome{}, err
	}
	defer rollbackTx(tx)

	command, err := readCommandDB(ctx, tx, orgID, commandID, true)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Outcome{}, ErrNotFound
		}
		return Outcome{}, err
	}
	if err := rejectCandidateSourceJournalLocked(
		ctx, tx, orgID, commandID, true,
	); err != nil {
		return Outcome{}, err
	}
	if command.State == CommandCompleted {
		outcome, err := readOutcomeDB(ctx, tx, orgID, commandID, true)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return Outcome{}, fmt.Errorf("%w: completed command has no outcome", ErrCorrupt)
			}
			return Outcome{}, err
		}
		if outcome.CandidatePublicationDigest != command.CandidatePublicationDigest {
			return Outcome{}, fmt.Errorf("%w: completed command and outcome candidates differ", ErrCorrupt)
		}
		if err := tx.Commit(); err != nil {
			return Outcome{}, err
		}
		return outcome, nil
	}
	if command.State != CommandReserved {
		return Outcome{}, fmt.Errorf("%w: unknown command state %q", ErrCorrupt, command.State)
	}

	writer, err := catalogwriter.Lock(ctx, tx, orgID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Outcome{}, fmt.Errorf("%w: reserved command has no writer row", ErrCorrupt)
		}
		return Outcome{}, err
	}
	if writer.RuntimeAudience != command.RuntimeAudience || writer.ActiveCommandID != commandID {
		return Outcome{}, ErrConflict
	}
	state, err := readActiveStateDB(ctx, tx, orgID, true)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Outcome{}, fmt.Errorf("%w: reserved command has no publication state", ErrCorrupt)
		}
		return Outcome{}, err
	}
	candidateDigest, err := readCandidate(ctx, tx, orgID, commandID)
	if err != nil {
		return Outcome{}, err
	}
	if candidateDigest != command.CandidatePublicationDigest {
		return Outcome{}, fmt.Errorf("%w: command and candidate rows disagree", ErrCorrupt)
	}
	candidate, err := readPublicationDB(ctx, tx, orgID, candidateDigest, true)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Outcome{}, fmt.Errorf("%w: candidate publication is missing", ErrCorrupt)
		}
		return Outcome{}, err
	}
	activeBinding, err := lockedActiveBinding(ctx, tx, orgID, state)
	if err != nil {
		return Outcome{}, err
	}
	if err := validateWriterCoherence(writer, activeBinding); err != nil {
		return Outcome{}, err
	}

	outcome := Outcome{
		OrgID:                      orgID,
		CommandID:                  commandID,
		CandidatePublicationDigest: candidateDigest,
		ObservedPrior:              state.Active,
		ResultingActive:            state.Active,
	}
	switch {
	case command.ExpectedPrior != state.Active:
		outcome.Result = ResultExpectedPriorMismatch
	case state.Active.Present && state.Active.Digest == candidateDigest:
		outcome.Result = ResultUnchanged
	default:
		if err := validateSelectionTransition(
			writer.CatalogRevision,
			writer.SecurityEpoch,
			activeBinding,
			candidate.Binding,
		); err != nil {
			return Outcome{}, fmt.Errorf("%w: reserved transition became invalid: %v", ErrCorrupt, err)
		}
		outcome.Result = ResultSelected
		outcome.ResultingActive = Reference{Present: true, Digest: candidateDigest}
	}
	outcome.OutcomeDigest, err = canonicalOutcomeDigest(outcome)
	if err != nil {
		return Outcome{}, fmt.Errorf("%w: cannot identify terminal outcome: %v", ErrCorrupt, err)
	}
	if state.HistorySequence == math.MaxUint64 {
		return Outcome{}, fmt.Errorf("%w: publication history sequence exhausted", ErrCorrupt)
	}
	sequence := state.HistorySequence + 1

	if outcome.Result == ResultSelected {
		if err := catalogwriter.Commit(ctx, tx, orgID, commandID, candidate.Binding.CatalogRevision, candidate.Binding.SecurityEpoch); err != nil {
			return Outcome{}, mapWriterError(err)
		}
	} else if err := catalogwriter.Release(ctx, tx, orgID, commandID); err != nil {
		return Outcome{}, mapWriterError(err)
	}
	if err := updateActiveState(ctx, tx, orgID, state, sequence, outcome); err != nil {
		return Outcome{}, err
	}
	if err := insertOutcome(ctx, tx, outcome); err != nil {
		return Outcome{}, err
	}
	if err := insertHistory(ctx, tx, sequence, outcome); err != nil {
		return Outcome{}, err
	}
	if err := completeCommand(ctx, tx, command); err != nil {
		return Outcome{}, err
	}
	if err := deleteCandidate(ctx, tx, orgID, commandID, candidateDigest); err != nil {
		return Outcome{}, err
	}
	if err := tx.Commit(); err != nil {
		return Outcome{}, err
	}
	return outcome, nil
}

func rejectCandidateSourceJournalLocked(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	asCorruption bool,
) error {
	var storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT WriterCommandId
FROM _System_CatalogCandidateSource
WHERE OrgId = ? AND WriterCommandId = ? FOR UPDATE`, orgID, commandID).Scan(
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	var mysqlError *mysql.MySQLError
	if errors.As(err, &mysqlError) && mysqlError.Number == 1146 {
		// During rolling schema deployment the source journal table may not
		// exist yet. It cannot contain a conflicting source command in that
		// state, and source Prepare performs the symmetric publication check
		// once the table is provisioned.
		return nil
	}
	if err != nil {
		return err
	}
	if asCorruption {
		return fmt.Errorf(
			"%w: publication command collides with a candidate-source journal",
			ErrCorrupt,
		)
	}
	return ErrConflict
}

func commandFromRequest(request BeginRequest, candidateDigest, requestDigest string) Command {
	return Command{
		OrgID:                      request.Candidate.OrgID,
		CommandID:                  request.CommandID,
		RequestDigest:              requestDigest,
		RuntimeAudience:            request.RuntimeAudience,
		CandidatePublicationDigest: candidateDigest,
		ExpectedPrior:              request.ExpectedPrior,
		State:                      CommandReserved,
	}
}

func commitExistingCommand(tx *sql.Tx, existing, wanted Command) (Command, error) {
	if existing.OrgID != wanted.OrgID ||
		existing.CommandID != wanted.CommandID ||
		existing.RequestDigest != wanted.RequestDigest ||
		existing.RuntimeAudience != wanted.RuntimeAudience ||
		existing.CandidatePublicationDigest != wanted.CandidatePublicationDigest ||
		existing.ExpectedPrior != wanted.ExpectedPrior {
		return Command{}, ErrConflict
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return existing, nil
}

func lockOrCreateActiveState(ctx context.Context, tx *sql.Tx, orgID string) (activeState, error) {
	state, err := readActiveStateDB(ctx, tx, orgID, true)
	if err == nil {
		return state, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return activeState{}, err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublicationActive
(OrgId, ActivePublicationPresent, ActivePublicationDigest, SelectedByCommandId, HistorySequence, CreatedDate, LastModifiedDate)
VALUES (?, FALSE, NULL, NULL, 0, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`, orgID)
	if err != nil {
		return activeState{}, err
	}
	return activeState{}, nil
}

func lockedActiveBinding(ctx context.Context, tx *sql.Tx, orgID string, state activeState) (*catalogpublicationv1.Binding, error) {
	if !state.Active.Present {
		return nil, nil
	}
	publication, err := readPublicationDB(ctx, tx, orgID, state.Active.Digest, true)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, fmt.Errorf("%w: active publication is missing", ErrCorrupt)
		}
		return nil, err
	}
	return &publication.Binding, nil
}

func validateWriterCoherence(writer catalogwriter.State, active *catalogpublicationv1.Binding) error {
	if active == nil {
		if writer.CatalogRevision != 0 || writer.SecurityEpoch != 0 {
			return fmt.Errorf("%w: org lineage exists without a backfilled publication", ErrNotReady)
		}
		return nil
	}
	if writer.CatalogRevision != active.CatalogRevision || writer.SecurityEpoch != active.SecurityEpoch {
		return fmt.Errorf("%w: writer lineage and active publication differ", ErrCorrupt)
	}
	return nil
}

func persistPublication(ctx context.Context, tx *sql.Tx, publication Publication) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublication
(PublicationDigest, BindingVersion, OrgId, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		publication.Digest,
		legacyPublicationBindingVersion,
		publication.Binding.OrgID,
		decimalText(publication.Binding.CatalogRevision),
		publication.Binding.CatalogSnapshotDigest,
		decimalText(publication.Binding.SchemaGeneration),
		decimalText(publication.Binding.EngineGeneration),
		publication.Binding.LayoutValidationViewDigest,
		publication.Binding.LayoutActivationSetDigest,
		decimalText(publication.Binding.SecurityEpoch),
		publication.Binding.LayoutRoutingSecurityViewDigest,
	)
	if err == nil {
		return requireOne(result)
	}
	stored, readErr := readPublicationDB(ctx, tx, publication.Binding.OrgID, publication.Digest, true)
	if readErr == nil {
		if stored != publication {
			return fmt.Errorf("%w: publication digest identifies different content", ErrCorrupt)
		}
		return nil
	}
	return err
}

func insertCommand(ctx context.Context, tx *sql.Tx, command Command) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublicationCommand
(OrgId, CommandId, RequestDigest, RuntimeAudience, CandidatePublicationDigest,
ExpectedPriorPresent, ExpectedPriorDigest, State, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, 'RESERVED', CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		command.OrgID, command.CommandID, command.RequestDigest, command.RuntimeAudience,
		command.CandidatePublicationDigest, command.ExpectedPrior.Present, nullableReference(command.ExpectedPrior))
	if err != nil {
		return err
	}
	return requireOne(result)
}

func insertCandidate(ctx context.Context, tx *sql.Tx, command Command) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublicationCandidate
(OrgId, CommandId, CandidatePublicationDigest, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		command.OrgID, command.CommandID, command.CandidatePublicationDigest)
	if err != nil {
		return err
	}
	return requireOne(result)
}

func readCandidate(ctx context.Context, tx *sql.Tx, orgID, commandID string) (string, error) {
	var storedCommand, digest string
	err := tx.QueryRowContext(ctx, `SELECT CommandId, CandidatePublicationDigest
FROM _System_CatalogPublicationCandidate WHERE OrgId = ? FOR UPDATE`, orgID).Scan(&storedCommand, &digest)
	if errors.Is(err, sql.ErrNoRows) {
		return "", fmt.Errorf("%w: reserved command has no candidate row", ErrCorrupt)
	}
	if err != nil {
		return "", err
	}
	if storedCommand != commandID {
		return "", fmt.Errorf("%w: candidate row belongs to another command", ErrCorrupt)
	}
	if err := validateReference("CandidatePublication", Reference{Present: true, Digest: digest}); err != nil {
		return "", fmt.Errorf("%w: %v", ErrCorrupt, err)
	}
	return digest, nil
}

func updateActiveState(ctx context.Context, tx *sql.Tx, orgID string, prior activeState, sequence uint64, outcome Outcome) error {
	var result sql.Result
	var err error
	if outcome.Result == ResultSelected {
		result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogPublicationActive
SET ActivePublicationPresent = TRUE, ActivePublicationDigest = ?, SelectedByCommandId = ?,
HistorySequence = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND HistorySequence = ?`, outcome.ResultingActive.Digest, outcome.CommandID,
			decimalText(sequence), orgID, decimalText(prior.HistorySequence))
	} else {
		result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogPublicationActive
SET HistorySequence = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND HistorySequence = ?`, decimalText(sequence), orgID, decimalText(prior.HistorySequence))
	}
	if err != nil {
		return err
	}
	return requireOne(result)
}

func insertOutcome(ctx context.Context, tx *sql.Tx, outcome Outcome) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublicationOutcome
(OrgId, CommandId, CandidatePublicationDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingActivePresent, ResultingActiveDigest,
OutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		outcome.OrgID, outcome.CommandID, outcome.CandidatePublicationDigest, string(outcome.Result),
		outcome.ObservedPrior.Present, nullableReference(outcome.ObservedPrior),
		outcome.ResultingActive.Present, nullableReference(outcome.ResultingActive), outcome.OutcomeDigest)
	if err != nil {
		return err
	}
	return requireOne(result)
}

func insertHistory(ctx context.Context, tx *sql.Tx, sequence uint64, outcome Outcome) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogPublicationHistory
(OrgId, SequenceNumber, CommandId, CandidatePublicationDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingActivePresent, ResultingActiveDigest,
OutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		outcome.OrgID, decimalText(sequence), outcome.CommandID, outcome.CandidatePublicationDigest, string(outcome.Result),
		outcome.ObservedPrior.Present, nullableReference(outcome.ObservedPrior),
		outcome.ResultingActive.Present, nullableReference(outcome.ResultingActive), outcome.OutcomeDigest)
	if err != nil {
		return err
	}
	return requireOne(result)
}

func completeCommand(ctx context.Context, tx *sql.Tx, command Command) error {
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogPublicationCommand
SET State = 'COMPLETED', LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND CommandId = ? AND State = 'RESERVED' AND CandidatePublicationDigest = ?`,
		command.OrgID, command.CommandID, command.CandidatePublicationDigest)
	if err != nil {
		return err
	}
	return requireOne(result)
}

func deleteCandidate(ctx context.Context, tx *sql.Tx, orgID, commandID, digest string) error {
	result, err := tx.ExecContext(ctx, `DELETE FROM _System_CatalogPublicationCandidate
WHERE OrgId = ? AND CommandId = ? AND CandidatePublicationDigest = ?`, orgID, commandID, digest)
	if err != nil {
		return err
	}
	return requireOne(result)
}

func nullableReference(reference Reference) any {
	if !reference.Present {
		return nil
	}
	return reference.Digest
}

func decimalText(value uint64) string {
	return strconv.FormatUint(value, 10)
}

func requireOne(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return ErrConflict
	}
	return nil
}

func mapWriterError(err error) error {
	if errors.Is(err, catalogwriter.ErrConflict) ||
		errors.Is(err, catalogwriter.ErrLineageExhausted) {
		return ErrConflict
	}
	return err
}

func rollbackTx(tx *sql.Tx) {
	_ = tx.Rollback()
}