package mail

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

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

const claimTestAccount = "bag"
const claimTestDomain = "shared.com"

// claimLogStore records every Get and Put. It embeds only release.Store, so it is
// not a ClaimMailStore and behaves as an unmarked store.
type claimLogStore struct {
	release.Store
	mu         sync.Mutex
	gets, puts []string
}

func (l *claimLogStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	l.mu.Lock()
	l.gets = append(l.gets, key)
	l.mu.Unlock()
	return l.Store.Get(ctx, key)
}
func (l *claimLogStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
	l.mu.Lock()
	l.puts = append(l.puts, key)
	l.mu.Unlock()
	return l.Store.Put(ctx, key, body, tag)
}
func (l *claimLogStore) reset() { l.mu.Lock(); l.gets, l.puts = nil, nil; l.mu.Unlock() }
func (l *claimLogStore) admissionGets() int {
	l.mu.Lock()
	defer l.mu.Unlock()
	n := 0
	for _, k := range l.gets {
		if strings.HasSuffix(k, "/admission.json") {
			n++
		}
	}
	return n
}
func (l *claimLogStore) putsTo(key string) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	n := 0
	for _, k := range l.puts {
		if k == key {
			n++
		}
	}
	return n
}

// countingProv counts every Provisioner call, reads included.
type countingProv struct {
	*MemProvisioner
	cmu  sync.Mutex
	hits map[string]int
}

func (c *countingProv) hit(name string) { c.cmu.Lock(); c.hits[name]++; c.cmu.Unlock() }
func (c *countingProv) total() int {
	c.cmu.Lock()
	defer c.cmu.Unlock()
	n := 0
	for _, v := range c.hits {
		n += v
	}
	return n
}
func (c *countingProv) reset() { c.cmu.Lock(); c.hits = map[string]int{}; c.cmu.Unlock() }
func (c *countingProv) EnsureTenant(ctx context.Context, name string) error {
	c.hit("EnsureTenant")
	return c.MemProvisioner.EnsureTenant(ctx, name)
}
func (c *countingProv) EnsureConfigSet(ctx context.Context, region, name, topic string) error {
	c.hit("EnsureConfigSet")
	return c.MemProvisioner.EnsureConfigSet(ctx, region, name, topic)
}
func (c *countingProv) SetIdentityConfigSet(ctx context.Context, region, domain, set string) error {
	c.hit("SetIdentityConfigSet")
	return c.MemProvisioner.SetIdentityConfigSet(ctx, region, domain, set)
}
func (c *countingProv) AssociateTenantResource(ctx context.Context, tenant, resource string) error {
	c.hit("AssociateTenantResource")
	return c.MemProvisioner.AssociateTenantResource(ctx, tenant, resource)
}
func (c *countingProv) DisassociateTenantResource(ctx context.Context, tenant, resource string) error {
	c.hit("DisassociateTenantResource")
	return c.MemProvisioner.DisassociateTenantResource(ctx, tenant, resource)
}
func (c *countingProv) EnsureIdentity(ctx context.Context, region, domain, from string, adopt bool) (Identity, error) {
	c.hit("EnsureIdentity")
	return c.MemProvisioner.EnsureIdentity(ctx, region, domain, from, adopt)
}
func (c *countingProv) IdentityStatus(ctx context.Context, region, domain string) (Status, error) {
	c.hit("IdentityStatus")
	return c.MemProvisioner.IdentityStatus(ctx, region, domain)
}
func (c *countingProv) DeleteIdentity(ctx context.Context, region, domain string) error {
	c.hit("DeleteIdentity")
	return c.MemProvisioner.DeleteIdentity(ctx, region, domain)
}
func (c *countingProv) EnsureUser(ctx context.Context, name, path string, identities, sets, domains []string) (string, error) {
	c.hit("EnsureUser")
	return c.MemProvisioner.EnsureUser(ctx, name, path, identities, sets, domains)
}
func (c *countingProv) MintKey(ctx context.Context, user string) (string, string, error) {
	c.hit("MintKey")
	return c.MemProvisioner.MintKey(ctx, user)
}
func (c *countingProv) SetKeyActive(ctx context.Context, user, key string, active bool) error {
	c.hit("SetKeyActive")
	return c.MemProvisioner.SetKeyActive(ctx, user, key, active)
}
func (c *countingProv) DeleteKey(ctx context.Context, user, key string) error {
	c.hit("DeleteKey")
	return c.MemProvisioner.DeleteKey(ctx, user, key)
}
func (c *countingProv) ReapKeys(ctx context.Context, user string, keep []string) error {
	c.hit("ReapKeys")
	return c.MemProvisioner.ReapKeys(ctx, user, keep)
}
func (c *countingProv) DeleteUser(ctx context.Context, user string) error {
	c.hit("DeleteUser")
	return c.MemProvisioner.DeleteUser(ctx, user)
}

