package host

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

// Ruling R-SPEC-43A-2 (Astra round 1 finding 3): before a cut seals the build
// state, the source box's consumed request record is merged into it under
// the lease (release.CarryRequest), on every lane: the legacy cut, the Phase
// 1 cut's final round and a dirty return's lane.

// carryCall is one CarryRequest call, host bucket/prefix.
type carryCall struct{ host, bucket, prefix string }

// The legacy cut carries the request before its byte-final round, after the
// final checkpoint, at the hibernate prefix the round copies; the carried
// state is what lands in staging.
func TestTheLegacyCutCarriesTheRequestBeforeItsFinalRound(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	h.f.put("data-ie", "instances/"+host+"/build-state.json", "before")
	var calls []carryCall
	h.cfg.CarryRequest = func(_ context.Context, host, bucket, prefix string) error {
		calls = append(calls, carryCall{host, bucket, prefix})
		h.f.ops = append(h.f.ops, "carry")
		h.f.put(bucket, prefix+"/build-state.json", "carried")
		return nil
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(calls) != 1 || calls[0] != (carryCall{host, "data-ie", "instances/" + host}) {
		t.Fatalf("carry calls %+v; want one at data-ie/instances/%s", calls, host)
	}
	ck, carry, cp := -1, -1, -1
	for i, op := range h.f.ops {
		switch {
		case strings.HasPrefix(op, "checkpoint ") && ck == -1:
			ck = i
		case op == "carry":
			carry = i
		case strings.HasPrefix(op, "copy ") && cp == -1:
			cp = i
		}
	}
	if !(ck < carry && carry < cp) {
		t.Fatalf("ops %v: want checkpoint, then the carry, then the final round", h.f.ops)
	}
	if got := h.f.bodies("data-uk", "moves/"+host+"/")["build-state.json"]; got != "carried" {
		t.Fatalf("staged build state %q, want the carried one", got)
	}
	if got := h.read(t, host); got.Move != nil {
		t.Fatalf("the cut did not complete: %+v", got.Move)
	}
}

// A carry that fails refuses the cut (the pass logs the round's error and
// goes on to other hosts): nothing is copied, the warm cache is not purged,
// and the move stays cutting for the next pass.
func TestAFailedCarryRefusesTheLegacyCut(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	db := filepath.Join(h.cfg.DataDir, host+".db")
	if err := writeFileForTest(db, "warm"); err != nil {
		t.Fatal(err)
	}
	carried := 0
	h.cfg.CarryRequest = func(context.Context, string, string, string) error { carried++; return errors.New("s3: 503") }
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if carried != 1 {
		t.Fatalf("carry asked %d times, want 1", carried)
	}
	if _, err := os.Stat(db); err != nil {
		t.Fatalf("the warm cache was purged past a failed carry: %v", err)
	}
	for _, op := range h.f.ops {
		if strings.HasPrefix(op, "copy ") {
			t.Fatalf("ops %v: the final round ran past a failed carry", h.f.ops)
		}
	}
	if got := h.read(t, host); got.Move == nil || got.Move.State != provision.MoveCutting {
		t.Fatalf("move %+v, want still cutting", got.Move)
	}
}

// The Phase 1 cut's final round carries the request at the claim holder's
// location before it copies the state for the seal.
func TestThePhase1CutCarriesTheRequestBeforeItsSeal(t *testing.T) {
	h := newCutHarness(t)
	storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
	h.ready(t)
	var calls []carryCall
	h.cfg.CarryRequest = func(_ context.Context, host, bucket, prefix string) error {
		calls = append(calls, carryCall{host, bucket, prefix})
		storeState(h.phase1Harness, bucket, prefix+"/build-state.json", "carried")
		return nil
	}
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding {
		t.Fatalf("cut did not finish: %+v", m)
	}
	if len(calls) == 0 || calls[len(calls)-1] != (carryCall{p1Host, "data-ie", cutRoot}) {
		t.Fatalf("carry calls %+v; want one at data-ie/%s", calls, cutRoot)
	}
	key := moves.BuildStateKey(p1Host, p1ID)
	if got := h.f.buckets["data-uk"][key].body; got != p1StateBody("carried") {
		t.Fatalf("the sealed lane holds %q, want the carried state", got)
	}
	if seal := h.seal(t); len(seal.State) != 1 || seal.State[0].VersionID != h.f.versionOf("data-uk", key) {
		t.Fatalf("seal state %+v", seal.State)
	}
}

// A dirty return carries the target's request record into the state beside
// the target's lease before its lane copies it back.
func TestTheReturnLaneCarriesTheRequest(t *testing.T) {
	t.Parallel()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
	h.flip(t)
	h.land(t)
	h.installed(t)
	h.step(t, provision.LandRestored, provision.LandProbing)
	h.tpos = h.n
	storeState(h.phase1Harness, "data-uk", cutRoot+"/build-state.json", "v8")
	h.uncover(t)
	h.rollBack(t)
	var calls []carryCall
	h.carry = func(_ context.Context, host, bucket, prefix string) error {
		calls = append(calls, carryCall{host, bucket, prefix})
		storeState(h.phase1Harness, bucket, prefix+"/build-state.json", "carried")
		return nil
	}
	h.rollbackPass(t)
	rb := h.rb(t)
	if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	if len(calls) == 0 || calls[len(calls)-1] != (carryCall{p1Host, "data-uk", cutRoot}) {
		t.Fatalf("carry calls %+v; want one at data-uk/%s", calls, cutRoot)
	}
	if _, body := returnSealState(t, h, rb.ID); body != p1StateBody("carried") {
		t.Fatalf("the return lane holds %q, want the carried state", body)
	}
}

// buildStates.carryRequest, the hook's production body: under the host's
// lock it merges root's consumed record into the object at the location
// given (release.CarryRequest), last_executed untouched, and writes only
// when something changes.
func TestCarryRequestWritesTheConsumedRecordAtTheLocation(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, astraHost, release.BuildState{LastExecuted: "A"}); err != nil {
		t.Fatal(err)
	}
	// The cut holds the row before it carries (ruling R-CARRY-FENCE).
	if err := reg.SetStatus(astraHost, "moving"); err != nil {
		t.Fatal(err)
	}
	key := release.BuildStateKey("inst/" + astraHost)
	etag := mustEtag(t, s, key)
	// No consumed file: nothing to carry, nothing written.
	if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil {
		t.Fatal(err)
	}
	if mustEtag(t, s, key) != etag {
		t.Fatal("a carry with no consumed file rewrote the object")
	}
	writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Completed: 1, CompletedFailed: true})
	if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil {
		t.Fatal(err)
	}
	st, _, err := getState(ctx, s, key, astraHost)
	if err != nil || st == nil || st.LastExecuted != "A" || st.Completed != 1 || !st.CompletedFailed {
		t.Fatalf("state %+v, %v; want A with R1 completed failed", st, err)
	}
	etag = mustEtag(t, s, key)
	if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil {
		t.Fatal(err)
	}
	if mustEtag(t, s, key) != etag {
		t.Fatal("a carry with nothing new rewrote the object")
	}
	// The location given wins over the hibernate prefix (a claim holder's).
	other := release.NewDirStore(t.TempDir())
	b.storeFor = storesOf(map[string]release.Store{"hib": s, "data": other})
	if err := b.carryRequest(ctx, astraHost, "data", "instances/"+astraHost+"/inc1"); err != nil {
		t.Fatal(err)
	}
	if st, _, _ := getState(ctx, other, release.BuildStateKey("instances/"+astraHost+"/inc1"), astraHost); st == nil || st.Completed != 1 {
		t.Fatalf("state at the location = %+v", st)
	}
}