package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// ApplyAllSourceRevisionAdvancePendingKind describes the exact mutation still
// required for one planned source-revision edge.
type ApplyAllSourceRevisionAdvancePendingKind string

const (
	ApplyAllSourceRevisionAdvancePendingAdvance      ApplyAllSourceRevisionAdvancePendingKind = "advance"
	ApplyAllSourceRevisionAdvancePendingRetry        ApplyAllSourceRevisionAdvancePendingKind = "retry"
	ApplyAllSourceRevisionAdvancePendingCanonicalize ApplyAllSourceRevisionAdvancePendingKind = "canonicalize"
)

// PendingApplyAllSourceRevisionAdvance is a read-only snapshot. ApplyAll must
// re-read the plan under its mutation lock before acting on it.
type PendingApplyAllSourceRevisionAdvance struct {
	Authorization   SourceRevisionAdvanceAuthorization
	Kind            ApplyAllSourceRevisionAdvancePendingKind
	InstalledStatus string
}

// ApplyAllSourceRevisionAdvancePlanStatus describes the whole plan. An empty
// Pending slice means every edge is already at its canonical target.
type ApplyAllSourceRevisionAdvancePlanStatus struct {
	Pending []PendingApplyAllSourceRevisionAdvance
}

func (s ApplyAllSourceRevisionAdvancePlanStatus) HasPending() bool {
	return len(s.Pending) > 0
}

type ApplyAllSourceRevisionAdvancePlanErrorStage string

const (
	ApplyAllSourceRevisionAdvancePlanValidate ApplyAllSourceRevisionAdvancePlanErrorStage = "validate"
	ApplyAllSourceRevisionAdvancePlanRead     ApplyAllSourceRevisionAdvancePlanErrorStage = "read"
	ApplyAllSourceRevisionAdvancePlanState    ApplyAllSourceRevisionAdvancePlanErrorStage = "state"
)

// ApplyAllSourceRevisionAdvancePlanError distinguishes invalid authoring from
// an installed-state read failure and from an unsupported installed state.
type ApplyAllSourceRevisionAdvancePlanError struct {
	Stage ApplyAllSourceRevisionAdvancePlanErrorStage
	Index int
	Name  string
	Err   error
}

func (e *ApplyAllSourceRevisionAdvancePlanError) Error() string {
	if e == nil {
		return "apply-all source revision advance plan failed"
	}
	label := fmt.Sprintf("apply-all source revision advance plan %s", e.Stage)
	if e.Name != "" {
		label += " for " + e.Name
	} else if e.Index >= 0 {
		label += fmt.Sprintf(" at index %d", e.Index)
	}
	return label + ": " + e.Err.Error()
}

func (e *ApplyAllSourceRevisionAdvancePlanError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Err
}

type normalizedApplyAllSourceRevisionAdvance struct {
	SourceRevisionAdvanceAuthorization
	index          int
	completeReplay bool
}

func markOfflineCandidateCompleteSourceRevisionReplays(
	plan []normalizedApplyAllSourceRevisionAdvance,
	names map[string]struct{},
) error {
	if len(names) == 0 {
		return nil
	}
	remaining := make(map[string]struct{}, len(names))
	for name := range names {
		name = strings.ToLower(strings.TrimSpace(name))
		if name == "" {
			return fmt.Errorf("offline candidate complete source-revision replay has an empty pack name")
		}
		remaining[name] = struct{}{}
	}
	for index := range plan {
		name := strings.ToLower(strings.TrimSpace(plan[index].Name))
		if _, ok := remaining[name]; !ok {
			continue
		}
		plan[index].completeReplay = true
		delete(remaining, name)
	}
	if len(remaining) != 0 {
		unknown := make([]string, 0, len(remaining))
		for name := range remaining {
			unknown = append(unknown, name)
		}
		sort.Strings(unknown)
		return fmt.Errorf("offline candidate complete source-revision replay is outside the normalized plan: %s", strings.Join(unknown, ", "))
	}
	return nil
}

// PendingApplyAllSourceRevisionAdvances returns a stale-tolerant, read-only
// snapshot. It performs no bootstrap-marker or installed-pack writes.
func PendingApplyAllSourceRevisionAdvances(ctx context.Context, db *sql.DB, options InstallPackOptions) (ApplyAllSourceRevisionAdvancePlanStatus, error) {
	plan, err := normalizeApplyAllSourceRevisionAdvancePlan(options.ApplyAllSourceRevisionAdvances)
	if err != nil {
		return ApplyAllSourceRevisionAdvancePlanStatus{}, err
	}
	snapshot, err := readApplyAllSourceRevisionAdvanceCurrentSnapshot(ctx, db, plan)
	if err != nil {
		return ApplyAllSourceRevisionAdvancePlanStatus{}, err
	}
	status := ApplyAllSourceRevisionAdvancePlanStatus{
		Pending: make([]PendingApplyAllSourceRevisionAdvance, 0, len(snapshot.entries)),
	}
	for _, entry := range snapshot.entries {
		if entry.pending != nil {
			status.Pending = append(status.Pending, *entry.pending)
		}
	}
	return status, nil
}