type claimDomainFixture struct {
	t      *testing.T
	ctx    context.Context
	base   release.Store
	log    *claimLogStore
	marked ClaimMailStore
	mem    *MemProvisioner
	prov   *countingProv
	now    time.Time
	mg     *memGate
	adm    *claimFixtureAdmitter
	states map[string]string
}

var claimTestOwner = ClaimMailOwner{Console: "c", Enrollment: "e", Controller: "one"}

// memGate is a test-only ClaimMailGate holding {state,id,revision} in memory.
type memGate struct {
	mu       sync.Mutex
	State    string
	id       string
	Revision int
}
type memGateKey struct{}

func (g *memGate) AcquireClaimMailGate(ctx context.Context, _ ClaimMailOwner, _ ClaimMailScope) (context.Context, string, error) {
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.State == "held" {
		return nil, "", ErrClaimMailPending
	}
	g.Revision++
	g.State = "held"
	g.id = fmt.Sprintf("gate-%d", g.Revision)
	return context.WithValue(ctx, memGateKey{}, g.id), g.id, nil
}
func (g *memGate) owns(ctx context.Context) bool {
	id, _ := ctx.Value(memGateKey{}).(string)
	return g.State == "held" && id == g.id
}
func (g *memGate) CheckClaimMailGate(ctx context.Context) error {
	g.mu.Lock()
	defer g.mu.Unlock()
	if !g.owns(ctx) {
		return ErrClaimMailPending
	}
	return nil
}
func (g *memGate) ReleaseClaimMailGate(ctx context.Context) error {
	g.mu.Lock()
	defer g.mu.Unlock()
	if !g.owns(ctx) {
		return ErrClaimMailPending
	}
	g.Revision++
	g.State = "free"
	return nil
}
func (g *memGate) snapshot() memGate {
	g.mu.Lock()
	defer g.mu.Unlock()
	return memGate{State: g.State, id: g.id, Revision: g.Revision}
}

// fixtureAdmission is the admitter's result: it adds no authority of its own.
type fixtureAdmission struct {
	ctx      context.Context
	released *int
	mu       *sync.Mutex
}

func (a fixtureAdmission) Context() context.Context                                   { return a.ctx }
func (a fixtureAdmission) Settle(context.Context, error, []hostops.MailAttempt) error { return nil }
func (a fixtureAdmission) Release(context.Context) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	*a.released++
	return nil
}

// claimFixtureAdmitter refuses an app whose fixture state is "purging". Each
// app it checks is logged as an admission read so the call-count assertions
// keep their meaning.
type claimFixtureAdmitter struct {
	f        *claimDomainFixture
	mu       sync.Mutex
	released int
}

func (a *claimFixtureAdmitter) AdmitClaimMailApps(ctx context.Context, _ ClaimMailHold, apps []ClaimMailApp) (ClaimMailAdmission, error) {
	for _, app := range apps {
		a.f.log.mu.Lock()
		a.f.log.gets = append(a.f.log.gets, "apps/"+app.Account+"/"+app.App+"/admission.json")
		a.f.log.mu.Unlock()
		if a.f.states[app.App] == "purging" {
			return nil, ErrClaimMailPending
		}
	}
	return fixtureAdmission{ctx: ctx, released: &a.released, mu: &a.mu}, nil
}

func (f *claimDomainFixture) useAdmitter(a ClaimMailAdmitter) {
	f.t.Helper()
	m, err := NewClaimMailStore(f.log, f.mg, a, claimTestOwner)
	if err != nil {
		f.t.Fatal(err)
	}
	f.marked = m
}

