package hostops

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"maps"
	"strings"
	"testing"
	"time"

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

// resolveControlStore is the ControlWriter's raw store in these tests. Only
// a fencing CAS writes the record through it, so fences counts those. It can
// fail record reads, and run a hook just before the first fence.
type resolveControlStore struct {
	release.Store
	recordKey   string
	failGet     bool
	fences      int
	beforeFence func()
}

func (s *resolveControlStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == s.recordKey && s.failGet {
		return nil, "", errors.New("record read: connection reset")
	}
	return s.Store.Get(ctx, key)
}

func (s *resolveControlStore) Put(ctx context.Context, key string, body []byte, match string) (string, error) {
	if key == s.recordKey {
		s.fences++
		if f := s.beforeFence; f != nil {
			s.beforeFence = nil
			f()
		}
	}
	return s.Store.Put(ctx, key, body, match)
}

// lostRecordPut is the raw store beneath the conditional record writer. Its
// first Put of the record loses its answer: with landed false it fails
// without writing, with landed true it writes and then fails. It keeps that
// Put's body and If-Match ETag, so a test can deliver it late.
type lostRecordPut struct {
	release.Store
	recordKey string
	landed    bool
	armed     bool
	body      []byte
	etag      string
}

func (s *lostRecordPut) Put(ctx context.Context, key string, body []byte, match string) (string, error) {
	if key != s.recordKey || !s.armed {
		return s.Store.Put(ctx, key, body, match)
	}
	s.armed = false
	s.body, s.etag = append([]byte(nil), body...), match
	if s.landed {
		if _, err := s.Store.Put(ctx, key, body, match); err != nil {
			return "", err
		}
	}
	return "", errors.New("record put: connection reset")
}

// resolveFixture is the moves door's failure on a claim-on console, through
// the production data handle: RegisterMove, BeginClaimMove, then a
// StartClaimMoveRecord whose record Put loses its answer after it was
// issued, leaving the move-start effect uncertain.
type resolveFixture struct {
	*moveStartFixture
	control *resolveControlStore
	lost    *lostRecordPut
	data    release.Store
	h       *StartHandle
	serving hostclaim.Observed
	moving  hostclaim.Observed
	start   provision.MoveStart
	before  []byte
	tag     string
}

func newResolveFixture(t *testing.T, landed bool) *resolveFixture {
	t.Helper()
	ctx := context.Background()
	f := newMoveStartFixture(t, nil)
	r := &resolveFixture{moveStartFixture: f}
	r.control = &resolveControlStore{Store: f.raw, recordKey: f.raw.recordKey}
	var err error
	if f.w, err = NewControlWriter(r.control, f.pub, func() time.Time { return f.now }); err != nil {
		t.Fatal(err)
	}
	r.lost = &lostRecordPut{Store: f.raw, recordKey: f.raw.recordKey, landed: landed, armed: true}
	records, err := NewConditionalRecordStore(r.lost, f.pub)
	if err != nil {
		t.Fatal(err)
	}
	check, err := NewProductionMutationCheck(f.raw, f.w)
	if err != nil {
		t.Fatal(err)
	}
	if r.data, err = NewMutationBoundary(true, records, check); err != nil {
		t.Fatal(err)
	}
	f.authority.Store, f.authority.Control = r.data, f.w
	f.authority.Directory = hostactors.Directory{Store: r.data, PublicKey: f.pub, Bucket: "bucket"}
	if r.before, r.tag, err = f.raw.Store.Get(ctx, f.raw.recordKey); err != nil {
		t.Fatal(err)
	}
	r.serving = f.read(t)
	if r.h, err = f.authority.RegisterMove(ctx, r.serving, "canonical", moveStartID); err != nil {
		t.Fatal(err)
	}
	if r.moving, err = BeginClaimMove(ctx, r.data, f.authority.Directory, r.h, r.serving, moveStartID, "operator", "uk-box", "uk", f.now); err != nil {
		t.Fatal(err)
	}
	r.start = provision.MoveStart{ID: moveStartID, Incarnation: f.inc, From: "ie", To: "uk", Cut: "auto", By: "operator", Now: f.now}
	if _, err = StartClaimMoveRecord(ctx, r.data, r.h, r.moving, r.start); err == nil {
		t.Fatal("the record start succeeded through a lost Put")
	}
	if r.lost.armed || r.lost.etag != r.tag {
		t.Fatalf("the start's record Put matched on %q, want the record's ETag %q", r.lost.etag, r.tag)
	}
	if e := r.effect(t); e.State != "uncertain" || e.Action != moveStartAction {
		t.Fatalf("move-start effect after the lost Put: %+v", e)
	}
	return r
}

