package console

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"net/url"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/geoip"
	"github.com/carlosframework/platform/internal/status"
)

func putGeoip(t *testing.T, base, token string, body io.Reader, size int64) (int, string) {
	t.Helper()
	req, _ := http.NewRequest(http.MethodPut, base+"/api/cli/system/geoip", body)
	req.ContentLength = size
	req.Header.Set("Content-Type", "application/octet-stream")
	req.Header.Set("Authorization", "Bearer "+token)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	b, _ := io.ReadAll(res.Body)
	return res.StatusCode, string(b)
}

func jobID(t *testing.T, body string) string {
	t.Helper()
	i := strings.Index(body, `"job":"`)
	if i < 0 {
		t.Fatalf("no job id in %q", body)
	}
	rest := body[i+len(`"job":"`):]
	return rest[:strings.IndexByte(rest, '"')]
}

func TestGeoipPublishIsAJobAndReturnsItsPointer(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-put")
	body := fixture(t, "GeoIP2-City-Test.mmdb")
	st, out := putGeoip(t, fx.ts.URL, owner, bytes.NewReader(body), int64(len(body)))
	if st != http.StatusAccepted {
		t.Fatalf("PUT = %d %s, want 202", st, out)
	}
	job := waitJob(t, fx, owner, jobID(t, out))
	if job["error"] != "" {
		t.Fatalf("job error = %v", job["error"])
	}
	res, _ := job["result"].(map[string]any)
	if res == nil || res["database_type"] != "GeoIP2-City" {
		t.Fatalf("job result = %v, want the published pointer", job["result"])
	}
	if _, _, err := fx.rel.Get(context.Background(), geoip.PointerKey); err != nil {
		t.Fatalf("pointer not written: %v", err)
	}
}

func TestGeoipPublishRefusalIsTheJobsError(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-junk")
	st, out := putGeoip(t, fx.ts.URL, owner, strings.NewReader("junk"), 4)
	if st != http.StatusAccepted {
		t.Fatalf("PUT junk = %d %s, want 202 (verification is the job's)", st, out)
	}
	job := waitJob(t, fx, owner, jobID(t, out))
	if e, _ := job["error"].(string); !strings.Contains(e, "not a valid MaxMind-format database") {
		t.Fatalf("job error = %q", e)
	}
}

func TestGeoipDoorsAreForOperatorsOnly(t *testing.T) {
	fx := newConsoleFixture(t)
	if err := fx.st.AddMember(OpsAccountID, "member@keymail.dev", RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "member@keymail.dev", "code-geoip-member")
	_, want := getRaw(t, fx.ts.URL+"/api/cli/system/no-such-door", member)
	if st, body := getRaw(t, fx.ts.URL+"/api/cli/system/geoip", member); st != http.StatusNotFound || body != want {
		t.Errorf("GET as a member = %d %q, want the bare 404", st, body)
	}
	if st, _ := putGeoip(t, fx.ts.URL, member, strings.NewReader("x"), 1); st != http.StatusNotFound {
		t.Errorf("PUT as a member = %d, want 404", st)
	}
}

