package park

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

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

func sweepFixture(t *testing.T) (*fixture, hostclaim.Observed, View, []lease.ObjectVersion) {
	f := newFixture(t)
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	obs, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations = nil
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
		c.Holder.Generation = "g3"
		c.Holder.Session = 3
		c.Holder.Admissions = []hostclaim.AdmissionRecord{
			{Session: 2, Generation: "g2", Predecessor: "g1", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
			{Session: 3, Generation: "g3", Predecessor: "g2", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	p := f.ob.Park.Location + "/"
	for _, g := range []string{"g0", "g1", "g2", "g3"} {
		f.b.Put("data", p+g+"/data", []byte(g))
	}
	f.b.Put("data", p+"generation.json", []byte("older"))
	markerID := f.b.Put("data", p+"generation.json", []byte("g1"))
	pinned := f.b.Put("data", p+"pinned/data", []byte("baseline"))
	f.b.Put("data", p+"pinned/data", []byte("newer"))
	classify := func(r Ref) (string, bool) {
		s := strings.TrimPrefix(r.Key, p)
		g, _, ok := strings.Cut(s, "/")
		if r.Key == p+"generation.json" {
			return "@marker", true
		}
		return g, ok
	}
	v := View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: "g1", Marker: Ref{p + "generation.json", markerID}, Pinned: map[Ref]bool{{p + "pinned/data", pinned}: true}, GenerationOf: classify}
	vs, err := f.e.Discover(t.Context(), f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	return f, obs, v, vs
}

func TestSweepMovesNamespaceIsSeparateFromLiveInstance(t *testing.T) {
	f := newFixture(t)
	allowFixtureMoveSweep(f)
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations = nil
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	liveKey := f.ob.Park.Location + "/0000/live"
	liveID := f.b.Put("data", liveKey, []byte("live"))
	for _, op := range []string{"move-1", f.ob.Park.Incarnation} {
		f.b.Put("data", "moves/"+f.w.Host+"/"+op+"/staging", []byte(op))
	}
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
	}
	d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit)
	if err != nil {
		t.Fatal(err)
	}
	want := []Candidate{{Location: "moves/" + f.w.Host, Incarnation: f.ob.Park.Incarnation}}
	if !reflect.DeepEqual(d.Candidates, want) {
		t.Fatalf("staging candidates=%+v, want %+v", d.Candidates, want)
	}
	vs, err := f.b.ListVersions(t.Context(), "data", "moves/"+f.w.Host+"/")
	if err != nil || len(vs) != 0 {
		t.Fatalf("staging residue: %+v %v", vs, err)
	}
	body, err := f.b.GetVersion(t.Context(), "data", liveKey, liveID)
	if err != nil || string(body) != "live" {
		t.Fatalf("live instance changed: %q %v", body, err)
	}
}

func TestSweepJoinsRunningHolderlessOperation(t *testing.T) {
	f := newFixture(t)
	f.seed()
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations = nil
		return hostclaim.FreeAfterPark("op")(c)
	})
	if err != nil {
		t.Fatal(err)
	}
	f.held.failed = true
	if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
		t.Fatalf("first region queue: %v", err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	first := obs.Claim.Obligations[0]
	second := f.e
	second.Region = "uk"
	b := leasetest.NewVersionedBucket()
	b.AddBucket("data")
	b.Put("data", f.ob.Park.Location+"/0000/data", []byte("second region"))
	second.Store = b
	second.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "uk"}, nil
	}
	second.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
		if o.Op != first.Op || o.Region != "uk" {
			return ErrUnsafe
		}
		return nil
	}
	if _, err := second.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
		t.Fatalf("second region did not join the running sweep: %v", err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	if obs.Claim.Holder != nil || obs.Claim.Op.ID != first.Op || len(obs.Claim.Obligations) != 2 || !reflect.DeepEqual(obs.Claim.Obligations[0], first) {
		t.Fatalf("joined sweep: %+v", obs.Claim)
	}
	joined := obs.Claim.Obligations[1]
	f.held.failed = false
	if err := f.e.Run(t.Context(), Work{f.w.Host, first.ID, first.Op}); err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || obs.Claim.Free() || len(obs.Claim.Obligations) != 1 {
		t.Fatalf("first completion freed an outstanding sweep: %+v %v", obs.Claim, err)
	}
	if err := second.Run(t.Context(), Work{f.w.Host, joined.ID, joined.Op}); err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || !obs.Claim.Free() {
		t.Fatalf("last completion did not free the sweep: %+v %v", obs.Claim, err)
	}
	deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + f.ob.Park.RequestLimit)
	for _, region := range []string{"ie", "uk"} {
		if !obs.Claim.QuietAfter[region].Equal(deadline) {
			t.Fatalf("quiet_after[%s]=%v, want %v", region, obs.Claim.QuietAfter[region], deadline)
		}
	}
}
func TestSweepProtectsUnfinishedRunAndPins(t *testing.T) {
	f, obs, v, vs := sweepFixture(t)
	d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil || len(d.Alerts) != 0 || len(d.Candidates) != 1 || d.Candidates[0].Generation != "g0" {
		t.Fatalf("%+v %v", d, err)
	}
	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.Fatal(err)
	}
	for _, x := range vs {
		if strings.Contains(x.Key, "/g0/") {
			continue
		}
		if _, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID); err != nil {
			t.Fatalf("live version lost: %+v %v", x, err)
		}
	}
}

