package park

import (
	"context"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
)

// View is a moves-owned, completely read snapshot bound to one claim ETag.
// GenerationOf is a pure classifier for any exact reference, including a
// version already deleted by a previous attempt. It does no storage I/O.
// Pinned includes the exact noncurrent versions needed by the installed root.
// MarkerKnown means its read succeeded; an absent marker may denote @root.
type View struct {
	HomeRegion string
	// ArchivedIncarnations labels proven retired locations independently of
	// the moves-sweep proof's incarnation. Missing entries never invent labels.
	ArchivedIncarnations                   map[string]string
	RefusedLocations                       map[string]error
	ClaimETag, Region, AuthoritativeRegion string
	MarkerKnown                            bool
	MarkerGeneration                       string
	Marker                                 Ref
	Pinned                                 map[Ref]bool
	FilePins                               map[Ref]bool
	GenerationOf                           func(Ref) (string, bool)
}
type Alert struct{ Location, Reason string }
type Candidate struct{ Location, Incarnation, Generation string }
type Decision struct {
	Candidates []Candidate
	Alerts     []Alert
}

func generation(s string) string {
	if s == "" {
		return hostclaim.RootGeneration
	}
	return s
}

func protected(c *hostclaim.Claim, v View, location string) (map[string]bool, error) {
	live := map[string]bool{}
	if c.Holder == nil || c.Op.Kind == hostclaim.KindParking {
		return live, nil
	}
	if v.AuthoritativeRegion == "" {
		return nil, ErrUnsafe
	}
	if !v.sameRegion(v.Region, v.AuthoritativeRegion) {
		return live, nil
	}
	if !v.MarkerKnown || v.GenerationOf == nil {
		return nil, ErrUnsafe
	}
	for r, on := range v.Pinned {
		if on {
			g, ok := v.GenerationOf(r)
			if !ok {
				return nil, ErrUnsafe
			}
			if v.FilePins[r] && strings.HasPrefix(r.Key, c.Holder.Location+"/files/") {
				continue
			}
			live[generation(g)] = true
		}
	}
	h := c.Holder
	if location != "" && location != h.Location && !strings.HasPrefix(location, h.Location+"/") && !strings.HasPrefix(h.Location, location+"/") {
		return live, nil
	}
	current := generation(h.Generation)
	marker := generation(v.MarkerGeneration)
	live[current] = true
	live[marker] = true
	if current == marker {
		return live, nil
	}
	// Walk the recorded chronological predecessor chain backwards. Warm
	// admissions do not form cold unfinished links and cannot bridge a gap.
	want, region := current, v.AuthoritativeRegion
	for i := len(h.Admissions) - 1; i >= 0; i-- {
		a := h.Admissions[i]
		if generation(a.Generation) != want || !v.sameRegion(a.Region, region) {
			continue
		}
		if a.Mode != "cold" {
			return nil, ErrUnsafe
		}
		pred := generation(a.Predecessor)
		if pred == want {
			return nil, ErrUnsafe
		}
		want = pred
		region = a.PredecessorRegion
		if region == "" || v.sameRegion(region, v.AuthoritativeRegion) {
			if a.Covered {
				return nil, ErrUnsafe
			}
			live[want] = true
		}
		if want == marker {
			if region != "" && !v.sameRegion(region, v.AuthoritativeRegion) {
				return nil, ErrUnsafe
			}
			return live, nil
		}
	}
	return nil, ErrUnsafe
}

// LiveGenerations is the region-qualified live generation set the engine
// protects at location: the holder's generation, the marker's, every
// generation the pinned versions belong to, and the unfinished cold
// admission chain walked back to the marker in the authoritative region.
// It is protected itself, exported so a view that needs the live set (the
// moves files live set) enumerates the same generations rather than a
// parallel walk. An error is protected's refusal.
func LiveGenerations(c *hostclaim.Claim, v View, location string) (map[string]bool, error) {
	if c == nil {
		return nil, ErrUnsafe
	}
	return protected(c, v, location)
}

