package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"errors"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/hostrelocate"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// finishRecorder is a Finish that records each host it is handed and
// answers err.
type finishRecorder struct {
	calls []string
	err   error
}

func (f *finishRecorder) finish(_ context.Context, host, op string) error {
	f.calls = append(f.calls, host+":"+op)
	return f.err
}

// Carry 8: the completion pass hands Finish only a claim serving on this
// box whose completion names this box. The completion-only runtime has no
// Box; it asks Self each pass, and a pass that cannot answer finishes
// nothing.
func TestCompletionPassFinishesOnlyWhatServesOnThisBox(t *testing.T) {
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "mine.example", hostclaim.Op{})
	completeOn(t, store, "mine.example", "rel-m", "dst", false)
	// Serving on spa, its completion naming spa: spa's to finish.
	passClaim(t, store, "theirs.example", hostclaim.Op{})
	completeOn(t, store, "theirs.example", "rel-t", "spa", false)
	// Serving on spa while its record names dst: not serving here, so not
	// this box's to finish, whatever the record says.
	passClaim(t, store, "moved.example", hostclaim.Op{})
	o := completeOn(t, store, "moved.example", "rel-x", "dst", false)
	if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error { c.Op.Exec = "spa"; return nil }); err != nil {
		t.Fatal(err)
	}
	// Serving on dst while its record names spa: the completion is spa's.
	passClaim(t, store, "foreign.example", hostclaim.Op{})
	o = completeOn(t, store, "foreign.example", "rel-f", "spa", false)
	if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error { c.Op.Exec = "dst"; return nil }); err != nil {
		t.Fatal(err)
	}
	f := &finishRecorder{}
	hosts := func() ([]string, error) {
		return []string{"mine.example", "theirs.example", "moved.example", "foreign.example"}, nil
	}
	r := &RelocationRuntime{Finish: f.finish, Hosts: hosts, Self: func(context.Context) (string, error) { return "dst", nil }}
	r.once(ctx, store)
	if !slices.Equal(f.calls, []string{"mine.example:rel-m"}) {
		t.Fatalf("Finish was handed %v, want only mine.example", f.calls)
	}
	// No answer to "which box is this": nothing is finished.
	f.calls = nil
	buf := captureRacyLog(t)
	r = &RelocationRuntime{Finish: f.finish, Hosts: hosts, Self: func(context.Context) (string, error) { return "", errors.New("no verified identity") }}
	r.once(ctx, store)
	if len(f.calls) != 0 || !strings.Contains(buf.String(), "no verified identity") {
		t.Fatalf("finished %v without knowing this box:\n%s", f.calls, buf.String())
	}
}

// Carry 8 and review M4: ErrChanged from the hold release, after the
// pass's own serving-here check passed, is never taken as finished: it is
// said once (a completion stuck that way keeps the barrier closed) and
// asked again next pass. Every other refusal is alerted once and retried
// too (carry 6: a probe-fence owner Finish cannot find is ErrRefused).
func TestCompletionPassRetriesARefusedFinishAndAChangedClaim(t *testing.T) {
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "a.example", hostclaim.Op{})
	completeOn(t, store, "a.example", "rel-a", "dst", false)
	buf := captureRacyLog(t)
	f := &finishRecorder{err: hostbox.ErrChanged}
	r := &RelocationRuntime{Finish: f.finish, Box: "dst"}
	r.once(ctx, store)
	r.once(ctx, store)
	if len(f.calls) != 2 || strings.Count(buf.String(), "relocate a.example: finish: "+hostbox.ErrChanged.Error()) != 1 {
		t.Fatalf("a changed claim: calls %v, log:\n%s", f.calls, buf.String())
	}
	f.calls, f.err = nil, errors.Join(hostrelocate.ErrRefused, errors.New("a.example has no probe-fence owner"))
	r.once(ctx, store)
	r.once(ctx, store)
	if len(f.calls) != 2 || strings.Count(buf.String(), "no probe-fence owner") != 1 {
		t.Fatalf("a refused Finish: calls %v (want a retry each pass), log:\n%s", f.calls, buf.String())
	}
	if obs, err := hostclaim.Read(ctx, store, "a.example"); err != nil {
		t.Fatal(err)
	} else if done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim); err != nil || !ok || done.Discharged {
		t.Fatalf("the record was marked over a refusal: %+v %v %v", done, ok, err)
	}
}

