package host

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"

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

// probeLog is a fake probe wake. Each call admits a writer on the claim as
// the claim's runtime would (a cold admission into the move id, or a fresh
// generation once the move id is admitted), journals the session's binding
// to the record's attempt as Task 5's attempts do (and, when openWindow is
// set, the attempt's opened time), then answers from its script. A passing
// call leaves a live entry for the admitted session.
type probeLog struct {
	t          *testing.T
	h          *landingHarness
	configs    []activator.ProbeConfig
	answers    []error
	before     func(n int)
	admissions []probeAdmission
	openWindow bool
	// noBind leaves the admission without Task 5's binding: a crash
	// between the admission CAS and Reserve(0).
	noBind bool
	last   activator.ProbeResult
	live   bool
}

// probeAdmission is one admission the fake made: the record attempt it was
// made under and the session it admitted.
type probeAdmission struct {
	attempt int
	session uint64
}

func (p *probeLog) wake(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) {
	p.configs = append(p.configs, cfg)
	n := len(p.configs)
	if p.before != nil {
		p.before(n)
	}
	m := p.h.read(p.t, p1Host).Move
	k := m.Landing.Attempts
	obs, err := hostclaim.Read(ctx, p.h.s, p1Host)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	h := obs.Claim.Holder
	gen := p1ID
	if h.WriterBox == p.h.dstBox {
		gen = "fresh" + strconv.Itoa(n)
	}
	// The predecessor's region is the probe owners' own rule: the last
	// admission's region, or the move's source's engine region.
	from, err := predecessorRegion(h, m.From)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	admitted, err := hostclaim.Update(ctx, p.h.s, obs, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: p1ID, ObservedPhase: provision.LandProbing,
		ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: p.h.dstBox, Mode: "cold", NewGeneration: gen, KeepStamp: h.Stamp > h.Session, Region: p.h.dstEngine, PredecessorRegion: from}))
	if err != nil {
		return activator.ProbeResult{}, err
	}
	ah := admitted.Claim.Holder
	p.admissions = append(p.admissions, probeAdmission{attempt: k, session: ah.Session})
	j, err := moves.ReadJournal(p.h.tdir, p1Host, p1ID)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	if !p.noBind {
		j.Mark(probeStepSession+strconv.FormatUint(ah.Session, 10), strconv.Itoa(k))
	}
	if p.openWindow {
		j.Mark(probeStepOpened+strconv.Itoa(k), p.h.clock.Now().UTC().Format(time.RFC3339Nano))
	}
	if err := j.Commit(p.h.tdir); err != nil {
		return activator.ProbeResult{}, err
	}
	p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation}
	if n <= len(p.answers) && p.answers[n-1] != nil {
		p.live = false
		return activator.ProbeResult{}, p.answers[n-1]
	}
	p.live = true
	return p.last, nil
}

// liveProbe is the activator's LiveProbe over the fake's last passing call.
func (p *probeLog) liveProbe(host string) (string, activator.ProbeResult, bool) {
	if host != p1Host || !p.live {
		return "", activator.ProbeResult{}, false
	}
	return p1ID, p.last, true
}

func failedAttempt(why string) error {
	return errors.Join(activator.ErrProbeFailed, errors.New(why))
}

// installedLanding is a landing at restored, with the cut's admitted probe
// attempts set to attempts (0 clears them) through the record's writer.
func installedLanding(t *testing.T, attempts int) (*landingHarness, *probeLog) {
	t.Helper()
	h := newLandingHarness(t)
	h.flip(t)
	h.land(t)
	h.installed(t)
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.CutAttempts = attempts
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	p := &probeLog{t: t, h: h}
	h.probe, h.probeLive = p.wake, p.liveProbe
	h.holdProbed = true
	return h, p
}

