// The plans pass (plans-caps-packs spec §7, plans 2): learn whether plans are
// on for this deployment and, only if they are, read the limits record of
// every account this box serves into a last-known-good cache the activator
// asks. Rides the same passes as telemetryPass.
//
// Self-hosters are never interrupted: with no marker (no catalog on the
// console) the pass reads one object and applies nothing, a box with
// CARLOS_PLAN_LIMITS=off reads nothing at all, and an account with no record
// runs exactly as it always has.
package host

import (
	"context"
	"log"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

// planLimitsSnapshot is immutable once stored; nil in planLimits means plans
// are off on this box.
type planLimitsSnapshot struct {
	byAccount map[string]plans.Limits
}

var planLimits atomic.Pointer[planLimitsSnapshot]

// plansPassMu serialises whole passes: the full pass and the five-minute pass
// both run plansPass, and a stalled pass must not publish over a newer one
// (Astra plans-2 finding 2). Readers never take it.
var plansPassMu sync.Mutex

// planLimitsFailures counts consecutive failing passes for log throttling.
var planLimitsFailures atomic.Int64

// planLimitsSwitchOn reads CARLOS_PLAN_LIMITS: off, false, 0 or no turn plans
// off on this box; anything else, including unset, leaves them to the marker.
func planLimitsSwitchOn(getenv func(string) string) bool {
	switch strings.ToLower(strings.TrimSpace(getenv("CARLOS_PLAN_LIMITS"))) {
	case "off", "false", "0", "no":
		return false
	}
	return true
}

// planLimitsFor is sqid's limits, or false when plans are off or the account
// has no record.
func planLimitsFor(sqid string) (plans.Limits, bool) {
	snap := planLimits.Load()
	if snap == nil {
		return plans.Limits{}, false
	}
	l, ok := snap.byAccount[sqid]
	return l, ok
}

func plansPass(ctx context.Context, s release.Store, reg *registry.Registry, getenv func(string) string) {
	plansPassMu.Lock()
	defer plansPassMu.Unlock()
	if getenv == nil {
		getenv = os.Getenv
	}
	if !planLimitsSwitchOn(getenv) {
		planLimits.Store(nil)
		return
	}
	marker, _, err := plans.GetMarker(ctx, s)
	if err != nil {
		planLimitsFailed("marker: %v", err)
		return // keep the last good cache
	}
	if marker == nil || !marker.Enabled {
		planLimits.Store(nil)
		planLimitsFailures.Store(0)
		return
	}
	routes, err := reg.List()
	if err != nil {
		planLimitsFailed("routes: %v", err)
		return
	}
	prev := planLimits.Load()
	next := &planLimitsSnapshot{byAccount: map[string]plans.Limits{}}
	seen := map[string]bool{}
	failed := false
	for _, rt := range routes {
		if rt.Kind == "status" {
			continue
		}
		sqid := release.RouteAccount(rt)
		if sqid == "" || seen[sqid] {
			continue
		}
		seen[sqid] = true
		rec, _, err := plans.GetLimits(ctx, s, sqid)
		switch {
		case err != nil:
			failed = true
			planLimitsFailed("%s: %v", sqid, err)
			if prev != nil {
				if l, ok := prev.byAccount[sqid]; ok {
					next.byAccount[sqid] = l
				}
			}
		case rec != nil:
			next.byAccount[sqid] = rec.Limits
		}
	}
	planLimits.Store(next)
	if !failed {
		planLimitsFailures.Store(0)
	}
}

// planLimitsFailed logs the first failure of a run and every tenth after it.
func planLimitsFailed(format string, args ...any) {
	if n := planLimitsFailures.Add(1); n == 1 || n%10 == 0 {
		log.Printf("agent: plans pass: "+format+" (keeping the last good limits)", args...)
	}
}

// planIdleFor is the activator's plans idle hook: the route's account's idle
// window from its limits, or 0 (no opinion) when plans are off or the account
// has no record.
func planIdleFor(rt registry.Route) time.Duration {
	l, ok := planLimitsFor(release.RouteAccount(rt))
	if !ok || l.IdleSeconds <= 0 {
		return 0
	}
	return time.Duration(l.IdleSeconds) * time.Second
}