// Codex R5-1: the slot is reserved before any of the body is read.
func TestSecondUploadIsRefusedBeforeItsBodyIsRead(t *testing.T) {
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-slot")
	pr, pw := io.Pipe()
	defer pw.Close()
	go func() { pw.Write(make([]byte, 1024)) }() // then stall
	first := make(chan struct{})
	go func() {
		// Not putGeoip: t.Fatal from a goroutine would exit it without
		// closing first. The first upload ends in an error by design.
		defer close(first)
		req, _ := http.NewRequest(http.MethodPut, fx.ts.URL+"/api/cli/system/geoip", pr)
		req.ContentLength = 1 << 20
		req.Header.Set("Authorization", "Bearer "+owner)
		if res, err := http.DefaultClient.Do(req); err == nil {
			res.Body.Close()
		}
	}()
	deadline := time.Now().Add(5 * time.Second)
	for {
		if !fx.srv.geoip.mu.TryLock() {
			break // the first upload holds the slot
		}
		fx.srv.geoip.mu.Unlock()
		if time.Now().After(deadline) {
			t.Fatal("the first upload never took the slot")
		}
		time.Sleep(5 * time.Millisecond)
	}
	var read atomic.Bool
	second := readSpy{r: strings.NewReader("second"), read: &read}
	if st, out := putGeoip(t, fx.ts.URL, owner, second, 6); st != http.StatusConflict {
		t.Fatalf("second PUT = %d %s, want 409", st, out)
	}
	// The client may have sent the bytes; what matters is the handler never
	// read them, which an unread body plus a 409 shows from the server's side:
	// the slot is still the first upload's.
	if fx.srv.geoip.mu.TryLock() {
		fx.srv.geoip.mu.Unlock()
		t.Fatal("the refusal released the first upload's slot")
	}
	pw.CloseWithError(io.ErrUnexpectedEOF)
	<-first
	_ = read.Load()
}

type readSpy struct {
	r    io.Reader
	read *atomic.Bool
}

func (s readSpy) Read(p []byte) (int, error) { s.read.Store(true); return s.r.Read(p) }

// Codex R4-2: the 202 arrives through a proxy with a short response-header
// timeout even when verification takes longer.
func TestGeoipPublishAnswersBeforeVerificationThroughABackhaul(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-backhaul")
	geoipCommitDelay = 300 * time.Millisecond
	t.Cleanup(func() { geoipCommitDelay = 0 })
	target, _ := url.Parse(fx.ts.URL)
	rp := httputil.NewSingleHostReverseProxy(target)
	rp.Transport = &http.Transport{ResponseHeaderTimeout: 100 * time.Millisecond}
	proxy := httptest.NewServer(rp)
	t.Cleanup(proxy.Close)
	body := fixture(t, "GeoIP2-City-Test.mmdb")
	st, out := putGeoip(t, proxy.URL, owner, bytes.NewReader(body), int64(len(body)))
	if st != http.StatusAccepted {
		t.Fatalf("PUT through the proxy = %d %s, want 202 before verification", st, out)
	}
	if job := waitJob(t, fx, owner, jobID(t, out)); job["error"] != "" {
		t.Fatalf("job = %v", job)
	}
}

func seedGeoBox(t *testing.T, fx *consoleFixture, label string, g *status.GeoIPState, at time.Time) {
	t.Helper()
	b := status.CollectBox(nil, status.Probes{Release: "v1", GeoIP: g}, at)
	b.Label = label
	if err := status.WriteBox(context.Background(), fx.rel, b); err != nil {
		t.Fatal(err)
	}
}

func seedEdgesRecord(t *testing.T, fx *consoleFixture, label string) {
	t.Helper()
	if _, err := fx.rel.Put(context.Background(), "control/edges/"+label+".json", []byte(`{"v":1,"box":"`+label+`","eip":"192.0.2.1","hosts":{}}`), ""); err != nil {
		t.Fatal(err)
	}
}

func TestGeoipStatusVerdict(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-status")
	ptr, err := fx.srv.geoip.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now()
	seedEdgesRecord(t, fx, "ie-1")
	seedEdgesRecord(t, fx, "uk-1")
	seedGeoBox(t, fx, "ie-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now)
	// Codex R3-4: a heartbeat just inside the next jittered tick is fresh.
	seedGeoBox(t, fx, "uk-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now.Add(-(status.SnapshotPeriod + 59*time.Second)))
	st, out := getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner)
	if st != http.StatusOK || out["ready"] != true {
		t.Fatalf("status = %d %v, want ready", st, out)
	}
	// An old binary, an expected box with no heartbeat, and a stale one.
	seedEdgesRecord(t, fx, "us-1")
	seedGeoBox(t, fx, "us-1", nil, now)
	seedEdgesRecord(t, fx, "au-1")
	seedEdgesRecord(t, fx, "sg-1")
	seedGeoBox(t, fx, "sg-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now.Add(-2*time.Hour))
	_, out = getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner)
	if out["ready"] != false {
		t.Fatalf("status = %v, want not ready", out)
	}
	blocking, _ := out["blocking"].([]any)
	joined := ""
	for _, b := range blocking {
		joined += b.(string) + "\n"
	}
	for _, want := range []string{"us-1: runs a binary", "au-1: no heartbeat", "sg-1: no heartbeat since", "control/edges/sg-1.json"} {
		if !strings.Contains(joined, want) {
			t.Errorf("blocking lacks %q:\n%s", want, joined)
		}
	}
}

func TestGeoipVerdictIncompleteReadIsNeverReady(t *testing.T) {
	ready, _, blocking := geoipVerdict(geoipWant{state: geoipOff}, geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{}, expectedComplete: false, heartbeatsComplete: true}, time.Now())
	if ready || len(blocking) == 0 {
		t.Fatalf("incomplete read = ready %v, blocking %v", ready, blocking)
	}
}