// newClaimDomainFixture seeds one shared domain whose Apps are members, through the
// unmarked store, then opens an admission for alpha, beta and gamma.
func newClaimDomainFixture(t *testing.T, members ...string) *claimDomainFixture {
	t.Helper()
	ctx := context.Background()
	base := release.NewDirStore(t.TempDir())
	mem := NewMemProvisioner("1")
	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
	for _, app := range members {
		if _, _, err := EnsureDomain(ctx, base, mem, dnsrec.NewMemWriter(), testDeployment(), claimTestAccount, app, claimTestDomain, now); err != nil {
			t.Fatalf("seed %s: %v", app, err)
		}
	}
	mem.SetIdentityStatus("eu-west-1", claimTestDomain, Status{DKIM: "SUCCESS", MailFrom: "SUCCESS", Sending: "SUCCESS"})
	log := &claimLogStore{Store: base}
	f := &claimDomainFixture{t: t, ctx: ctx, base: base, log: log, mem: mem, now: now, mg: &memGate{State: "free"}, states: map[string]string{},
		prov: &countingProv{MemProvisioner: mem, hits: map[string]int{}}}
	f.adm = &claimFixtureAdmitter{f: f}
	for _, app := range []string{"alpha", "beta", "gamma"} {
		f.setAdmission(app, "open")
	}
	f.useAdmitter(f.adm)
	f.reset()
	return f
}
func (f *claimDomainFixture) reset() { f.log.reset(); f.prov.reset() }

// setAdmission sets the state the fixture admitter reports: "open" or
// "purging".
func (f *claimDomainFixture) setAdmission(app, state string) { f.t.Helper(); f.states[app] = state }
func (f *claimDomainFixture) domainRaw() ([]byte, string) {
	f.t.Helper()
	raw, etag, err := f.base.Get(f.ctx, DomainKey(claimTestAccount, claimTestDomain))
	if err != nil {
		f.t.Fatal(err)
	}
	return raw, etag
}
func (f *claimDomainFixture) requireUntouched(raw []byte, etag string) {
	f.t.Helper()
	gotRaw, gotEtag := f.domainRaw()
	if !bytes.Equal(gotRaw, raw) || gotEtag != etag {
		f.t.Fatalf("domain record changed: etag %s -> %s", etag, gotEtag)
	}
	if n := f.log.putsTo(DomainKey(claimTestAccount, claimTestDomain)); n != 0 {
		f.t.Fatalf("domain puts=%d", n)
	}
	if n := f.prov.total(); n != 0 {
		f.t.Fatalf("backend calls=%d %v", n, f.prov.hits)
	}
}
func (f *claimDomainFixture) gate() memGate { f.t.Helper(); return f.mg.snapshot() }

// requireGateFree: one acquire and one release since the gate was first used.
func (f *claimDomainFixture) requireGateFree() {
	f.t.Helper()
	if g := f.gate(); g.State != "free" || g.Revision != 2 {
		f.t.Fatalf("gate not released cleanly: %+v", &g)
	}
}

func TestClaimMailPutDomainPreflightsEveryAffectedApp(t *testing.T) {
	cases := []struct {
		name     string
		members  []string
		purging  string
		proposed []string
	}{
		{"proposal drops the purging member", []string{"alpha", "beta"}, "beta", []string{"alpha"}},
		{"proposal adds a purging app", []string{"alpha"}, "gamma", []string{"alpha", "gamma"}},
		{"proposal is reordered", []string{"alpha", "beta", "gamma"}, "gamma", []string{"gamma", "beta", "alpha"}},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			f := newClaimDomainFixture(t, tc.members...)
			rec, etag, err := GetDomain(f.ctx, f.base, claimTestAccount, claimTestDomain)
			if err != nil {
				t.Fatal(err)
			}
			beforeRaw, beforeEtag := f.domainRaw()
			f.setAdmission(tc.purging, "purging")
			f.reset()
			rec.Apps = tc.proposed
			if _, err = PutDomain(f.ctx, f.marked, claimTestAccount, rec, etag); !errors.Is(err, ErrClaimMailPending) {
				t.Fatalf("PutDomain with a purging affected app: %v", err)
			}
			f.requireUntouched(beforeRaw, beforeEtag)
			f.requireGateFree()
			if f.log.admissionGets() == 0 {
				t.Fatal("preflight never read admissions")
			}
			// The gate is free and the refusal was about the purge only: reopening it lets the same write through.
			f.setAdmission(tc.purging, "open")
			if _, err = PutDomain(f.ctx, f.marked, claimTestAccount, rec, etag); err != nil {
				t.Fatalf("retry after refusal: %v", err)
			}
			if _, after := f.domainRaw(); after == beforeEtag {
				t.Fatal("control write did not change the domain record")
			}
		})
	}
}

