package console

import (
	"context"
	"errors"
	"os"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

const exampleCatalogPath = "../plans/testdata/example-catalog.json"

func examplePlanCatalog(t *testing.T) *plans.Catalog {
	t.Helper()
	c, err := plans.LoadCatalog(exampleCatalogPath)
	if err != nil {
		t.Fatal(err)
	}
	return c
}

// bumpedCatalog is the example catalog with free's running_things set to n
// under a new version.
func bumpedCatalog(t *testing.T, version string, n string) *plans.Catalog {
	t.Helper()
	b, err := os.ReadFile(exampleCatalogPath)
	if err != nil {
		t.Fatal(err)
	}
	s := strings.Replace(string(b), `"version": "2026-10-20-v1"`, `"version": "`+version+`"`, 1)
	s = strings.Replace(s, `"running_things": 3,`, `"running_things": `+n+`,`, 1)
	c, err := plans.ParseCatalog([]byte(s))
	if err != nil {
		t.Fatal(err)
	}
	return c
}

// planAccount adds a console account with a valid sqid and returns it.
func planAccount(t *testing.T, fx *consoleFixture, id int64) *Account {
	t.Helper()
	sqid := domains.Sqid(id)
	if err := fx.st.CreateAccount(id, sqid, "acct", "owner"+sqid+"@example.com", time.Unix(1_700_000_000, 0)); err != nil {
		t.Fatal(err)
	}
	acct, err := fx.st.AccountBySqid(sqid)
	if err != nil || acct == nil {
		t.Fatalf("account %s: %v", sqid, err)
	}
	return acct
}

func readLimits(t *testing.T, fx *consoleFixture, sqid string) (*plans.Record, string) {
	t.Helper()
	r, etag, err := plans.GetLimits(context.Background(), fx.rel, sqid)
	if err != nil {
		t.Fatal(err)
	}
	return r, etag
}

func TestPlanLimitsPassOffWithoutACatalog(t *testing.T) {
	fx := newConsoleFixture(t)
	acct := planAccount(t, fx, 501)
	fx.srv.planLimitsPass(context.Background(), planNow)
	if r, _ := readLimits(t, fx, acct.Sqid); r != nil {
		t.Fatalf("wrote a record with no catalog: %+v", r)
	}
	if row, _ := fx.st.AccountPlan(acct.ID); row != nil {
		t.Fatalf("inserted a plan row with no catalog: %+v", row)
	}
}

func TestPlanLimitsPassPublishesDefaults(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	acct := planAccount(t, fx, 501)
	fx.srv.planLimitsPass(context.Background(), planNow)
	r, _ := readLimits(t, fx, acct.Sqid)
	if r == nil {
		t.Fatal("no record published")
	}
	if r.Plan != "free" || r.Source != PlanSourceDefault || r.Limits.RunningThings != 3 || r.CatalogID != fx.srv.cfg.PlanCatalog.ID() {
		t.Fatalf("record %+v", r)
	}
	row, _ := fx.st.AccountPlan(acct.ID)
	if row.PublishedRevision != row.Revision || row.PublishedCatalog != fx.srv.cfg.PlanCatalog.ID() {
		t.Fatalf("row not marked published: %+v", row)
	}
}

func TestPlanLimitsPassRepublishesOnCatalogChange(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	acct := planAccount(t, fx, 501)
	fx.srv.planLimitsPass(context.Background(), planNow)
	next := bumpedCatalog(t, "2026-10-21-v2", "4")
	fx.srv.cfg.PlanCatalog = next
	fx.srv.planLimitsPass(context.Background(), planNow)
	r, etag := readLimits(t, fx, acct.Sqid)
	if r.CatalogID != next.ID() || r.Limits.RunningThings != 4 {
		t.Fatalf("not republished: %+v", r)
	}
	fx.srv.planLimitsPass(context.Background(), planNow)
	if _, again := readLimits(t, fx, acct.Sqid); again != etag {
		t.Fatal("an up-to-date record was rewritten")
	}
}

func TestPlanLimitsPassSkipsAnOrphanedAssignment(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	orphan := planAccount(t, fx, 501)
	other := planAccount(t, fx, 502)
	fx.srv.planLimitsPass(context.Background(), planNow)
	_, before := readLimits(t, fx, orphan.Sqid)
	if _, err := fx.st.SetAccountPlan(orphan.ID, "gold", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.st.SetAccountPlan(other.ID, "personal", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
		t.Fatal(err)
	}
	fx.srv.planLimitsPass(context.Background(), planNow)
	if _, after := readLimits(t, fx, orphan.Sqid); after != before {
		t.Fatal("the orphan's last good record was replaced")
	}
	if row, _ := fx.st.AccountPlan(orphan.ID); row.PublishedRevision == row.Revision {
		t.Fatal("the orphan was marked published")
	}
	if r, _ := readLimits(t, fx, other.Sqid); r.Plan != "personal" {
		t.Fatalf("the other account was not published: %+v", r)
	}
}

func TestPublishNeverOverwritesANewerRecord(t *testing.T) {
	fx := newConsoleFixture(t)
	cat := examplePlanCatalog(t)
	fx.srv.cfg.PlanCatalog = cat
	acct := planAccount(t, fx, 501)
	newer := &plans.Record{Account: acct.Sqid, CatalogID: cat.ID(), Plan: "personal", Source: PlanSourceOperator, Revision: 5}
	if _, err := plans.PutLimits(context.Background(), fx.rel, newer, ""); err != nil {
		t.Fatal(err)
	}
	for range 3 {
		if _, err := fx.st.SetAccountPlan(acct.ID, "free", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
			t.Fatal(err)
		}
	}
	row, _ := fx.st.AccountPlan(acct.ID)
	if err := fx.srv.publishAccountLimits(context.Background(), *row, planNow); err != nil {
		t.Fatal(err)
	}
	if r, _ := readLimits(t, fx, acct.Sqid); r.Revision != 5 || r.Plan != "personal" {
		t.Fatalf("newer record overwritten: %+v", r)
	}
	if row, _ := fx.st.AccountPlan(acct.ID); row.PublishedRevision != 3 {
		t.Fatalf("published_revision %d, want 3", row.PublishedRevision)
	}
}

// planFailingPuts is a bucket whose next n Puts fail.
type planFailingPuts struct {
	release.Store
	n atomic.Int32
}

func (f *planFailingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if f.n.Add(-1) >= 0 {
		return "", errors.New("bucket unavailable")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

func TestPublishRetriesAfterABucketFailure(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	acct := planAccount(t, fx, 501)
	fp := &planFailingPuts{Store: fx.rel}
	fx.srv.cfg.Releases = fp
	if _, err := fx.st.EnsureDefaultPlans("free", planNow); err != nil {
		t.Fatal(err)
	}
	fp.n.Store(1000) // every Put fails during the first pass
	fx.srv.planLimitsPass(context.Background(), planNow)
	if row, _ := fx.st.AccountPlan(acct.ID); row.PublishedRevision != 0 {
		t.Fatal("marked published although the write failed")
	}
	fp.n.Store(0)
	fx.srv.planLimitsPass(context.Background(), planNow)
	if r, _ := readLimits(t, fx, acct.Sqid); r == nil {
		t.Fatal("not published on the next pass")
	}
}

func TestEffectiveLimits(t *testing.T) {
	fx := newConsoleFixture(t)
	acct := planAccount(t, fx, 501)
	if l, err := fx.srv.EffectiveLimits(acct); l != nil || err != nil {
		t.Fatalf("no catalog: %+v, %v", l, err)
	}
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	l, err := fx.srv.EffectiveLimits(acct)
	if err != nil || l == nil || l.RunningThings != 3 {
		t.Fatalf("default: %+v, %v", l, err)
	}
	if _, err := fx.st.SetAccountPlan(acct.ID, "personal", map[string]int{"apps": 1}, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
		t.Fatal(err)
	}
	l, err = fx.srv.EffectiveLimits(acct)
	if err != nil || l.RunningThings != 30 {
		t.Fatalf("personal + apps: %+v, %v", l, err)
	}
}

// Orphaned assignments must not starve the rest: with a batch of two and
// the two lowest accounts orphaned, one pass still reaches the third.
func TestPlanLimitsPassOrphansDoNotStarveLaterAccounts(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
	defer func(n int) { planLimitsPassBatch = n }(planLimitsPassBatch)
	planLimitsPassBatch = 2
	a := planAccount(t, fx, 501)
	b := planAccount(t, fx, 502)
	c := planAccount(t, fx, 503)
	for _, acct := range []*Account{a, b} {
		if _, err := fx.st.SetAccountPlan(acct.ID, "gold", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
			t.Fatal(err)
		}
	}
	fx.srv.planLimitsPass(context.Background(), planNow)
	if r, _ := readLimits(t, fx, c.Sqid); r == nil {
		t.Fatal("orphaned accounts starved a later valid account")
	}
}