package hostrelocate

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

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

// driveToProbing runs a drain until dst's participant is in probing with
// its local install recorded and no attempt counted.
func driveToProbing(t *testing.T, f *fixture, src, dst *fakeBox) (Participant, Participant) {
	t.Helper()
	ctx := context.Background()
	if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
		t.Fatal(err)
	}
	ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
	pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
	for i := 0; i < 40; i++ {
		if o := f.claim(); o.Claim.Op.Phase == hostclaim.PhaseRelocatingProbing {
			return ps, pd
		}
		f.w.Step(ctx, host)
		ps.Step(ctx, host)
		if f.claim().Claim.Op.Phase == hostclaim.PhaseInstalled && f.claim().Claim.Op.Journal["local_install"] != "" {
			pd.Step(ctx, host) // ToProbing
			continue
		}
		pd.Step(ctx, host)
	}
	t.Fatalf("never probing: %+v", f.claim().Claim.Op)
	return ps, pd
}

// Ruling 48: while root refuses the relocating kind, the participant
// counts no attempt; once it answers, the next pass counts one.
func TestParticipantCountsNoAttemptWhileRootRefusesTheKind(t *testing.T) {
	f := newFixture(t)
	ctx := context.Background()
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", preflightErr: errors.New("root: relocating unsupported")}
	_, pd := driveToProbing(t, f, src, dst)
	for i := 0; i < 3; i++ {
		if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
			t.Fatalf("pass %d: %v", i, err)
		}
		if n := f.claim().Claim.Op.Journal["probe_attempts.dst"]; n != "" {
			t.Fatalf("an attempt was counted while root refused: %s", n)
		}
	}
	dst.preflightErr = nil
	if _, err := pd.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
		t.Fatal(err)
	}
	if n := f.claim().Claim.Op.Journal["probe_attempts.dst"]; n != "1" {
		t.Fatalf("attempts after root answered: %q", n)
	}
}

// Ruling 58: once P's opening failed, an attempt root cannot launch is not
// made; the operation goes probe-failed at once, with one attempt counted.
func TestParticipantFailsAProbeRootCannotRelaunch(t *testing.T) {
	f := newFixture(t)
	ctx := context.Background()
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", failProbes: true}
	_, pd := driveToProbing(t, f, src, dst)
	if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
		t.Fatalf("the first attempt: %v", err)
	}
	if !hostclaim.PreparedOpened(f.claim().Claim) {
		t.Fatal("P was not opened by the first attempt")
	}
	dst.preflightErr = ErrLaterAttemptUnlaunchable
	o, err := pd.Step(ctx, host)
	if err != nil {
		t.Fatal(err)
	}
	if o.Claim.Op.Phase != hostclaim.PhaseProbeFailed || o.Claim.Op.Journal["probe_attempts.dst"] != "1" || len(dst.certified) != 1 {
		t.Fatalf("phase %s attempts %q admissions %d", o.Claim.Op.Phase, o.Claim.Op.Journal["probe_attempts.dst"], len(dst.certified))
	}
}

// flakyStore is the claim store with one failure armed: the next Get (the
// read after a passed probe) or Put (the CAS that records it) fails once,
// as a transient store error would, and the store is itself again.
type flakyStore struct {
	release.Store
	armed bool
	puts  bool // fail a Put rather than a Get
}

var errFlaky = errors.New("the claim store failed once")

func (s *flakyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if s.armed && !s.puts {
		s.armed = false
		return nil, "", errFlaky
	}
	return s.Store.Get(ctx, key)
}

