package park

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"net/url"
	"reflect"
	"sort"
	"strings"
	"time"

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

const SignatureAge = 15 * time.Minute

var (
	ErrPending = errors.New("park: waiting for verification after horizon")
	ErrUnsafe  = errors.New("park: unsafe or incomplete input")
	ErrChanged = errors.New("park: authority changed")
)

type Ref struct{ Key, VersionID string }

func ref(v lease.ObjectVersion) Ref { return Ref{v.Key, v.VersionID} }

type Copy struct {
	Source      lease.ObjectVersion
	Destination Ref
}
type Manifest struct {
	V                                                                           int
	Host, Obligation, Op, Retirement, Region, Location, Incarnation, Generation string
	Archive                                                                     []Copy
	Delete                                                                      []lease.ObjectVersion
}
type Work struct{ Host, ID, Op string }

// Close stops renewal; Release conditionally removes only the held lock version.
// Acquire must renew until Close, and Check must refuse expired/lost leases.
type HeldLease interface {
	Check(context.Context) error
	Release(context.Context) error
	Close()
}
type Leases interface {
	Acquire(context.Context, string, string) (HeldLease, error)
}

type Engine struct {
	HomeRegion     string
	Store          lease.VersionedStore
	Claims         release.Store
	Bucket, Region string
	Leases         Leases
	Now            func() time.Time
	NewID          func() string
	// View is supplied by moves/Plan 7b, using precisely this fresh observation.
	View func(context.Context, hostclaim.Observed) (View, error)
	// Authorize checks phase-specific stop evidence/fences and chore ownership.
	// It must not manufacture stop evidence from a stopped parent alone.
	Authorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
	// VerifyMovesSweep supplies the precondition for listing-time moves evidence.
	VerifyMovesSweep func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (MoveSweepProof, error)
	// After is a test crash point, called after durable effects. Nil in callers.
	After func(string) error
}

func (e Engine) after(step string) error {
	if e.After != nil {
		return e.After(step)
	}
	return nil
}
func token(s string) bool {
	return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
}
func hex64(s string) bool { _, err := hex.DecodeString(s); return len(s) == 64 && err == nil }
func find(c *hostclaim.Claim, w Work) (hostclaim.Obligation, bool) {
	if c == nil {
		return hostclaim.Obligation{}, false
	}
	for _, o := range c.Obligations {
		if o.ID == w.ID && o.Op == w.Op {
			return o, true
		}
	}
	return hostclaim.Obligation{}, false
}
func validateScope(host string, o hostclaim.Obligation) error {
	p := o.Park
	if p.Validate(host) != nil || !token(o.Retirement) || o.Region == "" || o.Owner != "region:"+o.Region {
		return ErrUnsafe
	}
	if o.Kind != "park-replica" && o.Kind != "park-generation" {
		return ErrUnsafe
	}
	if o.Kind == "park-generation" && o.Generation == "" {
		return ErrUnsafe
	}
	return nil
}

// Bind persists recovery inputs before the first data-bucket effect. Existing
// Plan 1 chores have no Park field; callers supply real stop-evidence time and
// a deployment-wide upper bound enforced by every relevant request sender.
func (e Engine) Bind(ctx context.Context, w Work, p hostclaim.ParkScope) error {
	obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
	if err != nil {
		return err
	}
	o, ok := find(obs.Claim, w)
	if !ok {
		return ErrChanged
	}
	if o.ParkTarget != "" && o.ParkTarget != p.Location {
		return ErrUnsafe
	}
	if o.Park != nil {
		if o.Park.Location == p.Location && o.Park.Incarnation == p.Incarnation && o.Park.RequestLimit == p.RequestLimit && o.Park.EvidenceAt.Equal(p.EvidenceAt) {
			return nil
		}
		return ErrChanged
	}
	if obs.Claim.Holder == nil || (p.Location != obs.Claim.Holder.Location && !isMovesScope(w.Host, p.Location)) || p.Incarnation != obs.Claim.Holder.Incarnation {
		return ErrUnsafe
	}
	o.Park = &p
	if err := validateScope(w.Host, o); err != nil {
		return err
	}
	if e.Authorize == nil {
		return ErrUnsafe
	}
	if err := e.Authorize(ctx, obs, o); err != nil {
		return err
	}
	_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
		for i := range c.Obligations {
			if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
				c.Obligations[i].Park = &p
				return nil
			}
		}
		return ErrChanged
	})
	return err
}
func isMovesScope(host, location string) bool {
	base := "moves/" + host
	return location == base || strings.HasPrefix(location, base+"/")
}
func prefixes(location string) []string {
	return []string{location + "/"}
}

