package mail

import (
	"context"
	"errors"
	"fmt"
	"io"
	"slices"
	"sort"
	"sync"
	"sync/atomic"
	"time"

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

const releaseTimeout = 30 * time.Second

var ErrClaimMailPending = errors.New("mail: lifecycle resource operation pending")

type ClaimMailOwner struct{ Console, Enrollment, Controller string }
type ClaimMailScope struct{ Account, App, AppID, Operation string }

// ClaimMailApp names one app a gated mail call can affect.
type ClaimMailApp struct{ Account, App string }

// ClaimMailHold identifies the held shared gate an admission belongs to.
type ClaimMailHold struct {
	Gate  string
	Owner ClaimMailOwner
}

// ClaimMailGate is the shared resource gate. The production implementation
// is the admission engine's typed control writer; this package never sees
// the gate's storage key and cannot write it.
type ClaimMailGate interface {
	// AcquireClaimMailGate takes the free gate and returns a context that
	// carries the lease (the backend and bucket boundaries check it).
	AcquireClaimMailGate(ctx context.Context, owner ClaimMailOwner, scope ClaimMailScope) (held context.Context, gateID string, err error)
	// CheckClaimMailGate proves held still owns the exact held revision.
	CheckClaimMailGate(held context.Context) error
	// ReleaseClaimMailGate frees exactly the revision held wrote.
	ReleaseClaimMailGate(held context.Context) error
}

// ClaimMailAdmission is one outermost gated call's durable app authority.
type ClaimMailAdmission interface {
	// Context carries a live permit for every admitted app.
	Context() context.Context
	// Settle runs once, after the call joined: it revokes the permits and
	// records each app effect complete (no error, or no attempt reached a
	// mutating surface) or uncertain (an error after an attempt). attempts
	// is exactly what THIS admission's call reached, never another
	// admission's under the same gate (review P2-9); an uncertain effect
	// records them for the fenced reconciler.
	Settle(ctx context.Context, callErr error, attempts []hostops.MailAttempt) error
	// Release runs only after the shared gate's release CAS succeeded. It
	// ends each app's mutation operation. An uncertain effect is never
	// released; it waits for fenced reconciliation.
	Release(ctx context.Context) error
}

// ClaimMailAdmitter is called at most once per outermost gated call, while
// the gate is held, before the call's first backend, SES, DNS or store
// mutation. apps is complete, sorted and de-duplicated. A refusal leaves
// every admission unchanged.
type ClaimMailAdmitter interface {
	AdmitClaimMailApps(ctx context.Context, hold ClaimMailHold, apps []ClaimMailApp) (ClaimMailAdmission, error)
}

// ClaimMailStore marks the enabled mail path. Its inner store is the
// console's DATA boundary, never a raw bucket. Named private fields only.
type ClaimMailStore struct {
	data     release.Store
	gate     ClaimMailGate
	admitter ClaimMailAdmitter
	owner    ClaimMailOwner
}

var (
	_ release.Store              = ClaimMailStore{}
	_ release.ConditionalDeleter = ClaimMailStore{}
	_ release.Opener             = ClaimMailStore{}
	_ release.ModLister          = ClaimMailStore{}
	_ release.PageLister         = ClaimMailStore{}
	_ release.SizeLister         = ClaimMailStore{}
	_ release.BucketListDirs     = ClaimMailStore{}
)

func NewClaimMailStore(data release.Store, gate ClaimMailGate, admitter ClaimMailAdmitter, owner ClaimMailOwner) (ClaimMailStore, error) {
	if data == nil || gate == nil || admitter == nil || owner.Console == "" || owner.Enrollment == "" || owner.Controller == "" {
		return ClaimMailStore{}, ErrClaimMailPending
	}
	return ClaimMailStore{data: data, gate: gate, admitter: admitter, owner: owner}, nil
}

func (s ClaimMailStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	return s.data.Get(ctx, key)
}
func (s ClaimMailStore) List(ctx context.Context, prefix string) ([]string, error) {
	return s.data.List(ctx, prefix)
}
func (s ClaimMailStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	return s.data.ListDirs(ctx, prefix)
}
func (s ClaimMailStore) BucketShapedListDirs() bool { return release.HasBucketListDirs(s.data) }
func (s ClaimMailStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	return release.OpenObject(ctx, s.data, key)
}
func (s ClaimMailStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	if l, ok := s.data.(release.ModLister); ok {
		return l.ListMod(ctx, prefix)
	}
	return nil, ErrClaimMailPending
}
func (s ClaimMailStore) ListPage(ctx context.Context, prefix, after string, limit int) ([]string, error) {
	if l, ok := s.data.(release.PageLister); ok {
		return l.ListPage(ctx, prefix, after, limit)
	}
	return nil, ErrClaimMailPending
}
func (s ClaimMailStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
	if l, ok := s.data.(release.SizeLister); ok {
		return l.ListSizes(ctx, prefix)
	}
	return nil, ErrClaimMailPending
}
func (s ClaimMailStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
	etag, err := s.data.Put(ctx, key, body, tag)
	markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "Put", Key: key}, err)
	return etag, err
}
func (s ClaimMailStore) Delete(ctx context.Context, key string) error {
	err := s.data.Delete(ctx, key)
	markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "Delete", Key: key}, err)
	return err
}
func (s ClaimMailStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
	err := release.DeleteIfMatch(ctx, s.data, key, etag)
	markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "DeleteIfMatch", Key: key}, err)
	return err
}