func (s *flakyStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if s.armed && s.puts {
		s.armed = false
		return "", errFlaky
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// Final review A2 (Astra 2): after Probe returned passed, the claim read
// or the CAS that records the result fails once. The next pass recovers
// the result from the box's durable attempt record, keyed by the admitted
// session, and journals it: no Preflight, no second attempt, no
// probe-failed, and the live session is the one journaled. A record that
// cannot be read is retried, never taken as "no probe".
func TestParticipantRecoversAPassedProbeTheClaimDidNotRecord(t *testing.T) {
	for name, puts := range map[string]bool{"the claim read fails once": false, "the CAS fails once": true} {
		t.Run(name, func(t *testing.T) {
			f := newFixture(t)
			ctx := context.Background()
			flaky := &flakyStore{Store: f.s, puts: puts}
			src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", flaky: flaky}
			_, pd := driveToProbing(t, f, src, dst)
			pd.Claims = flaky
			if _, err := pd.Step(ctx, host); !errors.Is(err, errFlaky) {
				t.Fatalf("the pass after the probe passed: %v", err)
			}
			c := f.claim().Claim
			if !hostclaim.PreparedOpened(c) || c.Op.Journal["probe"] != "" || flaky.armed {
				t.Fatalf("after the failure: opened %v, probe %q, armed %v", hostclaim.PreparedOpened(c), c.Op.Journal["probe"], flaky.armed)
			}
			dst.preflightErr = ErrLaterAttemptUnlaunchable // what root would answer a new attempt
			dst.effects = nil
			dst.passedErr = errors.New("the attempt journal cannot be read")
			if _, err := pd.Step(ctx, host); !errors.Is(err, dst.passedErr) {
				t.Fatalf("an unreadable record: %v", err)
			}
			dst.passedErr = nil
			o, err := pd.Step(ctx, host)
			if err != nil {
				t.Fatalf("the recovering pass: %v", err)
			}
			var p hostclaim.Probe
			if o.Claim.Op.Phase != hostclaim.PhaseRelocatingProbing || json.Unmarshal([]byte(o.Claim.Op.Journal["probe"]), &p) != nil || p.Session != o.Claim.Holder.Session || p.Box != "dst" || p.At.IsZero() {
				t.Fatalf("phase %s probe %q", o.Claim.Op.Phase, o.Claim.Op.Journal["probe"])
			}
			if len(dst.certified) != 1 || o.Claim.Op.Journal["probe_attempts.dst"] != "1" || slices.Contains(dst.effects, "preflight") {
				t.Fatalf("admissions %d attempts %q effects %v", len(dst.certified), o.Claim.Op.Journal["probe_attempts.dst"], dst.effects)
			}
		})
	}
}

// Ruling 62: an install that writes the claim (the typed inventory) has its
// receipt recorded on a fresh read, not a conflict loop on the stale one.
func TestParticipantRecordsItsReceiptAfterAnEffectWroteTheClaim(t *testing.T) {
	f := newFixture(t)
	ctx := context.Background()
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", writesOnInstall: true}
	if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
		t.Fatal(err)
	}
	ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
	pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
	for i := 0; i < 40 && f.claim().Claim.Op.Phase != hostclaim.PhaseInstalled; i++ {
		f.w.Step(ctx, host)
		ps.Step(ctx, host)
		pd.Step(ctx, host)
	}
	if f.claim().Claim.Op.Phase != hostclaim.PhaseInstalled {
		t.Fatalf("never installed: %+v", f.claim().Claim.Op)
	}
	if _, err := pd.Step(ctx, host); err != nil {
		t.Fatalf("the receipt after an effect wrote the claim: %v", err)
	}
	c := f.claim().Claim
	if c.Op.Journal["local_install"] == "" || c.Op.Journal["test.inventory"] != "dst" {
		t.Fatalf("receipt %q inventory %q", c.Op.Journal["local_install"], c.Op.Journal["test.inventory"])
	}
}

// passProbe runs dst's first attempt to a journaled passing probe.
func passProbe(t *testing.T, f *fixture, pd Participant) {
	t.Helper()
	if _, err := pd.Step(context.Background(), host); err != nil {
		t.Fatalf("the probe: %v", err)
	}
	if f.claim().Claim.Op.Journal["probe"] == "" {
		t.Fatal("no probe journaled")
	}
}

// Ruling 67's first scenario: the execution lease is lost right after
// RecordProbe. The participant reopens a controller and ends nothing, so
// no probe_stop is ever journaled by it (the worker's bound decides).
func TestParticipantEndsNoProbeWithoutItsLease(t *testing.T) {
	f := newFixture(t)
	ctx := context.Background()
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
	_, pd := driveToProbing(t, f, src, dst)
	passProbe(t, f, pd)
	dst.restarted = true // the lease is gone; Ready reports no controller
	dst.effects = nil
	if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
		t.Fatalf("a step without the lease: %v", err)
	}
	if slices.Contains(dst.effects, "end-probe") || f.claim().Claim.Op.Journal["probe_stop"] != "" {
		t.Fatalf("ended the probe without a ready controller: %v", dst.effects)
	}
}

// Ruling 67's second scenario: root keeps refusing the stop. The error is
// returned and nothing is journaled; a later pass that gets the receipt
// records it.
func TestParticipantRecordsNoProbeStopWhileRootRefuses(t *testing.T) {
	f := newFixture(t)
	ctx := context.Background()
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
	_, pd := driveToProbing(t, f, src, dst)
	passProbe(t, f, pd)
	refused := errors.New("root: stop refused")
	dst.endProbeErr = refused
	for i := 0; i < 2; i++ {
		if _, err := pd.Step(ctx, host); !errors.Is(err, refused) || f.claim().Claim.Op.Journal["probe_stop"] != "" {
			t.Fatalf("pass %d: %v", i, err)
		}
	}
	dst.endProbeErr = nil
	if _, err := pd.Step(ctx, host); err != nil || f.claim().Claim.Op.Journal["probe_stop"] == "" {
		t.Fatalf("the stop once root answered: %v", err)
	}
}