// locationOf partitions legacy level/generation children from hex incarnation
// children. No retired-incarnations lookup.
func locationOf(host, key string) (string, error) {
	for _, root := range []string{"instances/", "moves/"} {
		base := root + host + "/"
		if !strings.HasPrefix(key, base) {
			continue
		}
		tail := strings.TrimPrefix(key, base)
		if tail == "" {
			return "", ErrUnsafe
		}
		if root == "moves/" {
			return root + host, nil
		}
		first, _, child := strings.Cut(tail, "/")
		if child && hex64(first) {
			return "instances/" + host + "/" + first, nil
		}
		return "instances/" + host, nil
	}
	return "", ErrUnsafe
}
func scopeContains(host, location, key string) (bool, error) {
	if isMovesScope(host, location) {
		return strings.HasPrefix(key, location+"/"), nil
	}
	loc, err := locationOf(host, key)
	return loc == location, err
}

// isBuildState reports whether key is the build state beside the lease at
// location (release.BuildStateKey; ruling R-PARK). At the live holder
// location it is the marker's twin: live state that belongs to no
// generation, passed over wherever the marker key is, and a whole-location
// park there refuses because of it as it refuses because of the marker.
// Callers ask it only of the holder's location; at a retired location, or
// in a moves lane, the same file name is residue like any other object.
func isBuildState(location, key string) bool {
	return key == release.BuildStateKey(location)
}
func isLock(host, location, key string) bool {
	for _, p := range prefixes(location) {
		if key == p+"lock.json" {
			return true
		}
	}
	return false
}
func (e Engine) list(ctx context.Context, host, location string) ([]lease.ObjectVersion, error) {
	var out []lease.ObjectVersion
	for _, p := range prefixes(location) {
		vs, err := e.Store.ListVersions(ctx, e.Bucket, p)
		if err != nil {
			return nil, err
		}
		out = append(out, vs...)
	}
	seen := map[Ref]bool{}
	filtered := out[:0]
	for _, v := range out {
		inScope, err := scopeContains(host, location, v.Key)
		if err != nil || v.VersionID == "" || seen[ref(v)] {
			return nil, ErrUnsafe
		}
		seen[ref(v)] = true
		if inScope && !isLock(host, location, v.Key) {
			filtered = append(filtered, v)
		}
	}
	sort.Slice(filtered, func(i, j int) bool {
		a, b := filtered[i], filtered[j]
		if a.Key == b.Key {
			return a.VersionID < b.VersionID
		}
		return a.Key < b.Key
	})
	return filtered, nil
}
func archiveBase(host string, o hostclaim.Obligation) string {
	return "parked/" + o.Park.Incarnation + "/" + host + "/" + o.Retirement + "/"
}
func destination(host string, o hostclaim.Obligation, v lease.ObjectVersion) string {
	for _, p := range prefixes(o.Park.Location) {
		if strings.HasPrefix(v.Key, p) {
			kind := strings.SplitN(p, "/", 2)[0]
			return archiveBase(host, o) + kind + "/" + strings.TrimPrefix(v.Key, p) + "@" + url.QueryEscape(v.VersionID)
		}
	}
	return ""
}
func (e Engine) current(ctx context.Context, key string) (lease.ObjectVersion, bool, error) {
	vs, err := e.Store.ListVersions(ctx, e.Bucket, key)
	if err != nil {
		return lease.ObjectVersion{}, false, err
	}
	var got lease.ObjectVersion
	ok := false
	for _, v := range vs {
		if v.Key == key && v.IsLatest {
			if ok || v.VersionID == "" {
				return got, false, ErrUnsafe
			}
			got, ok = v, true
		}
	}
	if ok && got.DeleteMarker {
		return got, false, ErrUnsafe
	}
	return got, ok, nil
}
func (e Engine) manifest(ctx context.Context, w Work, o hostclaim.Obligation) (*Manifest, error) {
	v, ok, err := e.current(ctx, archiveBase(w.Host, o)+"manifest.json")
	if err != nil || !ok {
		return nil, err
	}
	b, err := e.Store.GetVersion(ctx, e.Bucket, v.Key, v.VersionID)
	if err != nil {
		return nil, err
	}
	var m Manifest
	d := json.NewDecoder(bytes.NewReader(b))
	d.DisallowUnknownFields()
	if err := d.Decode(&m); err != nil {
		return nil, err
	}
	if _, err := d.Token(); err != io.EOF {
		return nil, ErrUnsafe
	}
	if m.V != 1 || m.Host != w.Host || m.Obligation != w.ID || m.Op != w.Op || m.Retirement != o.Retirement || m.Region != o.Region || m.Location != o.Park.Location || m.Incarnation != o.Park.Incarnation || m.Generation != o.Generation {
		return nil, ErrUnsafe
	}
	deletes := map[Ref]lease.ObjectVersion{}
	for _, v := range m.Delete {
		inScope, err := scopeContains(w.Host, o.Park.Location, v.Key)
		if err != nil || !inScope || v.VersionID == "" || isLock(w.Host, o.Park.Location, v.Key) {
			return nil, ErrUnsafe
		}
		if _, ok := deletes[ref(v)]; ok {
			return nil, ErrUnsafe
		}
		deletes[ref(v)] = v
	}
	copies := map[Ref]bool{}
	for _, a := range m.Archive {
		v, ok := deletes[ref(a.Source)]
		if !ok || v.DeleteMarker || !reflect.DeepEqual(v, a.Source) || copies[ref(v)] || a.Destination.Key != destination(w.Host, o, v) || a.Destination.VersionID == "" {
			return nil, ErrUnsafe
		}
		copies[ref(v)] = true
	}
	for r, v := range deletes {
		if !v.DeleteMarker && !copies[r] {
			return nil, ErrUnsafe
		}
	}
	return &m, nil
}
func (e Engine) observe(ctx context.Context, w Work, held HeldLease, expected hostclaim.Obligation) (hostclaim.Observed, View, error) {
	if err := held.Check(ctx); err != nil {
		return hostclaim.Observed{}, View{}, err
	}
	obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
	if err != nil {
		return obs, View{}, err
	}
	o, ok := find(obs.Claim, w)
	if !ok || !reflect.DeepEqual(o, expected) {
		return obs, View{}, ErrChanged
	}
	if err := e.Authorize(ctx, obs, o); err != nil {
		return obs, View{}, err
	}
	v, err := e.View(ctx, obs)
	if err != nil {
		return obs, v, err
	}
	if v.ClaimETag != obs.ETag || !e.OwnsRegion(v.Region) {
		return obs, v, ErrChanged
	}
	home, err := BoundHome(e.HomeRegion, obs.Claim.Op.Intent["home_region"])
	if err != nil {
		return obs, v, err
	}
	v.HomeRegion = home
	return obs, v, nil
}
func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, held HeldLease, vs []lease.ObjectVersion) (hostclaim.Observed, View, []lease.ObjectVersion, error) {
	obs, v, err := e.observe(ctx, w, held, o)
	if err != nil {
		return obs, v, nil, err
	}
	live, err := protected(obs.Claim, v, o.Park.Location)
	if err != nil {
		return obs, v, nil, err
	}
	out := make([]lease.ObjectVersion, 0, len(vs))
	for _, x := range vs {
		inScope, err := scopeContains(w.Host, o.Park.Location, x.Key)
		if err != nil || !inScope || x.VersionID == "" {
			return obs, v, nil, ErrUnsafe
		}
		if isLock(w.Host, o.Park.Location, x.Key) {
			return obs, v, nil, ErrUnsafe
		}
		if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && o.Park.Location == obs.Claim.Holder.Location && v.sameRegion(v.Region, v.AuthoritativeRegion) {
			if x.Key == v.Marker.Key || isBuildState(obs.Claim.Holder.Location, x.Key) {
				if o.Kind == "park-generation" {
					continue
				}
				return obs, v, nil, ErrUnsafe
			}
			g, ok := v.GenerationOf(ref(x))
			if !ok {
				return obs, v, nil, ErrUnsafe
			}
			g = generation(g)
			if o.Kind == "park-generation" && g != o.Generation {
				continue
			}
			if o.Kind == "park-generation" && !live[g] && v.Pinned[ref(x)] && v.FilePins[ref(x)] && strings.HasPrefix(x.Key, obs.Claim.Holder.Location+"/files/") {
				continue
			}
			if live[g] || v.Pinned[ref(x)] || ref(x) == v.Marker {
				return obs, v, nil, ErrUnsafe
			}
		} else if o.Kind == "park-generation" {
			// The holder location's build state is no generation's, as the
			// view's classifier answers for its marker: never this chore's.
			if obs.Claim.Holder != nil && isBuildState(obs.Claim.Holder.Location, x.Key) {
				continue
			}
			if v.GenerationOf == nil {
				return obs, v, nil, ErrUnsafe
			}
			g, ok := v.GenerationOf(ref(x))
			if !ok {
				return obs, v, nil, ErrUnsafe
			}
			if generation(g) != o.Generation {
				continue
			}
		}
		out = append(out, x)
	}
	return obs, v, out, nil
}

