package host

import (
	"context"
	"errors"
	"path/filepath"
	"reflect"
	"sort"
	"testing"
	"time"

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

func report(label string, at time.Time, pins map[string]status.BuildStatePins) status.Box {
	return status.Box{Label: label, GeneratedAt: at, Metrics: &status.BoxMetrics{BuildStates: pins}}
}

func kmPins(builds ...string) status.BuildStatePins {
	return status.BuildStatePins{Account: "bab", App: "km", Builds: builds, Complete: true}
}

// Astra I1 #6: a host whose build state its box could not read stops
// pruning for the app, even in a fresh, readable report.
func TestAnIncompleteBuildStatePrunesNothing(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	inc := kmPins()
	inc.Complete = false
	boxes := []status.Box{report("uk-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": inc})}
	if _, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("an unreadable build state did not stop pruning")
	}
}

func TestAnUnreadBuildStateIsReportedIncomplete(t *testing.T) {
	_, _, reg := testBuildStates(t) // a.example.com, never refreshed: copy unknown
	pins, err := buildStatePins(reg, false)
	if p, ok := pins["a.example.com"]; err != nil || !ok || p.Complete {
		t.Fatalf("report = %+v, %v, %v; want the host listed and incomplete", p, ok, err)
	}
	if err := reg.SetBuildState("a.example.com", release.BuildStateCopy{Known: true, ReadErr: "timeout"}.Column()); err != nil {
		t.Fatal(err)
	}
	if pins, _ := buildStatePins(reg, false); pins["a.example.com"].Complete {
		t.Fatal("a failed refresh was reported complete")
	}
	if err := reg.SetBuildState("a.example.com", "{"); err != nil {
		t.Fatal(err)
	}
	if pins, _ := buildStatePins(reg, false); pins["a.example.com"].Complete {
		t.Fatal("an unparseable copy was reported complete")
	}
}

func recordsStore(t *testing.T, recs ...provision.Record) release.Store {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	for _, r := range recs {
		if err := provision.CreateRecord(context.Background(), s, r); err != nil {
			t.Fatal(err)
		}
	}
	return s
}

var now0 = time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)

// Spec rev 11 §4.4: pins from records, build-state builds from box reports.
func TestPinsAndReportedBuildStatesAreProtected(t *testing.T) {
	s := recordsStore(t,
		provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable", Pin: "v1"},
		provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable", Region: "uk"})
	boxes := []status.Box{report("uk-1", now0.Add(-time.Hour), map[string]status.BuildStatePins{
		"a.example.com": kmPins(), "b.example.com": kmPins("v2", "v3")})}
	got, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km")
	if err != nil {
		t.Fatal(err)
	}
	for _, v := range []string{"v1", "v2", "v3"} {
		if !got[v] {
			t.Errorf("%s is not protected: %v", v, got)
		}
	}
}

func TestABoxSilentForSixHoursPrunesNothing(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	boxes := []status.Box{report("uk-1", now0.Add(-6*time.Hour-time.Minute), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})}
	if _, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a box hosting the app that has not reported in 6h did not stop pruning")
	}
}

func TestANewerReportForTheHostWins(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	boxes := []status.Box{
		report("uk-1", now0.Add(-48*time.Hour), map[string]status.BuildStatePins{"b.example.com": kmPins("v1")}), // the box it moved off
		report("ie-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")}),
	}
	got, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km")
	if err != nil || !got["v2"] {
		t.Fatalf("got %v, %v; want v2 from the newest report and no refusal", got, err)
	}
	// §4.4 protects every build named in a report: the older report's v1 is
	// kept too, which can only keep more.
	if !got["v1"] {
		t.Errorf("v1, named by the older report, is not protected: %v", got)
	}
}