type claimMailToken struct{}
type claimMailScopeKey struct{}
type claimMailCapsKey struct{}
type claimMailAttemptsKey struct{}
type claimMailCaps map[ClaimMailApp]struct{}

type claimMailHeld struct {
	ID    string
	Owner ClaimMailOwner
	// Unadmitted: a mutating surface was reached under the gate outside any
	// admission (no durable effect records it). The gate is then retained
	// on failure, since nothing else can say what happened.
	Unadmitted atomic.Bool
	Retain     atomic.Bool // an effect is unsettled: keep the gate
	mu         sync.Mutex
	releases   []func(context.Context) error
}

// claimMailAttempts is ONE admission's record of the surfaces its call
// reached. It travels in the admitted context, so nested gated calls add to
// the outermost admission and a later admission starts empty.
type claimMailAttempts struct {
	mu   sync.Mutex
	list []hostops.MailAttempt
	err  error
}

func (a *claimMailAttempts) snapshot() ([]hostops.MailAttempt, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	return slices.Clone(a.list), a.err
}

// markClaimMailDirty records an attempt on an unnamed surface (kept for the
// existing claim_gate_test.go fixture). Outside an admission it marks the
// hold Unadmitted; inside one it makes that admission's failure uncertain
// with an attempt the reconciler cannot read back.
func markClaimMailDirty(ctx context.Context) {
	markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "unknown", Method: "unknown"}, nil)
}

// markClaimMailEffect records that a mutation reached its surface, unless a
// boundary refused it before any effect. An unknown error may have acted.
func markClaimMailEffect(ctx context.Context, at hostops.MailAttempt, err error) {
	if err != nil && hostops.RefusedBeforeEffect(err) {
		return
	}
	if rec, ok := ctx.Value(claimMailAttemptsKey{}).(*claimMailAttempts); ok {
		rec.mu.Lock()
		rec.list = append(rec.list, at)
		rec.err = errors.Join(rec.err, err)
		rec.mu.Unlock()
		return
	}
	if held, ok := ctx.Value(claimMailToken{}).(*claimMailHeld); ok {
		held.Unadmitted.Store(true)
	}
}

func WithClaimMailScope(ctx context.Context, scope ClaimMailScope) context.Context {
	return context.WithValue(ctx, claimMailScopeKey{}, scope)
}
func ClaimMailScopeFrom(ctx context.Context) (ClaimMailScope, bool) {
	scope, ok := ctx.Value(claimMailScopeKey{}).(ClaimMailScope)
	return scope, ok
}