func (h *landingHarness) atPhase(t *testing.T, phase string) *provision.MoveState {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m.State != provision.MoveLanding || m.LandingPhase != phase || m.Reason != "" {
		t.Fatalf("the record is not landing/%s: %+v", phase, m)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != phase {
		t.Fatalf("the claim is not landing/%s: %+v", phase, c.Op)
	}
	return m
}

// TestLandingProbesWithinTheBudget (brief): two admitted attempts, the
// first fails and the second passes, so the landing is probed, record and
// claim; a third is never attempted. Each attempt is counted on the record
// and journalled before its call, and the probe is given the landing's
// deadline (H8), the build's manifest and the pins.
func TestLandingProbesWithinTheBudget(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	p.answers = []error{failedAttempt("not ready")}
	h.land(t)
	m := h.atPhase(t, provision.LandProbed)
	if len(p.configs) != 2 || m.Landing == nil || m.Landing.Attempts != 2 || h.landJournal(t).Attempt != 2 {
		t.Fatalf("attempts: calls %d record %+v journal %d", len(p.configs), m.Landing, h.landJournal(t).Attempt)
	}
	deadline, _ := landingDeadlineOf(m)
	want := activator.ProbeConfig{Host: p1Host, Op: p1ID, HoldOp: p1ID, Deadline: *deadline, Prepared: h.preparedManifest(t), Inputs: m.Inputs}
	for i, got := range p.configs {
		if !reflect.DeepEqual(got, want) {
			t.Fatalf("probe %d was given %+v, want %+v", i+1, got, want)
		}
	}
	// The passing attempt's admitted writer (fix round 1, Minor): its
	// session, stamp and generation survive the probed mirroring.
	if c := h.claim(t); len(p.admissions) != 2 || c.Holder.Session != p.last.Session || c.Holder.Stamp != p.last.Stamp || c.Holder.Generation != p.last.Generation || c.Holder.WriterBox != "uk-box" {
		t.Fatalf("the probed step did not keep the admitted writer %+v: holder %+v", p.last, c.Holder)
	}
	// probed is the probe's end: another pass probes nothing.
	h.land(t)
	if len(p.configs) != 2 {
		t.Fatalf("a pass at probed probed again: %d", len(p.configs))
	}
	h.atPhase(t, provision.LandProbed)
}

// preparedManifest is the build's install manifest from the op journal.
func (h *landingHarness) preparedManifest(t *testing.T) *moves.InstallManifest {
	t.Helper()
	m := h.installedManifest(t)
	return &m
}

func (h *landingHarness) installedManifest(t *testing.T) moves.InstallManifest {
	t.Helper()
	j := h.landJournal(t)
	var m moves.InstallManifest
	if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &m); err != nil {
		t.Fatal(err)
	}
	return m
}

// TestLandingExhaustedAttemptsMarkRollback (brief): every admitted attempt
// fails: the landing records the attempts-exhausted rollback sentence,
// which the next pass's rollback decision turns into the rollback.
func TestLandingExhaustedAttemptsMarkRollback(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	p.answers = []error{failedAttempt("one"), errors.Join(activator.ErrProbeFailed, activator.ErrAttemptOver)}
	h.land(t)
	if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackProbeAttempts || len(p.configs) != 2 {
		t.Fatalf("exhausted: %d calls, %+v", len(p.configs), m)
	}
	h.land(t)
	rolledBack(t, h, provision.RollbackProbeAttempts, provision.LandProbing)
	if len(p.configs) != 2 {
		t.Fatalf("a rolled-back landing probed again: %d", len(p.configs))
	}
}

// A changed configuration found by the probe (Task 5's ProbeInputsChanged,
// before or after the admission) is the configuration rollback, at once.
func TestLandingConfigChangedDuringTheProbeRollsBack(t *testing.T) {
	t.Parallel()
	for name, answer := range map[string]error{
		"before the admission": &ProbeInputsChanged{Reason: provision.RollbackConfigChanged},
		"after the admission":  errors.Join(activator.ErrProbeFailed, &ProbeInputsChanged{Reason: provision.RollbackConfigChanged}),
	} {
		t.Run(name, func(t *testing.T) {
			h, p := installedLanding(t, 3)
			p.answers = []error{answer}
			h.land(t)
			if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackConfigChanged || len(p.configs) != 1 {
				t.Fatalf("%d calls, %+v", len(p.configs), m)
			}
			h.land(t)
			rolledBack(t, h, provision.RollbackConfigChanged, provision.LandProbing)
		})
	}
}