func TestGeoipOff(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-off")
	if _, err := fx.srv.geoip.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"}); err != nil {
		t.Fatal(err)
	}
	req, _ := http.NewRequest(http.MethodDelete, fx.ts.URL+"/api/cli/system/geoip", nil)
	req.Header.Set("Authorization", "Bearer "+owner)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	res.Body.Close()
	if res.StatusCode != http.StatusOK {
		t.Fatalf("DELETE = %d", res.StatusCode)
	}
	if _, _, err := fx.rel.Get(context.Background(), geoip.PointerKey); err == nil {
		t.Fatal("pointer survived off")
	}
}

// Codex R8-1: identity is (namespace, label). A fresh platform box must not
// stand in for a fleet box of the same label, nor one fleet's for another's.
func TestGeoipVerdictKeysBoxesByNamespace(t *testing.T) {
	now := time.Now()
	fresh := func(label string) status.Box {
		b := status.CollectBox(nil, status.Probes{Release: "v1", GeoIP: &status.GeoIPState{Mode: "mirror", SHA256: "s"}}, now)
		b.Label = label
		return b
	}
	old := status.CollectBox(nil, status.Probes{Release: "v0"}, now)
	old.Label = "node-1"
	inv := geoipInventoryRead{
		expected:         map[geoipBoxKey]bool{{"", "node-1"}: true, {"tito", "node-1"}: true, {"bae", "node-1"}: true},
		boxes:            map[geoipBoxKey]status.Box{{"", "node-1"}: fresh("node-1"), {"tito", "node-1"}: fresh("node-1"), {"bae", "node-1"}: old},
		expectedComplete: true, heartbeatsComplete: true,
	}
	ready, _, blocking := geoipVerdict(geoipWant{state: geoipLive, sha: "s"}, inv, now)
	if ready || len(blocking) != 1 || !strings.HasPrefix(blocking[0], "bae/node-1: runs a binary") {
		t.Fatalf("ready=%v blocking=%v, want only bae/node-1 blocking", ready, blocking)
	}
	delete(inv.boxes, geoipBoxKey{"tito", "node-1"})
	_, _, blocking = geoipVerdict(geoipWant{state: geoipLive, sha: "s"}, inv, now)
	found := false
	for _, b := range blocking {
		found = found || strings.HasPrefix(b, "tito/node-1: no heartbeat")
	}
	if !found {
		t.Fatalf("a missing fleet box was satisfied by a same-label box elsewhere: %v", blocking)
	}
}

// Codex R8 on R1-16: a box with no edges record (retired) is listed but
// never blocks.
func TestGeoipVerdictRetiredBoxDoesNotBlock(t *testing.T) {
	now := time.Now()
	b := status.CollectBox(nil, status.Probes{Release: "v0"}, now.Add(-time.Hour))
	b.Label = "gone-1"
	inv := geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{{"", "gone-1"}: b}, expectedComplete: true, heartbeatsComplete: false}
	ready, rows, _ := geoipVerdict(geoipWant{state: geoipOff}, inv, now)
	if !ready || len(rows) != 1 {
		t.Fatalf("ready=%v rows=%v, want ready with the retired box listed", ready, rows)
	}
}

