package hostops

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"

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

func TestPlan9B8SequenceEveryRecordWriter(t *testing.T) {
	ctx := context.Background()
	f := newMoveStartFixture(t, nil)
	records, err := NewConditionalRecordStore(f.raw, f.pub)
	if err != nil {
		t.Fatal(err)
	}
	key := f.raw.recordKey
	// Ordinary typed writer: old absent sequence becomes one and then two.
	for want := uint64(1); want <= 2; want++ {
		if err := provision.SetRecordRegion(ctx, records, "bab", "app", f.host, "ie"); err != nil {
			t.Fatal(err)
		}
		body, _, _ := records.Get(ctx, key)
		rec, err := provision.DecodeRecord(body)
		if err != nil || rec.RecordSeq != want {
			t.Fatalf("ordinary: %+v %v", rec, err)
		}
	}
	// Both effect branches also sequence their writes, including PutFrom.
	for _, resource := range []string{key, "control-object:" + key} {
		body, tag, _ := records.Get(ctx, key)
		_, seq, _ := recordSequence(body)
		permitted := context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: resource})
		if _, err := release.PutObjectFrom(permitted, records, key, bytes.NewReader(body), int64(len(body)), tag); err != nil {
			t.Fatal(err)
		}
		next, _, _ := records.Get(ctx, key)
		_, got, _ := recordSequence(next)
		if got != seq+1 {
			t.Fatalf("effect %s: seq %d -> %d", resource, seq, got)
		}
	}
	// Lifecycle updates keep the same counter after the first journal Put.
	r := newResolveFixture(t, true)
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeComplete {
		t.Fatalf("resolve: %q %v", got, err)
	}
	lifecycle, err := NewConditionalRecordStore(r.raw, r.pub)
	if err != nil {
		t.Fatal(err)
	}
	if rec, err := setLifecycleRecordMove(ctx, lifecycle, r.moving, func(rec *provision.Record) error { return nil }); err != nil || rec.RecordSeq != 2 {
		t.Fatalf("returned lifecycle record: %+v %v", rec, err)
	}
	body, _ := r.record(t)
	_, seq, err := recordSequence(body)
	if err != nil || seq != 2 {
		t.Fatalf("lifecycle seq=%d err=%v", seq, err)
	}
	// Sidecar create and typed decode/rewrite preserve the counter too.
	side := provision.SidecarRecord{Schema: 1, Account: "bab", App: "app", Name: "worker", Host: provision.SidecarHost("worker", "app", "bab"), CreatedAt: f.now}
	key = provision.SidecarRecordKey("bab", "app", "worker")
	permitted := context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: key})
	body, _ = json.Marshal(side)
	tag, err := records.Put(permitted, key, body, "")
	if err != nil {
		t.Fatal(err)
	}
	body, _, _ = records.Get(ctx, key)
	if err := json.Unmarshal(body, &side); err != nil || side.RecordSeq != 1 {
		t.Fatalf("sidecar: %+v %v", side, err)
	}
	body, _ = json.Marshal(side)
	if _, err := records.Put(permitted, key, body, tag); err != nil {
		t.Fatal(err)
	}
	body, _, _ = records.Get(ctx, key)
	_, seq, err = recordSequence(body)
	if err != nil || seq != 2 {
		t.Fatalf("sidecar seq=%d %v", seq, err)
	}
}

func TestPlan9B8SequenceCannotBeRemovedLoweredOrChosen(t *testing.T) {
	for _, mode := range []string{"ordinary", "effect", "control-object", "stream"} {
		for _, counter := range []string{"", `,"record_seq":0`, `,"record_seq":1`, `,"record_seq":3`, `,"record_seq":null`, `,"record_seq":-1`, `,"record_seq":1.5`, `,"record_seq":18446744073709551616`, `,"Record_Seq":2`, `,"record_seq":2,"record_seq":2`} {
			t.Run(mode+counter, func(t *testing.T) {
				f := newMoveStartFixture(t, nil)
				ctx := context.Background()
				body, tag, _ := f.raw.Get(ctx, f.raw.recordKey)
				var fields map[string]json.RawMessage
				_ = json.Unmarshal(body, &fields)
				fields["record_seq"] = json.RawMessage(`2`)
				before, _ := json.Marshal(fields)
				tag, err := f.raw.Put(ctx, f.raw.recordKey, before, tag)
				if err != nil {
					t.Fatal(err)
				}
				delete(fields, "record_seq")
				body, _ = json.Marshal(fields)
				candidate := append(append([]byte(nil), body[:len(body)-1]...), []byte(counter+"}")...)
				records, err := NewConditionalRecordStore(f.raw, f.pub)
				if err != nil {
					t.Fatal(err)
				}
				switch mode {
				case "effect", "stream":
					ctx = context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: f.raw.recordKey})
				case "control-object":
					ctx = context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: "control-object:" + f.raw.recordKey})
				}
				if mode == "stream" {
					_, err = release.PutObjectFrom(ctx, records, f.raw.recordKey, bytes.NewReader(candidate), int64(len(candidate)), tag)
				} else {
					_, err = records.Put(ctx, f.raw.recordKey, candidate, tag)
				}
				if !errors.Is(err, ErrIdentity) {
					t.Fatalf("counter accepted: %s err=%v", candidate, err)
				}
				after, afterTag, _ := f.raw.Get(ctx, f.raw.recordKey)
				if !bytes.Equal(before, after) || afterTag != tag {
					t.Fatal("refusal changed record")
				}
			})
		}
	}
}

