package hostclaim

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"strings"
	"testing"
	"time"

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

// probedDrain drives a drain of the serving fixture to a passed probe on
// relTarget: released, pinned, installed, probing, P admitted, the probe
// journaled with its time.
func probedDrain(t *testing.T) (func(func(*Claim) error) Observed, func() Observed) {
	t.Helper()
	ctx := context.Background()
	s, cur := probedDrainIn(t)
	apply := func(tr func(*Claim) error) Observed {
		t.Helper()
		next, err := Update(ctx, s, cur, tr)
		if err != nil {
			t.Fatal(err)
		}
		cur = next
		return next
	}
	return apply, func() Observed { return cur }
}

// probedDrainIn is probedDrain's claim and its store.
func probedDrainIn(t *testing.T) (release.Store, Observed) {
	t.Helper()
	ctx := context.Background()
	s, o := relServing(t)
	o, err := StartRelocation(ctx, s, o, relStart(ReasonDrain, nil), relAuthority{})
	if err != nil {
		t.Fatal(err)
	}
	o = step(t, s, o, ack(relSource))
	o = step(t, s, o, ack(relTarget))
	o = step(t, s, o, ToReleasing("rel-1", relObserved))
	h := o.Claim.Holder
	o = step(t, s, o, JournalRelease("rel-1", Release{ReleasedBy: relSource, Enrollment: relEnrollment(relSource), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "control/stop/r.json", ReceiptSHA256: strings.Repeat("d", 64)}))
	if o, err = PinAfterRelease(ctx, s, o, "rel-1", relReleases{}, nil); err != nil {
		t.Fatal(err)
	}
	o = step(t, s, o, JournalPin("rel-1", relPin(o, 40, 7, relObserved)))
	o = toProbing(t, s, o, "P1", 40)
	o = step(t, s, o, AdmitWriter(Admission{ObservedOpID: "rel-1", ObservedPhase: PhaseRelocatingProbing, ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
		Box: relTarget, Mode: "cold", NewGeneration: "P1", KeepStamp: true, PredecessorCertified: true, Region: "ie", PredecessorRegion: "ie"}))
	o = step(t, s, o, RecordProbe("rel-1", Probe{Box: relTarget, Session: o.Claim.Holder.Session, At: relObserved.Add(time.Minute)}))
	return s, o
}

// copyOf is a deep copy, for trying a transition without writing it.
func copyOf(t *testing.T, c *Claim) *Claim {
	t.Helper()
	raw, err := json.Marshal(c)
	if err != nil {
		t.Fatal(err)
	}
	var out Claim
	if err := json.Unmarshal(raw, &out); err != nil {
		t.Fatal(err)
	}
	return &out
}

func stopOf(o Observed) ProbeStop {
	return ProbeStop{Box: relTarget, Session: o.Claim.Holder.Session, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("e", 64)}
}

// Ruling 49 step 4: the serving CAS requires the probe's stop, journaled
// for exactly the passed probe's box and session; it then records the
// completed relocation on the serving op (ruling 56).
func TestCompleteRelocationRequiresTheProbeStop(t *testing.T) {
	apply, cur := probedDrain(t)
	o := cur()
	var p Probe
	if !readJournal(o.Claim.Op, "probe", &p) || p.At.IsZero() {
		t.Fatalf("the probe has no time: %+v", p)
	}
	if err := CompleteRelocation("rel-1", counter())(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
		t.Fatalf("completed with the probe's session running: %v", err)
	}
	for name, s := range map[string]ProbeStop{
		"another session": {Box: relTarget, Session: o.Claim.Holder.Session + 1, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("e", 64)},
		"another box":     {Box: relSource, Session: o.Claim.Holder.Session, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("e", 64)},
		"no receipt":      {Box: relTarget, Session: o.Claim.Holder.Session, ReceiptSHA256: strings.Repeat("e", 64)},
		"a bad digest":    {Box: relTarget, Session: o.Claim.Holder.Session, Receipt: "fence-1", ReceiptSHA256: "short"},
	} {
		if err := RecordProbeStop("rel-1", s)(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
			t.Fatalf("%s: recorded: %v", name, err)
		}
	}
	stop := stopOf(o)
	o = apply(RecordProbeStop("rel-1", stop))
	if err := RecordProbeStop("rel-1", stop)(copyOf(t, o.Claim)); !errors.Is(err, ErrAlreadyRecorded) {
		t.Fatalf("a replay: %v", err)
	}
	other := stop
	other.Receipt = "fence-2"
	if err := RecordProbeStop("rel-1", other)(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
		t.Fatalf("a second, different stop: %v", err)
	}
	o = apply(CompleteRelocation("rel-1", counter()))
	done, ok, _ := ReadCompletedRelocation(o.Claim)
	if !ok || done.Op != "rel-1" || done.Stop != stop || done.Stamp != o.Claim.Holder.Stamp || done.Phase != PhaseRelocatingProbing || done.Generation != "P1" {
		t.Fatalf("completed relocation: %+v %v", done, ok)
	}
	if o.Claim.Op.Journal["released_move"] != "" {
		t.Fatal("a plain relocation named a move")
	}
}