func (r *resolveFixture) effect(t *testing.T) Effect {
	t.Helper()
	a, err := ReadAdmission(context.Background(), r.raw.Store, "bab", "app", "canonical")
	if err != nil {
		t.Fatal(err)
	}
	return a.Value.Effects[moveStartID]
}

func (r *resolveFixture) record(t *testing.T) ([]byte, string) {
	t.Helper()
	body, tag, err := r.raw.Store.Get(context.Background(), r.raw.recordKey)
	if err != nil {
		t.Fatal(err)
	}
	return body, tag
}

// latePut delivers the lost start Put now, with its original If-Match ETag.
func (r *resolveFixture) latePut() error {
	_, err := r.raw.Store.Put(context.Background(), r.raw.recordKey, r.lost.body, r.lost.etag)
	return err
}

// A Put that failed before landing: the record lacks the journal entry and
// still carries the ETag the Put matched on, so a fencing CAS rewrites the
// same record value under a new ETag. The late Put can then only fail.
func TestPlan9B8AcceptancePutLostBeforeLandingIsFencedNotApplied(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID)
	if err != nil || got != OutcomeNotApplied {
		t.Fatalf("ResolveMoveStart = %q, %v; want not-applied", got, err)
	}
	if r.control.fences != 1 {
		t.Fatalf("fencing CAS writes = %d, want 1", r.control.fences)
	}
	body, tag := r.record(t)
	if tag == r.tag {
		t.Fatal("the fence left the ETag the start Put matched on")
	}
	before, _, _ := recordSequence(r.before)
	after, seq, parseErr := recordSequence(body)
	delete(before, "record_seq")
	delete(after, "record_seq")
	if parseErr != nil || seq != 1 || !maps.EqualFunc(before, after, func(a, b json.RawMessage) bool { return bytes.Equal(a, b) }) {
		t.Fatalf("fence did not preserve the record and advance sequence: %s %v", body, parseErr)
	}
	if rec, err := provision.DecodeRecord(body); err != nil || rec.Move != nil {
		t.Fatalf("fenced record: %+v %v", rec.Move, err)
	}
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("the late start Put after the fence: %v", err)
	}
	if e := r.effect(t); e.State != "complete" || e.ResultReference != "not-applied" {
		t.Fatalf("effect = %+v, want complete not-applied", e)
	}
	// Asking again answers the recorded outcome and writes nothing.
	if got, err = ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied || r.control.fences != 1 {
		t.Fatalf("second ResolveMoveStart = %q, %v, fences %d", got, err, r.control.fences)
	}
}

// A Put that landed but whose response was lost: the record carries this
// move's journal entry, so the effect is complete and nothing is written.
func TestPlan9B8PutLandedResponseLostIsComplete(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, true)
	_, landed := r.record(t)
	got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID)
	if err != nil || got != OutcomeComplete {
		t.Fatalf("ResolveMoveStart = %q, %v; want complete", got, err)
	}
	if _, tag := r.record(t); r.control.fences != 0 || tag != landed {
		t.Fatalf("a landed start was fenced: fences %d", r.control.fences)
	}
	if e := r.effect(t); e.State != "complete" || e.ResultReference != moveStartID {
		t.Fatalf("effect = %+v, want complete with the move id", e)
	}
	// The start's replay reads the journaled record and writes nothing.
	rec, err := StartClaimMoveRecord(ctx, r.data, r.h, r.moving, r.start)
	if err != nil || rec.Move == nil || rec.Move.ID != moveStartID {
		t.Fatalf("replay = %+v, %v", rec, err)
	}
	if _, tag := r.record(t); tag != landed {
		t.Fatal("the replay wrote the record")
	}
}