func TestPlan9B8SequencePrecisionOverflowAndUnknownFields(t *testing.T) {
	const high = uint64(1<<53) + 1
	before := []byte(fmt.Sprintf(`{"record_seq":%d,"future":{"value":9007199254740993}}`, high))
	body, err := advanceRecordSequence(before, before)
	fields, seq, parseErr := recordSequence(body)
	if err != nil || parseErr != nil || seq != high+1 || string(fields["future"]) != `{"value":9007199254740993}` {
		t.Fatalf("precision: %s %v %v", body, err, parseErr)
	}
	max := []byte(`{"record_seq":18446744073709551615}`)
	if _, err := advanceRecordSequence(max, max); !errors.Is(err, ErrPending) {
		t.Fatalf("overflow: %v", err)
	}
}

func TestPlan9B8FlagOffBytesAndNonRecordBytesUnchanged(t *testing.T) {
	f := newMoveStartFixture(t, nil)
	ctx := context.Background()
	before, tag, _ := f.raw.Get(ctx, f.raw.recordKey)
	off, err := NewMutationBoundary(false, f.raw, nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := provision.SetRecordRegion(ctx, off, "bab", "app", f.host, "ie"); err != nil {
		t.Fatal(err)
	}
	after, got, _ := f.raw.Get(ctx, f.raw.recordKey)
	if !bytes.Equal(before, after) || got != tag || bytes.Contains(after, []byte(`"record_seq"`)) {
		t.Fatalf("flag-off changed: %s", after)
	}
	records, err := NewConditionalRecordStore(f.raw, f.pub)
	if err != nil {
		t.Fatal(err)
	}
	key := "apps/bab/app/metadata.json"
	body := []byte("{\"value\": 1}\n")
	if _, err := records.Put(ctx, key, body, ""); err != nil {
		t.Fatal(err)
	}
	after, _, _ = records.Get(ctx, key)
	if !bytes.Equal(body, after) {
		t.Fatal("non-record bytes changed")
	}
	side, _ := json.Marshal(provision.SidecarRecord{Name: "worker"})
	if strings.Contains(string(side), "record_seq") {
		t.Fatalf("legacy sidecar: %s", side)
	}
}

func TestPlan9B8ResolverRefusesMissingReplacedOrMalformedRecord(t *testing.T) {
	for _, mode := range []string{"missing", "incarnation", "null-sequence", "lower-sequence", "uncaptured", "exhausted"} {
		t.Run(mode, func(t *testing.T) {
			ctx := context.Background()
			r := newResolveFixture(t, false)
			body, tag := r.record(t)
			fields, _, _ := recordSequence(body)
			switch mode {
			case "missing":
				if err := r.raw.Store.Delete(ctx, r.raw.recordKey); err != nil {
					t.Fatal(err)
				}
			case "incarnation":
				fields["created_at"] = json.RawMessage(`"2026-10-08T00:00:00Z"`)
			case "null-sequence":
				fields["record_seq"] = json.RawMessage(`null`)
			case "lower-sequence":
				r.h.moveStart.puts.Load().bases = []uint64{1}
			case "uncaptured":
				r.h.moveStart.puts.Load().bases = nil
			case "exhausted":
				fields["record_seq"] = json.RawMessage(`18446744073709551615`)
				r.h.moveStart.puts.Load().bases = []uint64{^uint64(0)}
			}
			if mode == "incarnation" || mode == "null-sequence" || mode == "exhausted" {
				next, _ := json.Marshal(fields)
				if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, next, tag); err != nil {
					t.Fatal(err)
				}
			}
			if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain {
				t.Fatalf("refusal: %q %v", got, err)
			}
			if r.effect(t).State != "uncertain" || r.control.fences != 0 {
				t.Fatal("unproven resolution wrote or settled")
			}
		})
	}
}