// M1: a converted landing's completion names the move on the serving op,
// beside the completed relocation.
func TestCompleteRelocationJournalsReleasedMoveForAConvertedLanding(t *testing.T) {
	apply, cur := probedDrain(t)
	apply(func(c *Claim) error { c.Op.Journal["conversion"] = "committed-landing"; return nil })
	apply(RecordProbeStop("rel-1", stopOf(cur())))
	o := apply(CompleteRelocation("rel-1", counter()))
	if o.Claim.Op.Journal["released_move"] != "rel-1" {
		t.Fatalf("serving op journal: %v", o.Claim.Op.Journal)
	}
	if _, ok, err := ReadCompletedRelocation(o.Claim); !ok || err != nil {
		t.Fatal("no completed relocation beside the released move")
	}
}

// Ruling 60: installed starts its own bound.
func TestToInstalledAtJournalsItsStart(t *testing.T) {
	ctx := context.Background()
	s, o := relServing(t)
	o, err := StartRelocation(ctx, s, o, relStart(ReasonDrain, nil), relAuthority{})
	if err != nil {
		t.Fatal(err)
	}
	o = step(t, s, o, ack(relSource))
	o = step(t, s, o, ack(relTarget))
	o = step(t, s, o, ToReleasing("rel-1", relObserved))
	h := o.Claim.Holder
	o = step(t, s, o, JournalRelease("rel-1", Release{ReleasedBy: relSource, Enrollment: relEnrollment(relSource), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "control/stop/r.json", ReceiptSHA256: strings.Repeat("d", 64)}))
	if o, err = PinAfterRelease(ctx, s, o, "rel-1", relReleases{}, nil); err != nil {
		t.Fatal(err)
	}
	o = step(t, s, o, JournalPin("rel-1", relPin(o, 40, 7, relObserved)))
	o = step(t, s, o, ToInstalling("rel-1", Target{relTarget, relEnrollment(relTarget)}, "P1"))
	if err := ToInstalledAt("rel-1", relPrepared("P1", 40, 7), time.Time{})(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
		t.Fatalf("installed without a start: %v", err)
	}
	o = step(t, s, o, ToInstalledAt("rel-1", relPrepared("P1", 40, 7), relObserved))
	if o.Claim.Op.Phase != PhaseInstalled || o.Claim.Op.Journal["installed_at"] != relObserved.Format(time.RFC3339Nano) {
		t.Fatalf("installed: %s %v", o.Claim.Op.Phase, o.Claim.Op.Journal)
	}
}