// A late Put that lands between the read and the fence makes the fencing
// CAS lose; the reread finds the journal entry and the effect is complete.
func TestPlan9B8AcceptanceLatePutBeforeTheFenceIsComplete(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	r.control.beforeFence = func() {
		if err := r.latePut(); err != nil {
			t.Error(err)
		}
	}
	got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID)
	if err != nil || got != OutcomeComplete {
		t.Fatalf("ResolveMoveStart = %q, %v; want complete", got, err)
	}
	if r.control.fences != 1 {
		t.Fatalf("fencing CAS writes = %d, want the one that lost", r.control.fences)
	}
	body, _ := r.record(t)
	if !bytes.Equal(body, r.lost.body) {
		t.Fatal("the losing fence wrote the record")
	}
	if e := r.effect(t); e.State != "complete" || e.ResultReference != moveStartID {
		t.Fatalf("effect = %+v, want complete with the move id", e)
	}
}

// A concurrent unrelated write already moved the record off the ETag the
// failed Put matched on: that Put is fenced, and no fencing write is made.
func TestPlan9B8AcceptanceUnrelatedWriteIsAlreadyAFence(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	var rec provision.Record
	if err := json.Unmarshal(r.before, &rec); err != nil {
		t.Fatal(err)
	}
	rec.Channel = "beta"
	rec.RecordSeq = 1 // An unrelated claim-on writer advanced the counter.
	body, err := json.Marshal(rec)
	if err != nil {
		t.Fatal(err)
	}
	moved, err := r.raw.Store.Put(ctx, r.raw.recordKey, body, r.tag)
	if err != nil {
		t.Fatal(err)
	}
	got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID)
	if err != nil || got != OutcomeNotApplied {
		t.Fatalf("ResolveMoveStart = %q, %v; want not-applied", got, err)
	}
	if _, tag := r.record(t); r.control.fences != 0 || tag != moved {
		t.Fatalf("an already fenced Put was fenced again: fences %d", r.control.fences)
	}
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("the late start Put after the unrelated write: %v", err)
	}
	if e := r.effect(t); e.State != "complete" || e.ResultReference != "not-applied" {
		t.Fatalf("effect = %+v, want complete not-applied", e)
	}
}

// A record read error decides nothing: the effect stays uncertain and
// nothing is written. Once the read works, the same handle resolves it.
func TestPlan9B8ReadErrorIsUncertain(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	r.control.failGet = true
	got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID)
	if err == nil || got != OutcomeUncertain {
		t.Fatalf("ResolveMoveStart = %q, %v; want uncertain with an error", got, err)
	}
	if _, tag := r.record(t); r.control.fences != 0 || tag != r.tag {
		t.Fatal("an unread record was fenced")
	}
	if e := r.effect(t); e.State != "uncertain" {
		t.Fatalf("effect = %+v, want uncertain", e)
	}
	r.control.failGet = false
	if got, err = ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
		t.Fatalf("resolution after read recovery: %q, %v", got, err)
	}
}

// Only the handle whose effect issued the Puts resolves it, for its own
// move. Knowing the registration's id and owner tuple is not enough.
func TestPlan9B8RefusesAnotherHandleOrMove(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	otherApp := r.moving
	claim := *otherApp.Claim
	claim.Op.Intent = maps.Clone(claim.Op.Intent)
	claim.Op.Intent["app_id"] = "other"
	otherApp.Claim = &claim
	forged := &StartHandle{w: r.h.w, account: "bab", app: "app", appID: "canonical", id: moveStartID, owner: r.h.Owner(), epoch: r.h.Epoch()}
	for _, tc := range []struct {
		name string
		h    *StartHandle
		o    hostclaim.Observed
		id   string
	}{
		{"another move id", r.h, r.moving, "mv_" + strings.Repeat("b", 26)},
		{"same owner, no issue", forged, r.moving, moveStartID},
		{"nil handle", nil, r.moving, moveStartID},
		{"serving observation", r.h, r.serving, moveStartID},
		{"another app id", r.h, otherApp, moveStartID},
		{"absent observation", r.h, hostclaim.Observed{Host: r.host, Absent: true}, moveStartID},
		{"empty move id", r.h, r.moving, ""},
		{"another registration", &StartHandle{w: r.h.w}, r.moving, moveStartID},
		{"handle without writer", &StartHandle{id: moveStartID}, r.moving, moveStartID},
	} {
		if got, err := ResolveMoveStart(ctx, tc.h, tc.o, tc.id); got != OutcomeUncertain || !errors.Is(err, ErrIdentity) {
			t.Errorf("%s: %q, %v; want uncertain ErrIdentity", tc.name, got, err)
		}
	}
	if _, tag := r.record(t); r.control.fences != 0 || tag != r.tag {
		t.Fatal("a refused resolution wrote the record")
	}
	if e := r.effect(t); e.State != "uncertain" {
		t.Fatalf("a refused resolution settled the effect: %+v", e)
	}
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
		t.Fatalf("issuing handle resolution: %q, %v", got, err)
	}
}

