package host

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// ---- fakes -----------------------------------------------------------------

// failingListStore breaks the LISTING of one prefix — the control-bucket
// blip that must never reach the converger as "no edges" or "no records".
// Prefix-scoped because the pass now lists three of them (control/edges/,
// accounts/, apps/) and each is a separate fail-closed direction.
type failingListStore struct {
	release.Store
	prefix string
	err    error
}

func (f failingListStore) List(ctx context.Context, prefix string) ([]string, error) {
	if prefix == f.prefix {
		return nil, f.err
	}
	return f.Store.List(ctx, prefix)
}

// ListDirs fails the same prefix: the record readers walk apps/ by its
// directories now (provision.listAppSubtree), so "the apps listing fails"
// is a ListDirs of apps/ failing as much as a List of it.
func (f failingListStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	if prefix == f.prefix {
		return nil, f.err
	}
	return f.Store.ListDirs(ctx, prefix)
}

// failingGetStore lists fine but refuses ONE named key — the half-read that
// would otherwise fold to a short desired set and delete the missing box's
// records.
type failingGetStore struct {
	release.Store
	bad string
	err error
}

func (f failingGetStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == f.bad {
		return nil, "", f.err
	}
	return f.Store.Get(ctx, key)
}

// convergeRecorder captures what the pass handed the converger, and whether
// it was called at all — the assertion every fail-closed test makes.
type convergeRecorder struct {
	calls  int
	routes []registry.Route
	edges  [][]byte
	err    error
}

func (c *convergeRecorder) converge(_ context.Context, routes []registry.Route, edges [][]byte) error {
	c.calls++
	c.routes = routes
	c.edges = edges
	return c.err
}

func steeringTestStore(t *testing.T, files map[string][]byte) release.Store {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	for key, body := range files {
		if _, err := s.Put(context.Background(), key, body, ""); err != nil {
			t.Fatal(err)
		}
	}
	return s
}

// seedAccount writes the account manifest ListAllRecords' known-account
// guard (provision.ParseRecordKey) requires, and returns its sqid — the
// frozen encoding of the id, via domains.Sqid, never a literal.
func seedAccount(t *testing.T, s release.Store, id int64, name string) string {
	t.Helper()
	sqid := domains.Sqid(id)
	body := []byte(`{"id":` + strconv.FormatInt(id, 10) + `,"sqid":"` + sqid +
		`","name":"` + name + `","created_at":"2026-08-02T09:00:00Z"}`)
	if _, err := s.Put(context.Background(), "accounts/"+sqid+"/account.json", body, ""); err != nil {
		t.Fatal(err)
	}
	return sqid
}

// seedRecord writes one instance record at its canonical key.
func seedRecord(t *testing.T, s release.Store, sqid, app, host, steering, pool string) {
	t.Helper()
	rec := provision.Record{
		Schema: 1, Host: host, Slug: app, App: app, Account: sqid,
		Pool: pool, Channel: "canary/main", Steering: steering,
	}
	body, err := json.Marshal(rec)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), provision.RecordKey(sqid, app, host), body, ""); err != nil {
		t.Fatal(err)
	}
}

// ---- the pass: what it reads, and failing closed (H2) ----------------------

// steeringWorld seeds a whole small deployment: two boxes' edges files, one
// account, and three records — one steered, one unsteered, one steered under
// a pool.
func steeringWorld(t *testing.T) (release.Store, string) {
	t.Helper()
	s := steeringTestStore(t, map[string][]byte{
		"control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`),
		"control/edges/oregon.json": []byte(`{"v":1,"box":"oregon"}`),
	})
	sqid := seedAccount(t, s, 12, "Tito")
	seedRecord(t, s, sqid, "app", "app.example.com", "latency", "")
	seedRecord(t, s, sqid, "app", "plain.example.com", "", "")
	return s, sqid
}