// Astra round 6, finding 3: A to B completes, then a successor B to C
// asks to start before B has discharged its probe's participation (C10).
// Through the real start CAS: a drain or rebalance from B is refused, with
// the claim unchanged, until MarkCompletedDischarged; a failover that
// cites B's termination fence is not, since a dead B's slot binds nothing
// on C.
func TestRelocationSuccessorWaitsForTheCompletedDischarge(t *testing.T) {
	ctx := context.Background()
	successor := func(reason string, fence *FenceRef) RelocationStart {
		return RelocationStart{OpID: "rel-2", By: "worker", Now: relObserved.Add(time.Hour), Region: "ie", Reason: reason,
			SourceBox: relTarget, SourceEnrollment: relEnrollment(relTarget), TargetBox: relSpare, TargetEnrollment: relEnrollment(relSpare),
			DirectoryRevision: 9, AdmissionID: "adm-2", Fence: fence}
	}
	completed := completedDrain
	t.Run("a drain from B waits for B's discharge", func(t *testing.T) {
		s, o := completed(t)
		for _, reason := range []string{ReasonDrain, ReasonRebalance} {
			if _, err := StartRelocation(ctx, s, o, successor(reason, nil), relAuthority{}); !errors.Is(err, ErrPendingChore) {
				t.Fatalf("%s started before the discharge: %v", reason, err)
			}
		}
		if now, err := Read(ctx, s, o.Claim.Host); err != nil || now.ETag != o.ETag {
			t.Fatalf("a refused start wrote the claim: %v", err)
		}
		if _, err := Update(ctx, s, o, MarkCompletedDischarged("rel-9")); !errors.Is(err, ErrRelocation) {
			t.Fatalf("another op's discharge: %v", err)
		}
		o = step(t, s, o, MarkCompletedDischarged("rel-1"))
		if _, err := Update(ctx, s, o, MarkCompletedDischarged("rel-1")); !errors.Is(err, ErrAlreadyRecorded) {
			t.Fatalf("a repeat: %v", err)
		}
		if done, ok, err := ReadCompletedRelocation(o.Claim); err != nil || !ok || !done.Discharged || done.Stop != stopOf(o) {
			t.Fatalf("the mark changed the record: %+v", done)
		}
		next, err := StartRelocation(ctx, s, o, successor(ReasonDrain, nil), relAuthority{})
		if err != nil {
			t.Fatalf("the successor after the discharge: %v", err)
		}
		if next.Claim.Op.ID != "rel-2" || next.Claim.Op.Intent["source_box"] != relTarget || next.Claim.Op.Intent["target_box"] != relSpare {
			t.Fatalf("successor op %+v", next.Claim.Op)
		}
	})
	t.Run("a failover from a fenced B does not wait", func(t *testing.T) {
		s, o := completed(t)
		ref := putFence(t, s, relTarget)
		next, err := StartRelocation(ctx, s, o, successor(ReasonFailover, &ref), relAuthority{})
		if err != nil {
			t.Fatalf("failover from the fenced source: %v", err)
		}
		if next.Claim.Op.Exec != "none" || !slices.Contains(next.Claim.Lost, relTarget) {
			t.Fatalf("failover op %+v lost %v", next.Claim.Op, next.Claim.Lost)
		}
	})
	t.Run("no completed relocation, nothing to mark", func(t *testing.T) {
		_, o := relServing(t)
		if err := MarkCompletedDischarged("rel-1")(copyOf(t, o.Claim)); !errors.Is(err, ErrWrongState) {
			t.Fatalf("%v", err)
		}
	})
}

// completedDrain is A to B through every real transition, completed and
// not yet discharged on B.
func completedDrain(t *testing.T) (release.Store, Observed) {
	t.Helper()
	s, o := probedDrainIn(t)
	o = step(t, s, o, RecordProbeStop("rel-1", stopOf(o)))
	o = step(t, s, o, CompleteRelocation("rel-1", counter()))
	if done, ok, err := ReadCompletedRelocation(o.Claim); err != nil || !ok || done.Discharged || o.Claim.Op.Exec != relTarget {
		t.Fatalf("A to B did not complete undischarged: %+v %v", done, ok)
	}
	return s, o
}

