package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// seedInstance upserts a ready, provisioned exec instance: a relocatable
// host (spec 1 §2).
func seedInstance(t *testing.T, reg *registry.Registry, host string) {
	t.Helper()
	if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/" + host + ".sock",
		Provisioned: true, App: "jam", Account: "bab", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
}

// claimsBook answers claim reads from a map; a missing host is an absent
// claim, a host in errs fails.
type claimsBook struct {
	claims map[string]*hostclaim.Claim
	errs   map[string]bool
	reads  []string
}

func (b *claimsBook) read(_ context.Context, host string) (hostclaim.Observed, error) {
	b.reads = append(b.reads, host)
	if b.errs[host] {
		return hostclaim.Observed{}, errors.New("SlowDown")
	}
	c, ok := b.claims[host]
	if !ok {
		return hostclaim.Observed{Host: host, Absent: true}, nil
	}
	return hostclaim.Observed{Host: host, Claim: c, ETag: "e"}, nil
}

func serving(exec string, stamp uint64) *hostclaim.Claim {
	return &hostclaim.Claim{V: 2, Gen: 7, Holder: &hostclaim.Holder{Stamp: stamp},
		Op: hostclaim.Op{Kind: hostclaim.KindServing, ID: "op-1", Exec: exec}}
}

func publishWithClaims(t *testing.T, reg *registry.Registry, claims *edgesClaims) map[string]edgeHost {
	t.Helper()
	// These tests are about the hosts; the box's addresses come from a
	// fixed source, never from IMDS. nil claims is region HA off.
	if claims != nil {
		claims.addrs = staticAddresses{Public: "1.2.3.4"}
	}
	st := release.NewDirStore(t.TempDir())
	var last []byte
	if err := edgesFileWriteClaims(context.Background(), st, reg, edgesFile{Box: "annie", BoxID: "annie"}, "1.2.3.4", &last, false, claims); err != nil {
		t.Fatal(err)
	}
	body, _, err := st.Get(context.Background(), edgesFileKey("annie"))
	if err != nil {
		t.Fatal(err)
	}
	var f edgesFile
	if err := json.Unmarshal(body, &f); err != nil {
		t.Fatal(err)
	}
	return f.Hosts
}

func TestEdgesFileFollowsTheClaimsExecutor(t *testing.T) {
	reg := edgesTestRegistry(t)
	for _, h := range []string{"mine.bab.test", "theirs.bab.test", "between.bab.test", "unclaimed.bab.test"} {
		seedInstance(t, reg, h)
	}
	seedRoute(t, reg, "static.example.com") // not relocatable: never read, always home
	book := &claimsBook{claims: map[string]*hostclaim.Claim{
		"mine.bab.test":    serving("annie", 41),
		"theirs.bab.test":  serving("bert", 42),
		"between.bab.test": serving("", 43), // op.exec none: nobody serves it
	}}
	hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
	if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 || h.Role != "home" {
		t.Fatalf("executor's host = %+v, %v; want home stamped 41", h, ok)
	}
	for _, h := range []string{"theirs.bab.test", "between.bab.test"} {
		if _, ok := hosts[h]; ok {
			t.Errorf("%s published by a box its claim does not name", h)
		}
	}
	if h, ok := hosts["unclaimed.bab.test"]; !ok || h.Stamp != 0 {
		t.Errorf("a host with no claim = %+v, %v; want home, unstamped, as before", h, ok)
	}
	if h, ok := hosts["static.example.com"]; !ok || h.Stamp != 0 {
		t.Errorf("a service route = %+v, %v; want home, unstamped", h, ok)
	}
	for _, r := range book.reads {
		if r == "static.example.com" {
			t.Error("read a claim for a route that is not relocatable")
		}
	}
}

func TestEdgesFileWithholdsTheAliasOfAWithheldHost(t *testing.T) {
	reg := edgesTestRegistry(t)
	seedInstance(t, reg, "theirs.bab.test")
	if err := reg.UpsertAlias(registry.RouteAlias{Alias: "www.customer.com", Host: "theirs.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"}); err != nil {
		t.Fatal(err)
	}
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"theirs.bab.test": serving("bert", 42)}}
	hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
	if len(hosts) != 0 {
		t.Fatalf("hosts = %v, want none", hosts)
	}
}