// The runner's settle hook is scoped to an uncertain move-start effect of
// its own registration: another action and an exclusive runner refuse, and
// an effect still issued (its call not joined) is never resolved.
func TestPlan9B8SettleHookIsScopedToAnUncertainMoveStart(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	runner := r.h.Runner(r.h.w.now)
	e := r.effect(t)
	other := e
	other.Action = "move-relocating-decision"
	if err := runner.resolveUncertain(ctx, other, "not-applied"); !errors.Is(err, ErrContract) {
		t.Fatalf("another action: %v", err)
	}
	exclusive := runner
	exclusive.exclusive = true
	if err := exclusive.resolveUncertain(ctx, e, "not-applied"); !errors.Is(err, ErrContract) {
		t.Fatalf("an exclusive runner: %v", err)
	}
	issued := Effect{ID: "issued-effect", Resource: "apps/bab/app/other.json", Action: moveStartAction, AppID: "canonical", OperationID: moveStartID, Owner: r.h.Owner()}
	_, settle, err := runner.Issue(ctx, issued)
	if err != nil {
		t.Fatal(err)
	}
	if err := runner.resolveUncertain(ctx, issued, "not-applied"); !errors.Is(err, ErrPending) {
		t.Fatalf("resolved an issued effect: %v", err)
	}
	_ = settle("", errors.Join(ErrNotApplied, errors.New("never called")))
	if got := r.effect(t); got.State != "uncertain" {
		t.Fatalf("the refusals moved the move-start effect: %+v", got)
	}
}

// The moves door's acceptance scenario (moves-phase1-2 26d9e4d1,
// TestMoveDoorHostClaimRetryAfterAFailedRecordStart) in hostops: after a
// not-applied resolution the door releases the claim under the
// registration's grant, the registration drains, and a retry with a new
// move id starts the move.
func TestPlan9B8AcceptanceNotAppliedDrainsAndARetryStarts(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
		t.Fatalf("ResolveMoveStart = %q, %v", got, err)
	}
	releaseMove := func(c *hostclaim.Claim) error {
		if err := hostclaim.RequireMoveOp(moveStartID, hostclaim.KindMoving, hostclaim.PhaseMoving)(c); err != nil {
			return err
		}
		return hostclaim.ReleaseMove(moveStartID, "ie-box")(c)
	}
	if _, err := hostclaim.Update(r.h.Lifecycle(ctx), r.data, r.moving, releaseMove); err != nil {
		t.Fatalf("the door's claim release: %v", err)
	}
	if err := r.w.SettleRegistration(ctx, "bab", "app", "canonical", moveStartID, r.now.Add(RegistrationTTL+2*RegistrationSkew)); err != nil {
		t.Fatalf("the failed attempt's registration does not drain: %v", err)
	}
	a, err := ReadAdmission(ctx, r.raw.Store, "bab", "app", "canonical")
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := a.Value.Registrations[moveStartID]; ok {
		t.Fatal("the failed attempt's registration is still held")
	}
	retry := "mv_" + strings.Repeat("b", 26)
	o := r.read(t)
	h, err := r.authority.RegisterMove(ctx, o, "canonical", retry)
	if err != nil {
		t.Fatal(err)
	}
	moving, err := BeginClaimMove(ctx, r.data, r.authority.Directory, h, o, retry, "operator", "uk-box", "uk", r.now)
	if err != nil {
		t.Fatalf("retry start CAS: %v", err)
	}
	start := r.start
	start.ID = retry
	rec, err := StartClaimMoveRecord(ctx, r.data, h, moving, start)
	if err != nil || rec.Move == nil || rec.Move.ID != retry || rec.Move.State != provision.MovePreparing {
		t.Fatalf("retry record start = %+v, %v", rec, err)
	}
	if a, err = ReadAdmission(ctx, r.raw.Store, "bab", "app", "canonical"); err != nil {
		t.Fatal(err)
	}
	if e := a.Value.Effects[retry]; e.State != "complete" || e.ResultReference != retry {
		t.Fatalf("retry effect = %+v", e)
	}
	if e := a.Value.Effects[moveStartID]; e.State != "complete" || e.ResultReference != "not-applied" {
		t.Fatalf("failed attempt's effect = %+v", e)
	}
}