// Decide uses a complete namespace version listing supplied by Discover.
// LegacyIncarnation is a caller-supplied, evidenced archive label only. It is
// never used to discover locations, and its absence refuses legacy cleanup.
func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, legacyIncarnation string) (Decision, error) {
	var d Decision
	if obs.Absent || obs.Claim == nil || obs.ETag == "" || v.ClaimETag != obs.ETag || v.Region == "" {
		return d, ErrUnsafe
	}
	c := obs.Claim
	if err := c.Validate(); err != nil {
		return d, err
	}
	joining := c.Holder == nil && c.Op.Kind == hostclaim.KindParking && c.Op.Intent["park_sweep"] == "v1"
	if !c.Free() && c.Op.Kind != hostclaim.KindServing && !joining {
		return d, nil
	}
	for _, o := range c.Obligations {
		if v.sameRegion(o.Region, v.Region) {
			return d, nil
		}
	}
	live, liveErr := protected(c, v, "")
	if liveErr != nil && c.Holder != nil {
		d.Alerts = append(d.Alerts, Alert{c.Holder.Location, "marker does not match the claim or an unfinished admission run"})
	}
	groups := map[Candidate]bool{}
	refused := map[string]bool{}
	seen := map[Ref]bool{}
	for _, x := range versions {
		loc, err := locationOf(c.Host, x.Key)
		if err != nil || x.VersionID == "" || seen[ref(x)] {
			return Decision{}, ErrUnsafe
		}
		seen[ref(x)] = true
		if isLock(c.Host, loc, x.Key) {
			continue
		}
		if refusal := v.RefusedLocations[loc]; refusal != nil && !(c.Holder != nil && loc == c.Holder.Location && v.sameRegion(v.Region, v.AuthoritativeRegion)) {
			if !refused[loc] {
				d.Alerts = append(d.Alerts, Alert{loc, refusal.Error()})
				refused[loc] = true
			}
			continue
		}
		candidate := Candidate{Location: loc}
		if tail := strings.TrimPrefix(loc, "instances/"+c.Host+"/"); tail != loc {
			candidate.Incarnation = tail
		} else {
			candidate.Incarnation = legacyIncarnation
			if loc == "instances/"+c.Host && v.ArchivedIncarnations[loc] != "" {
				candidate.Incarnation = v.ArchivedIncarnations[loc]
			}
		}
		if c.Holder != nil && loc == c.Holder.Location && v.sameRegion(v.Region, v.AuthoritativeRegion) {
			if liveErr != nil {
				continue
			}
			if x.Key == v.Marker.Key || isBuildState(c.Holder.Location, x.Key) || v.Pinned[ref(x)] {
				continue
			}
			g, ok := v.GenerationOf(ref(x))
			if !ok {
				return Decision{}, ErrUnsafe
			}
			g = generation(g)
			if live[g] {
				continue
			}
			candidate.Generation = g
			candidate.Incarnation = c.Holder.Incarnation
		}
		if !hex64(candidate.Incarnation) {
			return Decision{}, ErrUnsafe
		}
		groups[candidate] = true
	}
	for c := range groups {
		d.Candidates = append(d.Candidates, c)
	}
	sort.Slice(d.Candidates, func(i, j int) bool {
		a, b := d.Candidates[i], d.Candidates[j]
		if a.Location == b.Location {
			return a.Generation < b.Generation
		}
		return a.Location < b.Location
	})
	return d, nil
}

// Discover expands both namespaces across every version page. This is a
// conservative superset of delimiter discovery: deriving first child prefixes
// from the complete version listing also discovers marker-only locations.
func (e Engine) Discover(ctx context.Context, host string) ([]lease.ObjectVersion, error) {
	var out []lease.ObjectVersion
	for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} {
		vs, err := e.Store.ListVersions(ctx, e.Bucket, p)
		if err != nil {
			return nil, err
		}
		out = append(out, vs...)
	}
	return out, nil
}

