package provision

import (
	"context"
	"errors"
	"reflect"
	"slices"
	"sort"
	"strings"
	"sync"
	"testing"

	"github.com/carlosframework/platform/internal/release"
)

// bucketShaped is a store that promises bucket-shaped ListDirs, as the S3
// store does, and records every listing asked of it. DirStore's own ListDirs
// has that shape; it just does not advertise it.
type bucketShaped struct {
	release.Store
	mu    sync.Mutex
	lists []string
	dirs  []string
	// failDirs and failList fail one prefix each.
	failDirs, failList string
}

func (s *bucketShaped) BucketShapedListDirs() bool { return true }

func (s *bucketShaped) List(ctx context.Context, prefix string) ([]string, error) {
	s.mu.Lock()
	s.lists = append(s.lists, prefix)
	s.mu.Unlock()
	if prefix == s.failList {
		return nil, errors.New("list refused")
	}
	return s.Store.List(ctx, prefix)
}

func (s *bucketShaped) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	s.mu.Lock()
	s.dirs = append(s.dirs, prefix)
	s.mu.Unlock()
	if prefix == s.failDirs {
		return nil, errors.New("listdirs refused")
	}
	return s.Store.ListDirs(ctx, prefix)
}

func (s *bucketShaped) listedAllOfApps() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, p := range s.lists {
		if p == "apps/" {
			return true
		}
	}
	return false
}

// fleetShaped answers ListDirs the way the fleet door does: scoped to one
// account, ListDirs("apps/") is that account's APP directories. It makes no
// bucket-shape promise, which is the point.
type fleetShaped struct {
	release.Store
	acct string
}

func (s fleetShaped) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	if prefix == "apps/" {
		return s.Store.ListDirs(ctx, "apps/"+s.acct+"/")
	}
	return s.Store.ListDirs(ctx, prefix)
}

// A bucket with everything a production one holds under apps/: records of
// both kinds, the logs and release artifacts that dwarf them, an account that
// has not materialized, a legacy account-less app, and apps named like the
// release plane's own words.
func seedAppListingBucket(t *testing.T) release.Store {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	for _, k := range []string{
		"apps/bab/shop/instances/a.example.com.json",
		"apps/bab/shop/instances/b.example.com.json",
		"apps/bab/shop/instances.json",
		"apps/bab/shop/sidecars/bot.json",
		"apps/bab/shop/logs/2026/10/05/chunk-1.jsonl",
		"apps/bab/shop/logs/2026/10/05/chunk-2.jsonl",
		"apps/bab/shop/releases/v1/manifest.json",
		"apps/bab/shop/channels/edge.json",
		"apps/bab/blog/instances/c.example.com.json",
		"apps/bab/blog/instances/nested/too/deep.json",
		"apps/bad/site/sidecars/w.json",
		"apps/bad/site/instances/d.example.com.json",
		"apps/zzz/ghost/instances/e.example.com.json", // account not materialized
		"apps/legacyapp/channels/edge.json",           // account-less
		"apps/bab/channels/instances/f.example.com.json",
		"apps/bab/releases/instances/g.example.com.json",
	} {
		if _, err := s.Put(context.Background(), k, []byte(`{}`), ""); err != nil {
			t.Fatalf("seed %s: %v", k, err)
		}
	}
	return s
}