// The newest report decides completeness too: an older box's incomplete
// entry for a host that has since moved does not block, and a newer
// incomplete one does even when an older one was complete.
func TestTheNewestReportDecidesCompleteness(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	inc := kmPins()
	inc.Complete = false
	moved := []status.Box{
		report("uk-1", now0.Add(-2*time.Hour), map[string]status.BuildStatePins{"b.example.com": inc}),
		report("ie-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")}),
	}
	if _, err := settingsProtected(context.Background(), s, moved, true, now0, "bab", "km"); err != nil {
		t.Fatalf("an older box's incomplete entry blocked: %v", err)
	}
	reversed := []status.Box{
		report("uk-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": inc}),
		report("ie-1", now0.Add(-2*time.Hour), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")}),
	}
	if _, err := settingsProtected(context.Background(), s, reversed, true, now0, "bab", "km"); err == nil {
		t.Fatal("the newest report's incomplete entry did not stop pruning")
	}
}

func TestOtherAppsReportsDoNotBlock(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable"})
	boxes := []status.Box{
		report("uk-1", now0.Add(-72*time.Hour), map[string]status.BuildStatePins{
			"x.example.com": {Account: "bab", App: "blog", Builds: []string{"b1"}}}),
		report("ie-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"a.example.com": kmPins("v1")}),
	}
	if _, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km"); err != nil {
		t.Fatalf("another app's silent box blocked km: %v", err)
	}
}

// Astra I2 #5: every provisioned instance of the app needs an entry. A
// fresh, readable report that simply omits the host (its map left out)
// prunes nothing, not "nothing to protect".
func TestAnOmittedHostPrunesNothing(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	boxes := []status.Box{report("uk-1", now0.Add(-time.Minute), nil)}
	if _, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a host no report names was treated as having no builds to protect")
	}
	// A report naming the host under another account is not this app's.
	other := []status.Box{report("uk-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{
		"b.example.com": {Account: "zzz", App: "km", Builds: []string{"v1"}, Complete: true}})}
	if _, err := settingsProtected(context.Background(), s, other, true, now0, "bab", "km"); err == nil {
		t.Fatal("another account's entry for the host counted as this app's")
	}
}

// Only a record that keeps a build state needs a report: a unit-backed or
// always-on instance, or a static route, has none to report.
func TestRecordsWithNoBuildStateNeedNoReport(t *testing.T) {
	off := false
	s := recordsStore(t,
		provision.Record{Schema: 1, Host: "u.example.com", App: "km", Account: "bab", Channel: "stable", Unit: "km.service"},
		provision.Record{Schema: 1, Host: "o.example.com", App: "km", Account: "bab", Channel: "stable", Hibernate: &off, Pin: "v7"},
		provision.Record{Schema: 1, Host: "s.example.com", App: "km", Account: "bab", Channel: "stable", Kind: "static"})
	got, err := settingsProtected(context.Background(), s, nil, true, now0, "bab", "km")
	if err != nil {
		t.Fatalf("records that keep no build state blocked: %v", err)
	}
	if !got["v7"] {
		t.Errorf("the pin of a record that keeps no build state is not protected: %v", got)
	}
}

// Astra I2 #5: a box that could not list its routes says so, and a fresh
// report saying so stops pruning even when an older report from another
// box still names the host (it may have moved onto the failing box).
func TestACollectionFailurePrunesNothing(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"})
	failed := report("uk-1", now0.Add(-time.Minute), nil)
	failed.Metrics.BuildStatesErr = "registry: database is locked"
	boxes := []status.Box{
		report("ie-1", now0.Add(-2*time.Hour), map[string]status.BuildStatePins{"b.example.com": kmPins("v1")}),
		failed,
	}
	if _, err := settingsProtected(context.Background(), s, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a box that could not collect its build states did not stop pruning")
	}
}

// registry.List opens its own connection, so a closed registry still lists:
// the failing list is a fake.
func TestTheCollectorReportsItsFailure(t *testing.T) {
	reg := fakeLister{err: errors.New("database is locked")}
	if pins, err := buildStatePins(reg, false); err == nil || pins != nil {
		t.Fatalf("a failed registry list returned %v, %v; want an error", pins, err)
	}
}

