package host

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

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
)

// probedLanding is a landing driven from the flip through every phase with
// the real transitions, to probed.
func probedLanding(t *testing.T) (*landingHarness, *probeLog) {
	t.Helper()
	h, p := installedLanding(t, 2)
	h.land(t)
	h.atPhase(t, provision.LandProbed)
	h.holdProbed = false
	return h, p
}

// committedOpened asserts the end state of the commit and the open: the
// record committed with opened_at, the claim committed with the admitted
// writer kept and the commit's parks queued, the row ready, one probe.
func committedOpened(t *testing.T, h *landingHarness, p *probeLog) {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m.State != provision.MoveCommitted || m.CommittedAt == nil || m.OpenedAt == nil {
		t.Fatalf("record: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted {
		t.Fatalf("claim: %+v", c.Op)
	}
	if c.Holder.Session != p.last.Session || c.Holder.Stamp != p.last.Stamp || c.Holder.Generation != p.last.Generation {
		t.Fatalf("the commit changed the admitted writer %+v: %+v", p.last, c.Holder)
	}
	kinds := map[string]int{}
	m2 := h.read(t, p1Host).Move
	for _, o := range c.Obligations {
		kinds[o.Kind+"/"+o.Region]++
		if o.Kind == "park-replica" && o.Region == "ie" && (o.Park == nil || o.Park.Location != c.Holder.Location || !o.Park.EvidenceAt.Equal(m2.FlippedAt.UTC())) {
			t.Fatalf("the source's park is not scoped from the flip: %+v", o.Park)
		}
		if o.Kind == "park-replica" && o.Region == "uk" && (o.Park == nil || o.Park.Location != "moves/"+p1Host+"/"+p1ID) {
			t.Fatalf("the staging park: %+v", o.Park)
		}
	}
	// The source region's location (scoped from the flip, H7) and the
	// target region's staging lane (scoped to the seal's bound).
	if kinds["park-replica/ie"] != 1 || kinds["park-replica/uk"] != 1 {
		t.Fatalf("the commit's parks: %+v", c.Obligations)
	}
	if h.route.Status != "ready" || h.route.HoldOp != "" {
		t.Fatalf("the row is not open: %+v", h.route)
	}
	if len(p.configs) != 1 {
		t.Fatalf("probed %d times", len(p.configs))
	}
}

// TestCommitThenOpenInOrder (brief; witness wCommitBeforeOpen): from the
// flip through every landing phase with the real transitions, the commit is
// the record's CAS, then the claim's; the probe's live entry is marked open,
// and only then is the hold released: the row is ready only after the record
// and the claim both say committed.
func TestCommitThenOpenInOrder(t *testing.T) {
	t.Parallel()
	h, p := probedLanding(t)
	h.land(t)
	committedOpened(t, h, p)
	want := []string{"mark:committed/committed/landing", "release:committed/committed/landing", "confirm:committed/committed/ready"}
	if !reflect.DeepEqual(h.openLog, want) {
		t.Fatalf("the open: %v, want %v", h.openLog, want)
	}
}

// TestOpenAfterCrashDoesNotReprobe (brief, Review Focus 3): a worker
// restarted on a record at committed with the claim still landing/probed
// writes the claim half, opens, and never calls the probe.
func TestOpenAfterCrashDoesNotReprobe(t *testing.T) {
	t.Parallel()
	h, p := probedLanding(t)
	h.landAt = "after-record-committed"
	h.crashLand(t)
	if m := h.read(t, p1Host).Move; m.State != provision.MoveCommitted || h.claim(t).Op.Phase != hostclaim.PhaseLandingProbed || h.route.Status != "landing" {
		t.Fatalf("the crash point: %+v claim %s row %s", m, h.claim(t).Op.Phase, h.route.Status)
	}
	h.freshTarget()
	h.land(t)
	committedOpened(t, h, p)
}

// The open repeats while the record is committed: a pass that finds the row
// already open changes nothing, opened_at included.
func TestCommittedOpenRepeatsConditionally(t *testing.T) {
	t.Parallel()
	h, p := probedLanding(t)
	h.land(t)
	committedOpened(t, h, p)
	opened := *h.read(t, p1Host).Move.OpenedAt
	h.clock.Advance(time.Minute)
	h.land(t)
	committedOpened(t, h, p)
	if m := h.read(t, p1Host).Move; !m.OpenedAt.Equal(opened) {
		t.Fatalf("opened_at moved: %v, was %v", m.OpenedAt, opened)
	}
}

// TestLateCommitRollsBack (brief, Review Focus 2; H8): a commit that would
// come after the landing's deadline is refused by the record's CAS and
// becomes the commit-deadline rollback, at the one conversion site; the row
// never opens.
func TestLateCommitRollsBack(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	h.landFault = func(point string) error {
		if point == "before-commit" {
			h.clock.Advance(cutLandingWindow + time.Minute)
		}
		return nil
	}
	h.land(t)
	if m := h.read(t, p1Host).Move; m.State != provision.MoveLanding || m.Reason != provision.RollbackCommitDeadline {
		t.Fatalf("a late commit: %+v", m)
	}
	h.landFault = nil
	h.land(t)
	rolledBack(t, h, provision.RollbackCommitDeadline, provision.LandProbed)
	if h.route.Status == "ready" || len(h.openLog) != 0 {
		t.Fatalf("a rolled-back landing opened: %+v %v", h.route, h.openLog)
	}
}

// m-2: a claim lagging at probing behind a probed record is walked to probed
// before the commit, on the admitted session, so Commit (probed only) holds.
func TestCommitReassertsALaggingProbedClaim(t *testing.T) {
	t.Parallel()
	h, p := probedLanding(t)
	rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = hostclaim.PhaseLandingProbing })
	h.land(t)
	committedOpened(t, h, p)
}