// Run performs one resumable event. ErrPending is a scheduler retry, never an
// acknowledgement. A newly rotated event is persisted before returning it.
func (e Engine) Run(ctx context.Context, w Work) error { return e.run(ctx, w, nil) }

// RunHeld borrows an exact location lease. The caller retains its renewal and
// release duties through publication. Only the configured lease resolver may
// borrow it; the same authorizer, census and engine effects still run.
func (e Engine) RunHeld(ctx context.Context, w Work, held HeldLease) error {
	if held == nil {
		return ErrUnsafe
	}
	return e.run(ctx, w, held)
}
func (e Engine) run(ctx context.Context, w Work, own HeldLease) error {
	if e.Store == nil || e.Claims == nil || e.Leases == nil || e.Now == nil || e.NewID == nil || e.View == nil || e.Authorize == nil || e.Bucket == "" || e.Region == "" {
		return ErrUnsafe
	}
	initial, err := hostclaim.Read(ctx, e.Claims, w.Host)
	if err != nil {
		return err
	}
	if initial.Absent {
		return ErrChanged
	}
	o, ok := find(initial.Claim, w)
	if !ok {
		return nil
	}
	if err := validateScope(w.Host, o); err != nil {
		return err
	}
	if !e.OwnsRegion(o.Region) {
		return ErrUnsafe
	}
	var held HeldLease
	if own == nil {
		held, err = e.Leases.Acquire(ctx, e.Bucket, o.Park.Location)
	} else {
		resolver, ok := e.Leases.(interface {
			AcquireHeld(context.Context, string, string, HeldLease) (HeldLease, error)
		})
		if !ok {
			return ErrUnsafe
		}
		held, err = resolver.AcquireHeld(ctx, e.Bucket, o.Park.Location, own)
	}
	if err != nil {
		return err
	}
	if held == nil {
		return ErrUnsafe
	}
	defer held.Close()
	if bound, ok := held.(interface {
		Context(context.Context) context.Context
	}); ok {
		ctx = bound.Context(ctx)
	}
	if _, _, err = e.observe(ctx, w, held, o); err != nil {
		return err
	}
	m, err := e.manifest(ctx, w, o)
	if err != nil {
		return err
	}
	if m == nil {
		vs, err := e.list(ctx, w.Host, o.Park.Location)
		if err != nil {
			return err
		}
		_, _, vs, err = e.selected(ctx, w, o, held, vs)
		if err != nil {
			return err
		}
		m = &Manifest{V: 1, Host: w.Host, Obligation: w.ID, Op: w.Op, Retirement: o.Retirement, Region: o.Region, Location: o.Park.Location, Incarnation: o.Park.Incarnation, Generation: o.Generation, Delete: vs}
		for _, v := range vs {
			if v.DeleteMarker {
				continue
			}
			if _, _, _, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v}); err != nil {
				return err
			}
			dst := destination(w.Host, o, v)
			id, _, err := e.Store.CopyVersion(ctx, e.Bucket, v.Key, v.VersionID, e.Bucket, dst, true)
			if errors.Is(err, lease.ErrDestinationExists) {
				got, ok, getErr := e.current(ctx, dst)
				if getErr != nil {
					return getErr
				}
				if !ok || got.Size != v.Size {
					return ErrUnsafe
				}
				id, err = got.VersionID, nil
			}
			if err != nil {
				return err
			}
			if id == "" {
				return ErrUnsafe
			}
			m.Archive = append(m.Archive, Copy{v, Ref{dst, id}})
			if err := e.after("copy"); err != nil {
				return err
			}
		}
		if _, _, _, err = e.selected(ctx, w, o, held, m.Delete); err != nil {
			return err
		}
		raw, err := json.Marshal(m)
		if err != nil {
			return err
		}
		_, err = e.Store.PutIfAbsent(ctx, e.Bucket, archiveBase(w.Host, o)+"manifest.json", raw)
		if errors.Is(err, lease.ErrExists) {
			m, err = e.manifest(ctx, w, o)
		}
		if err != nil {
			return err
		}
		if m == nil {
			return ErrUnsafe
		}
		if err := e.after("manifest"); err != nil {
			return err
		}
	}
	// Existing manifests go directly to their recorded deletes: never re-copy
	// vanished source versions. The live-set admission check still applies.
	for _, v := range m.Delete {
		_, _, selected, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v})
		if err != nil {
			return err
		}
		if len(selected) != 1 {
			return ErrUnsafe
		}
		if err := e.Store.DeleteVersion(ctx, e.Bucket, v.Key, v.VersionID); err != nil {
			return err
		}
		if err := e.after("delete"); err != nil {
			return err
		}
	}
	started := e.Now()
	vs, err := e.list(ctx, w.Host, o.Park.Location)
	if err != nil {
		return err
	}
	obs, view, vs, err := e.selected(ctx, w, o, held, vs)
	if err != nil {
		return err
	}
	if err := e.after("verify"); err != nil {
		return err
	}
	deadline := o.Park.EvidenceAt.Add(SignatureAge + o.Park.RequestLimit)
	if !started.After(deadline) {
		return ErrPending
	}
	generations, err := parkedGenerations(obs.Claim, view, []hostclaim.Obligation{o}, m.Delete)
	if err != nil {
		return err
	}
	cover := func(c *hostclaim.Claim) error {
		if c.Holder == nil {
			return nil
		}
		if err := hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, o.Region, generations)(c); err != nil {
			return err
		}
		if e.HomeRegion != "" && e.OwnsRegion(HomeRegion) {
			other := HomeRegion
			if o.Region == HomeRegion {
				other = e.HomeRegion
			}
			return hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, other, generations)(c)
		}
		return nil
	}
	if len(vs) > 0 {
		next := e.NewID()
		if !token(next) || next == o.Retirement {
			return ErrUnsafe
		}
		_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
			if err := cover(c); err != nil {
				return err
			}
			for i := range c.Obligations {
				if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
					c.Obligations[i].Retirement = next
					return nil
				}
			}
			return ErrChanged
		})
		if err != nil {
			return err
		}
		return ErrPending
	}
	if err := e.after("before-completion"); err != nil {
		return err
	}
	if err := held.Check(ctx); err != nil {
		return err
	}
	_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
		if err := cover(c); err != nil {
			return err
		}
		if c.QuietAfter == nil {
			c.QuietAfter = map[string]time.Time{}
		}
		if c.QuietAfter[o.Region].Before(deadline) {
			c.QuietAfter[o.Region] = deadline
		}
		if err := hostclaim.CompleteObligation(w.Op, w.ID)(c); err != nil {
			return err
		}
		if c.Holder == nil && len(c.Obligations) == 0 && c.Op.Kind == hostclaim.KindParking && c.Op.Intent["park_sweep"] == "v1" {
			return hostclaim.FreeAfterPark(c.Op.ID)(c)
		}
		return nil
	})
	if err != nil {
		return err
	}
	if err := e.after("completion"); err != nil {
		return err
	}
	return held.Release(ctx)
}

