package catalogpublication

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

	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
)

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

type rowsQuerier interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}

// activeState is the non-serving authority pointer and its org-local history
// allocator. Runtime serving must not use this row as an implicit selector.
type activeState struct {
	Active              Reference
	SelectedByCommandID string
	HistorySequence     uint64
}

// ReadActive returns the exact immutable publication currently recorded by the
// control-plane authority. It does not install or expose a Runtime selector.
func (a *Authority) ReadActive(ctx context.Context, orgID string) (Publication, error) {
	if err := validateOpaque("OrgID", orgID, 255); err != nil {
		return Publication{}, err
	}
	state, err := readActiveStateDB(ctx, a.db, orgID, false)
	if errors.Is(err, sql.ErrNoRows) || (err == nil && !state.Active.Present) {
		return Publication{}, fmt.Errorf("%w: active publication for org %q", ErrNotFound, orgID)
	}
	if err != nil {
		return Publication{}, err
	}
	publication, err := readPublicationDB(ctx, a.db, orgID, state.Active.Digest, false)
	if errors.Is(err, sql.ErrNoRows) {
		return Publication{}, fmt.Errorf("%w: active pointer %q has no immutable publication", ErrCorrupt, state.Active.Digest)
	}
	return publication, err
}

// ReadExact returns one immutable publication by its canonical digest.
func (a *Authority) ReadExact(ctx context.Context, orgID, digest string) (Publication, error) {
	if err := validateOpaque("OrgID", orgID, 255); err != nil {
		return Publication{}, err
	}
	if err := validateSHA256Digest(digest); err != nil {
		return Publication{}, fmt.Errorf("catalog publication: publication digest: %w", err)
	}
	publication, err := readPublicationDB(ctx, a.db, orgID, digest, false)
	if errors.Is(err, sql.ErrNoRows) {
		return Publication{}, fmt.Errorf("%w: publication %q for org %q", ErrNotFound, digest, orgID)
	}
	return publication, err
}

// ReadCommand returns the durable idempotency record for one command.
func (a *Authority) ReadCommand(ctx context.Context, orgID, commandID string) (Command, error) {
	if err := validateOpaque("OrgID", orgID, 255); err != nil {
		return Command{}, err
	}
	if !validCommandID(commandID) {
		return Command{}, fmt.Errorf("catalog publication: CommandID must be a lowercase UUIDv7")
	}
	command, err := readCommandDB(ctx, a.db, orgID, commandID, false)
	if errors.Is(err, sql.ErrNoRows) {
		return Command{}, fmt.Errorf("%w: command %q for org %q", ErrNotFound, commandID, orgID)
	}
	return command, err
}

// ReadOutcome returns the immutable terminal outcome for one command.
func (a *Authority) ReadOutcome(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")
	}
	outcome, err := readOutcomeDB(ctx, a.db, orgID, commandID, false)
	if errors.Is(err, sql.ErrNoRows) {
		return Outcome{}, fmt.Errorf("%w: outcome for command %q in org %q", ErrNotFound, commandID, orgID)
	}
	return outcome, err
}

// ReadHistory returns immutable terminal outcomes after an org-local sequence,
// ordered strictly ascending. Limits outside 1..1000 are rejected.
func (a *Authority) ReadHistory(ctx context.Context, orgID string, afterSequence uint64, limit int) ([]HistoryEntry, error) {
	if err := validateOpaque("OrgID", orgID, 255); err != nil {
		return nil, err
	}
	if limit < 1 || limit > 1000 {
		return nil, fmt.Errorf("catalog publication: history limit must be within 1..1000")
	}
	return readHistoryDB(ctx, a.db, orgID, afterSequence, limit)
}