func parkedGenerations(c *hostclaim.Claim, v View, chores []hostclaim.Obligation, versions []lease.ObjectVersion) ([]string, error) {
	if c.Holder == nil || len(c.Holder.Admissions) == 0 {
		return nil, nil
	}
	seen := map[string]bool{}
	for _, o := range chores {
		if o.Park.Location != c.Holder.Location || o.Park.Incarnation != c.Holder.Incarnation {
			continue
		}
		for _, x := range versions {
			// The holder location's build state belongs to no generation
			// (ruling R-PARK), so it names none parked, whichever region
			// parked it.
			if !strings.HasPrefix(x.Key, c.Holder.Location+"/") || isLock(c.Host, o.Park.Location, x.Key) || x.Key == v.Marker.Key || isBuildState(c.Holder.Location, x.Key) {
				continue
			}
			inScope, err := scopeContains(c.Host, o.Park.Location, x.Key)
			if err != nil {
				return nil, err
			}
			if !inScope {
				continue
			}
			if v.GenerationOf == nil {
				return nil, ErrUnsafe
			}
			g, ok := v.GenerationOf(ref(x))
			if !ok {
				return nil, ErrUnsafe
			}
			g = generation(g)
			if o.Kind != "park-generation" || o.Generation == g {
				seen[g] = true
			}
		}
	}
	var generations []string
	for g := range seen {
		generations = append(generations, g)
	}
	sort.Strings(generations)
	return generations, nil
}

// Sweep queues every decision in one claim CAS; Run is the common executor.
// EvidenceAt is genuine prior stop/fence evidence for the residue. A sweep
// uses listing-time evidence for moves/<host> only after VerifyMovesSweep
// proves that the newest retired move was fenced or released before listing.
// Other namespaces still require their original stop/fence evidence.
func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (Decision, error) {
	return e.sweep(ctx, hostclaim.Observed{Host: host}, nil, legacyIncarnation, evidenceAt, requestLimit, false)
}