func TestAnUnreadableRecordOrReportPrunesNothing(t *testing.T) {
	s := recordsStore(t, provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable"})
	if _, err := s.Put(context.Background(), provision.RecordKey("bab", "km", "z.example.com"), []byte("{"), ""); err != nil {
		t.Fatal(err)
	}
	if _, err := settingsProtected(context.Background(), s, nil, true, now0, "bab", "km"); err == nil {
		t.Fatal("an unreadable record did not stop pruning")
	}
	clean := recordsStore(t, provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable"})
	if _, err := settingsProtected(context.Background(), clean, nil, false, now0, "bab", "km"); err == nil {
		t.Fatal("an incomplete read of box reports did not stop pruning")
	}
}

// Ruling R-T9-STORE: provision.ListRecords(..., 0) reports truncated whenever
// a key exists, and it skips a record it cannot read. Every record is read,
// and any failure (a record whose body names another host, a failed Get, a
// failed listing) prunes nothing.
func TestEveryRecordIsReadAndAnyFailurePrunesNothing(t *testing.T) {
	ctx := context.Background()
	recs := []provision.Record{
		{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable", Pin: "v1"},
		{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable", Pin: "v2"},
		{Schema: 1, Host: "c.example.com", App: "km", Account: "bab", Channel: "stable", Pin: "v3"},
	}
	all := map[string]status.BuildStatePins{"a.example.com": kmPins(), "b.example.com": kmPins(), "c.example.com": kmPins()}
	boxes := []status.Box{report("uk-1", now0.Add(-time.Minute), all)}
	s := recordsStore(t, recs...)
	got, err := settingsProtected(ctx, s, boxes, true, now0, "bab", "km")
	if err != nil {
		t.Fatalf("three good records refused: %v", err)
	}
	if !got["v1"] || !got["v2"] || !got["v3"] {
		t.Fatalf("not every record's pin is protected: %v", got)
	}

	// A body that disagrees with its own path.
	moved := recordsStore(t, recs...)
	raw, _, err := moved.Get(ctx, provision.RecordKey("bab", "km", "a.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := moved.Put(ctx, provision.RecordKey("bab", "km", "d.example.com"), raw, ""); err != nil {
		t.Fatal(err)
	}
	if _, err := settingsProtected(ctx, moved, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a record that disagrees with its path did not stop pruning")
	}

	// A Get that fails.
	getFails := failingGetStore{Store: recordsStore(t, recs...), bad: provision.RecordKey("bab", "km", "b.example.com"), err: errors.New("SlowDown")}
	if _, err := settingsProtected(ctx, getFails, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a record that could not be read did not stop pruning")
	}

	// A listing that fails.
	listFails := failingListStore{Store: recordsStore(t, recs...), prefix: provision.RecordsPrefix("bab", "km"), err: errors.New("AccessDenied")}
	if _, err := settingsProtected(ctx, listFails, boxes, true, now0, "bab", "km"); err == nil {
		t.Fatal("a failed record listing did not stop pruning")
	}
}

func TestTheBoxReportsItsBuildStates(t *testing.T) {
	_, _, reg := testBuildStates(t)
	if err := reg.SetBuildState("a.example.com", release.BuildStateCopy{Known: true, Replica: true,
		State: &release.BuildState{LastExecuted: "v1", Resolved: &release.ResolvedRequest{ID: 2, Target: "v2"}}}.Column()); err != nil {
		t.Fatal(err)
	}
	pins, err := buildStatePins(reg, false)
	if err != nil {
		t.Fatal(err)
	}
	got := pins["a.example.com"]
	// testBuildStates' row has no account: RouteAccount's fallback is ops,
	// whose sqid is "bab", the account the housekeeper keys the app by.
	if got.Account != "bab" || got.App != "km" || !got.Complete || len(got.Builds) != 2 || got.Builds[0] != "v1" || got.Builds[1] != "v2" {
		t.Fatalf("report = %+v", got)
	}
}

// Task 4's review: the report names exactly the rows keepsBuildState
// selects, the rows refreshAll keeps a copy of. A unit-backed hibernating
// instance is never refreshed, so listing it would read as incomplete
// forever; an exec-backed worker is refreshed with host claims off, and
// never with them on.
func TestTheReportNamesExactlyTheRowsThatKeepABuildState(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	rows := []registry.Route{
		{Host: "a.example.com", Kind: "instance", App: "km", Account: "bab", Hibernate: true, Socket: "/run/carlos/a.sock"},
		{Host: "u.example.com", Kind: "instance", App: "km", Account: "bab", Hibernate: true, Socket: "/run/carlos/u.sock", Unit: "km.service"},
		{Host: "w.example.com", Kind: "sidecar", App: "km", Account: "bab", Socket: "/run/carlos/w.sock"},
		{Host: "t.example.com", Kind: "instance", App: "km", Account: "bab", Hibernate: true, Socket: "/run/carlos/t.sock", Tunnel: "fleet-1"},
		{Host: "r.example.com", Kind: "instance", App: "km", Account: "bab", Socket: "/run/carlos/r.sock"},
		{Host: "p.example.com", Kind: "instance", App: "edge", Hibernate: true, Socket: "/run/carlos/p.sock", Platform: true},
	}
	for _, rt := range rows {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	routes, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	for _, claims := range []bool{false, true} {
		pins, err := buildStatePins(reg, claims)
		if err != nil {
			t.Fatal(err)
		}
		var got, want []string
		for h := range pins {
			got = append(got, h)
		}
		for _, rt := range routes {
			if keepsBuildState(rt, claims) {
				want = append(want, rt.Host)
			}
		}
		sort.Strings(got)
		sort.Strings(want)
		if !reflect.DeepEqual(got, want) {
			t.Errorf("claims %v: report names %v, keepsBuildState selects %v", claims, got, want)
		}
		wantHosts := []string{"a.example.com", "w.example.com"}
		if claims {
			wantHosts = []string{"a.example.com"}
		}
		if !reflect.DeepEqual(got, wantHosts) {
			t.Errorf("claims %v: report names %v, want %v", claims, got, wantHosts)
		}
	}
}

// The reports are the deployment's and every fleet's, and load says when any
// could not be read.
func TestLoadAllBoxesReadsEveryFleet(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	if err := status.WriteBox(ctx, s, report("uk-1", now0, map[string]status.BuildStatePins{"a.example.com": kmPins("v1")})); err != nil {
		t.Fatal(err)
	}
	if err := status.WriteBox(ctx, release.NewPrefixStore(s, "fleets/acme/"), report("acme-1", now0, map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})); err != nil {
		t.Fatal(err)
	}
	boxes, complete, err := (&boxReports{status: s}).load(ctx)
	if err != nil || !complete {
		t.Fatalf("load = %v, %v", complete, err)
	}
	var labels []string
	for _, b := range boxes {
		labels = append(labels, b.Label)
	}
	sort.Strings(labels)
	if !reflect.DeepEqual(labels, []string{"acme-1", "uk-1"}) {
		t.Fatalf("read %v, want both the deployment's box and the fleet's", labels)
	}

	if _, err := s.Put(ctx, "fleets/beta/status/boxes/beta-1.json", []byte("{"), ""); err != nil {
		t.Fatal(err)
	}
	if _, complete, err := (&boxReports{status: s}).load(ctx); err != nil || complete {
		t.Fatalf("a corrupt fleet report read as complete (%v, %v)", complete, err)
	}

	// R-T9-FLEETS: a failed fleets listing is an incomplete read.
	fleetsFail := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "fleets/", err: errors.New("AccessDenied")}
	if _, complete, err := (&boxReports{status: fleetsFail}).load(ctx); err != nil || complete {
		t.Fatalf("a failed fleets listing read as complete (%v, %v)", complete, err)
	}
	rootFail := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "status/boxes/", err: errors.New("AccessDenied")}
	if _, _, err := (&boxReports{status: rootFail}).load(ctx); err == nil {
		t.Fatal("a failed listing of the platform's reports was not an error")
	}
	if _, _, err := (&boxReports{}).load(ctx); err == nil {
		t.Fatal("no report source at all was not an error")
	}
}

// retentionFixture is TestRetentionPassPrunesEligibleAndProtectsBoxPins'
// shape with 27 versions and keep 20, so v01..v07 are the candidates, and
// a provisioned, hibernating kass.example.com on this box at v07 (previous
// v06, good v05) whose record pins v04.
func retentionFixture(t *testing.T) (context.Context, *release.DirStore, *registry.Registry, func(string) bool) {
	t.Helper()
	ctx := context.Background()
	dir := t.TempDir()
	s := release.NewDirStore(dir)
	now := time.Now()
	base := now.Add(-30 * 24 * time.Hour)
	vname := func(i int) string { return "v" + string(rune('0'+i/10)) + string(rune('0'+i%10)) }
	for i := 1; i <= 27; i++ {
		stampKey(t, s, dir, release.ManifestKey("bab", "kass", vname(i)), base.Add(time.Duration(i)*time.Hour))
		stampKey(t, s, dir, release.ArtifactKey("bab", "kass", vname(i), "linux-arm64"), base.Add(time.Duration(i)*time.Hour))
	}
	if err := release.SetRetention(ctx, s, "bab", "kass", 20, "test", now); err != nil {
		t.Fatal(err)
	}
	if err := provision.CreateRecord(ctx, s, provision.Record{Schema: 1, Host: "kass.example.com", App: "kass", Account: "bab", Channel: "stable", Pin: "v04"}); err != nil {
		t.Fatal(err)
	}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	if err := reg.Upsert(registry.Route{
		Host: "kass.example.com", Kind: "instance", App: "kass", Account: "bab", Hibernate: true, Socket: "/run/carlos/kass.sock",
		Version: "v07", PrevVersion: "v06", GoodVersion: "v05", Status: "ready",
	}); err != nil {
		t.Fatal(err)
	}
	present := func(v string) bool {
		_, _, err := s.Get(ctx, release.ManifestKey("bab", "kass", v))
		return err == nil
	}
	return ctx, s, reg, present
}

// The pass itself: this box's Version, PrevVersion and GoodVersion (zdd-5
// obligation 9), the record's pin, and the reported last_executed and
// resolved T are all kept; only what nothing names is pruned.
func TestRetentionPassProtectsPinsGoodVersionAndReportedBuilds(t *testing.T) {
	t.Parallel()
	ctx, s, reg, present := retentionFixture(t)
	_, pub := retentionKeys(t)
	reports := release.NewDirStore(t.TempDir())
	if err := status.WriteBox(ctx, reports, report("uk-1", time.Now().Add(-time.Minute), map[string]status.BuildStatePins{
		"kass.example.com": {Account: "bab", App: "kass", Builds: []string{"v03", "v02"}, Complete: true}})); err != nil {
		t.Fatal(err)
	}
	if err := retentionPass(ctx, s, &boxReports{status: reports}, pub, reg); err != nil {
		t.Fatal(err)
	}
	if present("v01") {
		t.Error("v01 survived; nothing names it")
	}
	for _, v := range []string{"v02", "v03", "v04", "v05", "v06", "v07"} {
		if !present(v) {
			t.Errorf("%s was pruned", v)
		}
	}
}

// With no report naming the provisioned instance, or no readable reports at
// all, the pass prunes nothing for the app.
func TestRetentionPassPrunesNothingWithoutTheReports(t *testing.T) {
	t.Parallel()
	t.Run("no report names the host", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		if err := retentionPass(ctx, s, &boxReports{status: release.NewDirStore(t.TempDir())}, pub, reg); err != nil {
			t.Fatal(err)
		}
		if !present("v01") {
			t.Fatal("v01 was pruned with no report naming kass.example.com's build state")
		}
	})
	t.Run("the reports cannot be listed", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		reports := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "status/boxes/", err: errors.New("AccessDenied")}
		_ = retentionPass(ctx, s, &boxReports{status: reports}, pub, reg)
		if !present("v01") {
			t.Fatal("v01 was pruned with the box reports unreadable")
		}
	})
	t.Run("no report source", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		_ = retentionPass(ctx, s, nil, pub, reg)
		_ = retentionPass(ctx, s, &boxReports{}, pub, reg)
		if !present("v01") {
			t.Fatal("v01 was pruned with no report source")
		}
	})
}