func TestSweepUnfinishedRunRegionalCoverage(t *testing.T) {
	for _, scenario := range []string{"covered-foreign", "uncovered-foreign", "covered-local"} {
		t.Run(scenario, func(t *testing.T) {
			f, obs, v, _ := sweepFixture(t)
			_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
				c.Holder.Admissions = []hostclaim.AdmissionRecord{
					{Generation: "T", Predecessor: "g1", Region: "us", PredecessorRegion: "ie", Mode: "cold", Covered: scenario == "covered-local"},
					{Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: scenario == "covered-foreign"},
					{Generation: "g3", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold", Covered: scenario == "covered-foreign"},
				}
				return nil
			})
			if err != nil {
				t.Fatal(err)
			}
			for _, g := range []string{"T", "U"} {
				f.b.Put("data", f.ob.Park.Location+"/"+g+"/data", []byte(g))
			}
			before, err := f.e.Discover(t.Context(), f.w.Host)
			if err != nil {
				t.Fatal(err)
			}
			f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
				x := v
				x.ClaimETag = o.ETag
				return x, nil
			}
			snapshot := f.b.Snapshot()
			d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
			if err != nil {
				t.Fatal(err)
			}
			if scenario == "covered-local" {
				if len(d.Alerts) != 1 || len(d.Candidates) != 0 || f.b.Snapshot() != snapshot {
					t.Fatalf("covered local baseline did not refuse unchanged: %+v", d)
				}
				return
			}
			var got []string
			for _, c := range d.Candidates {
				got = append(got, c.Generation)
			}
			if len(d.Alerts) != 0 || !reflect.DeepEqual(got, []string{"T", "U", "g0", "g2"}) {
				t.Fatalf("foreign links blocked sweep or were protected: %+v", d)
			}
			for _, x := range before {
				g, _ := v.GenerationOf(ref(x))
				_, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID)
				if g == "T" || g == "U" || g == "g0" || g == "g2" {
					if !errors.Is(err, lease.ErrMirrorNotFound) {
						t.Fatalf("dead version retained: %+v %v", x, err)
					}
				} else if err != nil {
					t.Fatalf("live version lost: %+v %v", x, err)
				}
			}
			completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
			if err != nil || len(completed.Claim.Obligations) != 0 {
				t.Fatalf("sweep did not complete: %+v %v", completed, err)
			}
		})
	}
}
func TestSweepDisagreementAndBusyRefuse(t *testing.T) {
	f, obs, v, vs := sweepFixture(t)
	v.MarkerGeneration = "unrelated"
	d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil || len(d.Alerts) != 1 || len(d.Candidates) != 0 {
		t.Fatalf("%+v %v", d, err)
	}
	obs.Claim.Op.Kind = hostclaim.KindMoving
	obs.Claim.Op.ID = "moving"
	d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil || len(d.Candidates) != 0 {
		t.Fatalf("%+v %v", d, err)
	}
}
func TestSweepDiscoversOldNamespaceAndMoves(t *testing.T) {
	f, obs, v, vs := sweepFixture(t)
	old := strings.Repeat("a", 64)
	other := strings.Repeat("b", 64)
	for _, key := range []string{"instances/" + f.w.Host + "/" + old + "/0000/a", "moves/" + f.w.Host + "/" + other + "/g-op/a", "instances/" + f.w.Host + "/0000/legacy", "moves/" + f.w.Host + "/g-old/staging"} {
		id := f.b.Put("data", key, []byte("old"))
		vs = append(vs, lease.ObjectVersion{Key: key, VersionID: id})
	}
	d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil || len(d.Candidates) != 4 {
		t.Fatalf("%+v %v", d, err)
	}
	// Capped retired history is irrelevant to discovery.
	obs.Claim.RetiredIncarnations = nil
	v.Region = "us"
	d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
	if err != nil || len(d.Candidates) != 4 {
		t.Fatalf("foreign location not residue: %+v %v", d, err)
	}
}
func TestSweepFreeHostQueuesAndAcknowledges(t *testing.T) {
	f := newFixture(t)
	f.seed()
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Obligations = nil; return hostclaim.FreeAfterPark("op")(c) })
	if err != nil {
		t.Fatal(err)
	}
	d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
	if err != nil || len(d.Candidates) != 1 {
		t.Fatalf("%+v %v", d, err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || !obs.Claim.Free() || len(obs.Claim.Obligations) != 0 {
		t.Fatalf("%+v %v", obs, err)
	}
}
func TestSweepOperationOwnership(t *testing.T) {
	for _, serving := range []bool{true, false} {
		name := "holderless"
		if serving {
			name = "serving"
		}
		t.Run(name, func(t *testing.T) {
			f, obs, v, _ := sweepFixture(t)
			if !serving {
				_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
					c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "free"}
					return hostclaim.FreeAfterPark("free")(c)
				})
				if err != nil {
					t.Fatal(err)
				}
			}
			f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
				x := v
				x.ClaimETag = o.ETag
				return x, nil
			}
			ids := []string{"generated-op", "chore", "retirement"}
			wantOp, wantCalls := "generated-op", 3
			if serving {
				ids, wantOp, wantCalls = ids[1:], "session:3", 2
			}
			calls := 0
			f.e.NewID = func() string {
				if calls >= len(ids) {
					t.Fatal("unexpected generated operation id")
				}
				id := ids[calls]
				calls++
				return id
			}
			authorized := false
			f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
				authorized = true
				if o.Op != wantOp {
					return ErrUnsafe
				}
				return nil
			}
			// Stop after queueing, before Run can acknowledge the persisted chore.
			f.held.failed = true
			if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
				t.Fatal(err)
			}
			if !authorized || calls != wantCalls {
				t.Fatalf("authorized=%v ids=%d", authorized, calls)
			}
			queued, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
			if err != nil {
				t.Fatal(err)
			}
			if len(queued.Claim.Obligations) != 1 || queued.Claim.Obligations[0].Op != wantOp {
				t.Fatalf("wrong chore owner: %+v", queued.Claim)
			}
			if serving {
				if queued.Claim.Op.Kind != hostclaim.KindServing || queued.Claim.Holder.Session != 3 {
					t.Fatal("serving state changed")
				}
			} else if queued.Claim.Holder != nil || queued.Claim.Op.ID != wantOp || queued.Claim.Op.Intent["park_sweep"] != "v1" {
				t.Fatal("wrong holderless owner")
			}
		})
	}
}

func TestSweepPageFailureAndStaleView(t *testing.T) {
	f, obs, v, vs := sweepFixture(t)
	before := f.b.Snapshot()
	f.b.FailListAfterPages = 1
	if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err == nil {
		t.Fatal("short sweep acted")
	}
	if before != f.b.Snapshot() {
		t.Fatal("mutation")
	}
	v.ClaimETag = "old"
	if _, err := Decide(obs, v, vs, f.ob.Park.Incarnation); !errors.Is(err, ErrUnsafe) {
		t.Fatal(err)
	}
}