package host

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostenv"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/lbhealth"
	"github.com/carlosframework/platform/internal/status"
)

// stagedFacts walks a box through the joining stages, one per selectMode
// call: no invitation, invited, guard installed, guard-ready reported,
// then hosting.
type stagedFacts struct {
	mu    sync.Mutex
	calls int
	steps []joinFacts
}

func (s *stagedFacts) facts() modeFacts {
	pick := func() joinFacts {
		s.mu.Lock()
		defer s.mu.Unlock()
		i := min(s.calls/5, len(s.steps)-1) // each stage for five checks
		s.calls++
		return s.steps[i]
	}
	cur := joinFacts{}
	f := modeFacts{box: "ada-hiking", now: time.Now}
	f.enrollment = func() (hostfloor.Enrollment, error) { cur = pick(); return cur.facts().enrollment() }
	f.roster = func(ctx context.Context) (hostactors.Roster, error) { return cur.facts().roster(ctx) }
	f.guard = func() (hostfloor.Running, error) { return cur.facts().guard() }
	f.invited = func(ctx context.Context, b string) (bool, error) { return cur.facts().invited(ctx, b) }
	f.report = func(ctx context.Context, b string) (status.Box, error) { return cur.facts().report(ctx, b) }
	return f
}

func freePort(t *testing.T) string {
	t.Helper()
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Skipf("no TCP listener in this sandbox: %v", err)
	}
	addr := l.Addr().String()
	l.Close()
	return addr
}