// Review Focus 2: a claim that cannot be read never moves a host. It keeps
// the last decision, or, never read, publishes as it did before claims.
func TestEdgesFileKeepsTheLastDecisionWhenAClaimCannotBeRead(t *testing.T) {
	reg := edgesTestRegistry(t)
	seedInstance(t, reg, "mine.bab.test")
	seedInstance(t, reg, "theirs.bab.test")
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41), "theirs.bab.test": serving("bert", 42)}}
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	claims := newEdgesClaims("annie", book.read)
	claims.now = func() time.Time { return clock }
	publishWithClaims(t, reg, claims)
	clock = clock.Add(edgesClaimTTL)
	book.errs = map[string]bool{"mine.bab.test": true, "theirs.bab.test": true}
	hosts := publishWithClaims(t, reg, claims)
	if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 {
		t.Fatalf("mine = %+v, %v: an unreadable claim dropped or unstamped a served host", h, ok)
	}
	if _, ok := hosts["theirs.bab.test"]; ok {
		t.Fatal("an unreadable claim published a host this box was told not to")
	}

	fresh := newEdgesClaims("annie", book.read)
	if h, ok := publishWithClaims(t, reg, fresh)["mine.bab.test"]; !ok || h.Stamp != 0 {
		t.Fatalf("never read = %+v, %v; want home, unstamped", h, ok)
	}
}

func TestEdgesClaimsRereadOnlyAfterTheTTL(t *testing.T) {
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41)}}
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := newEdgesClaims("annie", book.read)
	c.now = func() time.Time { return clock }
	ctx := context.Background()
	c.decide(ctx, []string{"mine.bab.test"})
	clock = clock.Add(edgesClaimTTL - time.Second)
	c.decide(ctx, []string{"mine.bab.test"})
	if len(book.reads) != 1 {
		t.Fatalf("%d reads inside the TTL, want 1", len(book.reads))
	}
	book.claims["mine.bab.test"] = serving("bert", 50)
	clock = clock.Add(time.Second)
	if d := c.decide(ctx, []string{"mine.bab.test"})["mine.bab.test"]; !d.withhold {
		t.Fatalf("decision after the TTL = %+v; the move to another box was not seen", d)
	}
	c.decide(ctx, nil)
	if len(c.entries) != 0 {
		t.Fatalf("a host no longer served kept its entry: %v", c.entries)
	}
}

func TestEdgesClaimsBoundReadsPerPassNeverReadFirst(t *testing.T) {
	book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := newEdgesClaims("annie", book.read)
	c.now = func() time.Time { return clock }
	var hosts []string
	for i := 0; i < edgesClaimReadsPerPass+10; i++ {
		hosts = append(hosts, fmt.Sprintf("h%04d.bab.test", i))
	}
	ctx := context.Background()
	c.decide(ctx, hosts)
	if len(book.reads) != edgesClaimReadsPerPass {
		t.Fatalf("first pass read %d claims, want %d", len(book.reads), edgesClaimReadsPerPass)
	}
	book.reads = nil
	c.decide(ctx, hosts)
	if len(book.reads) != 10 {
		t.Fatalf("second pass read %d claims, want the 10 never read", len(book.reads))
	}
}

// Astra plan R1-2: an alias and a served catch-all are dialled through
// their primary's installation, so they carry its stamp. Unstamped, a dead
// predecessor's alias would tie with the successor's and the edge would
// keep the first box by name.
func TestEdgesFileStampsAliasesAndCatchAllsWithTheirPrimary(t *testing.T) {
	reg := edgesTestRegistry(t)
	seedInstance(t, reg, "mine.bab.test")
	for _, a := range []registry.RouteAlias{
		{Alias: "www.customer.com", Host: "mine.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"},
		{Alias: "*.customer.com", Host: "mine.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"},
	} {
		if err := reg.UpsertAlias(a); err != nil {
			t.Fatal(err)
		}
	}
	if err := reg.UpsertWildcardDomain(registry.WildcardDomain{Host: "customer.com", Account: "bab", App: "jam"}); err != nil {
		t.Fatal(err)
	}
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 57)}}
	hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
	for _, key := range []string{"mine.bab.test", "www.customer.com", "*.customer.com"} {
		if h, ok := hosts[key]; !ok || h.Stamp != 57 || h.Role != "home" {
			t.Errorf("%s = %+v, %v; want home at stamp 57", key, h, ok)
		}
	}
	// With no claim reader the file is what it always was: no stamps.
	for key, h := range publishWithClaims(t, reg, nil) {
		if h.Stamp != 0 {
			t.Errorf("%s stamped %d with region HA off", key, h.Stamp)
		}
	}
}

