package host

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

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

// The abort (moves spec §7.11; Plan 5 Task 1). The fixture is the Phase 1
// harness after one preparing round, so the lane holds staged objects in the
// target bucket (data-uk) and the source's own bucket (data-ie) holds its
// receipts, its ledger and a snapshot. The target region's box (uk-box) runs
// the lane park through the real park engine over the same buckets with a
// fake lease table. Everything else is real: the source's abort, hostclaim's
// Abort and ReleaseAbort, and provision's guarded transitions.

type abortHarness struct {
	*phase1Harness
	tdir    string
	tleases *fakeParkLeases
	tw      *movesWorker
	abAt    string
	abDone  bool
	abHits  []string
}

func newAbortHarness(t *testing.T) *abortHarness {
	t.Helper()
	h := &abortHarness{phase1Harness: newPhase1Harness(t), tdir: t.TempDir(), tleases: &fakeParkLeases{}}
	h.start(t, "manual")
	h.pass(t)
	if len(h.laneKeys()) == 0 {
		t.Fatal("the fixture staged nothing")
	}
	h.f.put("data-ie", moves.SnapshotsPrefix(p1Host, p1ID)+"files-x.sealed", "s")
	h.cfg.abFault = func(point string) error {
		h.abHits = append(h.abHits, point)
		if point == h.abAt && !h.abDone {
			h.abDone = true
			return errCrash
		}
		return nil
	}
	return h
}

// abort is the console's abort door: the guarded record CAS.
func (h *abortHarness) abort(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil {
		t.Fatal(err)
	}
}

// laneCfg is the target region's box with its lane park armed.
func (h *abortHarness) laneCfg() moveCfg {
	c := h.targetCfg()
	c.DataDir = h.tdir
	armParkFakes(&c, h.s, h.tleases, c.Region, roleTarget)
	return c
}

// tPass runs one pass as the target region's box.
func (h *abortHarness) tPass(t *testing.T) {
	t.Helper()
	if h.tw == nil {
		h.tw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	h.tw.cfg = h.laneCfg()
	if err := movesPass(context.Background(), h.s, h.tw); err != nil {
		t.Fatal(err)
	}
}

// laneChore is the lane's park the abort queued in the target region.
func (h *abortHarness) laneChore(t *testing.T) *hostclaim.Obligation {
	t.Helper()
	for _, o := range h.claim(t).Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) {
			return &o
		}
	}
	return nil
}

// laneKeys are the objects still in the move's lane in the target bucket.
func (h *abortHarness) laneKeys() []string {
	var out []string
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, moves.ScopeLocation(p1Host, p1ID)+"/") {
			out = append(out, k)
		}
	}
	return out
}

// archived counts the objects under one retirement's archive of the lane.
func (h *abortHarness) archived(inc, ret string) int {
	prefix := "parked/" + inc + "/" + p1Host + "/" + ret + "/moves/"
	n := 0
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			n++
		}
	}
	return n
}

// targetDeletes are every delete request anyone sent to the target bucket:
// a plain delete, a guarded staging delete, or a delete by version.
func (h *abortHarness) targetDeletes() []string {
	var out []string
	for _, op := range h.f.ops {
		if strings.HasPrefix(op, "delete data-uk/") {
			out = append(out, op)
		}
	}
	for _, c := range h.f.stagingCalls() {
		if c.Op == "delete" && c.Bucket == "data-uk" {
			out = append(out, "staging delete "+c.Key)
		}
	}
	out = append(out, h.vs.callsOn("delete-version data-uk/")...)
	return out
}

func (h *abortHarness) pastHorizon(t *testing.T, o *hostclaim.Obligation) {
	t.Helper()
	until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit)
	if d := until.Sub(h.clock.now); d > 0 {
		h.clock.Advance(d + time.Second)
	}
}