// claimMailPlan says which apps one gated call can affect. Apps come from the
// arguments (the proposed members). Domain adds the persisted members of
// that domain (fresh read) and the app named on this account's global claim
// for it. AppDomains adds the same for every domain that app refers to.
//
// The claim's app is part of the persisted set because the claim is what
// the backend boundary binds an SES identity and its DNS to, and the claim
// write is scoped to it: an EnsureDomain re-run by a sibling app, a
// VerifyDomain stamp or a RemoveDomain release all touch it. Resolving it
// here admits it before any SES, DNS or store call, instead of letting the
// boundary refuse halfway through a walk. A claim held by another account
// adds nothing: the walk refuses or lapses it, and a takeover's write needs
// that account's authority, which no admission here grants.
//
// Refresh adds what a downstream RefreshPolicies or EnsureUser will need. A
// policy is written from the app's granted domains, and the backend boundary
// authorizes each granted domain against ALL its members. So every verified
// domain naming a refreshed app (and Domain itself, which may verify during
// the call) contributes its members and its claim app, and those members are
// refreshed in turn: the closure is the TRANSITIVE fixed point across shared
// domains (D1/A, D2/A+B, D3/B+C admits A, B and C). An unverified domain other
// than Domain grants nothing and adds nothing. Without it a nested call
// refuses after the SES and domain effects of the outer call, retaining the
// shared gate.
type claimMailPlan struct {
	Apps       []string
	Domain     string
	AppDomains string
	Refresh    bool
}

func claimPutDomainPlan(rec *DomainRecord) claimMailPlan {
	if rec == nil {
		return claimMailPlan{}
	}
	return claimMailPlan{Apps: rec.Apps, Domain: rec.Domain}
}

func (p claimMailPlan) resolve(ctx context.Context, s release.Store, acct string) ([]ClaimMailApp, error) {
	if !release.ValidAccount(acct) {
		return nil, fmt.Errorf("mail: invalid account %q", acct)
	}
	set := map[string]struct{}{}
	for _, app := range p.Apps {
		set[app] = struct{}{}
	}
	claimApp := func(domain string) error {
		c, err := getDomainClaim(ctx, s, domain)
		if err != nil {
			return err
		}
		if c != nil && c.Account == acct && c.App != "" {
			set[c.App] = struct{}{}
		}
		return nil
	}
	if p.Domain != "" {
		if !ValidDomain(p.Domain) {
			return nil, fmt.Errorf("mail: invalid domain %q", p.Domain)
		}
		rec, _, err := GetDomain(ctx, s, acct, p.Domain)
		if err != nil {
			return nil, err
		}
		if rec != nil {
			for _, app := range rec.Apps {
				set[app] = struct{}{}
			}
		}
		if err = claimApp(p.Domain); err != nil {
			return nil, err
		}
	}
	if p.AppDomains != "" {
		// A complete listing: a malformed record aborts the census.
		recs, err := StrictListDomains(ctx, s, acct)
		if err != nil {
			return nil, err
		}
		for _, rec := range recs {
			if slices.Contains(rec.Apps, p.AppDomains) {
				for _, app := range rec.Apps {
					set[app] = struct{}{}
				}
				if err = claimApp(rec.Domain); err != nil {
					return nil, err
				}
			}
		}
	}
	if p.Refresh {
		refreshed := map[string]struct{}{}
		for _, app := range p.Apps {
			refreshed[app] = struct{}{}
		}
		if p.Domain != "" {
			rec, _, err := GetDomain(ctx, s, acct, p.Domain)
			if err != nil {
				return nil, err
			}
			if rec != nil {
				for _, app := range rec.Apps {
					refreshed[app] = struct{}{}
				}
			}
		}
		// A complete listing: an unreadable or malformed record aborts.
		recs, err := StrictListDomains(ctx, s, acct)
		if err != nil {
			return nil, err
		}
		// Fixed point: a domain in the closure pulls all its members in, and a
		// newly pulled member can in turn use another relevant domain, so
		// repeat until a pass adds no app. Each domain is taken at most once.
		taken := map[string]struct{}{}
		for changed := true; changed; {
			changed = false
			for _, rec := range recs {
				if _, done := taken[rec.Domain]; done {
					continue
				}
				if len(rec.VerifiedARNs()) == 0 && rec.Domain != p.Domain {
					continue
				}
				touched := false
				for _, app := range rec.Apps {
					if _, ok := refreshed[app]; ok {
						touched = true
						break
					}
				}
				if !touched {
					continue
				}
				taken[rec.Domain] = struct{}{}
				for _, app := range rec.Apps {
					set[app] = struct{}{}
					if _, ok := refreshed[app]; !ok {
						refreshed[app] = struct{}{}
						changed = true
					}
				}
				if err = claimApp(rec.Domain); err != nil {
					return nil, err
				}
			}
		}
	}
	names := make([]string, 0, len(set))
	for app := range set {
		if !release.ValidApp(app) {
			return nil, fmt.Errorf("mail: invalid app %q", app)
		}
		names = append(names, app)
	}
	sort.Strings(names)
	out := make([]ClaimMailApp, len(names))
	for i, app := range names {
		out[i] = ClaimMailApp{Account: acct, App: app}
	}
	return out, nil
}