// Astra plan R1-4: claims that keep failing go to the back of the queue,
// so they cannot hold every read of every pass.
func TestEdgesClaimsAFailingClaimDoesNotStarveTheRest(t *testing.T) {
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"zz-healthy.bab.test": serving("annie", 7)}, errs: map[string]bool{}}
	var hosts []string
	for i := 0; i < edgesClaimReadsPerPass; i++ {
		h := fmt.Sprintf("a%04d.bab.test", i)
		book.errs[h] = true
		hosts = append(hosts, h)
	}
	hosts = append(hosts, "zz-healthy.bab.test")
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := newEdgesClaims("annie", book.read)
	c.now = func() time.Time { return clock }
	ctx := context.Background()
	c.decide(ctx, hosts) // the 256 failing claims sort first and all fail
	clock = clock.Add(edgesClaimPassEvery)
	if d, ok := c.decide(ctx, hosts)["zz-healthy.bab.test"]; !ok || d.stamp != 7 {
		t.Fatalf("healthy claim = %+v, %v: starved by claims that cannot be read", d, ok)
	}
}

// Astra plan R1-5: the loop's own schedule, with a stable registry, keeps
// every claim of a 2,000-host box fresh. Timer passes alone come every
// five to six minutes and read 256 claims each.
func TestEdgesLoopRefreshesEveryClaimWithinTheTTL(t *testing.T) {
	t.Parallel()
	const n = 2000
	book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
	var hosts []string
	for i := 0; i < n; i++ {
		h := fmt.Sprintf("h%04d.bab.test", i)
		book.claims[h] = serving("annie", uint64(i+1))
		hosts = append(hosts, h)
	}
	start := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	clock := start
	claims := newEdgesClaims("annie", book.read)
	claims.now = func() time.Time { return clock }
	state := &edgesLoopState{sched: newEdgesFileSchedule(start, func() time.Duration { return 5*time.Minute + 30*time.Second }), claims: claims}
	lastRead := map[string]time.Time{}
	worst := time.Duration(0)
	for clock = start; clock.Before(start.Add(3 * time.Hour)); clock = clock.Add(edgesFileWatchInterval) {
		if !state.due(clock, "fp", true) {
			continue
		}
		book.reads = nil
		claims.decide(context.Background(), hosts)
		for _, h := range book.reads {
			lastRead[h] = clock
		}
		state.done(clock, true)
		if clock.Sub(start) < time.Hour {
			continue // warm-up: the first reads of every host
		}
		for _, h := range hosts {
			if age := clock.Sub(lastRead[h]); age > worst {
				worst = age
			}
		}
	}
	if limit := edgesClaimTTL + 5*time.Minute; worst > limit {
		t.Fatalf("a claim went %s without a read, want at most %s", worst, limit)
	}
}

// Astra plan R2-1: a claim pass is a publication. While the bucket refuses
// every write, claim passes wait out the schedule's backoff and their own
// failures move it; once writes succeed, the 30-second claim passes resume.
func TestEdgesLoopClaimPassesObeyTheRetryBackoff(t *testing.T) {
	t.Parallel()
	book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
	var hosts []string
	for i := 0; i < 4000; i++ { // enough that reads stay due throughout
		h := fmt.Sprintf("h%04d.bab.test", i)
		book.claims[h] = serving("annie", 1)
		hosts = append(hosts, h)
	}
	start := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	clock := start
	claims := newEdgesClaims("annie", book.read)
	claims.now = func() time.Time { return clock }
	state := &edgesLoopState{sched: newEdgesFileSchedule(start, func() time.Duration { return 5*time.Minute + 30*time.Second }), claims: claims}
	failUntil := start.Add(20 * time.Minute)
	var during, after []time.Time
	for clock = start; clock.Before(failUntil.Add(8 * time.Minute)); clock = clock.Add(edgesFileWatchInterval) {
		if !state.due(clock, "fp", true) {
			continue
		}
		claims.decide(context.Background(), hosts)
		ok := !clock.Before(failUntil)
		if ok {
			after = append(after, clock)
		} else {
			during = append(during, clock)
		}
		state.done(clock, ok)
	}
	// Unthrottled, a claim pass every 30 seconds would be 40 attempts in
	// 20 minutes. The backoff (2 s doubling to 5 min) plus the timer allows
	// about a dozen.
	t.Logf("%d attempts while failing, %d after", len(during), len(after))
	if len(during) > 16 {
		t.Fatalf("%d publication attempts while every write failed, want at most 16: %v", len(during), during)
	}
	// Recovery waits out the backoff (at most five minutes), then the
	// 30-second claim passes resume.
	if len(after) < 3 || after[len(after)-1].Sub(after[len(after)-2]) != edgesClaimPassEvery {
		t.Fatalf("passes after writes recovered = %v, want the 30-second claim passes back within eight minutes", after)
	}
}