// Codex R9-1: an unreadable pointer is never ready, and off requires every
// mirror to have dropped its generation.
func TestGeoipVerdictPointerStates(t *testing.T) {
	now := time.Now()
	mirror := func(sha string) status.Box {
		b := status.CollectBox(nil, status.Probes{Release: "v1", GeoIP: &status.GeoIPState{Mode: "mirror", SHA256: sha}}, now)
		b.Label = "ie-1"
		return b
	}
	inv := func(sha string) geoipInventoryRead {
		return geoipInventoryRead{expected: map[geoipBoxKey]bool{{"", "ie-1"}: true},
			boxes: map[geoipBoxKey]status.Box{{"", "ie-1"}: mirror(sha)}, expectedComplete: true, heartbeatsComplete: true}
	}
	if ready, _, _ := geoipVerdict(geoipWant{state: geoipUnknown}, inv("a"), now); ready {
		t.Fatal("an unreadable pointer reported ready")
	}
	if ready, _, b := geoipVerdict(geoipWant{state: geoipOff}, inv("a"), now); ready || len(b) != 1 || !strings.Contains(b[0], "still has") {
		t.Fatalf("off with a mirror still loaded = %v %v", ready, b)
	}
	if ready, _, _ := geoipVerdict(geoipWant{state: geoipOff}, inv(""), now); !ready {
		t.Fatal("off with every mirror empty must be ready")
	}
	if ready, _, _ := geoipVerdict(geoipWant{state: geoipLive, sha: "b"}, inv("a"), now); ready {
		t.Fatal("a mirror on another sha reported ready")
	}
}

// Codex R9-2: off never waits behind a publish.
func TestGeoipOffDuringAPublishIs409(t *testing.T) {
	withFixtureProbes(t)
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-offbusy")
	st, err := fx.srv.geoip.Stage(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")))
	if err != nil {
		t.Fatal(err)
	}
	defer fx.srv.geoip.Abandon(st)
	req, _ := http.NewRequest(http.MethodDelete, fx.ts.URL+"/api/cli/system/geoip", nil)
	req.Header.Set("Authorization", "Bearer "+owner)
	done := make(chan int, 1)
	go func() {
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			done <- 0
			return
		}
		res.Body.Close()
		done <- res.StatusCode
	}()
	select {
	case code := <-done:
		if code != http.StatusConflict {
			t.Fatalf("off during a publish = %d, want 409", code)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("off waited behind the publish")
	}
}

// Codex R10 on R9-1: a pointer past D1 by the clock is "off" for the
// verdict even before the daily job marks it.
func TestGeoipStatusTreatsAnUnmarkedExpiredPointerAsOff(t *testing.T) {
	fx := newConsoleFixture(t)
	owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-expired")
	sha := strings.Repeat("c", 64)
	ptr := geoip.Pointer{V: 1, SHA256: sha, Key: geoip.BlobKey(sha), Size: 1, DatabaseType: "GeoLite2-City",
		BuildEpoch: time.Now().Add(-40 * 24 * time.Hour).Unix(), Origin: "manual"}
	b, _ := marshalPointer(ptr)
	if _, err := fx.rel.Put(context.Background(), geoip.PointerKey, b, ""); err != nil {
		t.Fatal(err)
	}
	seedEdgesRecord(t, fx, "ie-1")
	seedGeoBox(t, fx, "ie-1", &status.GeoIPState{Mode: "mirror", SHA256: sha}, time.Now())
	_, out := getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner)
	if out["ready"] != false || !strings.Contains(fmt.Sprint(out["blocking"]), "still has") {
		t.Fatalf("expired pointer with a loaded mirror = %v", out)
	}
	seedGeoBox(t, fx, "ie-1", &status.GeoIPState{Mode: "mirror"}, time.Now())
	if _, out = getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner); out["ready"] != true {
		t.Fatalf("expired pointer with every mirror empty = %v, want ready", out)
	}
}