// Fix round 1, M4 (ruling 62): sameInstallation accepts a fresh read that
// differs only in an effect's own bookkeeping, and refuses one whose
// installation moved: another stamp, install stamp, prepared P, session
// or phase, or a stuck mark. (The claim store itself refuses a changed
// incarnation, or an installed op's executor moved off its target.)
func TestSameInstallationRefusesAMovedInstallation(t *testing.T) {
	ctx := context.Background()
	installed := func(t *testing.T) (*fixture, hostclaim.Observed, Participant) {
		f := newFixture(t)
		releasedAndInstalling(t, f)
		f.worker(ErrNotYet) // P is not built yet
		p := f.claim().Claim.Op.Journal["prepared"]
		f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
		if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalled {
			t.Fatalf("not installed: %s", o.Claim.Op.Phase)
		}
		return f, f.claim(), Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: &fakeBox{f: f, box: "dst"}, Now: f.clock}
	}
	t.Run("the effect's own bookkeeping", func(t *testing.T) {
		f, obs, p := installed(t)
		f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.inventory"] = "dst"; return nil })
		fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled)
		if err != nil || fresh.ETag == obs.ETag || fresh.Claim.Op.Journal["test.inventory"] != "dst" {
			t.Fatalf("a fresh read of the same installation: %v", err)
		}
	})
	for name, change := range map[string]func(*hostclaim.Claim){
		"another stamp":         func(c *hostclaim.Claim) { c.Holder.Stamp++ },
		"another install stamp": func(c *hostclaim.Claim) { c.Op.Journal["install_stamp"] += "0" },
		"another prepared P":    func(c *hostclaim.Claim) { c.Op.Journal["prepared"] = "P9" },
		"another session":       func(c *hostclaim.Claim) { c.Holder.Session++ },
		"stuck":                 func(c *hostclaim.Claim) { c.Op.Journal["stuck"] = "operator" },
	} {
		t.Run(name, func(t *testing.T) {
			f, obs, p := installed(t)
			f.participant(func(c *hostclaim.Claim) error { change(c); return nil })
			if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled); !errors.Is(err, ErrRefused) || !errors.Is(err, ErrNotYet) {
				t.Fatalf("a moved installation was accepted: %v", err)
			}
		})
	}
	t.Run("another phase", func(t *testing.T) {
		_, obs, p := installed(t)
		if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseRelocatingProbing); !errors.Is(err, ErrRefused) {
			t.Fatalf("another phase was accepted: %v", err)
		}
	})
}

// Fix round 1, I3 and M1: a journaled release that is unreadable, or that
// names another writer than the predecessor, is a claim defect. The
// participant reads it before any attempt is counted: the operation is
// stuck for an operator, no attempt is spent, and no probe is made.
func TestParticipantRefusesADamagedReleaseBeforeCountingAnAttempt(t *testing.T) {
	ctx := context.Background()
	for name, damage := range map[string]func(*hostclaim.Claim) error{
		"unparseable": func(c *hostclaim.Claim) error { c.Op.Journal["release"] = `{"released_by":`; return nil },
		"another session": func(c *hostclaim.Claim) error {
			var rel hostclaim.Release
			if err := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel); err != nil {
				return err
			}
			rel.Session++
			raw, err := json.Marshal(rel)
			c.Op.Journal["release"] = string(raw)
			return err
		},
	} {
		t.Run(name, func(t *testing.T) {
			f := newFixture(t)
			src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
			_, pd := driveToProbing(t, f, src, dst)
			if _, err := hostclaim.Update(ctx, f.s, f.claim(), damage); err != nil {
				t.Fatal(err)
			}
			if _, err := pd.Step(ctx, host); err != nil {
				t.Fatal(err)
			}
			c := f.claim().Claim
			if c.Op.Journal["stuck"] == "" || c.Op.Journal["probe_attempts.dst"] != "" || len(dst.certified) != 0 {
				t.Fatalf("stuck %q attempts %q probes %d", c.Op.Journal["stuck"], c.Op.Journal["probe_attempts.dst"], len(dst.certified))
			}
		})
	}
	// A sound release certifies P's opening, and only it.
	f := newFixture(t)
	src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
	_, pd := driveToProbing(t, f, src, dst)
	if certified, err := CertifiedFor(f.claim().Claim); err != nil || !certified {
		t.Fatalf("a drain's release: %v %v", certified, err)
	}
	if _, err := pd.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
		t.Fatal(err)
	}
	if certified, err := CertifiedFor(f.claim().Claim); err != nil || certified {
		t.Fatalf("after P's opening: %v %v", certified, err)
	}
}