// Final review 4: a store that never answers cannot stall the edges file.
// The pass has a deadline; a read it cuts short keeps the last decision as
// any failed read does, and the reads not yet made wait for a later pass.
func TestEdgesClaimsPassHasADeadline(t *testing.T) {
	book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41), "theirs.bab.test": serving("bert", 42)}}
	hang := false
	read := func(ctx context.Context, host string) (hostclaim.Observed, error) {
		if hang {
			select { // a Get that blocks until the context ends
			case <-ctx.Done():
				return hostclaim.Observed{}, ctx.Err()
			case <-time.After(2 * time.Second): // so a missing deadline fails rather than hangs
				return hostclaim.Observed{}, errors.New("no deadline")
			}
		}
		return book.read(ctx, host)
	}
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := newEdgesClaims("annie", read)
	c.now = func() time.Time { return clock }
	c.passDeadline = 100 * time.Millisecond
	c.logf = func(string, ...any) {}
	routes := []registry.Route{
		{Host: "mine.bab.test", Kind: "instance", Socket: "/run/carlos/mine.sock", Provisioned: true},
		{Host: "theirs.bab.test", Kind: "instance", Socket: "/run/carlos/theirs.sock", Provisioned: true},
	}
	served := func() map[string]edgeHost {
		return map[string]edgeHost{"mine.bab.test": {Role: "home"}, "theirs.bab.test": {Role: "home"}}
	}
	applyEdgesClaims(context.Background(), c, routes, served())
	clock = clock.Add(edgesClaimTTL)
	hang = true
	hosts := served()
	start := time.Now()
	applyEdgesClaims(context.Background(), c, routes, hosts)
	if took := time.Since(start); took > time.Second {
		t.Fatalf("the pass took %s against a store that never answers, want about its 100ms deadline", took)
	}
	if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 {
		t.Fatalf("mine = %+v, %v: a read cut short by the deadline moved a served host", h, ok)
	}
	if _, ok := hosts["theirs.bab.test"]; ok {
		t.Fatal("a read cut short by the deadline published a host this box was told not to")
	}
	if d := newEdgesClaims("annie", read).passDeadline; d != 20*time.Second {
		t.Fatalf("the default pass deadline is %s, want 20s", d)
	}
}

// Final review 4: a host hostclaim.Key rejects is never read; it publishes
// as it did before claims.
func TestEdgesClaimsNeverReadAHostKeyRejects(t *testing.T) {
	const bad = "under_score.bab.test"
	if _, err := hostclaim.Key(bad); err == nil {
		t.Fatalf("hostclaim.Key accepts %q; the test needs a host it rejects", bad)
	}
	book := &claimsBook{claims: map[string]*hostclaim.Claim{bad: serving("bert", 42)}}
	c := newEdgesClaims("annie", book.read)
	routes := []registry.Route{{Host: bad, Kind: "instance", Socket: "/run/carlos/x.sock", Provisioned: true}}
	hosts := map[string]edgeHost{bad: {Role: "home"}}
	applyEdgesClaims(context.Background(), c, routes, hosts)
	if len(book.reads) != 0 {
		t.Fatalf("read claims for %v", book.reads)
	}
	if h, ok := hosts[bad]; !ok || h.Stamp != 0 {
		t.Fatalf("%s = %+v, %v; want home, unstamped, as before", bad, h, ok)
	}
}

// Final review 4: claim read failures are logged on the first failing
// pass and every 60th after it, and once when reads recover.
func TestEdgesClaimsLogReadFailuresThrottled(t *testing.T) {
	book := &claimsBook{claims: map[string]*hostclaim.Claim{}, errs: map[string]bool{"a.bab.test": true}}
	clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := newEdgesClaims("annie", book.read)
	c.now = func() time.Time { return clock }
	var lines []string
	c.logf = func(format string, a ...any) { lines = append(lines, fmt.Sprintf(format, a...)) }
	for i := 0; i < 120; i++ {
		c.decide(context.Background(), []string{"a.bab.test"})
		clock = clock.Add(edgesClaimTTL)
	}
	if len(lines) != 3 { // passes 1, 60 and 120
		t.Fatalf("%d lines for 120 failing passes, want 3: %q", len(lines), lines)
	}
	book.errs = nil
	c.decide(context.Background(), []string{"a.bab.test"})
	clock = clock.Add(edgesClaimTTL)
	c.decide(context.Background(), []string{"a.bab.test"})
	if len(lines) != 4 {
		t.Fatalf("lines after recovery = %q, want one recovery line", lines)
	}
}