// The healthy pass: edges bodies in sorted listing order, and routes derived
// from the BUCKET RECORDS — not from any registry.
func TestSteeringPassReadsEdgesFilesAndRecords(t *testing.T) {
	quietLog(t)
	s, _ := steeringWorld(t)
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v", err)
	}
	if rec.calls != 1 {
		t.Fatalf("converge calls = %d, want 1", rec.calls)
	}
	if len(rec.edges) != 2 {
		t.Fatalf("edges = %d bodies, want 2", len(rec.edges))
	}
	// Listing order is sorted, so the fold is deterministic tick to tick.
	if !strings.Contains(string(rec.edges[0]), "dublin") || !strings.Contains(string(rec.edges[1]), "oregon") {
		t.Fatalf("edges = %q, want dublin then oregon (sorted)", rec.edges)
	}
	// Only the steered record becomes a route; the unsteered one is dropped
	// here because the fold would skip it silently anyway.
	if len(rec.routes) != 1 {
		t.Fatalf("routes = %+v, want only the steered record", rec.routes)
	}
	got := rec.routes[0]
	if got.Host != "app.example.com" || got.Steering != "latency" || got.Pool != "" {
		t.Fatalf("route = %+v, want the three fields the fold reads", got)
	}
}

// FINDING 1's pin (review, 2026-08-23). The steered host here is the
// HOME-BOXED class — an instance record with no Addr, which everywhereClass
// materializes on ONE box only, so it is absent from every other box's
// registry. It must still produce a record from whichever box holds the
// housekeeper lease, because the lease moves and DNS must not.
//
// The proof is structural as much as behavioural: steeringPass takes no
// registry at all, so there is no local view for the answer to depend on.
// Before the fix this test could not even have been written — the pass would
// have needed a registry, and the host would have been missing from it.
func TestSteeringPassSteersHomeBoxedRecordWithNoLocalRegistry(t *testing.T) {
	quietLog(t)
	s := steeringTestStore(t, map[string][]byte{
		"control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`),
	})
	sqid := seedAccount(t, s, 12, "Tito")
	// No Addr, no Unit, default hibernation: home-boxed, exactly the class
	// the rollout's stage-5 checklist steers.
	seedRecord(t, s, sqid, "exec-app", "home.example.com", "latency", "")
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v", err)
	}
	if len(rec.routes) != 1 || rec.routes[0].Host != "home.example.com" {
		t.Fatalf("routes = %+v, want the home-boxed steered host — a lease migration must not withdraw it", rec.routes)
	}
}

// The loader drops Steering=="" (the fold skips those silently anyway) but
// must NOT become a second, quieter policy gate: an invalid value, a
// wildcard host and a pool-scoped row all go DOWN to the fold, whose loud
// documented refusals are the operator's only signal that they exist.
func TestSteeringPassPassesNonLatencySteeringValuesDown(t *testing.T) {
	quietLog(t)
	s := steeringTestStore(t, map[string][]byte{
		"control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`),
	})
	sqid := seedAccount(t, s, 12, "Tito")
	seedRecord(t, s, sqid, "app", "weird.example.com", "weighted", "")
	seedRecord(t, s, sqid, "app", "pooled.example.com", "latency", "eu")
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v", err)
	}
	if len(rec.routes) != 2 {
		t.Fatalf("routes = %+v, want both rows delivered for the fold to refuse loudly", rec.routes)
	}
}

// H2, direction one: a broken edges LISTING must not converge.
func TestSteeringPassFailsClosedOnEdgesListError(t *testing.T) {
	quietLog(t)
	boom := errors.New("bucket down")
	base, _ := steeringWorld(t)
	s := failingListStore{Store: base, prefix: "control/edges/", err: boom}
	rec := &convergeRecorder{}

	err := steeringPass(context.Background(), s, rec.converge)
	if err == nil {
		t.Fatal("steeringPass returned nil on a failed edges listing")
	}
	if !errors.Is(err, boom) {
		t.Fatalf("err = %v, want it to wrap the store error", err)
	}
	if rec.calls != 0 {
		t.Fatalf("converge ran %d times after a failed listing, want 0 — never converge on a partial fold", rec.calls)
	}
}