// H10: a cut that recorded no probe attempts refuses the probe with its
// recorded sentence; nothing is probed.
func TestLandingProbeNeedsTheCutsAttempts(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 0)
	h.land(t)
	if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackProbeUnpriced || len(p.configs) != 0 {
		t.Fatalf("%d calls, %+v", len(p.configs), m)
	}
	h.land(t)
	rolledBack(t, h, provision.RollbackProbeUnpriced, provision.LandProbing)
}

// H8: probing and probed check the landing's deadline at entry. A deadline
// that passes during an attempt ends the probing with the deadline's
// sentence, and no further attempt is made; a probed landing past its
// deadline rolls back.
func TestLandingProbingChecksTheDeadline(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 3)
	p.before = func(int) { h.clock.Advance(cutLandingWindow + time.Minute) }
	p.answers = []error{errors.Join(activator.ErrProbeFailed, activator.ErrAttemptOver)}
	h.land(t)
	if len(p.configs) != 1 {
		t.Fatalf("probed past the deadline: %d calls", len(p.configs))
	}
	h.land(t)
	rolledBack(t, h, provision.RollbackLandingDeadline, provision.LandProbing)

	h2, _ := installedLanding(t, 3)
	h2.land(t)
	h2.atPhase(t, provision.LandProbed)
	h2.clock.Advance(cutLandingWindow + time.Minute)
	h2.land(t)
	rolledBack(t, h2, provision.RollbackLandingDeadline, provision.LandProbed)
}