// waiting asserts the abort is still waiting for the lane park: the record
// aborting, the claim moving/aborting under the source, the chore owed.
func (h *abortHarness) waiting(t *testing.T) {
	t.Helper()
	if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting {
		t.Fatalf("the record is not aborting: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseAborting || c.Op.ID != p1ID || c.Op.Exec != "ie-box" {
		t.Fatalf("the claim while the abort waits: %+v", c.Op)
	}
}

// closed asserts the end: the record at rest in ie with LastMove aborted,
// the claim serving under the source with this move recorded as released.
func (h *abortHarness) closed(t *testing.T) {
	t.Helper()
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.Region != "ie" || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "aborted" {
		t.Fatalf("the record after the abort: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" || hostclaim.RequireReleasedMove(p1ID)(c) != nil || len(c.Obligations) != 0 {
		t.Fatalf("the claim after the abort: %+v obligations %+v", c.Op, c.Obligations)
	}
}

// ownStateGone asserts the source retired its own state for the move: the
// snapshots, receipts and ledger in its bucket, and its journal directory.
func (h *abortHarness) ownStateGone(t *testing.T) {
	t.Helper()
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) || k == moves.ReceiptsKey(p1Host, p1ID) || k == moves.BuildStateReceiptsKey(p1Host, p1ID) || k == moves.LedgerKey(p1Host, p1ID) {
			t.Fatalf("%s survived the abort", k)
		}
	}
	if _, err := os.Stat(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID)); !os.IsNotExist(err) {
		t.Fatalf("the source's journal directory is still there: %v", err)
	}
}

// TestAbortParksFromTheTargetAndWaits (Review Focus 2, §7.11, joint N13): the
// source's passes never send a delete to the target bucket; the source takes
// the claim to moving/aborting with the lane's park queued for the target
// region, and retires its own snapshots (by version), receipts, ledger and
// journal. The target region's box archives moves/<host>/<id>/... under
// parked/<inc>/<host>/<ret>/moves/... and acknowledges only after the
// horizon. Only then does the source release the claim to serving under
// itself, and the record clear.
func TestAbortParksFromTheTargetAndWaits(t *testing.T) {
	h := newAbortHarness(t)
	lane := h.laneKeys()
	h.abort(t)
	deletes := len(h.targetDeletes())
	acquired := h.lease.acquired
	decided := h.clock.now
	h.pass(t)
	h.waiting(t)
	o := h.laneChore(t)
	if o == nil {
		t.Fatal("no lane park queued")
	}
	if o.Owner != "region:uk" || o.Park.Incarnation != h.inc || o.Park.RequestLimit != moves.ParkRequestLimit {
		t.Fatalf("the lane park: %+v %+v", o, o.Park)
	}
	if min := decided.Add(moves.SignatureLife + moves.DefaultSkew); o.Park.EvidenceAt.Before(min) || o.Park.EvidenceAt.After(min.Add(time.Second)) {
		t.Fatalf("the park's horizon starts at %v, want the decision's time plus a signature's life and the skew, %v", o.Park.EvidenceAt, min)
	}
	if h.lease.acquired != acquired+1 || h.lease.released != h.lease.acquired {
		t.Fatalf("the staging lease: acquired %d (was %d), released %d", h.lease.acquired, acquired, h.lease.released)
	}
	h.ownStateGone(t)
	if got := h.vs.callsOn("delete-version data-ie/" + moves.SnapshotsPrefix(p1Host, p1ID)); len(got) == 0 {
		t.Fatal("the snapshots were not retired by version")
	}
	if got := h.laneKeys(); len(got) != len(lane) {
		t.Fatalf("the source touched the lane: %d keys, had %d", len(got), len(lane))
	}
	// The source waits: another pass changes nothing and deletes nothing there.
	gen := h.claim(t).Gen
	h.pass(t)
	h.waiting(t)
	if h.claim(t).Gen != gen {
		t.Fatal("a waiting source wrote the claim")
	}
	if got := h.targetDeletes(); len(got) != deletes {
		t.Fatalf("the source sent deletes to the target bucket: %v", got[deletes:])
	}
	// The target region's box archives the lane; the horizon has not passed.
	h.tPass(t)
	if left := h.laneKeys(); len(left) != 0 {
		t.Fatalf("the lane still holds %v", left)
	}
	if n := h.archived(o.Park.Incarnation, o.Retirement); n < len(lane) {
		t.Fatalf("%d archived, the lane had %d", n, len(lane))
	}
	if h.laneChore(t) == nil {
		t.Fatal("the lane park was acknowledged before its horizon")
	}
	h.pass(t)
	h.waiting(t)
	// Past the horizon the target acknowledges; the source then releases.
	h.pastHorizon(t, o)
	h.tPass(t)
	if h.laneChore(t) != nil {
		t.Fatal("the lane park was not acknowledged after its horizon")
	}
	h.waiting(t)
	h.pass(t)
	h.closed(t)
	before := h.claim(t)
	h.pass(t)
	h.tPass(t)
	if after := h.claim(t); after.Gen != before.Gen {
		t.Fatalf("a pass after the abort wrote the claim (%d to %d)", before.Gen, after.Gen)
	}
}

