package park

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

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

// Ruling R-PARK (settings-4 Task 8b): <holder.Location>/build-state.json,
// every version of it, is live state beside the lease, the marker's twin.
// The engine skips it wherever it skips the marker key at the holder
// location, and refuses a whole-location park of the live holder location
// because of it, exactly as for the marker. Anywhere else (a retired
// location, a moves lane) it is residue like any other object.

// TestHolderBuildStateIsNeverACandidate: a sweep never makes the live
// location's build state a candidate and never parks a version of it; the
// same key at a retired location is a candidate and is parked with the
// location, and a lane's copy stays a moves/<host> candidate.
func TestHolderBuildStateIsNeverACandidate(t *testing.T) {
	f, obs, v, _ := sweepFixture(t)
	key := release.BuildStateKey(f.ob.Park.Location)
	older := f.b.Put("data", key, []byte("older"))
	current := f.b.Put("data", key, []byte("current"))
	retiredInc := strings.Repeat("a", 64)
	retired := "instances/" + f.w.Host + "/" + retiredInc
	stale := release.BuildStateKey(retired)
	f.b.Put("data", stale, []byte("stale"))
	lane := "moves/" + f.w.Host + "/mv_1/state/build-state.json"
	f.b.Put("data", lane, []byte("lane"))
	vs, err := f.e.Discover(t.Context(), f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil {
		t.Fatalf("Decide: %v", err)
	}
	want := []Candidate{
		{Location: f.ob.Park.Location, Incarnation: f.ob.Park.Incarnation, Generation: "g0"},
		{Location: retired, Incarnation: retiredInc},
		{Location: "moves/" + f.w.Host, Incarnation: f.ob.Park.Incarnation},
	}
	if len(d.Alerts) != 0 || len(d.Candidates) != len(want) {
		t.Fatalf("decision %+v, want %+v", d, want)
	}
	for _, w := range want {
		found := false
		for _, c := range d.Candidates {
			found = found || c == w
		}
		if !found {
			t.Fatalf("decision %+v lacks %+v", d.Candidates, w)
		}
	}
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		x := v
		x.ClaimETag = o.ETag
		return x, nil
	}
	if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err != nil {
		t.Fatalf("Sweep: %v", err)
	}
	for _, id := range []string{older, current} {
		if _, err := f.b.GetVersion(t.Context(), "data", key, id); err != nil {
			t.Fatalf("the live build state version %s was parked: %v", id, err)
		}
	}
	left, err := f.b.ListVersions(t.Context(), "data", retired+"/")
	if err != nil || len(left) != 0 {
		t.Fatalf("the retired location still holds %+v (%v)", left, err)
	}
}

// twinFixture is a holder location, serving, holding one generation that
// is not live (g0) beside its build state, with one chore of kind at the
// holder location, seen from an authoritative region or not.
func twinFixture(t *testing.T, kind string, authoritative, withState bool) (*fixture, string) {
	t.Helper()
	f := newFixture(t)
	p := f.ob.Park.Location + "/"
	f.ob.Kind = kind
	if kind == "park-generation" {
		f.ob.Generation = "g0"
	}
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
		c.Holder.Generation, c.Holder.Session = "g1", 1
		c.Holder.Admissions = []hostclaim.AdmissionRecord{{Session: 1, Generation: "g1", Mode: "cold", Region: "ie"}}
		c.Obligations = []hostclaim.Obligation{f.ob}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	f.b.Put("data", p+"g0/data", []byte("g0"))
	key := release.BuildStateKey(f.ob.Park.Location)
	if withState {
		f.b.Put("data", key, []byte("older"))
		f.b.Put("data", key, []byte("current"))
	}
	classify := func(r Ref) (string, bool) {
		rel, ok := strings.CutPrefix(r.Key, p)
		if !ok {
			return "", false
		}
		g, _, ok := strings.Cut(rel, "/")
		return g, ok
	}
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		v := View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk", GenerationOf: classify}
		if authoritative {
			v.AuthoritativeRegion, v.MarkerKnown, v.MarkerGeneration = "ie", true, "g1"
			v.Marker = Ref{Key: p + "generation.json"}
		}
		return v, nil
	}
	return f, key
}

// TestHolderBuildStateIsParkedOnlyAsTheMarkerIs: at the live holder
// location a generation's park passes over the build state in either
// region; a whole-location park refuses because of it in the authoritative
// region (the control without it completes), and in a region that is not
// authoritative parks it with everything else, as it parks the marker.
func TestHolderBuildStateIsParkedOnlyAsTheMarkerIs(t *testing.T) {
	versionsAt := func(t *testing.T, f *fixture, key string) int {
		t.Helper()
		vs, err := f.b.ListVersions(t.Context(), "data", key)
		if err != nil {
			t.Fatal(err)
		}
		n := 0
		for _, v := range vs {
			if v.Key == key {
				n++
			}
		}
		return n
	}
	for name, authoritative := range map[string]bool{"authoritative": true, "not authoritative": false} {
		t.Run(name+": a generation's park passes it over", func(t *testing.T) {
			f, key := twinFixture(t, "park-generation", authoritative, true)
			if err := f.e.Run(t.Context(), f.w); err != nil {
				t.Fatalf("Run: %v", err)
			}
			f.done0(t)
			if n := versionsAt(t, f, key); n != 2 {
				t.Fatalf("%d versions of the build state left, want both", n)
			}
			if n := versionsAt(t, f, f.ob.Park.Location+"/g0/data"); n != 0 {
				t.Fatalf("g0 was not parked: %d versions", n)
			}
		})
	}
	t.Run("authoritative: a whole-location park refuses because of it", func(t *testing.T) {
		control, _ := twinFixture(t, "park-replica", true, false)
		if err := control.e.Run(t.Context(), control.w); err != nil {
			t.Fatalf("control without the build state: %v", err)
		}
		f, _ := twinFixture(t, "park-replica", true, true)
		before := f.b.Snapshot()
		if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrUnsafe) {
			t.Fatalf("Run = %v, want ErrUnsafe", err)
		}
		if f.b.Snapshot() != before {
			t.Fatal("the refused park changed storage")
		}
	})
	t.Run("not authoritative: a whole-location park parks it", func(t *testing.T) {
		f, key := twinFixture(t, "park-replica", false, true)
		if err := f.e.Run(t.Context(), f.w); err != nil {
			t.Fatalf("Run: %v", err)
		}
		f.done0(t)
		if n := versionsAt(t, f, key); n != 0 {
			t.Fatalf("%d versions of the build state left, want none", n)
		}
	})
}

// done0 asserts the chore is acknowledged.
func (f *fixture) done0(t *testing.T) {
	t.Helper()
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	if len(obs.Claim.Obligations) != 0 {
		t.Fatalf("obligation still open: %+v", obs.Claim.Obligations)
	}
}