// claimMailAdmitted must run inside WithClaimMailResource on a marked store.
func claimMailAdmitted(ctx context.Context, s release.Store, acct string, plan claimMailPlan, call func(context.Context) error) error {
	held, ok := ctx.Value(claimMailToken{}).(*claimMailHeld)
	if !ok {
		return ErrClaimMailPending
	}
	marked, ok := s.(ClaimMailStore)
	if !ok || marked.admitter == nil {
		return ErrClaimMailPending
	}
	apps, err := plan.resolve(ctx, s, acct)
	if err != nil {
		return err
	}
	if installed, nested := ctx.Value(claimMailCapsKey{}).(claimMailCaps); nested {
		// Nested: only the preflighted set is usable. No admission read, no
		// second hook call, no widening.
		for _, a := range apps {
			if _, covered := installed[a]; !covered {
				return ErrClaimMailPending
			}
		}
		return call(ctx)
	}
	if len(apps) == 0 {
		return ErrClaimMailPending
	}
	caps := claimMailCaps{}
	for _, a := range apps {
		caps[a] = struct{}{}
	}
	admission, err := marked.admitter.AdmitClaimMailApps(ctx, ClaimMailHold{Gate: held.ID, Owner: held.Owner}, slices.Clone(apps))
	if err != nil {
		return err
	}
	if admission == nil || admission.Context() == nil {
		return ErrClaimMailPending
	}
	// The admitted context descends from ctx, so it still carries the held
	// gate token and the gate lease.
	rec := &claimMailAttempts{}
	admitted := context.WithValue(context.WithValue(admission.Context(), claimMailCapsKey{}, caps), claimMailAttemptsKey{}, rec)
	result := call(admitted)
	attempts, effectErr := rec.snapshot()
	result = errors.Join(result, effectErr)
	if err := admission.Settle(ctx, result, attempts); err != nil {
		// An unsettled effect retains both the app authority and the gate.
		held.Retain.Store(true)
		return errors.Join(result, err)
	}
	if result != nil && len(attempts) != 0 {
		held.Retain.Store(true)
		return result
	}
	held.mu.Lock()
	held.releases = append(held.releases, admission.Release)
	held.mu.Unlock()
	return result
}

// WithClaimMailResource serializes one enabled mail mutation under the
// shared gate. An unmarked store runs the original body with no gate,
// admission or domain read. The gate is released only when no effect is
// retained; the app authorities are released only after that release CAS.
func WithClaimMailResource(ctx context.Context, s release.Store, call func(context.Context) error) error {
	marked, enabled := s.(ClaimMailStore)
	if !enabled {
		return call(ctx)
	}
	if marked.gate == nil || marked.admitter == nil {
		return ErrClaimMailPending
	}
	if nested, ok := ctx.Value(claimMailToken{}).(*claimMailHeld); ok {
		if nested.Owner != marked.owner {
			return ErrClaimMailPending
		}
		if err := marked.gate.CheckClaimMailGate(ctx); err != nil {
			return err
		}
		return call(ctx)
	}
	scope, _ := ClaimMailScopeFrom(ctx)
	heldCtx, gateID, err := marked.gate.AcquireClaimMailGate(ctx, marked.owner, scope)
	if err != nil {
		return err
	}
	held := &claimMailHeld{ID: gateID, Owner: marked.owner}
	result := call(context.WithValue(heldCtx, claimMailToken{}, held))
	if held.Retain.Load() || (result != nil && held.Unadmitted.Load()) {
		// Retained for fenced reconciliation; no timer or heartbeat frees it.
		return result
	}
	saveCtx, cancel := context.WithTimeout(context.WithoutCancel(heldCtx), releaseTimeout)
	defer cancel()
	if err = marked.gate.ReleaseClaimMailGate(saveCtx); err != nil {
		// The gate stays held, so its app authorities stay held too.
		return errors.Join(result, err)
	}
	held.mu.Lock()
	releases := held.releases
	held.mu.Unlock()
	var errs []error
	for _, end := range releases {
		errs = append(errs, end(saveCtx))
	}
	return errors.Join(append(errs, result)...)
}