package host

// These tests share the package-level plans cache, so they stay serial.

import (
	"context"
	"errors"
	"slices"
	"sync"
	"testing"
	"time"

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

// plansCountingStore records every Get and can fail chosen keys.
type plansCountingStore struct {
	release.Store
	mu   sync.Mutex
	gets []string
	fail map[string]bool
}

func (c *plansCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	c.mu.Lock()
	c.gets = append(c.gets, key)
	failing := c.fail[key]
	c.mu.Unlock()
	if failing {
		return nil, "", errors.New("bucket unavailable")
	}
	return c.Store.Get(ctx, key)
}

func (c *plansCountingStore) reset() {
	c.mu.Lock()
	c.gets, c.fail = nil, map[string]bool{}
	c.mu.Unlock()
}

func plansWorld(t *testing.T) (*plansCountingStore, *registry.Registry) {
	t.Helper()
	t.Cleanup(func() { planLimits.Store(nil) })
	s := &plansCountingStore{Store: release.NewDirStore(t.TempDir()), fail: map[string]bool{}}
	reg := edgesTestRegistry(t)
	for _, rt := range []registry.Route{
		{Host: "a.example", Kind: "instance", Account: "bab", App: "web", Status: "ready"},
		{Host: "b.example", Kind: "instance", Account: "bdk", App: "shop", Status: "ready"},
		{Host: "status.example", Kind: "status", Status: "ready"},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	return s, reg
}

func putMarker(t *testing.T, s release.Store, enabled bool) {
	t.Helper()
	ctx := context.Background()
	_, etag, err := plans.GetMarker(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := plans.PutMarker(ctx, s, &plans.Marker{Enabled: enabled, CatalogID: "v1@0123456789ab"}, etag); err != nil {
		t.Fatal(err)
	}
}

func putAccountLimits(t *testing.T, s release.Store, sqid string, idle int) {
	t.Helper()
	ctx := context.Background()
	_, etag, err := plans.GetLimits(ctx, s, sqid)
	if err != nil {
		t.Fatal(err)
	}
	r := &plans.Record{Account: sqid, CatalogID: "v1@0123456789ab", Plan: "free", Revision: 1, Limits: plans.Limits{IdleSeconds: idle}}
	if _, err := plans.PutLimits(ctx, s, r, etag); err != nil {
		t.Fatal(err)
	}
}

func plansNoEnv(string) string { return "" }

func TestPlanLimitsSwitch(t *testing.T) {
	for v, want := range map[string]bool{"": true, "on": true, "off": false, "OFF": false, "false": false, "0": false, "no": false, " off ": false} {
		if got := planLimitsSwitchOn(func(string) string { return v }); got != want {
			t.Errorf("CARLOS_PLAN_LIMITS=%q: on = %v, want %v", v, got, want)
		}
	}
}

func TestPlansPassOffReadsOnlyTheMarker(t *testing.T) {
	s, reg := plansWorld(t)
	putAccountLimits(t, s, "bab", 90)
	s.reset()
	plansPass(context.Background(), s, reg, plansNoEnv)
	if !slices.Equal(s.gets, []string{plans.MarkerKey()}) {
		t.Fatalf("reads %v, want only the marker", s.gets)
	}
	if _, ok := planLimitsFor("bab"); ok {
		t.Fatal("limits applied with no marker")
	}
}

func TestPlansPassBoxSwitchOffReadsNothing(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	plansPass(context.Background(), s, reg, plansNoEnv)
	if _, ok := planLimitsFor("bab"); !ok {
		t.Fatal("setup: limits not loaded")
	}
	s.reset()
	plansPass(context.Background(), s, reg, func(string) string { return "off" })
	if len(s.gets) != 0 {
		t.Fatalf("switched-off box read %v", s.gets)
	}
	if _, ok := planLimitsFor("bab"); ok {
		t.Fatal("switched-off box kept limits")
	}
}

func TestPlansPassLoadsServedAccounts(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	putAccountLimits(t, s, "bdk", 900)
	putAccountLimits(t, s, "zzz", 60)
	s.reset()
	plansPass(context.Background(), s, reg, plansNoEnv)
	if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
		t.Fatalf("bab: %+v, %v", l, ok)
	}
	if l, ok := planLimitsFor("bdk"); !ok || l.IdleSeconds != 900 {
		t.Fatalf("bdk: %+v, %v", l, ok)
	}
	if slices.Contains(s.gets, plans.LimitsKey("zzz")) {
		t.Fatal("read the limits of an account with no routes on this box")
	}
}

func TestPlansPassAbsentRecordMeansNoLimits(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	plansPass(context.Background(), s, reg, plansNoEnv)
	if _, ok := planLimitsFor("bdk"); ok {
		t.Fatal("an account with no record got limits")
	}
}

func TestPlansPassKeepsLastGoodOnErrors(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	putAccountLimits(t, s, "bdk", 900)
	plansPass(context.Background(), s, reg, plansNoEnv)

	s.fail[plans.MarkerKey()] = true
	plansPass(context.Background(), s, reg, plansNoEnv)
	if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
		t.Fatalf("a marker read error dropped bab: %+v, %v", l, ok)
	}

	s.reset()
	putAccountLimits(t, s, "bdk", 600)
	s.fail[plans.LimitsKey("bab")] = true
	plansPass(context.Background(), s, reg, plansNoEnv)
	if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
		t.Fatalf("a record read error dropped bab: %+v, %v", l, ok)
	}
	if l, _ := planLimitsFor("bdk"); l.IdleSeconds != 600 {
		t.Fatalf("bdk not refreshed: %+v", l)
	}
}

func TestPlansPassClearsWhenDisabled(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	plansPass(context.Background(), s, reg, plansNoEnv)
	putMarker(t, s, false)
	plansPass(context.Background(), s, reg, plansNoEnv)
	if _, ok := planLimitsFor("bab"); ok {
		t.Fatal("limits kept after plans were disabled")
	}
}

// blockingLimitsStore blocks reads of one limits key until released, and
// says when a read has reached it.
type blockingLimitsStore struct {
	release.Store
	key     string
	reached chan struct{}
	release chan struct{}
	once    sync.Once
}

func (b *blockingLimitsStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == b.key {
		b.once.Do(func() { close(b.reached); <-b.release })
	}
	return b.Store.Get(ctx, key)
}

// Astra plans-2 finding 2: the full pass and the five-minute pass both run
// plansPass. A pass that read an enabled marker and stalled must not install
// its snapshot after a newer pass saw plans disabled.
func TestPlansPassesDoNotRestoreDisabledLimits(t *testing.T) {
	s, reg := plansWorld(t)
	putMarker(t, s, true)
	putAccountLimits(t, s, "bab", 90)
	b := &blockingLimitsStore{Store: s.Store, key: plans.LimitsKey("bab"), reached: make(chan struct{}), release: make(chan struct{})}

	aDone := make(chan struct{})
	go func() { plansPass(context.Background(), b, reg, plansNoEnv); close(aDone) }()
	<-b.reached // pass A read the enabled marker and is stalled on bab's record

	putMarker(t, s, false)
	bDone := make(chan struct{})
	go func() { plansPass(context.Background(), s.Store, reg, plansNoEnv); close(bDone) }()
	select { // without ordering, pass B finishes now and clears the cache
	case <-bDone:
	case <-time.After(200 * time.Millisecond):
	}
	close(b.release)
	<-aDone
	<-bDone
	if _, ok := planLimitsFor("bab"); ok {
		t.Fatal("a stalled pass restored limits after plans were disabled")
	}
}