func parsedRecords(keys []string, parse func(string) bool) []string {
	var out []string
	for _, k := range keys {
		if parse(k) {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}

// On a bucket-shaped store the walk finds exactly the records the full
// listing did, for both record kinds, and never lists the whole of apps/.
func TestAppSubtreeWalkMatchesTheFullListing(t *testing.T) {
	inner := seedAppListingBucket(t)
	known := map[string]bool{"bab": true, "bad": true}
	all, err := inner.List(context.Background(), "apps/")
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range []struct {
		sub   string
		parse func(string) bool
	}{
		{"instances/", func(k string) bool { _, _, _, ok := ParseRecordKey(k, known); return ok }},
		{"sidecars/", func(k string) bool { _, _, _, ok := ParseSidecarKey(k, known); return ok }},
	} {
		s := &bucketShaped{Store: inner}
		got, partial, err := listAppSubtree(context.Background(), s, known, c.sub)
		if err != nil || partial {
			t.Fatalf("%s: partial=%v err=%v", c.sub, partial, err)
		}
		if g, w := parsedRecords(got, c.parse), parsedRecords(all, c.parse); !reflect.DeepEqual(g, w) {
			t.Fatalf("%s: the walk found %v, the full listing %v", c.sub, g, w)
		}
		for _, p := range s.lists {
			if !strings.HasSuffix(p, "/"+c.sub) {
				t.Fatalf("%s: listed %q — the walk lists only <app>/%s prefixes", c.sub, p, c.sub)
			}
		}
	}
}

// A store that makes no bucket-shape promise is listed the old way. The fleet
// door's ListDirs("apps/") is its account's app directories; walked as if it
// were account directories, it yields no records at all, which reconcile
// would read as every route retired.
func TestAppSubtreeListsTheOldWayWithoutTheBucketShape(t *testing.T) {
	fleet := fleetShaped{Store: seedAppListingBucket(t), acct: "bab"}
	known := map[string]bool{"bab": true}
	parse := func(k string) bool { _, _, _, ok := ParseRecordKey(k, known); return ok }
	got, partial, err := listAppSubtree(context.Background(), fleet, known, "instances/")
	if err != nil || partial {
		t.Fatalf("partial=%v err=%v", partial, err)
	}
	want := []string{
		"apps/bab/blog/instances/c.example.com.json",
		"apps/bab/shop/instances/a.example.com.json",
		"apps/bab/shop/instances/b.example.com.json",
	}
	if g := parsedRecords(got, parse); !reflect.DeepEqual(g, want) {
		t.Fatalf("records on a fleet-shaped store = %v, want %v", g, want)
	}
	if release.HasBucketListDirs(release.NewDirStore(t.TempDir())) {
		t.Fatal("DirStore promises bucket-shaped ListDirs; it walks its whole root per List, so it must be listed the old way")
	}
}

// One account's or one app's listing failing is partial, not fatal: the rest
// still come back, and the caller is told some were not seen.
func TestAppSubtreeOneListingFailingIsPartial(t *testing.T) {
	known := map[string]bool{"bab": true, "bad": true}
	parse := func(k string) bool { _, _, _, ok := ParseRecordKey(k, known); return ok }
	for _, c := range []struct {
		name string
		s    *bucketShaped
		want []string
	}{
		{"an app", &bucketShaped{failList: "apps/bab/blog/instances/"}, []string{
			"apps/bab/shop/instances/a.example.com.json",
			"apps/bab/shop/instances/b.example.com.json",
			"apps/bad/site/instances/d.example.com.json",
		}},
		{"an account", &bucketShaped{failDirs: "apps/bab/"}, []string{
			"apps/bad/site/instances/d.example.com.json",
		}},
	} {
		c.s.Store = seedAppListingBucket(t)
		got, partial, err := listAppSubtree(context.Background(), c.s, known, "instances/")
		if err != nil {
			t.Fatalf("%s: one failed listing failed the walk: %v", c.name, err)
		}
		if !partial {
			t.Fatalf("%s: a failed listing was not reported as partial", c.name)
		}
		if g := parsedRecords(got, parse); !reflect.DeepEqual(g, c.want) {
			t.Fatalf("%s: records = %v, want %v", c.name, g, c.want)
		}
	}
}

// All three record readers, on a bucket-shaped store with real records: they
// find the records, mint the same route they mint off a plain store, and
// never list the whole of apps/.
func TestRecordReadersWalkABucketShapedStore(t *testing.T) {
	inner, sqid := setup(t)
	if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}

	plainReg := regWith(t)
	if err := Reconcile(ctx, inner, plainReg, cfg); err != nil {
		t.Fatal(err)
	}
	s := &bucketShaped{Store: inner}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	want, got := *mustGet(t, plainReg, "acme.go.tito.io"), *mustGet(t, reg, "acme.go.tito.io")
	want.CreatedAt, got.CreatedAt = 0, 0
	if got != want {
		t.Fatalf("minted over the walk = %+v, over the full listing %+v", got, want)
	}

	recs, err := ListAllRecords(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	if len(recs) != 1 || recs[0].Host != "acme.go.tito.io" {
		t.Fatalf("ListAllRecords = %+v, want the one acme record", recs)
	}
	if err := ReconcileSidecars(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if s.listedAllOfApps() {
		t.Fatalf("a record reader listed the whole of apps/: %v", s.lists)
	}
}

// Reconcile treats a partial walk as it treats an unreadable record: the
// routes it could not see are not retired.
func TestReconcileKeepsRoutesItCouldNotList(t *testing.T) {
	inner, sqid := setup(t)
	if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	reg := regWith(t)
	if err := Reconcile(ctx, inner, reg, cfg); err != nil {
		t.Fatal(err)
	}
	s := &bucketShaped{Store: inner, failList: RecordsPrefix(sqid, "tito")}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt == nil {
		t.Fatalf("a route whose records could not be listed was removed: %+v, %v", rt, err)
	}
}

// ListAllRecords refuses a partial walk rather than report a short record
// set: the moves and steering passes act on what it returns, and a missing
// record reads to them as an instance that no longer exists.
func TestListAllRecordsRefusesAPartialWalk(t *testing.T) {
	inner, sqid := setup(t)
	if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	s := &bucketShaped{Store: inner, failList: RecordsPrefix(sqid, "tito")}
	if recs, err := ListAllRecords(ctx, s); err == nil {
		t.Fatalf("ListAllRecords over a failed app listing = %+v, nil; want an error", recs)
	}
}

// ListAllRecordsTolerant reports what it could read and says it is
// incomplete, where ListAllRecords refuses: an app it could not list, a
// record it could not Get.
func TestListAllRecordsTolerantSkipsWhatItCannotRead(t *testing.T) {
	inner, sqid := setup(t)
	first := acmeRecord(sqid)
	first.Host, first.Slug = "one.example.com", "one"
	second := acmeRecord(sqid)
	second.Host, second.Slug = "two.example.com", "two"
	for _, rec := range []Record{first, second} {
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	if recs, complete, err := ListAllRecordsTolerant(ctx, inner); err != nil || !complete || len(recs) != 2 {
		t.Fatalf("a clean read = %d records, complete %v, %v", len(recs), complete, err)
	}
	s := &countingGets{Store: inner, failKey: RecordKey(sqid, first.App, first.Host)}
	recs, complete, err := ListAllRecordsTolerant(ctx, s)
	if err != nil || complete || len(recs) != 1 || recs[0].Host != second.Host {
		t.Fatalf("one record unreadable = %+v, complete %v, %v", recs, complete, err)
	}
	if _, err := ListAllRecords(ctx, s); err == nil {
		t.Fatal("ListAllRecords read past an unreadable record")
	}
	walk := &bucketShaped{Store: inner, failList: RecordsPrefix(sqid, first.App)}
	if recs, complete, err := ListAllRecordsTolerant(ctx, walk); err != nil || complete || len(recs) != 0 {
		t.Fatalf("an app that will not list = %+v, complete %v, %v", recs, complete, err)
	}
}

// The sidecar reader over the walk mints the sidecar it mints over the full
// listing, and keeps it on the next pass. Returning no sidecar keys at all
// would otherwise pass every other test here.
func TestSidecarReconcileOverTheWalkMintsAndKeeps(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{
		Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}, Channel: "stable"})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	s := &bucketShaped{Store: f.store}
	cfg := ReconcileConfig{DataDir: f.dir}
	for pass := 1; pass <= 2; pass++ {
		if err := ReconcileSidecars(ctx, s, f.reg, cfg); err != nil {
			t.Fatalf("pass %d: %v", pass, err)
		}
		hasRoute(t, f.reg, f.host("bot", "eleven"), "the walk lost the sidecar")
	}
	if s.listedAllOfApps() {
		t.Fatalf("the sidecar reader listed the whole of apps/: %v", s.lists)
	}
}

// reversedList answers List in reverse order and makes no bucket-shape
// promise: Store.List does not promise an order.
type reversedList struct{ release.Store }

func (s reversedList) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := s.Store.List(ctx, prefix)
	slices.Reverse(keys)
	return keys, err
}

// ListAllRecords returns records in a stable order whatever order the store
// lists in, on the fallback path as on the walk.
func TestListAllRecordsIsSortedOnTheFallback(t *testing.T) {
	inner, sqid := setup(t)
	for _, slug := range []string{"acme", "zeta", "beta"} {
		rec := acmeRecord(sqid)
		rec.Host, rec.Slug = slug+".go.tito.io", slug
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	recs, err := ListAllRecords(ctx, reversedList{inner})
	if err != nil {
		t.Fatal(err)
	}
	var hosts []string
	for _, r := range recs {
		hosts = append(hosts, r.Host)
	}
	if !slices.IsSorted(hosts) {
		t.Fatalf("ListAllRecords over a reverse-listing store = %v, want sorted", hosts)
	}
}