package host

import (
	"context"
	"crypto/ecdh"
	"errors"
	"log"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// The config guard (amadan #213). A box takes an instance's lease, or mints
// its record, only if it can provide that instance's config.
//
// The failure it ends: a box with no CARLOS_SECRETS_KEY (or the wrong one)
// still admitted and woke socket-backed shared-pool instances in its region.
// Its config sync loops are off without the key, and ExecBackend.instanceEnv
// treats a missing <host>.config.env as optional, so the instance started
// without its secrets or store config. kpmg was down 1h54m on 2026-10-01
// because the UK box had no key. The console cannot prove what reconcile will
// admit, so the invariant lives here, at admission.
//
// One guard per process, built from CARLOS_SECRETS_KEY and the deploy store,
// and asked through Gate from three places, each BEFORE the side effect it
// guards: activator.ExecBackend.Acquire (before lease.Acquire),
// provision.Reconcile and provision.ReconcileSidecars (before a mint). The
// decision itself is appconfig.CanProvideConfig; this type adds what that
// pure function cannot: the box's key, the store, a short refusal cache, the
// once-per-host log and the list the status snapshot publishes.

const (
	// configGuardRefuseTTL bounds how long a refusal is answered from memory.
	// A request storm on a refused host would otherwise cost a set of bucket
	// reads per request, and the reconcile pass asks about every unminted
	// record every tick. It is also the retry interval after a store error:
	// the box can tell again within this long of the store coming back. An
	// admission is never cached: wakes are rare, and the answer must be
	// current when it is taken.
	configGuardRefuseTTL = 30 * time.Second
	// configGuardListCap bounds the refusal list in the status snapshot; the
	// total is published beside it so a capped list reads as long.
	configGuardListCap = 100
	// configGuardRefusalMaxAge is how long a host stays in the published list
	// without being asked about again. A wake-time refusal is only re-asked
	// when someone requests the host, so this is what makes the list mean
	// "currently refused" rather than "ever refused since boot".
	configGuardRefusalMaxAge = 24 * time.Hour
)

// errNoDeployStore is openDeployStoreEnv's answer for a box with no
// deployment store configured at all: no bucket, no dir, no fleet identity.
//
// That is NOT proof that nothing is sealed. Config lives in the deployment
// store, but CARLOS_HIBERNATE_BUCKET enables leases independently of it, so a
// box can wake instances whose sealed config sits in a store it was never told
// about. The guard therefore refuses (no_config_source) unless the operator
// has said, with CARLOS_NO_SEALED_CONFIG=1 (or "true"), that this deployment
// has no sealed config: the config-free development setup. The flag only
// matters when there is no store; with one, config is read as usual.
var errNoDeployStore = errors.New("neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set")

type guardRefusal struct {
	reason string
	since  time.Time
	last   time.Time
	count  int
}

type guardVerdict struct {
	err   error
	until time.Time
}

type configGuard struct {
	priv      *ecdh.PrivateKey
	keyState  appconfig.KeyState
	openStore func(context.Context) (release.Store, error)
	now       func() time.Time
	// noSealedConfig is the operator's explicit "this deployment has no
	// sealed config" (CARLOS_NO_SEALED_CONFIG), honoured only when there is
	// no deployment store at all.
	noSealedConfig bool

	mu       sync.Mutex
	bypassed map[string]bool // hosts already logged as admitted by the bypass
	store    release.Store
	verdicts map[string]guardVerdict // host -> cached refusal
	refused  map[string]*guardRefusal
}

// newConfigGuard reads CARLOS_SECRETS_KEY once, the way the config loops do.
// A key that does not parse is logged here, once, at arming; per-host lines
// say only that the host was refused.
func newConfigGuard(getenv func(string) string, openStore func(context.Context) (release.Store, error)) *configGuard {
	g := &configGuard{
		openStore: openStore,
		now:       time.Now,
		verdicts:  map[string]guardVerdict{},
		refused:   map[string]*guardRefusal{},
	}
	g.noSealedConfig = noSealedConfigFlag(getenv)
	raw := strings.TrimSpace(getenv("CARLOS_SECRETS_KEY"))
	switch {
	case raw == "":
		g.keyState = appconfig.KeyAbsent
	default:
		priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
		if err != nil {
			log.Printf("agent: config guard: %v", err)
			g.keyState = appconfig.KeyUnparseable
		} else {
			g.priv, g.keyState = priv, appconfig.KeyOK
		}
	}
	return g
}

// noSealedConfigFlag is the operator's CARLOS_NO_SEALED_CONFIG, read the
// one way the guard and the instance key's configuration reader share.
func noSealedConfigFlag(getenv func(string) string) bool {
	switch strings.ToLower(strings.TrimSpace(getenv("CARLOS_NO_SEALED_CONFIG"))) {
	case "1", "true":
		return true
	}
	return false
}

// needsBoxConfig reports whether a route is one this box would spawn with a
// config.env of its own: a socket-backed instance or a sidecar worker that
// belongs to an app. Static sites, addr-backed routes, Backing=unit routes
// (whose config the root adopt pass delivers), platform routes and app-less
// rows take no part: the guard must not touch them.
//
// The exclusion is Backing and nothing else, because Backing is what
// SplitBackend dispatches on. A socket-backed hibernating instance may carry a
// Unit name with an empty Backing (the record validator and `carlos route
// -unit` allow it) and still reaches ExecBackend.Acquire.
func needsBoxConfig(rt registry.Route) bool {
	if rt.App == "" || rt.Platform || rt.Backing == registry.BackingUnit {
		return false
	}
	switch {
	case rt.IsSidecar():
		return true
	case rt.Kind == "instance":
		return rt.Socket != ""
	}
	return false
}

// Gate is the seam both ExecBackend.ConfigGate and ReconcileConfig.ConfigGate
// are wired to. nil means this box can provide rt's config (or none is
// needed); an error means refuse, and is an *appconfig.ConfigRefusal.
func (g *configGuard) Gate(ctx context.Context, rt registry.Route) error {
	if !needsBoxConfig(rt) {
		return nil
	}
	now := g.now()
	if err := g.cached(rt.Host, now); err != nil {
		g.note(rt.Host, err, now)
		return err
	}
	err := g.decide(ctx, rt)
	if err != nil {
		g.mu.Lock()
		g.verdicts[rt.Host] = guardVerdict{err: err, until: now.Add(configGuardRefuseTTL)}
		g.mu.Unlock()
		g.note(rt.Host, err, now)
		return err
	}
	g.clear(rt.Host)
	return nil
}

func (g *configGuard) cached(host string, now time.Time) error {
	g.mu.Lock()
	defer g.mu.Unlock()
	v, ok := g.verdicts[host]
	if !ok {
		return nil
	}
	if !now.Before(v.until) {
		delete(g.verdicts, host)
		return nil
	}
	return v.err
}

func (g *configGuard) decide(ctx context.Context, rt registry.Route) error {
	s, err := g.deployStore(ctx)
	if errors.Is(err, errNoDeployStore) {
		if g.noSealedConfig {
			// The operator said this deployment has none. Said loudly, once
			// per host: this is the one path that admits without looking.
			g.noteBypass(rt.Host)
			return nil
		}
		return &appconfig.ConfigRefusal{Reason: appconfig.ReasonNoSource, Err: err}
	}
	if err != nil {
		return &appconfig.ConfigRefusal{Reason: appconfig.ReasonStoreError, Err: err}
	}
	return appconfig.CanProvideConfig(ctx, s, g.priv, g.keyState, release.RouteAccount(rt), rt.App, rt.Environment, rt.Host)
}

// configGuardBypassStatus is the private status word published while the
// CARLOS_NO_SEALED_CONFIG bypass is in effect.
const configGuardBypassStatus = "bypassed (CARLOS_NO_SEALED_CONFIG)"

// noteBypass logs, once per host, that a placed instance was admitted without
// its sealed config being checked.
func (g *configGuard) noteBypass(host string) {
	g.mu.Lock()
	if g.bypassed == nil {
		g.bypassed = map[string]bool{}
	}
	seen := g.bypassed[host]
	g.bypassed[host] = true
	g.mu.Unlock()
	if !seen {
		log.Printf("agent: config guard: BYPASS: admitting %s WITHOUT checking its sealed config, because CARLOS_NO_SEALED_CONFIG is set and this box has no deployment store. Never set it on a box that serves tenants", host)
	}
}

// Bypassed reports whether the CARLOS_NO_SEALED_CONFIG bypass is in effect:
// the flag is set AND the box has no deployment store configured. A flag set
// beside a store changes nothing, so it does not count.
func (g *configGuard) Bypassed() bool {
	if !g.noSealedConfig {
		return false
	}
	_, err := g.deployStore(context.Background())
	return errors.Is(err, errNoDeployStore)
}

// Status is the private status-block word for the guard's mode: the bypass
// words while it is in effect, "" otherwise.
func (g *configGuard) Status() string {
	if g.Bypassed() {
		return configGuardBypassStatus
	}
	return ""
}

// Announce logs the guard's mode once at startup. The bypass is the only mode
// worth shouting about, and it is shouted whether or not it is in effect, so a
// flag copied into the wrong host.env is visible either way.
func (g *configGuard) Announce() {
	switch {
	case g.Bypassed():
		log.Printf("agent: config guard: BYPASSED. CARLOS_NO_SEALED_CONFIG is set and no deployment store is configured, so this box admits placed exec instances without checking their sealed config. Intended for config-free development only; it is listed in this box's status snapshot")
	case g.noSealedConfig:
		log.Printf("agent: config guard: CARLOS_NO_SEALED_CONFIG is set but a deployment store is configured, so it is ignored and sealed config is checked as usual")
	}
}

// deployStore opens the store on first use and keeps it. A failure is not
// kept, so the next ask (after the refusal TTL) tries again.
func (g *configGuard) deployStore(ctx context.Context) (release.Store, error) {
	g.mu.Lock()
	s := g.store
	g.mu.Unlock()
	if s != nil {
		return s, nil
	}
	s, err := g.openStore(ctx)
	if err != nil {
		return nil, err
	}
	g.mu.Lock()
	g.store = s
	g.mu.Unlock()
	return s, nil
}

// note records a refusal for the snapshot and logs it: once per host, and
// again only when the reason changes. The detail in the log is the operator's;
// the snapshot carries the closed reason word alone.
func (g *configGuard) note(host string, err error, now time.Time) {
	reason := appconfig.ReasonStoreError
	var r *appconfig.ConfigRefusal
	if errors.As(err, &r) {
		reason = r.Reason
	}
	g.mu.Lock()
	e := g.refused[host]
	fresh := e == nil || e.reason != reason
	if e == nil {
		e = &guardRefusal{since: now}
		g.refused[host] = e
	}
	e.reason, e.last = reason, now
	e.count++
	g.mu.Unlock()
	if fresh {
		log.Printf("agent: config guard: refusing %s: %v. A box that cannot open an instance's sealed config does not take its lease or mint its record, so a keyed sibling can; it is listed in this box's status snapshot", host, err)
	}
}

// clear forgets a host that now passes, saying so once if it was refused.
func (g *configGuard) clear(host string) {
	g.mu.Lock()
	_, was := g.refused[host]
	delete(g.refused, host)
	delete(g.verdicts, host)
	g.mu.Unlock()
	if was {
		log.Printf("agent: config guard: %s can be provided from this box again", host)
	}
}

// Refusals is what the status snapshot publishes: the currently refused
// hosts, newest first and capped, plus how many there are in all. A host not
// asked about for configGuardRefusalMaxAge is dropped.
func (g *configGuard) Refusals() ([]status.ConfigRefusal, int) {
	now := g.now()
	g.mu.Lock()
	defer g.mu.Unlock()
	var out []status.ConfigRefusal
	for host, e := range g.refused {
		if now.Sub(e.last) > configGuardRefusalMaxAge {
			delete(g.refused, host)
			continue
		}
		out = append(out, status.ConfigRefusal{Host: host, Reason: e.reason, Since: e.since, Last: e.last, Count: e.count})
	}
	total := len(out)
	sort.Slice(out, func(i, j int) bool {
		if !out[i].Last.Equal(out[j].Last) {
			return out[i].Last.After(out[j].Last)
		}
		return out[i].Host < out[j].Host
	})
	if len(out) > configGuardListCap {
		out = out[:configGuardListCap]
	}
	return out, total
}

// armedConfigGuard is the process's one guard, set by RunAgent before any
// loop that asks it starts. Nil is "not armed": every consumer treats that as
// today's behaviour, so tests and callers that never arm one are unaffected.
var armedConfigGuard atomic.Pointer[configGuard]

func armConfigGuard(g *configGuard) { armedConfigGuard.Store(g) }

// configGate is the armed guard's Gate, or nil when none is armed.
func configGate() func(context.Context, registry.Route) error {
	if g := armedConfigGuard.Load(); g != nil {
		return g.Gate
	}
	return nil
}

// configGuardProbe feeds status.Probes.ConfigGuard: "" unless the bypass is
// in effect.
func configGuardProbe() string {
	if g := armedConfigGuard.Load(); g != nil {
		return g.Status()
	}
	return ""
}

// configRefusalProbe feeds status.Probes.ConfigRefusals.
func configRefusalProbe() ([]status.ConfigRefusal, int) {
	if g := armedConfigGuard.Load(); g != nil {
		return g.Refusals()
	}
	return nil, 0
}