type sequenceFenceFailure struct {
	release.Store
	key       string
	land      bool
	conflicts bool
	calls     int
}

func (s *sequenceFenceFailure) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
	if key != s.key {
		return s.Store.Put(ctx, key, body, tag)
	}
	s.calls++
	if s.conflicts {
		return "", release.ErrCASConflict
	}
	if s.land {
		if _, err := s.Store.Put(ctx, key, body, tag); err != nil {
			return "", err
		}
	}
	return "", errors.New("fence answer lost")
}
func TestPlan9B8FenceFailureRemainsUncertainAndCanRecover(t *testing.T) {
	for _, land := range []bool{false, true} {
		t.Run(fmt.Sprint(land), func(t *testing.T) {
			ctx := context.Background()
			r := newResolveFixture(t, false)
			failure := &sequenceFenceFailure{Store: r.control, key: r.raw.recordKey, land: land}
			r.h.w.raw = failure
			if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain {
				t.Fatalf("lost fence: %q %v", got, err)
			}
			if r.effect(t).State != "uncertain" {
				t.Fatal("lost fence settled")
			}
			r.h.w.raw = r.control
			if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
				t.Fatalf("retry: %q %v", got, err)
			}
			if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
				t.Fatalf("late Put: %v", err)
			}
		})
	}
	r := newResolveFixture(t, false)
	failure := &sequenceFenceFailure{Store: r.control, key: r.raw.recordKey, conflicts: true}
	r.h.w.raw = failure
	if got, err := ResolveMoveStart(context.Background(), r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain || failure.calls != 3 {
		t.Fatalf("bounded conflicts: %q %v calls=%d", got, err, failure.calls)
	}
	if r.effect(t).State != "uncertain" {
		t.Fatal("conflicts settled")
	}
}

func TestPlan9B8CapturesEveryCASAttemptBase(t *testing.T) {
	ctx := context.Background()
	r := newResolveFixture(t, false)
	p := r.h.moveStart.puts.Load()
	// Model updateRecord's retry reading another revision before its own Put.
	body, tag := r.record(t)
	next, err := advanceRecordSequence(body, body)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, next, tag); err != nil {
		t.Fatal(err)
	}
	failed := &lostRecordPut{Store: r.raw, recordKey: r.raw.recordKey, armed: true}
	recorder := moveStartRecorder{s: failed, puts: p}
	current, etag, err := recorder.Get(ctx, r.raw.recordKey)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := recorder.Put(ctx, r.raw.recordKey, current, etag); err == nil {
		t.Fatal("lost retry succeeded")
	}
	if high, ok := p.highestBase(); !ok || high != 1 || len(p.bases) != 2 {
		t.Fatalf("bases: %v", p.bases)
	}
	if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied || r.control.fences != 1 {
		t.Fatalf("retry fence: %q %v fences=%d", got, err, r.control.fences)
	}
	if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, failed.body, failed.etag); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("late retry: %v", err)
	}
}

func TestPlan9B8TerminalJournalStillProvesComplete(t *testing.T) {
	for _, outcome := range []string{"done", "rolled_back", "aborted", "different-id"} {
		t.Run(outcome, func(t *testing.T) {
			ctx := context.Background()
			r := newResolveFixture(t, outcome != "different-id")
			id := moveStartID
			want := OutcomeComplete
			if outcome == "different-id" {
				id = "mv_" + strings.Repeat("b", 26)
				want = OutcomeNotApplied
			}
			// Complete describes the record-start effect, including a move that
			// subsequently aborted or rolled back. The journal retains its outcome.
			_, err := provision.SetRecordMove(provision.WithRecordSequence(ctx), r.raw.Store, "bab", "app", r.host, func(rec *provision.Record) error {
				rec.LastMove = &provision.MoveOutcome{ID: id, Outcome: outcome, At: r.now}
				rec.Move = nil
				return nil
			})
			if err != nil {
				t.Fatal(err)
			}
			if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != want {
				t.Fatalf("terminal evidence: %q %v want %q", got, err, want)
			}
			result := moveStartID
			if want == OutcomeNotApplied {
				result = "not-applied"
			}
			if r.control.fences != 0 || r.effect(t).ResultReference != result {
				t.Fatal("wrong terminal resolution")
			}
		})
	}
}