// H2, direction two, and the one the shrink guard was written for: the
// listing succeeded, so the pass KNOWS how many boxes there are — and one of
// their files would not come back.
func TestSteeringPassFailsClosedOnEdgesGetError(t *testing.T) {
	quietLog(t)
	boom := errors.New("object unreadable")
	base, _ := steeringWorld(t)
	s := failingGetStore{Store: base, bad: "control/edges/oregon.json", err: boom}
	rec := &convergeRecorder{}

	err := steeringPass(context.Background(), s, rec.converge)
	if err == nil {
		t.Fatal("steeringPass returned nil with one edges file unreadable")
	}
	if !errors.Is(err, boom) {
		t.Fatalf("err = %v, want it to wrap the store error", err)
	}
	if !strings.Contains(err.Error(), "control/edges/oregon.json") {
		t.Fatalf("err = %v, want the unreadable key named", err)
	}
	if rec.calls != 0 {
		t.Fatalf("converge ran %d times after a failed get, want 0", rec.calls)
	}
}

// H2, direction three, re-pointed at the RECORDS read (FINDING 1): the
// records are now the routes half of the fold, so every way that read can
// come back short is a way a live steered host gets deleted. Four of them,
// each fatal.
func TestSteeringPassFailsClosedOnRecordsRead(t *testing.T) {
	boom := errors.New("bucket down")
	cases := []struct {
		name  string
		wrap  func(release.Store) release.Store
		named string
	}{
		{
			// Without accounts/, ParseRecordKey's known-account guard rejects
			// EVERY record key — the fold would see no steered hosts at all.
			name: "accounts listing fails",
			wrap: func(s release.Store) release.Store {
				return failingListStore{Store: s, prefix: "accounts/", err: boom}
			},
		},
		{
			// The subtle one: ONE account's manifest is unreadable, so only
			// THAT account's records vanish. Silent, partial, and exactly the
			// shape that deletes one tenant's DNS.
			name: "an account manifest is unreadable",
			wrap: func(s release.Store) release.Store {
				return failingGetStore{Store: s, bad: "accounts/" + domains.Sqid(12) + "/account.json", err: boom}
			},
		},
		{
			name: "apps listing fails",
			wrap: func(s release.Store) release.Store {
				return failingListStore{Store: s, prefix: "apps/", err: boom}
			},
		},
		{
			name: "a record object will not Get",
			wrap: func(s release.Store) release.Store {
				return failingGetStore{Store: s, bad: provision.RecordKey(domains.Sqid(12), "app", "app.example.com"), err: boom}
			},
			named: "app.example.com",
		},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			quietLog(t)
			base, _ := steeringWorld(t)
			rec := &convergeRecorder{}

			err := steeringPass(context.Background(), c.wrap(base), rec.converge)
			if err == nil {
				t.Fatalf("steeringPass returned nil when %s", c.name)
			}
			if rec.calls != 0 {
				t.Fatalf("converge ran %d times, want 0 — a short records read must never converge", rec.calls)
			}
			if c.named != "" && !strings.Contains(err.Error(), c.named) {
				t.Fatalf("err = %v, want the unreadable record named", err)
			}
		})
	}
}

// The content-vs-visibility split, records half: a record that READS but does
// not decode is skipped and logged, and the pass converges on the rest. One
// malformed body must not blind the deployment to every other steered host.
func TestSteeringPassSkipsUnparseableRecord(t *testing.T) {
	logs := captureLog(t)
	s, sqid := steeringWorld(t)
	if _, err := s.Put(context.Background(), provision.RecordKey(sqid, "app", "broken.example.com"), []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v — an unparseable record is not a read failure", err)
	}
	if rec.calls != 1 {
		t.Fatalf("converge calls = %d, want 1", rec.calls)
	}
	if len(rec.routes) != 1 || rec.routes[0].Host != "app.example.com" {
		t.Fatalf("routes = %+v, want the healthy steered record still folded", rec.routes)
	}
	if !strings.Contains(logs(), "broken.example.com") {
		t.Fatalf("log = %q, want the malformed record named", logs())
	}
}