func TestClaimMailVerifyDomainPreflightsEveryMemberBeforeBackend(t *testing.T) {
	verify := func(f *claimDomainFixture) (*DomainRecord, error) {
		return VerifyDomain(f.ctx, f.marked, f.prov, testDeployment(), claimTestAccount, claimTestDomain, "eu-west-1", f.now)
	}
	t.Run("one member purging", func(t *testing.T) {
		f := newClaimDomainFixture(t, "alpha", "beta", "gamma")
		beforeRaw, beforeEtag := f.domainRaw()
		f.setAdmission("beta", "purging")
		f.reset()
		rec, err := verify(f)
		if !errors.Is(err, ErrClaimMailPending) || rec != nil {
			t.Fatalf("VerifyDomain: rec=%v err=%v", rec, err)
		}
		f.requireUntouched(beforeRaw, beforeEtag)
		f.requireGateFree()
		if f.log.admissionGets() == 0 {
			t.Fatal("preflight never read admissions")
		}
		f.setAdmission("beta", "open")
		f.reset()
		rec, err = verify(f)
		if err != nil {
			t.Fatalf("VerifyDomain after reopen: %v", err)
		}
		if rec.Regions["eu-west-1"].Status != StatusVerified {
			t.Fatalf("status=%q", rec.Regions["eu-west-1"].Status)
		}
		if f.prov.total() == 0 {
			t.Fatal("control run never reached the backend, so the zero count above proves nothing")
		}
	})
	t.Run("membership is read fresh at call time", func(t *testing.T) {
		f := newClaimDomainFixture(t, "alpha")
		// beta joins after the fixture exists and is already purging when the call starts.
		if _, _, err := EnsureDomain(f.ctx, f.base, f.mem, dnsrec.NewMemWriter(), testDeployment(), claimTestAccount, "beta", claimTestDomain, f.now); err != nil {
			t.Fatal(err)
		}
		f.mem.SetIdentityStatus("eu-west-1", claimTestDomain, Status{DKIM: "SUCCESS", MailFrom: "SUCCESS", Sending: "SUCCESS"})
		beforeRaw, beforeEtag := f.domainRaw()
		f.setAdmission("beta", "purging")
		f.reset()
		if _, err := verify(f); !errors.Is(err, ErrClaimMailPending) {
			t.Fatalf("VerifyDomain: %v", err)
		}
		f.requireUntouched(beforeRaw, beforeEtag)
		f.requireGateFree()
	})
}

func TestClaimMailDomainWrappersUnmarkedDoNoNewReads(t *testing.T) {
	f := newClaimDomainFixture(t, "alpha", "beta")
	f.setAdmission("beta", "purging")
	rec, etag, err := GetDomain(f.ctx, f.base, claimTestAccount, claimTestDomain)
	if err != nil {
		t.Fatal(err)
	}
	f.reset()
	// f.log is not a ClaimMailStore: the purging member must not matter.
	if _, err = PutDomain(f.ctx, f.log, claimTestAccount, rec, etag); err != nil {
		t.Fatalf("unmarked PutDomain: %v", err)
	}
	if len(f.log.gets) != 0 {
		t.Fatalf("unmarked PutDomain read %v", f.log.gets)
	}
	f.reset()
	if _, err = VerifyDomain(f.ctx, f.log, f.prov, testDeployment(), claimTestAccount, claimTestDomain, "eu-west-1", f.now); err != nil {
		t.Fatalf("unmarked VerifyDomain: %v", err)
	}
	for _, k := range f.log.gets {
		if strings.HasSuffix(k, "/admission.json") {
			t.Fatalf("unmarked VerifyDomain read %s", k)
		}
	}
	if g := f.gate(); g.Revision != 0 {
		t.Fatalf("unmarked call used the gate: %+v", &g)
	}
}