// Unarmed: with no release of the hold the committed arm waits; the commit
// itself is the record's and is made.
func TestCommittedOpenWaitsUnarmed(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	h.noRelease = true
	h.land(t)
	if m := h.read(t, p1Host).Move; m.State != provision.MoveCommitted || m.OpenedAt != nil || h.route.Status != "landing" {
		t.Fatalf("an unarmed open: %+v %+v", m, h.route)
	}
	_ = context.Background
}

// The commit's scopes are Plan 5 Task 2's helpers' (H7): the source's
// location from the flip, the staging lane to the seal's bound.
func TestCommitScopesAreThePlan5Helpers(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	h.land(t)
	rec := h.read(t, p1Host)
	c := h.claim(t)
	seal, err := journalledSeal(c)
	if err != nil {
		t.Fatal(err)
	}
	wantSource := moves.SourceLocationScope(*c.Holder, *rec.Move.FlippedAt)
	wantStaging := moves.StagingScope(p1Host, p1ID, c.Holder.Incarnation, seal.Bound())
	var source, staging bool
	for _, o := range c.Obligations {
		if o.Kind != "park-replica" || o.Park == nil {
			continue
		}
		switch *o.Park {
		case wantSource:
			source = true
		case wantStaging:
			staging = true
		}
	}
	if !source || !staging {
		t.Fatalf("the commit's parks are not the helpers' %+v and %+v: %+v", wantSource, wantStaging, c.Obligations)
	}
}

// Task 7 fix round 1, finding 3: the open is proved, never assumed. An
// unheld row that is not ready is no replay of the open, and a release that
// matched nothing is refused: opened_at is not stamped in either case.
func TestCommittedOpenIsProved(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	h.route.HoldOp = ""
	h.land(t)
	if m := h.read(t, p1Host).Move; m.State != provision.MoveCommitted || m.OpenedAt != nil {
		t.Fatalf("an unheld row that is not ready was taken as open: %+v", m)
	}

	h2, _ := probedLanding(t)
	h2.releaseNoop = true
	h2.land(t)
	if m := h2.read(t, p1Host).Move; m.State != provision.MoveCommitted || m.OpenedAt != nil || h2.route.Status == "ready" {
		t.Fatalf("a release that matched nothing was taken as the open: %+v %+v", m, h2.route)
	}
	// The release happened and the row is ready: the next pass stamps it.
	h2.releaseNoop = false
	h2.land(t)
	if m := h2.read(t, p1Host).Move; m.OpenedAt == nil || h2.route.Status != "ready" {
		t.Fatalf("the open after the refusal: %+v %+v", m, h2.route)
	}
}

// Final review, ruling 3: the probe entry's idle protection ends only on a
// proved release. A release that fails confirms nothing and stamps
// nothing; the retried open that succeeds confirms it.
func TestCommittedOpenConfirmsOnlyAProvedRelease(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	h.releaseErr = errors.New("the registry is busy")
	h.land(t)
	for _, step := range h.openLog {
		if strings.HasPrefix(step, "confirm:") {
			t.Fatalf("a failed release was confirmed: %v", h.openLog)
		}
	}
	if m := h.read(t, p1Host).Move; m.OpenedAt != nil {
		t.Fatalf("a failed release was stamped: %+v", m)
	}
	h.releaseErr = nil
	h.land(t)
	if last := h.openLog[len(h.openLog)-1]; !strings.HasPrefix(last, "confirm:") || h.read(t, p1Host).Move.OpenedAt == nil {
		t.Fatalf("the retried open: %v", h.openLog)
	}
}