package console

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
	"github.com/carlosframework/platform/internal/status"
)

func seedReadyBox(t *testing.T, fx *consoleFixture, label, placement string, protocol int, floor, prevAllowed, mode string, claim bool) {
	t.Helper()
	putSeedEdges(t, fx, label, `{"v":1,"box":"`+label+`","eip":"1.2.3.4","placement":"`+placement+
		`","hibernation_configured":true,"hosts":{},"seeds_protocol":`+strconv.Itoa(protocol)+
		`,"seeds_floor_min_release":"`+floor+`","prev_release":"v0.27.1","prev_seeds_allowed_for":"`+prevAllowed+`"}`)
	putSeedHeartbeat(t, fx, label, mode, claim, time.Now())
}

// putSeedEdges writes a box's edges file, rewriting it in place: a test
// moves one box through several states, and putEdges only creates.
func putSeedEdges(t *testing.T, fx *consoleFixture, label, body string) {
	t.Helper()
	key := "control/edges/" + label + ".json"
	_, etag, err := fx.rel.Get(context.Background(), key)
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), key, []byte(body), etag); err != nil {
		t.Fatal(err)
	}
}

// putSeedHeartbeat writes the box's status heartbeat as of at; mode "" is
// a box with no layout report.
func putSeedHeartbeat(t *testing.T, fx *consoleFixture, label, mode string, claim bool, at time.Time) {
	t.Helper()
	putSeedHeartbeatIn(t, fx.rel, label, mode, claim, at)
}

func putSeedHeartbeatIn(t *testing.T, store release.Store, label, mode string, claim bool, at time.Time) {
	t.Helper()
	var layout *status.LayoutReport
	if mode != "" {
		layout = &status.LayoutReport{Mode: mode}
	}
	// A release, as every real heartbeat carries: without one a box with no
	// layout report would publish no metrics at all.
	b := status.CollectBox(nil, status.Probes{Release: "v0.27.1", Layout: layout, HostClaimEnabled: claim}, at)
	b.Label = label
	if err := status.WriteBox(context.Background(), store, b); err != nil {
		t.Fatal(err)
	}
}

func publishSeedsFloorFx(t *testing.T, fx *consoleFixture, rel string) {
	t.Helper()
	if err := seeds.PublishFloor(context.Background(), fx.rel, seeds.Floor{Min: seeds.RequiredProtocol, MinRelease: rel}); err != nil {
		t.Fatal(err)
	}
}