func readHistoryDB(ctx context.Context, query rowsQuerier, orgID string, afterSequence uint64, limit int) ([]HistoryEntry, error) {
	rows, err := query.QueryContext(ctx, `SELECT SequenceNumber, OrgId, CommandId,
CandidatePublicationDigest, Result, ObservedPriorPresent, ObservedPriorDigest,
ResultingActivePresent, ResultingActiveDigest, OutcomeDigest
FROM _System_CatalogPublicationHistory
WHERE OrgId = ? AND SequenceNumber > ?
ORDER BY SequenceNumber ASC LIMIT ?`, orgID, strconv.FormatUint(afterSequence, 10), limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	entries := make([]HistoryEntry, 0)
	previous := afterSequence
	for rows.Next() {
		var rawSequence []byte
		var outcome Outcome
		var rawResult string
		var observedPresent, resultingPresent bool
		var observedDigest, resultingDigest sql.NullString
		if err := rows.Scan(&rawSequence, &outcome.OrgID, &outcome.CommandID,
			&outcome.CandidatePublicationDigest, &rawResult,
			&observedPresent, &observedDigest, &resultingPresent, &resultingDigest,
			&outcome.OutcomeDigest); err != nil {
			return nil, fmt.Errorf("%w: history row cannot be decoded: %v", ErrCorrupt, err)
		}
		sequence, err := parseDurableUint64("history SequenceNumber", rawSequence)
		if err != nil {
			return nil, err
		}
		if previous == ^uint64(0) || sequence != previous+1 {
			return nil, fmt.Errorf("%w: history sequence %d does not immediately follow %d", ErrCorrupt, sequence, previous)
		}
		if outcome.OrgID != orgID {
			return nil, fmt.Errorf("%w: history org %q does not exactly match requested org %q", ErrCorrupt, outcome.OrgID, orgID)
		}
		outcome.Result = Result(rawResult)
		outcome.ObservedPrior, err = durableReference("history ObservedPrior", observedPresent, observedDigest)
		if err != nil {
			return nil, err
		}
		outcome.ResultingActive, err = durableReference("history ResultingActive", resultingPresent, resultingDigest)
		if err != nil {
			return nil, err
		}
		if err := validateDurableOutcome(outcome); err != nil {
			return nil, err
		}
		entries = append(entries, HistoryEntry{Sequence: sequence, Outcome: outcome})
		previous = sequence
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return entries, nil
}

func readPublicationDB(ctx context.Context, query rowQuerier, orgID, digest string, forUpdate bool) (Publication, error) {
	statement := `SELECT PublicationDigest, BindingVersion, OrgId, CatalogRevision,
CatalogSnapshotDigest, SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest
FROM _System_CatalogPublication WHERE OrgId = ? AND PublicationDigest = ?` + lockSuffix(forUpdate)
	var publication Publication
	var rawCatalogRevision, rawSchemaGeneration, rawEngineGeneration, rawSecurityEpoch []byte
	var bindingVersion, ownerVerificationReceiptSetDigest sql.NullString
	err := query.QueryRowContext(ctx, statement, orgID, digest).Scan(
		&publication.Digest, &bindingVersion, &publication.Binding.OrgID,
		&rawCatalogRevision, &publication.Binding.CatalogSnapshotDigest,
		&rawSchemaGeneration, &rawEngineGeneration,
		&publication.Binding.LayoutValidationViewDigest,
		&ownerVerificationReceiptSetDigest,
		&publication.Binding.LayoutActivationSetDigest,
		&rawSecurityEpoch, &publication.Binding.LayoutRoutingSecurityViewDigest,
	)
	if err != nil {
		return Publication{}, err
	}
	if publication.Binding.OrgID != orgID {
		return Publication{}, fmt.Errorf("%w: publication org %q does not exactly match requested org %q", ErrCorrupt, publication.Binding.OrgID, orgID)
	}
	if publication.Digest != digest {
		return Publication{}, fmt.Errorf("%w: publication digest %q does not exactly match requested digest %q", ErrCorrupt, publication.Digest, digest)
	}
	if !bindingVersion.Valid || bindingVersion.String != legacyPublicationBindingVersion {
		return Publication{}, fmt.Errorf("%w: publication binding version is %q, want %q", ErrCorrupt, bindingVersion.String, legacyPublicationBindingVersion)
	}
	if ownerVerificationReceiptSetDigest.Valid {
		return Publication{}, fmt.Errorf("%w: v1 publication has an owner-verification receipt-set digest", ErrCorrupt)
	}
	publication.Binding.CatalogRevision, err = parseDurableUint64("publication CatalogRevision", rawCatalogRevision)
	if err != nil {
		return Publication{}, err
	}
	publication.Binding.SchemaGeneration, err = parseDurableUint64("publication SchemaGeneration", rawSchemaGeneration)
	if err != nil {
		return Publication{}, err
	}
	publication.Binding.EngineGeneration, err = parseDurableUint64("publication EngineGeneration", rawEngineGeneration)
	if err != nil {
		return Publication{}, err
	}
	publication.Binding.SecurityEpoch, err = parseDurableUint64("publication SecurityEpoch", rawSecurityEpoch)
	if err != nil {
		return Publication{}, err
	}
	computed, err := catalogpublicationv1.BindingDigest(publication.Binding)
	if err != nil {
		return Publication{}, fmt.Errorf("%w: invalid immutable publication %q: %v", ErrCorrupt, digest, err)
	}
	if computed != publication.Digest {
		return Publication{}, fmt.Errorf("%w: immutable publication digest is %q, stored %q", ErrCorrupt, computed, publication.Digest)
	}
	return publication, nil
}

func readCommandDB(ctx context.Context, query rowQuerier, orgID, commandID string, forUpdate bool) (Command, error) {
	statement := `SELECT OrgId, CommandId, RequestDigest, RuntimeAudience,
CandidatePublicationDigest, ExpectedPriorPresent, ExpectedPriorDigest, State
FROM _System_CatalogPublicationCommand WHERE OrgId = ? AND CommandId = ?` + lockSuffix(forUpdate)
	var command Command
	var priorPresent bool
	var priorDigest sql.NullString
	var rawState string
	err := query.QueryRowContext(ctx, statement, orgID, commandID).Scan(
		&command.OrgID, &command.CommandID, &command.RequestDigest, &command.RuntimeAudience,
		&command.CandidatePublicationDigest, &priorPresent, &priorDigest, &rawState,
	)
	if err != nil {
		return Command{}, err
	}
	if command.OrgID != orgID || command.CommandID != commandID {
		return Command{}, fmt.Errorf("%w: command durable identity does not exactly match its lookup", ErrCorrupt)
	}
	if !validCommandID(command.CommandID) {
		return Command{}, fmt.Errorf("%w: invalid durable CommandID %q", ErrCorrupt, command.CommandID)
	}
	if err := validateOpaque("RuntimeAudience", command.RuntimeAudience, 255); err != nil {
		return Command{}, fmt.Errorf("%w: %v", ErrCorrupt, err)
	}
	if err := validateSHA256Digest(command.CandidatePublicationDigest); err != nil {
		return Command{}, fmt.Errorf("%w: invalid candidate publication digest: %v", ErrCorrupt, err)
	}
	command.ExpectedPrior, err = durableReference("command ExpectedPrior", priorPresent, priorDigest)
	if err != nil {
		return Command{}, err
	}
	command.State = CommandState(rawState)
	if command.State != CommandReserved && command.State != CommandCompleted {
		return Command{}, fmt.Errorf("%w: unknown command state %q", ErrCorrupt, rawState)
	}
	if err := validateSHA256Digest(command.RequestDigest); err != nil {
		return Command{}, fmt.Errorf("%w: invalid request digest: %v", ErrCorrupt, err)
	}
	computed := publicationCommandDigest(command.RuntimeAudience, command.ExpectedPrior, command.CandidatePublicationDigest)
	if computed != command.RequestDigest {
		return Command{}, fmt.Errorf("%w: command request digest is %q, stored %q", ErrCorrupt, computed, command.RequestDigest)
	}
	return command, nil
}

func readOutcomeDB(ctx context.Context, query rowQuerier, orgID, commandID string, forUpdate bool) (Outcome, error) {
	statement := `SELECT OrgId, CommandId, CandidatePublicationDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingActivePresent,
ResultingActiveDigest, OutcomeDigest
FROM _System_CatalogPublicationOutcome WHERE OrgId = ? AND CommandId = ?` + lockSuffix(forUpdate)
	var outcome Outcome
	var rawResult string
	var observedPresent, resultingPresent bool
	var observedDigest, resultingDigest sql.NullString
	err := query.QueryRowContext(ctx, statement, orgID, commandID).Scan(
		&outcome.OrgID, &outcome.CommandID, &outcome.CandidatePublicationDigest, &rawResult,
		&observedPresent, &observedDigest, &resultingPresent, &resultingDigest,
		&outcome.OutcomeDigest,
	)
	if err != nil {
		return Outcome{}, err
	}
	if outcome.OrgID != orgID || outcome.CommandID != commandID {
		return Outcome{}, fmt.Errorf("%w: outcome durable identity does not exactly match its lookup", ErrCorrupt)
	}
	outcome.Result = Result(rawResult)
	outcome.ObservedPrior, err = durableReference("outcome ObservedPrior", observedPresent, observedDigest)
	if err != nil {
		return Outcome{}, err
	}
	outcome.ResultingActive, err = durableReference("outcome ResultingActive", resultingPresent, resultingDigest)
	if err != nil {
		return Outcome{}, err
	}
	if err := validateDurableOutcome(outcome); err != nil {
		return Outcome{}, err
	}
	return outcome, nil
}

func readActiveStateDB(ctx context.Context, query rowQuerier, orgID string, forUpdate bool) (activeState, error) {
	statement := `SELECT OrgId, ActivePublicationPresent, ActivePublicationDigest,
SelectedByCommandId, HistorySequence
FROM _System_CatalogPublicationActive WHERE OrgId = ?` + lockSuffix(forUpdate)
	var storedOrg string
	var state activeState
	var present bool
	var digest, selectedBy sql.NullString
	var rawSequence []byte
	err := query.QueryRowContext(ctx, statement, orgID).Scan(
		&storedOrg, &present, &digest, &selectedBy, &rawSequence,
	)
	if err != nil {
		return activeState{}, err
	}
	if storedOrg != orgID {
		return activeState{}, fmt.Errorf("%w: active-state org %q does not exactly match requested org %q", ErrCorrupt, storedOrg, orgID)
	}
	state.Active, err = durableReference("active publication", present, digest)
	if err != nil {
		return activeState{}, err
	}
	if !present {
		if selectedBy.Valid {
			return activeState{}, fmt.Errorf("%w: absent active publication has a selecting command", ErrCorrupt)
		}
	} else {
		if !selectedBy.Valid || !validCommandID(selectedBy.String) {
			return activeState{}, fmt.Errorf("%w: active publication has an invalid selecting command", ErrCorrupt)
		}
		state.SelectedByCommandID = selectedBy.String
	}
	state.HistorySequence, err = parseDurableUint64("active HistorySequence", rawSequence)
	if err != nil {
		return activeState{}, err
	}
	return state, nil
}

func durableReference(label string, present bool, digest sql.NullString) (Reference, error) {
	if present != digest.Valid {
		return Reference{}, fmt.Errorf("%w: %s presence and digest NULL state disagree", ErrCorrupt, label)
	}
	reference := Reference{Present: present}
	if digest.Valid {
		reference.Digest = digest.String
	}
	if err := validateReference(label, reference); err != nil {
		return Reference{}, fmt.Errorf("%w: %v", ErrCorrupt, err)
	}
	return reference, nil
}

func validateDurableOutcome(outcome Outcome) error {
	computed, err := canonicalOutcomeDigest(outcome)
	if err != nil {
		return fmt.Errorf("%w: invalid durable outcome: %v", ErrCorrupt, err)
	}
	if err := validateSHA256Digest(outcome.OutcomeDigest); err != nil {
		return fmt.Errorf("%w: invalid outcome digest: %v", ErrCorrupt, err)
	}
	if computed != outcome.OutcomeDigest {
		return fmt.Errorf("%w: outcome digest is %q, stored %q", ErrCorrupt, computed, outcome.OutcomeDigest)
	}
	return nil
}

func parseDurableUint64(label string, raw []byte) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("%w: %s is empty", ErrCorrupt, label)
	}
	for _, character := range raw {
		if character < '0' || character > '9' {
			return 0, fmt.Errorf("%w: %s is not an unsigned decimal", ErrCorrupt, label)
		}
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%w: %s exceeds uint64: %v", ErrCorrupt, label, err)
	}
	return value, nil
}

func lockSuffix(forUpdate bool) string {
	if forUpdate {
		return " FOR UPDATE"
	}
	return ""
}