// Joining publishes no edges file and admits nothing: it answers the load
// balancer 503 throughout, asks root to arm the guard once invited,
// publishes its report once armed, and returns as soon as it can host.
func TestJoiningServes503UntilItCanHostAndPublishesNoEdgesFile(t *testing.T) {
	deploy, data := t.TempDir(), t.TempDir()
	t.Setenv("CARLOS_DEPLOYMENT_DIR", deploy)
	t.Setenv("CARLOS_DATA_DIR", data)
	t.Cleanup(func() { lbhealth.Default.SetHosting(false) })
	addr := freePort(t)
	f := enrolledFacts()
	f.report = freshReport(time.Now())
	steps := &stagedFacts{steps: []joinFacts{{}, {invited: true}, guardInstalled(), func() joinFacts { g := guardInstalled(); g.report = freshReport(time.Now()); return g }(), f}}
	j := productionJoining([]string{"-http", addr, "-https", ":0"}, steps.facts(), os.Getenv)
	j.every = 20 * time.Millisecond
	var mu sync.Mutex
	var reports int
	records := make(chan struct{}, 1)
	j.record = func(context.Context) { records <- struct{}{} }
	j.report = func(context.Context) error { mu.Lock(); reports++; mu.Unlock(); return nil }
	j.logf = t.Logf
	lbhealth.Default.SetHosting(true) // a stale flag from an earlier process state is cleared
	done := make(chan modeDecision, 1)
	go func() { d, _ := j.run(t.Context()); done <- d }()
	// The health check answers 503, naming hosting, while joining.
	var body string
	for i := 0; i < 100; i++ {
		resp, err := http.Get("http://" + addr + lbhealth.Path)
		if err == nil {
			b, _ := io.ReadAll(resp.Body)
			resp.Body.Close()
			if resp.StatusCode != http.StatusServiceUnavailable {
				t.Fatalf("joining box answered %d", resp.StatusCode)
			}
			body = string(b)
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if !strings.Contains(body, lbhealth.CondHosting) {
		t.Fatalf("health body %q", body)
	}
	if resp, err := http.Get("http://" + addr + "/"); err != nil || resp.StatusCode != http.StatusNotFound {
		t.Fatalf("a joining box served something other than its health check: %v %v", resp, err)
	}
	select {
	case d := <-done:
		if d.Mode != modeHosting {
			t.Fatalf("returned %+v", d)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("joining never returned")
	}
	<-records
	if _, err := os.Stat(filepath.Join(data, joinRequestMarker)); err != nil {
		t.Fatalf("root was never asked to arm the guard: %v", err)
	}
	mu.Lock()
	if reports == 0 {
		t.Fatal("the guard-ready report was never published")
	}
	mu.Unlock()
	if lbhealth.Default.Hosting() {
		t.Fatal("joining marked the box hosting")
	}
	err := filepath.WalkDir(deploy, func(p string, d fs.DirEntry, err error) error {
		if err == nil && strings.Contains(filepath.ToSlash(p), "control/edges") {
			t.Errorf("joining wrote %s", p)
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
}

func TestDecideAgentMode(t *testing.T) {
	on := func(k string) string {
		switch k {
		case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA":
			return "1"
		}
		return ""
	}
	noHA := func(k string) string {
		if k == "CARLOS_HOST_CLAIM" {
			return "1"
		}
		return ""
	}
	never := func(context.Context, func(string) string) (modeFacts, error) {
		t.Fatal("facts read with joining mode off")
		return modeFacts{}, nil
	}
	noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil }
	quiet := func(string, ...any) {}
	ctx := context.Background()
	for name, env := range map[string]func(string) string{"claim off": func(string) string { return "" }, "region HA off": noHA} {
		if !decideAgentMode(ctx, env, never, noJoin, time.Millisecond, quiet) {
			t.Fatalf("%s: not hosting", name)
		}
	}
	open := func(f joinFacts) func(context.Context, func(string) string) (modeFacts, error) {
		return func(context.Context, func(string) string) (modeFacts, error) { return f.facts(), nil }
	}
	if !decideAgentMode(ctx, on, open(enrolledFacts()), noJoin, time.Millisecond, quiet) {
		t.Fatal("an enrolled, armed box does not host")
	}
	var joined modeDecision
	join := func(_ context.Context, _ modeFacts, d modeDecision) error { joined = d; return nil }
	if decideAgentMode(ctx, on, open(joinFacts{invited: true}), join, time.Millisecond, quiet) || joined.Stage != stageInvited {
		t.Fatalf("an invited box: joined %+v", joined)
	}
	// Unknown, then known: the gate waits, asks again, and logs the
	// reason once.
	reads, logs := 0, 0
	flaky := func(context.Context, func(string) string) (modeFacts, error) {
		reads++
		f := enrolledFacts()
		if reads < 3 {
			f.rosterErr = errors.New("connection reset")
		}
		return f.facts(), nil
	}
	if !decideAgentMode(ctx, on, flaky, noJoin, time.Millisecond, func(string, ...any) { logs++ }) || reads != 3 || logs != 1 {
		t.Fatalf("flaky roster: %d reads, %d log lines", reads, logs)
	}
	// Stopped while unknown: no claim agent.
	stopped, cancel := context.WithCancel(ctx)
	cancel()
	failing := func(context.Context, func(string) string) (modeFacts, error) {
		return modeFacts{}, errors.New("no bucket")
	}
	if decideAgentMode(stopped, on, failing, noJoin, time.Hour, quiet) {
		t.Fatal("a stopped gate started the claim agent")
	}
}

// A failed carlos system join is retried (Task 7's review): the path unit
// fires only when the marker changes, so joining touches it again once a
// minute for as long as the box stays invited with its guard not armed,
// and never more often. Once the guard is armed it stops asking.
func TestJoiningAsksRootAgainEachMinuteWhileInvited(t *testing.T) {
	t.Cleanup(func() { lbhealth.Default.SetHosting(false) })
	var mu sync.Mutex
	clock, checks := joinNow, 0
	var touches []time.Time
	f := joinFacts{invited: true}.facts()
	armedAfter := 40 // checks; then the guard is armed and not yet enrolled
	f.enrollment = func() (hostfloor.Enrollment, error) {
		mu.Lock()
		defer mu.Unlock()
		if checks >= armedAfter {
			return guardInstalled().facts().enrollment()
		}
		return joinFacts{}.facts().enrollment()
	}
	f.roster = guardInstalled().facts().roster
	f.guard = guardInstalled().facts().guard
	f.now = func() time.Time {
		mu.Lock()
		defer mu.Unlock()
		checks++
		clock = clock.Add(20 * time.Second) // each check, 20 seconds on
		return clock
	}
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	reported := make(chan struct{}, 1)
	j := joining{
		facts:  f,
		every:  time.Millisecond,
		serve:  func(ctx context.Context) error { <-ctx.Done(); return nil },
		record: func(context.Context) {},
		report: func(context.Context) error {
			select {
			case reported <- struct{}{}:
			default:
			}
			return nil
		},
		requestArm: func() error {
			mu.Lock()
			touches = append(touches, clock)
			mu.Unlock()
			return nil
		},
		logf: t.Logf,
	}
	done := make(chan struct{})
	go func() { defer close(done); _, _ = j.run(ctx) }()
	select {
	case <-reported:
	case <-time.After(10 * time.Second):
		t.Fatal("the guard was armed and joining never published its report")
	}
	// A few more checks with the guard armed: no more touches.
	for {
		mu.Lock()
		n := checks
		mu.Unlock()
		if n >= armedAfter+10 {
			break
		}
		time.Sleep(time.Millisecond)
	}
	cancel()
	<-done
	mu.Lock()
	defer mu.Unlock()
	// 40 checks 20 seconds apart while invited: about 13 minutes, so a
	// touch at the first check and then one each minute.
	if len(touches) < 12 || len(touches) > 14 {
		t.Fatalf("%d touches over %d invited checks: %v", len(touches), armedAfter, touches)
	}
	for i := 1; i < len(touches); i++ {
		if gap := touches[i].Sub(touches[i-1]); gap < joinNudgeEvery || gap >= joinNudgeEvery+20*time.Second {
			t.Fatalf("touch %d came %s after the one before; want once a minute", i, gap)
		}
	}
	if last := touches[len(touches)-1]; last.After(joinNow.Add(time.Duration(armedAfter) * 20 * time.Second)) {
		t.Fatalf("root was asked again at %s, after the guard was armed", last)
	}
}

// Any error opening the facts (an empty group reference, say) is unknown:
// the gate waits and asks again, and neither joins nor hosts until a
// read answers (Task 6's review).
func TestDecideAgentModeWaitsOutAFactsError(t *testing.T) {
	on := func(k string) string {
		switch k {
		case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA":
			return "1"
		}
		return ""
	}
	for name, then := range map[string]joinFacts{"enrolled": enrolledFacts(), "invited": {invited: true}} {
		t.Run(name, func(t *testing.T) {
			reads, logs, joins := 0, 0, 0
			open := func(context.Context, func(string) string) (modeFacts, error) {
				reads++
				if reads < 3 {
					return modeFacts{}, errors.New("this box's group reference could not be read")
				}
				return then.facts(), nil
			}
			join := func(_ context.Context, _ modeFacts, d modeDecision) error {
				if reads < 3 {
					t.Fatalf("joined on read %d: %+v", reads, d)
				}
				joins++
				return nil
			}
			hosting := decideAgentMode(context.Background(), on, open, join, time.Millisecond, func(string, ...any) { logs++ })
			if reads != 3 || logs < 1 {
				t.Fatalf("%d reads, %d log lines", reads, logs)
			}
			wantHosting, wantJoins := name == "enrolled", 1
			if wantHosting {
				wantJoins = 0
			}
			if hosting != wantHosting || joins != wantJoins {
				t.Fatalf("hosting %t, %d joins; want %t, %d", hosting, joins, wantHosting, wantJoins)
			}
		})
	}
}

// Task 6's review (R-A): a group box resolves its reference before the
// gate and before runAgent arms any loop, whatever the switches say, since
// those loops read metricsBoxLabel once and keep it. IMDS that fails twice
// holds everything back until the third read; every failure is logged with
// its cause. Any other box asks IMDS nothing and starts at once.
func TestAGroupBoxResolvesItsReferenceBeforeTheAgentStarts(t *testing.T) {
	quick := func(int) time.Duration { return time.Millisecond }
	never := func(context.Context, func(string) string) (modeFacts, error) {
		t.Fatal("facts read with joining mode off")
		return modeFacts{}, nil
	}
	enrolled := func(context.Context, func(string) string) (modeFacts, error) { return enrolledFacts().facts(), nil }
	noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil }
	for name, tc := range map[string]struct {
		env    map[string]string
		facts  func(context.Context, func(string) string) (modeFacts, error)
		marked bool
	}{
		"group box, switches off": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference}, facts: never, marked: true},
		"group box, switches on": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference, "CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1", "CARLOS_REGION": "eu-west-1"},
			facts: enrolled, marked: true},
		"other box, switches off": {env: map[string]string{}, facts: never},
		"other box, switches on":  {env: map[string]string{"CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1"}, facts: enrolled},
	} {
		t.Run(name, func(t *testing.T) {
			withGroupIdentity(t, "", "", nil)
			asked := 0
			groupIdentity = func() (string, string, error) {
				asked++
				if asked < 3 {
					return "", "", errors.New("IMDS answered 500")
				}
				return "123456789012", "i-0a61f580df1acfa65", nil
			}
			getenv := groupEnv(tc.env)
			var lines []string
			logf := func(format string, args ...any) { lines = append(lines, fmt.Sprintf(format, args...)) }
			gatedAt, ranAt, label := -1, -1, ""
			gate := func() bool {
				gatedAt = asked
				return decideAgentMode(context.Background(), getenv, tc.facts, noJoin, time.Millisecond, logf)
			}
			run := func() { ranAt = asked; label = metricsBoxLabel(getenv, "machine-1") }
			agentEntry(getenv, resolveGroupBoxReference, quick, logf, gate, run)
			want, ref := 0, status.Label("machine-1")
			if tc.marked {
				want, ref = 3, hostactors.GroupBoxReference("123456789012", "i-0a61f580df1acfa65")
			}
			if gatedAt != want || ranAt != want {
				t.Fatalf("gate after %d IMDS reads, agent after %d; want %d", gatedAt, ranAt, want)
			}
			if label != ref {
				t.Fatalf("the agent started with label %q, want %q", label, ref)
			}
			failures := 0
			for _, l := range lines {
				if strings.Contains(l, "IMDS answered 500") {
					failures++
				}
			}
			if tc.marked && failures != 2 {
				t.Fatalf("logged %d failures with their cause, want 2: %q", failures, lines)
			}
		})
	}
}

func TestAwaitGroupReferenceStopsWithItsContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	failing := func() (string, error) { return "", errors.New("IMDS answered 500") }
	if awaitGroupReference(ctx, failing, func(int) time.Duration { return time.Hour }, func(string, ...any) {}) {
		t.Fatal("a stopped wait reported the reference resolved")
	}
	var got []time.Duration
	for n := 1; n <= 8; n++ {
		got = append(got, groupReferenceBackoff(n))
	}
	want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 30 * time.Second, 30 * time.Second, 30 * time.Second}
	if fmt.Sprint(got) != fmt.Sprint(want) {
		t.Fatalf("backoff %v, want %v", got, want)
	}
}