// The same split, edges half: the host loader must NOT pre-judge content —
// a body that will not parse is delivered verbatim and the fold decides.
func TestSteeringPassPassesUnparseableBodiesDown(t *testing.T) {
	quietLog(t)
	s := steeringTestStore(t, map[string][]byte{
		"control/edges/dublin.json": []byte("{not json"),
	})
	seedAccount(t, s, 12, "Tito")
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v — an unparseable body is not a read failure", err)
	}
	if rec.calls != 1 {
		t.Fatalf("converge calls = %d, want 1 — parse failures are the fold's business", rec.calls)
	}
	if len(rec.edges) != 1 {
		t.Fatalf("edges = %d bodies, want the unparseable one delivered verbatim", len(rec.edges))
	}
}

// Objects under the edges prefix that this deployment's edges loop did not
// write are not read at all — the same key-shape filter internal/edge's
// homeMap applies. A stray object must not be able to fail the pass forever.
func TestSteeringPassIgnoresForeignKeysUnderThePrefix(t *testing.T) {
	quietLog(t)
	s := steeringTestStore(t, map[string][]byte{
		"control/edges/dublin.json":       []byte(`{"v":1,"box":"dublin"}`),
		"control/edges/README.txt":        []byte("hello"),
		"control/edges/nested/inner.json": []byte(`{"v":1,"box":"nested"}`),
	})
	seedAccount(t, s, 12, "Tito")
	rec := &convergeRecorder{}

	if err := steeringPass(context.Background(), s, rec.converge); err != nil {
		t.Fatalf("steeringPass: %v", err)
	}
	if len(rec.edges) != 1 {
		t.Fatalf("edges = %d bodies, want only control/edges/<box>.json", len(rec.edges))
	}
}

// ---- the duty: dormancy and the housekeeper gate ---------------------------

// The dormancy proof: with no zone id the duty returns nil having called
// NOTHING — the pass closure holds every read and every AWS call, so "the
// pass never ran" is "zero reads, zero API calls" structurally.
func TestSteeringDutyDormantDoesNoWork(t *testing.T) {
	quietLog(t)
	ran := 0
	duty := steeringDuty("", func(context.Context) error { ran++; return nil })

	if err := duty.run(context.Background()); err != nil {
		t.Fatalf("dormant duty returned %v, want nil", err)
	}
	if ran != 0 {
		t.Fatalf("dormant duty ran the pass %d times, want 0", ran)
	}
}

// FINDING 3 (review, 2026-08-23): totality on a production DNS write path.
// A zone id with no pass is unreachable today — steeringLoop leaves pass nil
// in exactly the branch that leaves zoneID empty — but if that ever stopped
// being true, the duty must decline rather than panic the agent's goroutine
// on a nil call.
func TestSteeringDutyWithNoPassDeclines(t *testing.T) {
	quietLog(t)
	duty := steeringDuty("Z0ABCDEF1234", nil)
	if err := duty.run(context.Background()); err != nil {
		t.Fatalf("duty with a nil pass returned %v, want nil (and no panic)", err)
	}
}

func TestSteeringDutyRunsThePassWhenZoneSet(t *testing.T) {
	quietLog(t)
	ran := 0
	boom := errors.New("converge failed")
	duty := steeringDuty("Z0ABCDEF1234", func(context.Context) error { ran++; return boom })

	if err := duty.run(context.Background()); !errors.Is(err, boom) {
		t.Fatalf("duty error = %v, want the pass's error propagated to the housekeeper log", err)
	}
	if ran != 1 {
		t.Fatalf("pass ran %d times, want 1", ran)
	}
}