// TestAbortScopeIsTheWritersHorizon (witness wStagingQuiet; ruling K5): the
// abort's horizon covers every staging request the source signed before it,
// and a staging PUT signed before the abort that lands after the park's
// first pass is found by the verification listing and parked as a further
// retirement event. The park acknowledges only after abort_bound + H.
func TestAbortScopeIsTheWritersHorizon(t *testing.T) {
	h := newAbortHarness(t)
	signed := h.f.stagingCalls()
	if len(signed) == 0 {
		t.Fatal("the fixture signed no staging request")
	}
	h.abort(t)
	h.pass(t)
	o := h.laneChore(t)
	if o == nil {
		t.Fatal("no lane park queued")
	}
	bound := o.Park.EvidenceAt
	for _, c := range signed {
		if c.SignAt.Add(moves.SignatureLife + moves.DefaultSkew).After(bound) {
			t.Fatalf("a staging request signed at %v can land after the abort's horizon %v", c.SignAt, bound)
		}
	}
	// The park's first pass archives the lane.
	h.tPass(t)
	first := o.Retirement
	if len(h.laneKeys()) != 0 || h.laneChore(t) == nil {
		t.Fatalf("after the first pass: lane %v", h.laneKeys())
	}
	// A request signed before the abort lands now, in the lane.
	late := moves.ScopeLocation(p1Host, p1ID) + "/replica/0000/0000000000000009-0000000000000009.ltx"
	h.f.store("data-uk", late, fakeMoveObject{lease.MirrorObject{ETag: "late", Size: 4}, "late"})
	h.f.setVersion("data-uk", late, h.f.nextVersion())
	// One second before abort_bound + H nothing is acknowledged, whatever
	// the listing says.
	deadline := bound.Add(park.SignatureAge + o.Park.RequestLimit)
	if d := deadline.Add(-time.Second).Sub(h.clock.now); d > 0 {
		h.clock.Advance(d)
	}
	h.tPass(t)
	if h.laneChore(t) == nil {
		t.Fatal("the lane park was acknowledged before abort_bound + H")
	}
	h.pass(t)
	h.waiting(t)
	// Past the horizon the listing finds the late write: a further retirement.
	h.pastHorizon(t, o)
	h.tPass(t)
	o2 := h.laneChore(t)
	if o2 == nil {
		t.Fatal("the park acknowledged over the late write")
	}
	if o2.Retirement == first || !o2.Park.EvidenceAt.Equal(bound) {
		t.Fatalf("the further event: retirement %q (first %q), horizon %v", o2.Retirement, first, o2.Park.EvidenceAt)
	}
	h.tPass(t)
	if h.laneChore(t) != nil {
		t.Fatal("the further retirement was not acknowledged")
	}
	if _, ok := h.f.buckets["data-uk"][late]; ok {
		t.Fatal("the late write is still in the lane")
	}
	if n := h.archived(o.Park.Incarnation, o2.Retirement); n != 1 {
		t.Fatalf("the further retirement archived %d objects, want the late write", n)
	}
	if q := h.claim(t).QuietAfter["uk"]; !q.Equal(deadline) {
		t.Fatalf("quiet_after %v, want %v", q, deadline)
	}
	h.pass(t)
	h.closed(t)
}