// Carry 1: a completion record that cannot be read is alerted and never
// read as "no completion": Finish is not called over it, and the next pass
// reads it again.
func TestCompletionPassAlertsAnUnreadableRecord(t *testing.T) {
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "a.example", hostclaim.Op{})
	o := completeOn(t, store, "a.example", "rel-a", "dst", false)
	if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error {
		c.Op.Journal[hostclaim.JournalCompleted] = "not a record"
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	buf := captureRacyLog(t)
	f := &finishRecorder{}
	r := &RelocationRuntime{Finish: f.finish, Box: "dst"}
	r.once(ctx, store)
	if len(f.calls) != 0 || !strings.Contains(buf.String(), "relocate a.example: finish:") {
		t.Fatalf("an unreadable record: calls %v, log:\n%s", f.calls, buf.String())
	}
}

// Carry 3: the sweep never runs beside a relocation journal write. A
// write in flight (its directory made, its temporary not yet renamed)
// holds relocationJournals shared; the pass's sweep waits for it, then
// keeps the journal it committed. Without the lock the sweep removes the
// directory as a write's leftover under the write.
func TestRelocatePassSweepWaitsForAJournalWrite(t *testing.T) {
	t.Parallel()
	ctx := t.Context()
	store := release.NewDirStore(t.TempDir())
	passClaim(t, store, "a.example", relocatingOp("rel-a"))
	data := t.TempDir()
	dir := filepath.Join(data, "relocate", "ops", "rel-a")
	relocationJournals.RLock()
	locked := true
	defer func() {
		if locked {
			relocationJournals.RUnlock()
		}
	}()
	if err := os.MkdirAll(dir, 0o700); err != nil {
		t.Fatal(err)
	}
	temp := filepath.Join(dir, ".attempts.json.123")
	if err := os.WriteFile(temp, []byte("{"), 0o600); err != nil {
		t.Fatal(err)
	}
	r := &RelocationRuntime{Box: "dst", DataDir: data}
	done := make(chan struct{})
	go func() { r.once(ctx, store); close(done) }()
	select {
	case <-done:
		t.Fatal("the sweep ran beside a journal write")
	case <-time.After(300 * time.Millisecond):
	}
	// The write lands: its journal committed, its temporary gone.
	if err := os.WriteFile(filepath.Join(dir, "attempts.json"), []byte(`{"v":1,"host":"a.example","op":"rel-a"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.Remove(temp); err != nil {
		t.Fatal(err)
	}
	relocationJournals.RUnlock()
	locked = false
	<-done
	if _, err := os.Stat(filepath.Join(dir, "attempts.json")); err != nil {
		t.Fatalf("the running op's journal was swept: %v", err)
	}
}

// Carry 3, the writers' half: every relocation journal write this process
// makes outside the pass's own steps holds relocationJournals shared, so
// it waits while a sweep holds it.
func TestRelocationJournalWritesWaitForTheSweep(t *testing.T) {
	t.Parallel()
	data := t.TempDir()
	fence, err := relocationProbeFencePath(data, "rel-a")
	if err != nil {
		t.Fatal(err)
	}
	positions := ReleasePositionsFile{Dir: filepath.Join(data, "relocate", "release")}
	for name, write := range map[string]func() error{
		"attempts": func() error {
			return relocationAttemptJournal{V: 1, Host: "a.example", Op: "rel-a", Opened: map[string]time.Time{}, Sessions: map[string]int{}, Starts: map[string]relocationStart{}}.write(data)
		},
		"probe fence": func() error { _, err := relocationProbeFenceOwner(fence, "a.example", 7); return err },
		"positions": func() error {
			return positions.Record(t.Context(), ReleasePosition{Request: StopRequest{Host: "a.example", OperationID: "rel-a"}, PID: 1, ProcessStart: "1"})
		},
	} {
		t.Run(name, func(t *testing.T) {
			relocationJournals.Lock()
			done := make(chan error, 1)
			go func() { done <- write() }()
			select {
			case err := <-done:
				relocationJournals.Unlock()
				t.Fatalf("wrote beside the sweep: %v", err)
			case <-time.After(300 * time.Millisecond):
			}
			relocationJournals.Unlock()
			if err := <-done; err != nil {
				t.Fatal(err)
			}
		})
	}
}

// Carry 4: the full runtime's config carries the agent's held-stage
// evidence: the release key it was given and the edge's pointer digest
// ratchet (pointerRatchet(false, ...), never root's), with the box's
// secrets key when it has one. No release key, or a secrets key that does
// not parse, refuses.
func TestRelocationRuntimeConfigBindsTheAgentsHeldStageEvidence(t *testing.T) {
	pub, _, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	a := &ClaimAssembly{}
	a.Guard.DataDir = t.TempDir()
	opt := t.TempDir()
	env := map[string]string{}
	getenv := func(k string) string { return env[k] }
	s := relocationStores{Files: release.NewDirStore(t.TempDir()), Versions: leasetest.NewVersionedBucket(), Bucket: "data-ie", Region: "ie", LeaseRegion: "eu-west-1"}
	discharge := func(context.Context, hostclaim.Observed) error { return nil }
	build := func(in relocationRuntimeInputs) (RelocationAssemblyConfig, error) {
		return relocationRuntimeConfig(getenv, a, ClaimProviders{Selector: hostbox.WorkSource{Box: "dst", Region: "ie"}}, s, &registry.Registry{}, &relocationProbeActivator{}, hostactors.Directory{}, in, discharge)
	}
	cfg, err := build(relocationRuntimeInputs{Pub: pub, OptDir: opt})
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(cfg.ReleasePub, pub) || cfg.Ratchet == nil || cfg.Ratchet != pointerRatchet(false, opt, a.Guard.DataDir, getenv) || cfg.Ratchet == pointerRatchet(true, opt, a.Guard.DataDir, getenv) || cfg.SecretsKey != nil {
		t.Fatalf("held-stage evidence: pub %v ratchet %p secrets %v", cfg.ReleasePub, cfg.Ratchet, cfg.SecretsKey)
	}
	if _, err := build(relocationRuntimeInputs{OptDir: opt}); !errors.Is(err, hostwrite.ErrUnavailable) {
		t.Fatalf("no release key: %v", err)
	}
	env["CARLOS_SECRETS_KEY"] = "not a key"
	if _, err := build(relocationRuntimeInputs{Pub: pub, OptDir: opt}); err == nil {
		t.Fatal("a secrets key that does not parse was accepted")
	}
}

// With claim mode and region HA on and the relocation owners bound into
// the wake's dispatcher, NewRelocationRuntime builds the full pass: this
// box's worker in its roster region, which starts nothing (its Admission
// and Census refuse), the participant, the J3 conversion and the hold
// release. A roster entry that is not this box's enrollment, no release
// key, or no data bucket refuses rather than build a pass over a guess.
func TestRelocationRuntimeBuildsTheFullPassWithTheOwnersBound(t *testing.T) {
	ctx := t.Context()
	w := newLandingWorld(t, true, nil)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	data := t.TempDir()
	a := &ClaimAssembly{Claims: w.store, Guard: hostguard.Guard{DataDir: data}, RetirementAuthority: &hostops.ProductionAuthority{Store: w.store, Directory: w.directory},
		Wake: &ClaimWakeAssembly{Inputs: ClaimOwnerDispatch{Relocation: &ClaimKindOwners{}}}}
	local := &productionClaimLocal{legacyParticipation: true, assembly: a}
	env := map[string]string{"CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1", "CARLOS_HIBERNATE_BUCKET": "data-ie"}
	getenv := func(k string) string { return env[k] }
	pub, _, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	in := relocationRuntimeInputs{Enrollment: hostfloor.Enrollment{Box: "wrk", Enrollment: w.enrollment["wrk"]}, Pub: pub, OptDir: t.TempDir(),
		Files: func(context.Context, string) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil },
		Versioned: func(context.Context, string, string) (lease.VersionedStore, error) {
			return leasetest.NewVersionedBucket(), nil
		}}
	providers := ClaimProviders{Selector: hostbox.WorkSource{Box: "wrk", Region: "ie"}}
	build := func(in relocationRuntimeInputs) (*RelocationRuntime, error) {
		return NewRelocationRuntime(ctx, getenv, w.store, local, providers, reg, &relocationProbeActivator{}, in)
	}
	r, err := build(in)
	if err != nil {
		t.Fatal(err)
	}
	worker, ok := r.Worker.(hostrelocate.Worker)
	if !ok || r.Participant == nil || r.Convert == nil || r.Finish == nil || r.Box != "wrk" || r.DataDir != data || r.Hosts != nil {
		t.Fatalf("not the full pass: %+v", r)
	}
	// Review M7: the held-stage evidence reaches the full pass's config.
	if !bytes.Equal(r.config.ReleasePub, pub) || r.config.Ratchet == nil || r.config.Ratchet != pointerRatchet(false, in.OptDir, data, getenv) {
		t.Fatalf("the full pass's held-stage evidence: pub %v ratchet %p", r.config.ReleasePub, r.config.Ratchet)
	}
	if worker.Self.Box != "wrk" || worker.Self.Enrollment != w.enrollment["wrk"] || worker.Region != "ie" {
		t.Fatalf("the worker is not this box in its roster region: %+v %q", worker.Self, worker.Region)
	}
	if _, err := worker.Admission.Register(ctx, hostclaim.Observed{}, hostrelocate.Owner{}, ""); !errors.Is(err, errRelocationNotStarted) {
		t.Fatalf("the pass's admission registered a start: %v", err)
	}
	if err := worker.Census(ctx, "wrk", nil); !errors.Is(err, errRelocationNotStarted) {
		t.Fatalf("the pass's census passed a start: %v", err)
	}
	for name, change := range map[string]func(*relocationRuntimeInputs){
		"another enrollment": func(in *relocationRuntimeInputs) { in.Enrollment.Enrollment = strings.Repeat("f", 32) },
		"not in the roster":  func(in *relocationRuntimeInputs) { in.Enrollment.Box = "gone" },
		"no release key":     func(in *relocationRuntimeInputs) { in.Pub = nil },
	} {
		bad := in
		change(&bad)
		if r, err := build(bad); err == nil || r != nil {
			t.Fatalf("%s: built %+v", name, r)
		}
	}
	// Region HA off: the completion-only runtime, though the owners are
	// bound (ruling 82; the dormant constraint).
	env["CARLOS_REGION_HA"] = ""
	if r, err := build(in); err != nil || r == nil || r.Worker != nil || r.Participant != nil || r.Convert != nil || r.Finish == nil {
		t.Fatalf("region HA off: %+v %v", r, err)
	}
	env["CARLOS_REGION_HA"] = "1"
	delete(env, "CARLOS_HIBERNATE_BUCKET")
	if r, err := build(in); !errors.Is(err, hostwrite.ErrUnavailable) || r != nil {
		t.Fatalf("no data bucket: %+v %v", r, err)
	}
}