// Parked reports whether a retirement's archive is complete: its manifest
// exists and every version in the manifest's deletion set is gone. This is
// what the landing's parking phase waits for (moves spec §7.6); it is not the
// acknowledgement, which still waits for the horizon and a clean listing in
// Run. It takes no lease and writes nothing.
//
// Anything it cannot read, or a manifest that does not match the chore, is an
// error and never true. A chore that is no longer in the claim returns
// ErrChanged: it has been acknowledged (or never existed), and the caller
// decides which from the claim it already holds.
func (e Engine) Parked(ctx context.Context, w Work) (bool, error) {
	if e.Store == nil || e.Claims == nil || e.Bucket == "" {
		return false, ErrUnsafe
	}
	obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
	if err != nil {
		return false, err
	}
	if obs.Absent {
		return false, ErrChanged
	}
	o, ok := find(obs.Claim, w)
	if !ok {
		return false, ErrChanged
	}
	if o.Park == nil {
		return false, ErrUnsafe
	}
	if err := validateScope(w.Host, o); err != nil {
		return false, err
	}
	m, err := e.manifest(ctx, w, o)
	if err != nil || m == nil {
		return false, err
	}
	// Each distinct listed key is listed once: a version that is still there
	// shows up under its key, and a listing that fails is an error.
	listed := map[string]map[string]bool{}
	var keys []string
	for _, d := range m.Delete {
		if listed[d.Key] == nil {
			listed[d.Key] = map[string]bool{}
			keys = append(keys, d.Key)
		}
		listed[d.Key][d.VersionID] = true
	}
	for _, key := range keys {
		vs, err := e.Store.ListVersions(ctx, e.Bucket, key)
		if err != nil {
			return false, err
		}
		for _, v := range vs {
			if v.Key != key {
				continue
			}
			if v.VersionID == "" {
				return false, ErrUnsafe
			}
			if listed[key][v.VersionID] {
				return false, nil
			}
		}
	}
	return true, nil
}