// TestAbortWaitsForAnUnreachableTargetRegion (ruling K21): with no box of
// the target region running the lane park, the claim stays moving/aborting
// and nothing releases, however long the source waits. A delete cannot
// begin over it, because the claim is not serving.
func TestAbortWaitsForAnUnreachableTargetRegion(t *testing.T) {
	h := newAbortHarness(t)
	h.abort(t)
	h.pass(t)
	o := h.laneChore(t)
	if o == nil {
		t.Fatal("no lane park queued")
	}
	for i := 0; i < 4; i++ {
		h.clock.Advance(12 * time.Hour)
		h.pass(t)
		h.waiting(t)
		if h.laneChore(t) == nil {
			t.Fatal("the lane park vanished with no target pass")
		}
	}
	c := h.claim(t)
	if err := hostclaim.StartGuard(c, []string{"ie"}, []string{"ie-box"}, ""); err == nil {
		t.Fatal("a new operation could start over the waiting abort")
	}
	// An unarmed target box parks nothing either: the abort's lane park goes
	// through laneParkTarget (movesfinal.go's dispatch for an aborting
	// target), whose arming is the shared lane predicate, revertLaneArmed,
	// which requires all three of the claim's park hooks (Plan 5 Task 9a,
	// M17). Each one missing leaves the arm unarmed, with its standing
	// notice, and the lane untouched.
	for name, unarm := range map[string]func(*moveCfg){
		"no park leases":     func(c *moveCfg) { c.ParkLeases = nil },
		"no park view":       func(c *moveCfg) { c.ParkView = nil },
		"no park authorizer": func(c *moveCfg) { c.ParkAuthorize = nil },
	} {
		unarmed := h.laneCfg()
		unarm(&unarmed)
		logs := captureLog(t)
		w := &movesWorker{cfg: unarmed, mirrors: map[string]lease.MirrorStore{}}
		if err := movesPass(context.Background(), h.s, w); err != nil {
			t.Fatal(err)
		}
		if h.laneChore(t) == nil || len(h.laneKeys()) == 0 {
			t.Fatalf("%s: an unarmed target box parked the lane", name)
		}
		if !strings.Contains(logs(), "this box's park engine is not armed") {
			t.Fatalf("%s: the abort's lane park was tried by an unarmed box:\n%s", name, logs())
		}
	}
	// Once the region returns, the abort completes.
	h.tPass(t)
	h.pastHorizon(t, o)
	h.tPass(t)
	h.pass(t)
	h.closed(t)
}

// TestAbortReplays: a crash after any of the abort's side effects, then a
// restarted agent, still reaches the close, with the journalled bound named
// once (the lane park's scope never changes), one lane park queued, and the
// source's own state retired.
func TestAbortReplays(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"after-abort-bound", "after-abort-claim", "after-abort-own-state", "after-abort-journal", "after-release-abort"} {
		t.Run(point, func(t *testing.T) {
			h := newAbortHarness(t)
			h.abort(t)
			h.abAt = point
			if err := movesPass(context.Background(), h.s, h.w()); err != nil {
				t.Fatal(err)
			}
			for i := 0; i < 6 && !h.abDone; i++ {
				h.tPass(t)
				if o := h.laneChore(t); o != nil {
					h.pastHorizon(t, o)
				}
				h.pass(t)
			}
			if !h.abDone {
				t.Fatalf("never reached %s (%v)", point, h.abHits)
			}
			var bound time.Time
			if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID); err == nil {
				if raw := j.Steps[abortStepBound]; raw != "" {
					bound, _ = time.Parse(time.RFC3339Nano, raw)
				}
			}
			if o := h.laneChore(t); o != nil {
				if !bound.IsZero() && !o.Park.EvidenceAt.Equal(bound) {
					t.Fatalf("the park's horizon %v is not the journalled bound %v", o.Park.EvidenceAt, bound)
				}
				bound = o.Park.EvidenceAt
			}
			h.clock.Advance(time.Minute)
			h.worker1 = nil
			for i := 0; i < 8 && h.read(t, p1Host).Move != nil; i++ {
				h.pass(t)
				if o := h.laneChore(t); o != nil {
					if !bound.IsZero() && !o.Park.EvidenceAt.Equal(bound) {
						t.Fatalf("the replay named another horizon: %v, journalled %v", o.Park.EvidenceAt, bound)
					}
					h.tPass(t)
					h.pastHorizon(t, o)
					h.tPass(t)
				}
			}
			h.closed(t)
			h.ownStateGone(t)
		})
	}
}

// An aborting record never flipped: its region is the source's, so the
// target region's box has neither the row nor a journal of the move. The
// participant rule answers it as the target on the state alone, as for a
// revert: its one piece of work is the lane park (ruling K6). The source's
// region answers nothing without the row (its abort is phase1MoveRound's),
// and a third region nothing.
func TestAbortingTargetIsAParticipant(t *testing.T) {
	cfg := moveCfg{Region: "uk", DataDir: t.TempDir(), OwnsRoute: func(string) bool { return false }}
	rec := provision.Record{Host: p1Host, Region: "ie", Move: &provision.MoveState{ID: p1ID, State: provision.MoveAborting, From: "ie", To: "uk"}}
	if got := movesParticipantRecord(cfg, rec); got != roleTarget {
		t.Fatalf("an aborting record's target: %q", got)
	}
	cfg.Region = "ie"
	if got := movesParticipantRecord(cfg, rec); got != roleNone {
		t.Fatalf("the source's region without the row: %q", got)
	}
	cfg.Region = "fr"
	if got := movesParticipantRecord(cfg, rec); got != roleNone {
		t.Fatalf("a third region: %q", got)
	}
}