// The housekeeper contract: one box converges the deployment's zone, not
// every box. A non-holder does nothing and says nothing.
func TestSteeringDutyRunsOnlyOnTheHousekeeper(t *testing.T) {
	quietLog(t)
	ran := 0
	duty := steeringDuty("Z0ABCDEF1234", func(context.Context) error { ran++; return nil })

	runHousekeeperDuties(noGate, duty)
	if ran != 0 {
		t.Fatalf("pass ran %d times on a non-holder, want 0", ran)
	}
	runHousekeeperDuties(heldGate, duty)
	if ran != 1 {
		t.Fatalf("pass ran %d times on the holder, want 1", ran)
	}
}

// ---- the loop: arming ------------------------------------------------------

// Mirrors TestMintMeshKeyLoopIsAPlatformBoxJob: a fleet box must skip before
// ever opening a store.
func TestSteeringLoopIsAPlatformBoxJob(t *testing.T) {
	const skipLine = "agent: steering loop is a platform-box job — skipped on a fleet box"

	logs := captureLog(t)
	door := newDoorFake(t)
	for k, v := range fleetDoorEnv(door.url) {
		t.Setenv(k, v)
	}
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", "")

	steeringLoop(nil) // must return; a fleet box never reaches the loop

	if got := logs(); !strings.Contains(got, skipLine) {
		t.Fatalf("log = %q, want the skip line verbatim", got)
	}
	if hits := door.seen(); len(hits) != 0 {
		t.Fatalf("door saw %+v, want zero — the skip must land BEFORE the store opens", hits)
	}
}

// Arming independence, the mesh-key loop's own regression guard applied here:
// a box missing CARLOS_RELEASE_PUBKEY (retentionLoop's gate, unrelated to
// steering) still WINS the housekeeper election, so a steering duty gated on
// that var would silently never converge fleet-wide.
func TestSteeringLoopArmsWithoutReleasePubkey(t *testing.T) {
	t.Setenv("CARLOS_FLEET", "")
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
	t.Setenv("CARLOS_RELEASE_PUBKEY", "") // deliberately unset — the whole point
	t.Setenv("CARLOS_STEERING_ZONE_ID", "")

	buf := captureRacyLog(t)
	go steeringLoop(nil)

	waitFor(t, "the steering loop to arm despite no CARLOS_RELEASE_PUBKEY", func() bool {
		return strings.Contains(buf.String(), "agent: steering loop armed")
	})
}

// Today's boxes: the env is unset, so the loop arms and says so ONCE, and its
// duty is the dormant one — no store reads, no registry reads, no AWS client
// ever constructed.
func TestSteeringLoopDormantWhenZoneUnset(t *testing.T) {
	t.Setenv("CARLOS_FLEET", "")
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
	t.Setenv("CARLOS_STEERING_ZONE_ID", "")

	buf := captureRacyLog(t)
	go steeringLoop(nil)

	waitFor(t, "the dormant line", func() bool {
		return strings.Contains(buf.String(), "steering: zone unset — dormant")
	})
	if n := strings.Count(buf.String(), "steering: zone unset — dormant"); n != 1 {
		t.Fatalf("dormant line logged %d times, want exactly 1 (at arm)", n)
	}
}

// A malformed zone id is dormancy too, not a crash loop: SteeringZoneID logs
// it loudly and returns "", and the loop arms dormant on that.
func TestSteeringLoopDormantWhenZoneMalformed(t *testing.T) {
	t.Setenv("CARLOS_FLEET", "")
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
	t.Setenv("CARLOS_STEERING_ZONE_ID", "/hostedzone/Z0ABCDEF1234")

	buf := captureRacyLog(t)
	go steeringLoop(nil)

	waitFor(t, "the dormant line for a malformed zone id", func() bool {
		return strings.Contains(buf.String(), "steering: zone unset — dormant")
	})
	if !strings.Contains(buf.String(), "is not a bare hosted-zone id") {
		t.Fatalf("log = %q, want the malformed id named loudly", buf.String())
	}
}