// Why the fence rewrites the value with different bytes: the DirStore's
// ETag, like a single-part S3 ETag, is a content hash, so a byte-identical
// rewrite keeps the ETag a late Put matched on.
func TestPlan9B8IdenticalBytesKeepAContentETag(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	key := provision.RecordKey("bab", "app", "shop.example.com")
	tag, err := s.Put(ctx, key, []byte(`{"schema":1}`), "")
	if err != nil {
		t.Fatal(err)
	}
	same, err := s.Put(ctx, key, []byte(`{"schema":1}`), tag)
	if err != nil || same != tag {
		t.Fatalf("identical rewrite: %q, %v (was %q)", same, err, tag)
	}
	moved, err := s.Put(ctx, key, []byte("{\"schema\":1}\n"), same)
	if err != nil || moved == same {
		t.Fatalf("newline rewrite: %q, %v", moved, err)
	}
}

// A changed ETag without a sequence increment is not proof. The resolver
// must perform its own sequence CAS before acknowledging not-applied.
func TestPlan9B8RefusesUnprovenContentFence(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, append(r.before, '\n'), r.tag); err != nil {
		t.Fatal(err)
	}
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied || r.control.fences != 1 {
		t.Fatalf("content-only change was accepted without a sequence fence: %q %v fences=%d", got, err, r.control.fences)
	}
	r.ordinaryRewrite(t)
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("late Put: %v", err)
	}
}

// ordinaryRewrite releases the claim with the existing grant, then exercises
// the real ordinary record writer and claim-on conditional store. Its
// attempted semantic value is exactly the original record's value.
func (r *resolveFixture) ordinaryRewrite(t *testing.T) {
	t.Helper()
	ctx := context.Background()
	if _, err := hostclaim.Update(r.h.Lifecycle(ctx), r.data, r.moving, hostclaim.ReleaseMove(moveStartID, "ie-box")); err != nil {
		t.Fatal(err)
	}
	records, err := NewConditionalRecordStore(r.raw, r.pub)
	if err != nil {
		t.Fatal(err)
	}
	if err = provision.SetRecordRegion(ctx, records, "bab", "app", r.host, "ie"); err != nil {
		t.Fatal(err)
	}
}

func TestPlan9B8AcceptanceOrdinaryRewriteCannotReopenFenceABA(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
		t.Fatalf("resolve: %q %v", got, err)
	}
	r.ordinaryRewrite(t)
	body, tag := r.record(t)
	fields, seq, err := recordSequence(body)
	original, _, _ := recordSequence(r.before)
	delete(fields, "record_seq")
	delete(original, "record_seq")
	if err != nil || seq != 2 || tag == r.tag || !maps.EqualFunc(fields, original, func(a, b json.RawMessage) bool { return bytes.Equal(a, b) }) {
		t.Fatalf("ordinary rewrite restored conditional identity: seq=%d tag=%q body=%s err=%v", seq, tag, body, err)
	}
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("delayed Put survived fence and ordinary rewrite: %v", err)
	}
}

func TestPlan9B8AcceptanceBoxRewriteFencesWithoutClaimContext(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	// First fence introduces the sequence; the next box write must advance it
	// even without WithRecordSequence or a console conditional-record wrapper.
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
		t.Fatalf("resolve: %q %v", got, err)
	}
	if _, err := provision.SetRecordMove(ctx, r.raw.Store, "bab", "app", r.host, func(*provision.Record) error { return nil }); err != nil {
		t.Fatal(err)
	}
	body, _ := r.record(t)
	_, seq, err := recordSequence(body)
	if err != nil || seq != 2 {
		t.Fatalf("box seq=%d %v", seq, err)
	}
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("box reopened fence: %v", err)
	}
}

func TestPlan9B8AcceptanceBoxAdvanceProvesNotApplied(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	if _, err := provision.SetRecordMove(provision.WithRecordSequence(ctx), r.raw.Store, "bab", "app", r.host, func(*provision.Record) error { return nil }); err != nil {
		t.Fatal(err)
	}
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied || r.control.fences != 0 {
		t.Fatalf("box proof: %q %v fences=%d", got, err, r.control.fences)
	}
	if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("late Put: %v", err)
	}
}