// requireApplyAllSourceRevisionAdvancePlanCompleted is the lock-held final
// readiness gate for a full ApplyAll. The completed marker must not become the
// restart source of truth while any distribution-authorized source edge still
// requires mutation.
func requireApplyAllSourceRevisionAdvancePlanCompleted(ctx context.Context, db *sql.DB, options InstallPackOptions) error {
	status, err := PendingApplyAllSourceRevisionAdvances(ctx, db, options)
	if err != nil {
		return err
	}
	if !status.HasPending() {
		return nil
	}

	pending := make([]string, 0, len(status.Pending))
	for _, candidate := range status.Pending {
		pending = append(pending, fmt.Sprintf(
			"%s(%s:%s->%s,status=%s)",
			candidate.Authorization.Name,
			candidate.Kind,
			candidate.Authorization.FromRevision,
			candidate.Authorization.ToRevision,
			candidate.InstalledStatus,
		))
	}
	return fmt.Errorf(
		"%d planned source revision advance(s) remain pending: %s",
		len(status.Pending), strings.Join(pending, ", "),
	)
}

func normalizeApplyAllSourceRevisionAdvancePlan(raw []SourceRevisionAdvanceAuthorization) ([]normalizedApplyAllSourceRevisionAdvance, error) {
	plan := make([]normalizedApplyAllSourceRevisionAdvance, 0, len(raw))
	seenNames := make(map[string]int, len(raw))
	seenSources := make(map[string]int, len(raw))
	for index, candidate := range raw {
		candidate.Name = strings.TrimSpace(candidate.Name)
		candidate.FromRevision = strings.TrimSpace(candidate.FromRevision)
		candidate.ToRevision = strings.TrimSpace(candidate.ToRevision)
		kind, uri, err := canonicalPackSourceIdentity(candidate.Kind, candidate.URI)
		if err != nil || !packs.IsRevisionAddressedPackSourceKind(kind) {
			if err == nil {
				err = fmt.Errorf("source kind %q is not eligible for an exact named-repository revision advance", kind)
			}
			return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, err)
		}
		candidate.Kind, candidate.URI = kind, uri
		if candidate.Name == "" || candidate.URI == "" || candidate.FromRevision == "" || candidate.ToRevision == "" || candidate.FromRevision == candidate.ToRevision {
			return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("name, source identity, and distinct from/to revisions are required"))
		}
		if kind == string(packs.PackSourceKindReleaseRepo) {
			if err := packs.ValidateReleaseRepoRevision(candidate.FromRevision); err != nil {
				return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("from revision: %w", err))
			}
			if err := packs.ValidateReleaseRepoRevision(candidate.ToRevision); err != nil {
				return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("to revision: %w", err))
			}
		}
		if kind == string(packs.PackSourceKindOfficialRepo) && candidate.Name != candidate.URI {
			return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("official repository URI %q must equal pack name", candidate.URI))
		}
		nameKey := strings.ToLower(candidate.Name)
		if previous, duplicate := seenNames[nameKey]; duplicate {
			return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("duplicates plan entry %d by pack name", previous))
		}
		seenNames[nameKey] = index
		sourceKey := strings.ToLower(candidate.Kind + "\x00" + candidate.URI)
		if previous, duplicate := seenSources[sourceKey]; duplicate {
			return nil, applyAllSourceRevisionAdvancePlanError(ApplyAllSourceRevisionAdvancePlanValidate, index, candidate.Name, fmt.Errorf("duplicates plan entry %d by source identity", previous))
		}
		seenSources[sourceKey] = index
		plan = append(plan, normalizedApplyAllSourceRevisionAdvance{SourceRevisionAdvanceAuthorization: candidate, index: index})
	}
	return plan, nil
}

func classifyApplyAllSourceRevisionAdvance(row installedPackRow, advance SourceRevisionAdvanceAuthorization) (*PendingApplyAllSourceRevisionAdvance, error) {
	rowKind, rowURI, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if !row.SourceKind.Valid || !row.SourceURI.Valid || err != nil {
		return nil, fmt.Errorf("installed source identity is unavailable or unsupported")
	}
	if rowKind != advance.Kind || rowURI != advance.URI {
		return nil, fmt.Errorf("installed source identity %s:%s does not match planned %s:%s", rowKind, rowURI, advance.Kind, advance.URI)
	}
	if !row.SourceRevision.Valid {
		return nil, fmt.Errorf("installed source revision is unavailable")
	}
	revision := strings.TrimSpace(row.SourceRevision.String)
	canonicalStored := strings.TrimSpace(row.SourceKind.String) == advance.Kind && strings.TrimSpace(row.SourceURI.String) == advance.URI
	pending := &PendingApplyAllSourceRevisionAdvance{Authorization: advance, InstalledStatus: row.Status}
	switch row.Status {
	case installedPackStatusActive:
		if revision == advance.FromRevision {
			pending.Kind = ApplyAllSourceRevisionAdvancePendingAdvance
			return pending, nil
		}
		if revision == advance.ToRevision {
			if canonicalStored {
				return nil, nil
			}
			pending.Kind = ApplyAllSourceRevisionAdvancePendingCanonicalize
			return pending, nil
		}
		return nil, fmt.Errorf("active installed source revision %q matches neither planned endpoint", revision)
	case installedPackStatusFailed, installedPackStatusInstalling:
		if revision != advance.ToRevision {
			return nil, fmt.Errorf("retryable status %q is allowed only at target revision %q, got %q", row.Status, advance.ToRevision, revision)
		}
		pending.Kind = ApplyAllSourceRevisionAdvancePendingRetry
		return pending, nil
	default:
		return nil, fmt.Errorf("installed status %q is not eligible for source revision convergence", row.Status)
	}
}

func applyAllSourceRevisionAdvancePlanError(stage ApplyAllSourceRevisionAdvancePlanErrorStage, index int, name string, err error) error {
	return &ApplyAllSourceRevisionAdvancePlanError{Stage: stage, Index: index, Name: strings.TrimSpace(name), Err: err}
}