// kassPins is retentionFixture's host as this box's own collector reports it.
func kassPins(builds ...string) func() (map[string]status.BuildStatePins, error) {
	return func() (map[string]status.BuildStatePins, error) {
		return map[string]status.BuildStatePins{"kass.example.com": {Account: "bab", App: "kass", Builds: builds, Complete: true}}, nil
	}
}

// Ruling R-T9-SOLE: a directory-backed deployment is one box by
// construction and publishes no status. Its pass takes this box's own
// collector as a fresh report: what it names is protected, and a collector
// that fails, or that omits a provisioned host, prunes nothing.
func TestSoleModeReadsThisBoxsOwnReport(t *testing.T) {
	t.Parallel()
	t.Run("protects what this box names", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		if err := retentionPass(ctx, s, &boxReports{deploy: s, local: kassPins("v03", "v02")}, pub, reg); err != nil {
			t.Fatal(err)
		}
		if present("v01") {
			t.Error("v01 survived; nothing names it")
		}
		for _, v := range []string{"v02", "v03", "v04", "v05", "v06", "v07"} {
			if !present(v) {
				t.Errorf("%s was pruned", v)
			}
		}
	})
	t.Run("a failed collection", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		failing := func() (map[string]status.BuildStatePins, error) { return nil, errors.New("database is locked") }
		boxes, _, err := (&boxReports{local: failing}).load(ctx)
		if err != nil || len(boxes) != 1 || boxes[0].Metrics == nil || boxes[0].Metrics.BuildStatesErr != "database is locked" {
			t.Fatalf("load = %+v, %v; want this box's report saying it could not collect", boxes, err)
		}
		_ = retentionPass(ctx, s, &boxReports{deploy: s, local: failing}, pub, reg)
		if !present("v01") {
			t.Fatal("v01 was pruned while this box could not collect its build states")
		}
	})
	t.Run("the host is not named", func(t *testing.T) {
		ctx, s, reg, present := retentionFixture(t)
		_, pub := retentionKeys(t)
		none := func() (map[string]status.BuildStatePins, error) { return nil, nil }
		_ = retentionPass(ctx, s, &boxReports{deploy: s, local: none}, pub, reg)
		if !present("v01") {
			t.Fatal("v01 was pruned with kass.example.com named by no report")
		}
	})
	t.Run("fleets come from the deployment store", func(t *testing.T) {
		ctx := context.Background()
		deploy := release.NewDirStore(t.TempDir())
		if err := status.WriteBox(ctx, release.NewPrefixStore(deploy, "fleets/acme/"), report("acme-1", now0, map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})); err != nil {
			t.Fatal(err)
		}
		boxes, complete, err := (&boxReports{deploy: deploy, local: kassPins("v03"), now: func() time.Time { return now0 }}).load(ctx)
		if err != nil || !complete || len(boxes) != 2 {
			t.Fatalf("load = %+v, %v, %v; want this box and the fleet's", boxes, complete, err)
		}
		var local status.Box
		for _, b := range boxes {
			if b.Label != "acme-1" {
				local = b
			}
		}
		if !local.GeneratedAt.Equal(now0) || local.Metrics == nil || local.Metrics.BuildStates["kass.example.com"].App != "kass" {
			t.Fatalf("this box's report = %+v, want it fresh and naming kass.example.com", local)
		}
	})
}