// The abort's crash boundary between the claim's release and the record's
// clear: a round that finds the record aborting and the claim serving
// completes the record only when the serving op records this move as the
// one it released (RequireReleasedMove). A serving claim without that
// evidence was released by someone else, and the record is left alone.
func TestAbortCompletesOnlyWithItsReleaseEvidence(t *testing.T) {
	ctx := context.Background()
	h := newAbortHarness(t)
	h.abort(t)
	obs, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	scope := moves.AbortScope(p1Host, p1ID, h.inc, h.clock.now.Add(time.Hour))
	if obs, err = hostclaim.Update(ctx, h.s, obs, hostclaim.Abort(p1ID, "uk", scope, obligationID)); err != nil {
		t.Fatal(err)
	}
	if obs, err = hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, obs.Claim.Obligations[0].ID)); err != nil {
		t.Fatal(err)
	}
	if _, err = hostclaim.Update(ctx, h.s, obs, hostclaim.ReleaseAbort(p1ID, "ie-box")); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	h.closed(t)
	h.ownStateGone(t)

	// A serving claim with no evidence of this release: left alone.
	h2 := newAbortHarness(t)
	h2.abort(t)
	rewriteClaimOp(t, h2.s, func(op *hostclaim.Op) { *op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "ie-box"} })
	h2.pass(t)
	if got := h2.read(t, p1Host); got.Move == nil || got.Move.State != provision.MoveAborting {
		t.Fatalf("a serving claim without evidence completed the record: %+v", got.Move)
	}
}

// Fix round 1 (task-1-review.md, Important): release never follows from an
// inference that the source's cleanup finished. After the claim's CAS, the
// source retires its own state on every pass before ReleaseAbort, and any
// failure to do so blocks the release until it succeeds.
func TestAbortReleaseWaitsForTheSourcesCleanup(t *testing.T) {
	// crashAfterClaim stops the source right after the Abort CAS, then lets
	// the target region park and acknowledge the lane, so only the source's
	// own cleanup stands between the claim and its release.
	crashAfterClaim := func(t *testing.T) *abortHarness {
		t.Helper()
		h := newAbortHarness(t)
		h.abort(t)
		h.abAt = "after-abort-claim"
		if err := movesPass(context.Background(), h.s, h.w()); err != nil {
			t.Fatal(err)
		}
		if !h.abDone {
			t.Fatalf("never reached after-abort-claim (%v)", h.abHits)
		}
		o := h.laneChore(t)
		if o == nil {
			t.Fatal("no lane park queued")
		}
		h.tPass(t)
		h.pastHorizon(t, o)
		h.tPass(t)
		if h.laneChore(t) != nil {
			t.Fatal("the lane park was not acknowledged")
		}
		h.worker1 = nil
		return h
	}
	t.Run("the journal is gone but the state is not", func(t *testing.T) {
		h := crashAfterClaim(t)
		if err := os.RemoveAll(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		h.closed(t)
		h.ownStateGone(t)
	})
	t.Run("the journal cannot be inspected", func(t *testing.T) {
		if os.Geteuid() == 0 {
			t.Skip("root reads and removes past any mode")
		}
		h := crashAfterClaim(t)
		hostDir := filepath.Dir(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID))
		if err := os.Chmod(hostDir, 0); err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() { os.Chmod(hostDir, 0o755) })
		for i := 0; i < 3; i++ {
			movesPass(context.Background(), h.s, h.w())
			h.waiting(t)
		}
		if err := os.Chmod(hostDir, 0o755); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		h.closed(t)
		h.ownStateGone(t)
	})
	t.Run("the snapshots cannot be listed", func(t *testing.T) {
		h := crashAfterClaim(t)
		h.vs.deny = func(op, bucket, key string) error {
			if op == "list-versions" && bucket == "data-ie" {
				return errors.New("injected 503")
			}
			return nil
		}
		for i := 0; i < 3; i++ {
			movesPass(context.Background(), h.s, h.w())
			h.waiting(t)
		}
		h.vs.deny = nil
		h.pass(t)
		h.closed(t)
		h.ownStateGone(t)
	})
}