// Astra round 7, finding 2: the discharge barrier is StartGuard's too, so
// a move from B (and a delete) after A to B completes is refused, with the
// claim unchanged, until B marks the completion discharged; then the real
// move start CAS succeeds. A completion box on lost holds nothing.
func TestMoveSuccessorWaitsForTheCompletedDischarge(t *testing.T) {
	ctx := context.Background()
	s, o := completedDrain(t)
	move := BeginAdmittedMove("mv-1", "console", relTarget, relSpare, "uk", relObserved.Add(time.Hour),
		MoveAdmission{AppID: "app-1", AdmissionID: "mv-1", SourceEnrollment: relEnrollment(relTarget), TargetEnrollment: relEnrollment(relSpare)})
	if _, err := Update(ctx, s, o, move); !errors.Is(err, ErrPendingChore) || !errors.Is(err, ErrRelocationFinishing) {
		t.Fatalf("a move started before B's discharge: %v", err)
	}
	if _, err := Update(ctx, s, o, BeginDelete("del-1", "console", "etag-1", relObserved.Add(time.Hour), []string{"ie"}, []string{relTarget}, "")); !errors.Is(err, ErrPendingChore) {
		t.Fatalf("a delete started before B's discharge: %v", err)
	}
	if now, err := Read(ctx, s, o.Claim.Host); err != nil || now.ETag != o.ETag {
		t.Fatalf("a refused start wrote the claim: %v", err)
	}
	lost := copyOf(t, o.Claim)
	lost.Lost = append(lost.Lost, relTarget)
	if err := StartGuard(lost, []string{"uk"}, []string{relSpare}, ""); err != nil {
		t.Fatalf("a lost completion box still holds the barrier: %v", err)
	}
	o = step(t, s, o, MarkCompletedDischarged("rel-1"))
	next, err := Update(ctx, s, o, move)
	if err != nil {
		t.Fatalf("the move after the discharge: %v", err)
	}
	if next.Claim.Op.Kind != KindMoving || next.Claim.Op.ID != "mv-1" || next.Claim.Op.Exec != relTarget {
		t.Fatalf("move op %+v", next.Claim.Op)
	}
}

// Fix round 1, I1: a completed_relocation entry that is present but cannot
// be read is an error, never "none". Every start from serving refuses over
// it (a failover citing a fence included, since which box the record binds
// is unknown), the discharge mark refuses, and the claim is left as it is.
func TestAnUnreadableCompletedRelocationRefusesEveryStart(t *testing.T) {
	ctx := context.Background()
	for name, raw := range map[string]string{
		"unparseable":   `{"op":"rel-1","stop":`,
		"no box":        `{"op":"rel-1","stop":{"session":4,"receipt":"fence-1"}}`,
		"no session":    `{"op":"rel-1","stop":{"box":"` + relTarget + `","receipt":"fence-1"}}`,
		"no op":         `{"stop":{"box":"` + relTarget + `","session":4,"receipt":"fence-1"}}`,
		"empty":         ``,
		"another shape": `["rel-1"]`,
	} {
		t.Run(name, func(t *testing.T) {
			s, o := completedDrain(t)
			o = step(t, s, o, func(c *Claim) error { c.Op.Journal[JournalCompleted] = raw; return nil })
			if _, ok, err := ReadCompletedRelocation(o.Claim); ok || !errors.Is(err, ErrRelocation) {
				t.Fatalf("read as %v, %v", ok, err)
			}
			successor := RelocationStart{OpID: "rel-2", By: "worker", Now: relObserved.Add(time.Hour), Region: "ie", Reason: ReasonDrain,
				SourceBox: relTarget, SourceEnrollment: relEnrollment(relTarget), TargetBox: relSpare, TargetEnrollment: relEnrollment(relSpare),
				DirectoryRevision: 9, AdmissionID: "adm-2"}
			if _, err := StartRelocation(ctx, s, o, successor, relAuthority{}); !errors.Is(err, ErrRelocation) {
				t.Fatalf("a drain started over an unreadable completion: %v", err)
			}
			ref := putFence(t, s, relTarget)
			failover := successor
			failover.Reason, failover.Fence = ReasonFailover, &ref
			if _, err := StartRelocation(ctx, s, o, failover, relAuthority{}); !errors.Is(err, ErrRelocation) {
				t.Fatalf("a failover started over an unreadable completion: %v", err)
			}
			if err := StartGuard(copyOf(t, o.Claim), []string{"uk"}, []string{relSpare}, ""); !errors.Is(err, ErrRelocation) {
				t.Fatalf("StartGuard: %v", err)
			}
			if err := MarkCompletedDischarged("rel-1")(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
				t.Fatalf("the discharge mark over an unreadable completion: %v", err)
			}
			if now, err := Read(ctx, s, o.Claim.Host); err != nil || now.ETag != o.ETag {
				t.Fatalf("a refused start wrote the claim: %v", err)
			}
		})
	}
}