// SweepObserved binds retirement evidence to the exact claim observation and
// complete version census from which it was verified. Any claim or namespace
// change requires new evidence; it is never folded into the old timestamp.
func (e Engine) SweepObserved(ctx context.Context, expected hostclaim.Observed, versions []lease.ObjectVersion, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (Decision, error) {
	if expected.Host == "" || expected.Absent || expected.Claim == nil || expected.ETag == "" {
		return Decision{}, ErrUnsafe
	}
	return e.sweep(ctx, expected, versions, legacyIncarnation, evidenceAt, requestLimit, true)
}

func sameCensus(a, b []lease.ObjectVersion) bool {
	if len(a) != len(b) {
		return false
	}
	indexed := make(map[Ref]lease.ObjectVersion, len(a))
	for _, version := range a {
		r := ref(version)
		if r.Key == "" || r.VersionID == "" {
			return false
		}
		if _, duplicate := indexed[r]; duplicate {
			return false
		}
		indexed[r] = version
	}
	seen := make(map[Ref]bool, len(b))
	for _, version := range b {
		r := ref(version)
		want, ok := indexed[r]
		if !ok || seen[r] || !reflect.DeepEqual(want, version) {
			return false
		}
		seen[r] = true
	}
	return true
}

func (e Engine) sweep(ctx context.Context, expected hostclaim.Observed, expectedVersions []lease.ObjectVersion, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration, bound bool) (Decision, error) {
	var empty Decision
	if e.View == nil || e.NewID == nil || e.Authorize == nil || e.Now == nil {
		return empty, ErrUnsafe
	}
	host := expected.Host
	obs, err := hostclaim.Read(ctx, e.Claims, host)
	if err != nil {
		return empty, err
	}
	if bound && (obs.ETag != expected.ETag || !reflect.DeepEqual(obs.Claim, expected.Claim)) {
		return empty, ErrChanged
	}
	if obs.Absent || obs.Claim == nil {
		return empty, ErrUnsafe
	}
	v, err := e.View(ctx, obs)
	if err != nil {
		return empty, err
	}
	home, err := BoundHome(e.HomeRegion, obs.Claim.Op.Intent["home_region"])
	if err != nil {
		return empty, err
	}
	v.HomeRegion = home
	listingStarted := e.Now()
	vs, err := e.Discover(ctx, host)
	listedAt := e.Now()
	if err != nil {
		return empty, err
	}
	if bound && !sameCensus(vs, expectedVersions) {
		return empty, ErrChanged
	}
	d, err := Decide(obs, v, vs, legacyIncarnation)
	if err != nil || len(d.Candidates) == 0 {
		return d, err
	}
	var movesEvidence time.Time
	candidates := d.Candidates[:0]
	for _, candidate := range d.Candidates {
		if candidate.Location == "moves/"+host {
			if e.VerifyMovesSweep == nil || listedAt.Before(listingStarted) {
				continue
			}
			detached, copyErr := moveSweepObservation(obs)
			if copyErr != nil {
				return d, copyErr
			}
			proof, verifyErr := e.VerifyMovesSweep(ctx, detached, append([]lease.ObjectVersion(nil), vs...))
			if verifyErr != nil || proof.Check(obs, candidate.Incarnation, listingStarted) != nil {
				continue
			}
			movesEvidence = listedAt
		}
		candidates = append(candidates, candidate)
	}
	d.Candidates = candidates
	if len(d.Candidates) == 0 {
		return d, nil
	}
	var op string
	if obs.Claim.Holder != nil {
		op = fmt.Sprintf("session:%d", obs.Claim.Holder.Session)
	} else if obs.Claim.Op.Kind == hostclaim.KindParking {
		op = obs.Claim.Op.ID
	} else {
		op = e.NewID()
		if !token(op) {
			return d, ErrUnsafe
		}
	}
	chores := make([]hostclaim.Obligation, 0, len(d.Candidates))
	for _, x := range d.Candidates {
		boundAt := evidenceAt
		if x.Location == "moves/"+host {
			boundAt = movesEvidence
		}
		o := hostclaim.Obligation{ID: e.NewID(), Op: op, Kind: "park-replica", Region: e.Region, Generation: x.Generation, Retirement: e.NewID(), Owner: "region:" + e.Region, Park: &hostclaim.ParkScope{Location: x.Location, Incarnation: x.Incarnation, EvidenceAt: boundAt, RequestLimit: requestLimit}}
		if x.Generation != "" {
			o.Kind = "park-generation"
		}
		if !token(o.ID) || validateScope(host, o) != nil {
			return d, ErrUnsafe
		}
		if err := e.Authorize(ctx, obs, o); err != nil {
			return d, err
		}
		chores = append(chores, o)
	}
	generations, err := parkedGenerations(obs.Claim, v, chores, vs)
	if err != nil {
		return d, err
	}
	if bound {
		fresh, discoverErr := e.Discover(ctx, host)
		if discoverErr != nil {
			return d, discoverErr
		}
		if !sameCensus(fresh, expectedVersions) {
			return d, ErrChanged
		}
	}
	_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
		if c.Holder != nil {
			if err := hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, e.Region, generations)(c); err != nil {
				return err
			}
		}
		if c.Free() {
			c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: op, Intent: map[string]string{"park_sweep": "v1"}}
		}
		c.Obligations = append(c.Obligations, chores...)
		return nil
	})
	if err != nil {
		return d, err
	}
	for _, o := range chores {
		if err := e.Run(ctx, Work{host, o.ID, o.Op}); err != nil {
			return d, fmt.Errorf("sweep %s: %w", o.ID, err)
		}
	}
	return d, nil
}

// ResidueLocations uses the same namespace and lock classifier as Decide.
func ResidueLocations(host string, versions []lease.ObjectVersion) ([]string, error) {
	seen := map[string]bool{}
	for _, v := range versions {
		loc, err := locationOf(host, v.Key)
		if err != nil || v.VersionID == "" {
			return nil, ErrUnsafe
		}
		if !isLock(host, loc, v.Key) {
			seen[loc] = true
		}
	}
	out := make([]string, 0, len(seen))
	for loc := range seen {
		out = append(out, loc)
	}
	sort.Strings(out)
	return out, nil
}