func TestSeedRegionReady(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusConflict || !strings.Contains(ref.Msg, "no seeds floor yet") {
		t.Fatalf("no floor: %+v", ref)
	}
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref != nil {
		t.Fatalf("a ready region: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-2", "", 0, "v0.27.0", "v0.27.0", "nested", false)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusConflict || !strings.Contains(ref.Msg, "kass-2") || !strings.Contains(ref.Msg, "below the seeds floor") {
		t.Fatalf("a box on protocol 0: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-2", "", 1, "v0.27.0", "", "nested", false)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || !strings.Contains(ref.Msg, "below the seeds floor") {
		t.Fatalf("a carlos.prev without a seeds verdict: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "flat", false)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || !strings.Contains(ref.Msg, "nested") {
		t.Fatalf("a flat box: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", true)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || !strings.Contains(ref.Msg, "claim control") {
		t.Fatalf("a claim-armed box: %+v", ref)
	}
	fx.srv.cfg.HostClaim = true
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || !strings.Contains(ref.Msg, "This console runs with claim control on") {
		t.Fatalf("a claim console: %+v", ref)
	}
}

func TestSeedRegionReadyOneBoxForImports(t *testing.T) {
	fx := newConsoleFixture(t)
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	seedReadyBox(t, fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	if ref := fx.srv.seedRegionReady(context.Background(), "", false); ref != nil {
		t.Fatalf("two ready boxes, --from: %+v", ref)
	}
	if ref := fx.srv.seedRegionReady(context.Background(), "", true); ref == nil || ref.Status != http.StatusConflict || !strings.Contains(ref.Msg, "more than one box") {
		t.Fatalf("two ready boxes, import: %+v", ref)
	}
}

func TestSeedRegionReadyRefusesAnUnconfirmedBox(t *testing.T) {
	fx := newConsoleFixture(t)
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "", false) // no layout report
	if ref := fx.srv.seedRegionReady(context.Background(), "", false); ref == nil || ref.Status != http.StatusServiceUnavailable {
		t.Fatalf("a box with no layout report: %+v", ref)
	}
}

func TestSeedRegionReadyIgnoresOtherRegions(t *testing.T) {
	fx := newConsoleFixture(t)
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	seedReadyBox(t, fx, "uk-1", "uk", 0, "", "", "flat", true)
	if ref := fx.srv.seedRegionReady(context.Background(), "", false); ref != nil {
		t.Fatalf("a bad box in uk blocked the home region: %+v", ref)
	}
	if ref := fx.srv.seedRegionReady(context.Background(), "uk", false); ref == nil {
		t.Fatal("uk's own bad box did not block uk")
	}
}

// Every missing, zero or unparseable field is not ready. An old box's edges
// file reads as zeros, and a box that could not read its own local floor
// publishes the protocol but an empty floor: neither may pass, whatever the
// rest of the file says.
func TestSeedRegionReadyFailsClosedOnTheEdgesFile(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	for _, tc := range []struct{ name, body string }{
		{"an old box", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{}}`},
		{"an old box keeping no carlos.prev", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"prev_release":"none","seeds_floor_min_release":"v0.27.0"}`},
		{"an unreadable local floor", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"prev_release":"none"}`},
		{"a stale local floor", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.26.0","prev_release":"none"}`},
		{"a carlos.prev vouched for an older floor", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.27.0","prev_release":"v0.27.1","prev_seeds_allowed_for":"v0.26.0"}`},
		{"no carlos.prev answer at all", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.27.0"}`},
	} {
		putSeedEdges(t, fx, "kass-2", tc.body)
		putSeedHeartbeat(t, fx, "kass-2", "nested", false, time.Now())
		ref := fx.srv.seedRegionReady(ctx, "", false)
		want := "kass-2 in the home region are below the seeds floor. Roll the platform there, then publish the seeds floor again."
		if ref == nil || ref.Status != http.StatusConflict || ref.Msg != want {
			t.Errorf("%s: %+v", tc.name, ref)
		}
	}
	putSeedEdges(t, fx, "kass-2", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.27.0","prev_release":"none"}`)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref != nil {
		t.Fatalf("a box keeping no carlos.prev: %+v", ref)
	}
}

// A box in the edges files with no heartbeat at all, or a stale one, is
// unconfirmed: never nested, never ready.
func TestSeedRegionReadyRefusesAMissingOrStaleHeartbeat(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	putSeedEdges(t, fx, "kass-2", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.27.0","prev_release":"none"}`)
	want := "could not confirm every box in the home region. Try again in a minute."
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusServiceUnavailable || ref.Msg != want {
		t.Fatalf("a box with no heartbeat: %+v", ref)
	}
	putSeedHeartbeat(t, fx, "kass-2", "nested", false, time.Now().Add(-time.Hour))
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusServiceUnavailable || ref.Msg != want {
		t.Fatalf("a box with a stale heartbeat: %+v", ref)
	}
	putSeedHeartbeat(t, fx, "kass-2", "nested", false, time.Now())
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref != nil {
		t.Fatalf("both boxes fresh: %+v", ref)
	}
	// A region with no edges file at all is unconfirmed too, named.
	if ref := fx.srv.seedRegionReady(ctx, "uk", false); ref == nil || ref.Status != http.StatusServiceUnavailable || ref.Msg != "could not confirm every box in uk. Try again in a minute." {
		t.Fatalf("a region with no box: %+v", ref)
	}
}

// A published floor that PublishFloor would never write is unreadable, not
// a floor: an empty min_release would match every box that could not read
// its own.
func TestSeedRegionReadyRefusesAnUnusableFloor(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	seedReadyBox(t, fx, "kass-1", "", 1, "", "", "nested", false)
	for _, body := range []string{`{"min":1,"min_release":""}`, `{"min":1}`, `{"min":1,"min_release":"next"}`, `not json`} {
		_, etag, err := fx.rel.Get(ctx, seeds.FloorKey)
		if err != nil && !errors.Is(err, os.ErrNotExist) {
			t.Fatal(err)
		}
		if _, err := fx.rel.Put(ctx, seeds.FloorKey, []byte(body), etag); err != nil {
			t.Fatal(err)
		}
		if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusServiceUnavailable || ref.Msg != "could not read the seeds floor. Try again." {
			t.Errorf("floor %s: %+v", body, ref)
		}
	}
}

// The refusals name every box they found, sorted, and the region by its
// name: the configured home region's, not the record's "".
func TestSeedRegionReadyNamesBoxesAndRegion(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.HomeRegion = "eu"
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-3", "eu", 1, "v0.27.0", "v0.27.0", "flat", false)
	seedReadyBox(t, fx, "kass-1", "eu", 1, "v0.27.0", "v0.27.0", "flat", false)
	seedReadyBox(t, fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", true)
	want := "kass-1, kass-3 in eu do not run instances nested yet. A seeded instance needs every box there nested."
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusConflict || ref.Msg != want {
		t.Fatalf("two flat boxes: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-3", "eu", 1, "v0.27.0", "v0.27.0", "nested", true)
	seedReadyBox(t, fx, "kass-1", "eu", 1, "v0.27.0", "v0.27.0", "nested", false)
	want = "kass-2, kass-3 in eu run with claim control on. Seeded instances are not supported there yet."
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusConflict || ref.Msg != want {
		t.Fatalf("two claim boxes: %+v", ref)
	}
	if ref := fx.srv.seedRegionReady(ctx, "", true); ref == nil || ref.Status != http.StatusConflict || ref.Msg != "eu has more than one box. An import needs a region with one box for now." {
		t.Fatalf("an import into three boxes: %+v", ref)
	}
}

// A customer fleet's heartbeat is not a platform box's: a fresh, nested,
// claim-free fleet heartbeat under the same label cannot stand in for a
// stale platform box.
func TestSeedRegionReadyIgnoresFleetHeartbeats(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	putSeedHeartbeat(t, fx, "kass-1", "nested", false, time.Now().Add(-time.Hour))
	reg, err := registry.Open(fx.regPath)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	if err := reg.UpsertFleet(registry.Fleet{Name: "pi", Account: fx.sqid, Bucket: "pi-bucket"}); err != nil {
		t.Fatal(err)
	}
	putSeedHeartbeatIn(t, release.NewPrefixStore(fx.rel, "fleets/pi/"), "kass-1", "nested", false, time.Now())
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusServiceUnavailable {
		t.Fatalf("a fleet heartbeat covered a stale platform box: %+v", ref)
	}
}

// Two platform heartbeats claiming one label are ambiguous: unconfirmed,
// whichever the read happens to see last.
func TestSeedRegionReadyRefusesADuplicateLabel(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	b := status.CollectBox(nil, status.Probes{Release: "v0.27.1", Layout: &status.LayoutReport{Mode: "flat"}}, time.Now())
	b.Label = "kass-1"
	body, err := json.Marshal(b)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(ctx, "status/boxes/kass-1-old.json", body, ""); err != nil {
		t.Fatal(err)
	}
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusServiceUnavailable {
		t.Fatalf("two heartbeats for kass-1: %+v", ref)
	}
}

// The floor's own min protocol binds too, when it is above the one this
// console requires.
func TestSeedRegionReadyHonoursTheFloorsProtocol(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	if err := seeds.PublishFloor(ctx, fx.rel, seeds.Floor{Min: seeds.RequiredProtocol + 1, MinRelease: "v0.27.0"}); err != nil {
		t.Fatal(err)
	}
	seedReadyBox(t, fx, "kass-1", "", seeds.RequiredProtocol, "v0.27.0", "v0.27.0", "nested", false)
	want := "kass-1 in the home region are below the seeds floor. Roll the platform there, then publish the seeds floor again."
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref == nil || ref.Status != http.StatusConflict || ref.Msg != want {
		t.Fatalf("a box below the floor's protocol: %+v", ref)
	}
	seedReadyBox(t, fx, "kass-1", "", seeds.RequiredProtocol+1, "v0.27.0", "v0.27.0", "nested", false)
	if ref := fx.srv.seedRegionReady(ctx, "", false); ref != nil {
		t.Fatalf("a box at the floor's protocol: %+v", ref)
	}
}

// A caller may name the home region rather than pass the record's "".
func TestSeedRegionReadyAcceptsTheHomeRegionsName(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.HomeRegion = "eu"
	ctx := context.Background()
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	for _, region := range []string{"", "eu"} {
		if ref := fx.srv.seedRegionReady(ctx, region, false); ref != nil {
			t.Errorf("region %q: %+v", region, ref)
		}
	}
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "flat", false)
	want := "kass-1 in eu do not run instances nested yet. A seeded instance needs every box there nested."
	if ref := fx.srv.seedRegionReady(ctx, "eu", false); ref == nil || ref.Msg != want {
		t.Fatalf("region by name, a flat box: %+v", ref)
	}
}

func TestSeedsFloorDoorIsForSystemOperators(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-seeds-member")
	_, want := getRaw(t, fx.ts.URL+"/api/cli/system/no-such-door", member)
	if st, body := postRaw(t, fx.ts.URL+"/api/cli/system/seeds-floor", map[string]string{"min_release": "v0.27.0"}, member); st != http.StatusNotFound || body != want {
		t.Fatalf("a member = %d %q, want the bare 404", st, body)
	}
	owner := cliToken(t, fx, "paul@keymail.dev", "code-seeds-owner")
	for _, bad := range []string{"latest", "", "0.27.0"} {
		if st, out := postJSON(t, fx.ts.URL+"/api/cli/system/seeds-floor", map[string]any{"min_release": bad}, owner); st != http.StatusBadRequest {
			t.Fatalf("a bad release %q = %d %v, want 400", bad, st, out)
		}
	}
	if f, err := seeds.ReadFloor(context.Background(), fx.rel); err != nil || f != nil {
		t.Fatalf("a refused publish wrote a floor: %+v, %v", f, err)
	}
	if st, out := postJSON(t, fx.ts.URL+"/api/cli/system/seeds-floor", map[string]any{"min_release": "v0.27.0"}, owner); st != http.StatusOK || out["min_release"] != "v0.27.0" {
		t.Fatalf("publish = %d %v", st, out)
	}
	st, out := postJSON(t, fx.ts.URL+"/api/cli/system/seeds-floor", map[string]any{"min_release": "v0.26.0"}, owner)
	if st != http.StatusConflict || !strings.Contains(str(t, out, "error"), "already v0.27.0") {
		t.Fatalf("a lower floor = %d %v, want 409 naming v0.27.0", st, out)
	}
	f, err := seeds.ReadFloor(context.Background(), fx.rel)
	if err != nil || f == nil || f.MinRelease != "v0.27.0" || f.Min != seeds.RequiredProtocol {
		t.Fatalf("floor = %+v, %v", f, err)
	}
}

func TestPlan14SeedsFloorMixedFleet(t *testing.T) {
	fx := newConsoleFixture(t)
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	for _, extra := range []string{``, `,"moves_floor_state":"accepted"`, `,"moves_floor_state":"accepted","seeds_floor_state":"quarantined","seeds_floor_generation":1`, `,"moves_floor_state":"accepted","seeds_floor_state":"accepted","seeds_floor_generation":1`} {
		putSeedEdges(t, fx, "kass-2", `{"v":1,"box":"kass-2","eip":"1.2.3.4","hosts":{},"seeds_protocol":1,"seeds_floor_min_release":"v0.27.0","prev_release":"none"`+extra+`}`)
		putSeedHeartbeat(t, fx, "kass-2", "nested", false, time.Now())
		ref := fx.srv.seedRegionReady(t.Context(), "", false)
		accepted := extra == "" || strings.Contains(extra, `"seeds_floor_state":"accepted"`)
		if (ref == nil) != accepted {
			t.Fatal("mixed fleet", extra, ref)
		}
	}
}