package catalogpublication

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

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
)

// AdoptStoppedWriterState atomically selects a complete immutable publication
// that names the exact Catalog writer state already committed by Catalog
// reconcile. It exists only for a quiesced, stopped deployment: unlike Begin
// and Commit it neither reserves nor advances the writer axes.
//
// The method fails closed when any writer is active, the expected active
// publication changed, the candidate mixes lineage, or the active-to-candidate
// transition is not an exact step. An exact completed retry returns its durable
// outcome without consulting newer state.
func (a *Authority) AdoptStoppedWriterState(
	ctx context.Context,
	request StoppedAdoptionRequest,
) (Outcome, error) {
	candidateDigest, _, err := validateBeginRequest(BeginRequest{
		CommandID:       request.CommandID,
		RuntimeAudience: request.RuntimeAudience,
		Candidate:       request.Candidate,
	})
	if err != nil {
		return Outcome{}, err
	}

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

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

	writer, err := catalogwriter.Lock(ctx, tx, request.Candidate.OrgID)
	if errors.Is(err, sql.ErrNoRows) {
		return Outcome{}, ErrNotReady
	}
	if err != nil {
		return Outcome{}, mapWriterError(err)
	}
	// Re-check after acquiring the org fence. A concurrent exact adoption may
	// have completed while this transaction waited for the writer row.
	existing, err = readCommandDB(
		ctx, tx, request.Candidate.OrgID, request.CommandID, true,
	)
	if err == nil {
		return recoverStoppedAdoption(ctx, tx, existing, request, candidateDigest)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return Outcome{}, err
	}
	if err := rejectCandidateSourceJournalLocked(
		ctx, tx, request.Candidate.OrgID, request.CommandID, false,
	); err != nil {
		return Outcome{}, err
	}
	if writer.RuntimeAudience != request.RuntimeAudience || writer.ActiveCommandID != "" {
		return Outcome{}, ErrConflict
	}

	state, err := lockOrCreateActiveState(ctx, tx, request.Candidate.OrgID)
	if err != nil {
		return Outcome{}, err
	}
	activeBinding, err := lockedActiveBinding(ctx, tx, request.Candidate.OrgID, state)
	if err != nil {
		return Outcome{}, err
	}
	if err := validateStoppedWriterAdoption(
		writer.CatalogRevision,
		writer.SecurityEpoch,
		activeBinding,
		request.Candidate,
	); err != nil {
		return Outcome{}, fmt.Errorf("%w: invalid stopped publication adoption: %v", ErrConflict, err)
	}
	beginRequest := stoppedAdoptionBeginRequest(request, state.Active)
	_, requestDigest, err := validateBeginRequest(beginRequest)
	if err != nil {
		return Outcome{}, err
	}
	wanted := commandFromRequest(beginRequest, candidateDigest, requestDigest)
	if err := persistPublication(
		ctx, tx, Publication{Digest: candidateDigest, Binding: request.Candidate},
	); err != nil {
		return Outcome{}, err
	}
	if err := insertCommand(ctx, tx, wanted); err != nil {
		return Outcome{}, err
	}

	outcome := Outcome{
		OrgID:                      request.Candidate.OrgID,
		CommandID:                  request.CommandID,
		CandidatePublicationDigest: candidateDigest,
		ObservedPrior:              state.Active,
		ResultingActive:            state.Active,
	}
	if state.Active.Present && state.Active.Digest == candidateDigest {
		outcome.Result = ResultUnchanged
	} else {
		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 err := updateActiveState(
		ctx, tx, request.Candidate.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, wanted); err != nil {
		return Outcome{}, err
	}
	if err := tx.Commit(); err != nil {
		return Outcome{}, err
	}
	return outcome, nil
}

func recoverStoppedAdoption(
	ctx context.Context,
	tx *sql.Tx,
	existing Command,
	request StoppedAdoptionRequest,
	candidateDigest string,
) (Outcome, error) {
	beginRequest := stoppedAdoptionBeginRequest(request, existing.ExpectedPrior)
	_, requestDigest, err := validateBeginRequest(beginRequest)
	if err != nil {
		return Outcome{}, err
	}
	wanted := commandFromRequest(beginRequest, candidateDigest, requestDigest)
	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 Outcome{}, ErrConflict
	}
	if err := rejectCandidateSourceJournalLocked(
		ctx, tx, wanted.OrgID, wanted.CommandID, true,
	); err != nil {
		return Outcome{}, err
	}
	if existing.State != CommandCompleted {
		return Outcome{}, ErrNotReady
	}
	outcome, err := readOutcomeDB(ctx, tx, wanted.OrgID, wanted.CommandID, true)
	if errors.Is(err, sql.ErrNoRows) {
		return Outcome{}, fmt.Errorf("%w: completed command has no outcome", ErrCorrupt)
	}
	if err != nil {
		return Outcome{}, err
	}
	if outcome.CandidatePublicationDigest != existing.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
}

func stoppedAdoptionBeginRequest(
	request StoppedAdoptionRequest,
	expectedPrior Reference,
) BeginRequest {
	return BeginRequest{
		CommandID:       request.CommandID,
		RuntimeAudience: request.RuntimeAudience,
		ExpectedPrior:   expectedPrior,
		Candidate:       request.Candidate,
	}
}