// Ruling R-T9-FLEETS: fleet reports are read from the status store and from
// the deployment store (a fleet box that publishes through the store door
// lands in the deployment bucket), and the union is used. A report found in
// both is read twice, which the newest-per-host rule tolerates. A failed
// listing or read in either store is an incomplete read.
func TestFleetReportsComeFromBothStores(t *testing.T) {
	ctx := context.Background()
	st, dep := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
	both := report("both-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"a.example.com": kmPins("v1")})
	for _, w := range []struct {
		s release.Store
		b status.Box
	}{
		{release.NewPrefixStore(st, "fleets/acme/"), report("acme-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})},
		{release.NewPrefixStore(dep, "fleets/door/"), report("door-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"c.example.com": kmPins("v3")})},
		{release.NewPrefixStore(st, "fleets/same/"), both},
		{release.NewPrefixStore(dep, "fleets/same/"), both},
	} {
		if err := status.WriteBox(ctx, w.s, w.b); err != nil {
			t.Fatal(err)
		}
	}
	boxes, complete, err := (&boxReports{status: st, deploy: dep}).load(ctx)
	if err != nil || !complete {
		t.Fatalf("load = %v, %v", complete, err)
	}
	var labels []string
	for _, b := range boxes {
		labels = append(labels, b.Label)
	}
	sort.Strings(labels)
	if !reflect.DeepEqual(labels, []string{"acme-1", "both-1", "both-1", "door-1"}) {
		t.Fatalf("read %v, want the union of both stores' fleets", labels)
	}
	s := recordsStore(t,
		provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable"},
		provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"},
		provision.Record{Schema: 1, Host: "c.example.com", App: "km", Account: "bab", Channel: "stable"})
	got, err := settingsProtected(ctx, s, boxes, complete, now0, "bab", "km")
	if err != nil || !got["v1"] || !got["v2"] || !got["v3"] {
		t.Fatalf("got %v, %v; want every fleet's builds and no refusal", got, err)
	}

	deployListFails := failingListStore{Store: dep, prefix: "fleets/", err: errors.New("AccessDenied")}
	if _, complete, err := (&boxReports{status: st, deploy: deployListFails}).load(ctx); err != nil || complete {
		t.Fatalf("a failed fleets listing in the deployment store read as complete (%v, %v)", complete, err)
	}
	deployGetFails := failingGetStore{Store: dep, bad: "fleets/door/status/boxes/door-1.json", err: errors.New("SlowDown")}
	if _, complete, err := (&boxReports{status: st, deploy: deployGetFails}).load(ctx); err != nil || complete {
		t.Fatalf("a failed fleet report read in the deployment store read as complete (%v, %v)", complete, err)
	}
	statusListFails := failingListStore{Store: st, prefix: "fleets/", err: errors.New("AccessDenied")}
	if _, complete, err := (&boxReports{status: statusListFails, deploy: dep}).load(ctx); err != nil || complete {
		t.Fatalf("a failed fleets listing in the status store read as complete (%v, %v)", complete, err)
	}
}

// CARLOS_STATUS_PREFIX rebases the platform's own reports, as the status
// loop writes them; fleet reports stay at fleets/<name>/ from the bucket
// root, where fleet boxes write them.
func TestTheStatusPrefixRebasesThePlatformReports(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	if err := status.WriteBox(ctx, release.NewPrefixStore(s, "x/"), report("uk-1", now0, nil)); err != nil {
		t.Fatal(err)
	}
	if err := status.WriteBox(ctx, s, report("someone-else", now0, nil)); err != nil {
		t.Fatal(err)
	}
	if err := status.WriteBox(ctx, release.NewPrefixStore(s, "fleets/acme/"), report("acme-1", now0, nil)); err != nil {
		t.Fatal(err)
	}
	boxes, complete, err := (&boxReports{status: s, prefix: "x/"}).load(ctx)
	if err != nil || !complete {
		t.Fatalf("load = %v, %v", complete, err)
	}
	var labels []string
	for _, b := range boxes {
		labels = append(labels, b.Label)
	}
	sort.Strings(labels)
	if !reflect.DeepEqual(labels, []string{"acme-1", "uk-1"}) {
		t.Fatalf("read %v, want the prefixed platform report and the fleet's", labels)
	}
}

// retentionReports is retentionLoop's choice of report source, from the
// environment.
func TestRetentionReportsFromTheEnvironment(t *testing.T) {
	ctx := context.Background()
	deploy := release.NewDirStore(t.TempDir())
	local := kassPins("v1")
	env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } }

	r, err := retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "deploy", "CARLOS_STATUS_BUCKET": "status", "CARLOS_STATUS_PREFIX": "x/"}), deploy, local)
	if err != nil || r.status == nil || r.prefix != "x/" || r.deploy == nil || r.local != nil {
		t.Fatalf("separate status bucket: %+v, %v; want status, prefix x/, the deployment store for fleets, no local report", r, err)
	}
	r, err = retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "deploy"}), deploy, local)
	if err != nil || r.status == nil || r.deploy != nil || r.local != nil {
		t.Fatalf("status in the deployment bucket: %+v, %v; want the bucket's fleets read once", r, err)
	}
	r, err = retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_DIR": "/srv/deploy"}), deploy, local)
	if err != nil || r.status != nil || r.deploy == nil || r.local == nil {
		t.Fatalf("directory-backed: %+v, %v; want this box's own report and the deployment store's fleets", r, err)
	}
	if _, err := retentionReports(ctx, env(map[string]string{}), deploy, local); err == nil {
		t.Fatal("no status store and not a sole housekeeper: want an error")
	}
}