// TestLandingProbeCrashRestartsOnlyTheProbe (brief, §7.13): a crash after
// an attempt is counted and before its probe returns restarts only the
// probe: a fresh worker probes the same attempt again (its window never
// opened), and local and restored are not rerun, their journal steps and
// .files-gen untouched. An attempt whose window had opened and passed
// counts as failed, and the next one is made.
func TestLandingProbeCrashRestartsOnlyTheProbe(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	steps := map[string]string{}
	for k, v := range h.landJournal(t).Steps {
		steps[k] = v
	}
	marker := filepath.Join(h.tdir, p1Host, archive.MarkerName)
	before, err := os.Stat(marker)
	if err != nil {
		t.Fatal(err)
	}
	renders, timers := h.renders, len(h.replaced)
	h.landAt = "before-probe-1"
	h.crashLand(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || m.Landing.Attempts != 1 || h.landJournal(t).Attempt != 1 || len(p.configs) != 0 {
		t.Fatalf("the crash point: %+v journal %d calls %d", m, h.landJournal(t).Attempt, len(p.configs))
	}
	h.freshTarget()
	h.land(t)
	m := h.atPhase(t, provision.LandProbed)
	if len(p.configs) != 1 || m.Landing.Attempts != 1 {
		t.Fatalf("after the restart: %d calls, record %+v", len(p.configs), m.Landing)
	}
	after, err := os.Stat(marker)
	if err != nil || !after.ModTime().Equal(before.ModTime()) {
		t.Fatalf(".files-gen was written again: %v", err)
	}
	j := h.landJournal(t)
	for k, v := range steps {
		if j.Steps[k] != v {
			t.Fatalf("the journal step %s changed", k)
		}
	}
	for k := range j.Steps {
		if _, ok := steps[k]; !ok && !strings.HasPrefix(k, "probe_") {
			t.Fatalf("a non-probe step %s was added", k)
		}
	}
	if h.renders != renders || len(h.replaced) != timers {
		t.Fatalf("configured or local ran again: renders %d->%d timers %d->%d", renders, h.renders, timers, len(h.replaced))
	}

	// An attempt whose window opened and passed before the crash counts as
	// failed: the restarted worker makes the next one.
	h2, p2 := installedLanding(t, 2)
	h2.landAt = "before-probe-1"
	h2.crashLand(t)
	j2 := h2.landJournal(t)
	j2.Mark("probe_opened_1", h2.clock.Now().Add(-31*time.Second).UTC().Format(time.RFC3339Nano))
	if err := j2.Commit(h2.tdir); err != nil {
		t.Fatal(err)
	}
	h2.freshTarget()
	h2.land(t)
	if m := h2.atPhase(t, provision.LandProbed); len(p2.configs) != 1 || m.Landing.Attempts != 2 {
		t.Fatalf("the spent attempt was retried: %d calls, record %+v", len(p2.configs), m.Landing)
	}
}

// Unarmed (production until claim Plan 11's launcher and Plan 5 Task 9):
// with no probe wake the landing waits at restored and writes nothing.
func TestLandingUnarmedProbeWaitsAtRestored(t *testing.T) {
	t.Parallel()
	h, _ := installedLanding(t, 3)
	h.probe = nil
	before := h.read(t, p1Host).Move.UpdatedAt
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandRestored || !m.UpdatedAt.Equal(before) {
		t.Fatalf("an unarmed probe moved the landing: %+v", m)
	}
}

// H10: Task 5's Admitted hook reads the admission's recorded count, and
// refuses a record that has none.
func TestMovesRecordAdmittedAttempts(t *testing.T) {
	ctx := context.Background()
	if n, err := MovesRecordAdmittedAttempts(ctx, &provision.Record{Move: &provision.MoveState{CutAttempts: 2}}); err != nil || n != 2 {
		t.Fatalf("recorded: %d %v", n, err)
	}
	for _, rec := range []*provision.Record{nil, {}, {Move: &provision.MoveState{}}} {
		if _, err := MovesRecordAdmittedAttempts(ctx, rec); err == nil {
			t.Fatalf("no recorded count was admitted: %+v", rec)
		}
	}
}

// Fix round 1, ruling 1: a passing probe whose probing to probed record
// write fails is not probed again. probe_passed_<k> names the admitted
// writer; the next pass validates it against the claim and the activator's
// live entry and retries only the phase writes, record then claim.
func TestLandingProbePassReplaysTheRecordWrite(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	h.landAt = "before-record-probed"
	h.crashLand(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || len(p.configs) != 1 {
		t.Fatalf("the crash point: %d calls, %+v", len(p.configs), m)
	}
	if !h.landJournal(t).Done("probe_passed_1") {
		t.Fatalf("the pass was not journalled before the record write: %v", h.landJournal(t).Steps)
	}
	h.land(t)
	h.atPhase(t, provision.LandProbed)
	if len(p.configs) != 1 || len(p.admissions) != 1 {
		t.Fatalf("the replay admitted again: %d calls, %d admissions", len(p.configs), len(p.admissions))
	}
	if c := h.claim(t); c.Holder.Session != p.last.Session || c.Holder.Generation != p1ID {
		t.Fatalf("the probed writer: %+v", c.Holder)
	}

	// The session is gone (the entry did not survive): the attempt is
	// consumed and failed, and the next one is made.
	h2, p2 := installedLanding(t, 2)
	h2.landAt = "before-record-probed"
	h2.crashLand(t)
	p2.live = false
	h2.land(t)
	if m := h2.atPhase(t, provision.LandProbed); len(p2.configs) != 2 || m.Landing.Attempts != 2 || p2.admissions[1].attempt != 2 {
		t.Fatalf("a lost passing session: %d calls, record %+v, admissions %+v", len(p2.configs), m.Landing, p2.admissions)
	}
}

// Fix round 1, ruling 2 (§7.7): one cold admission per budgeted attempt. A
// crash after the attempt's admission, with its window still open and no
// failure journalled, consumes the attempt: the restart never admits again
// under the same index.
func TestLandingProbeAdmittedAttemptIsConsumedOnRestart(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	p.openWindow = true
	p.answers = []error{errors.New("the controller died during the attempt")}
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || len(p.admissions) != 1 || p.admissions[0].attempt != 1 || h.landJournal(t).Done("probe_failed_1") {
		t.Fatalf("the crash: %+v admissions %+v", m, p.admissions)
	}
	h.freshTarget()
	h.land(t)
	h.atPhase(t, provision.LandProbed)
	if len(p.admissions) != 2 || p.admissions[1].attempt != 2 {
		t.Fatalf("an index was admitted twice: %+v", p.admissions)
	}

	// With one attempt admitted, the restart has none left.
	h1, p1 := installedLanding(t, 1)
	p1.openWindow = true
	p1.answers = []error{errors.New("the controller died during the attempt")}
	h1.land(t)
	h1.freshTarget()
	h1.land(t)
	if m := h1.read(t, p1Host).Move; m.Reason != provision.RollbackProbeAttempts || len(p1.admissions) != 1 {
		t.Fatalf("the only attempt was admitted again: %+v admissions %+v", m, p1.admissions)
	}
}

// Fix round 2, ruling 1: the claim, not the binding, says whether attempt k
// was admitted. A crash after the admission CAS and before Reserve's
// binding leaves probe_admitting_<k> and a claim whose writer session moved
// on: the restart consumes k and never admits under it again.
func TestLandingProbeAdmissionWithoutBindingIsConsumed(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	p.noBind = true
	p.answers = []error{errors.New("the controller died after the admission")}
	h.land(t)
	if len(p.admissions) != 1 || !h.landJournal(t).Done("probe_admitting_1") || attemptAdmitted(h.landJournal(t), 1) {
		t.Fatalf("the crash: admissions %+v journal %v", p.admissions, h.landJournal(t).Steps)
	}
	h.freshTarget()
	p.noBind = false
	h.land(t)
	h.atPhase(t, provision.LandProbed)
	if len(p.admissions) != 2 || p.admissions[1].attempt != 2 {
		t.Fatalf("index 1 was admitted twice: %+v", p.admissions)
	}

	// probe_admitting_<k> with the claim unchanged (no admission happened):
	// the same index is used.
	h2, p2 := installedLanding(t, 2)
	h2.landAt = "before-probe-1"
	h2.crashLand(t)
	if !h2.landJournal(t).Done("probe_admitting_1") {
		t.Fatalf("no admitting step before the call: %v", h2.landJournal(t).Steps)
	}
	h2.freshTarget()
	h2.land(t)
	if m := h2.atPhase(t, provision.LandProbed); m.Landing.Attempts != 1 || len(p2.admissions) != 1 || p2.admissions[0].attempt != 1 {
		t.Fatalf("an unadmitted attempt was not reused: %+v %+v", m.Landing, p2.admissions)
	}
}

// Fix round 2, ruling 2: a probe that passed but whose probe_passed_<k>
// write failed is found through the activator's live entry: its writer is
// the claim's, so the attempt passed, and the next pass reaches probed
// without a second ProbeWake.
func TestLandingProbePassFoundThroughTheLiveEntry(t *testing.T) {
	t.Parallel()
	h, p := installedLanding(t, 2)
	h.landAt = "before-journal-passed"
	h.crashLand(t)
	if j := h.landJournal(t); j.Done("probe_passed_1") || len(p.configs) != 1 || !p.live {
		t.Fatalf("the crash point: %v calls %d live %v", j.Steps, len(p.configs), p.live)
	}
	h.land(t)
	h.atPhase(t, provision.LandProbed)
	if len(p.configs) != 1 || len(p.admissions) != 1 || !h.landJournal(t).Done("probe_passed_1") {
		t.Fatalf("probed again: %d calls %d admissions", len(p.configs), len(p.admissions))
	}
	if c := h.claim(t); c.Holder.Session != p.last.Session {
		t.Fatalf("the probed writer: %+v", c.Holder)
	}
}