package host

// The box agent's reads through the S3 change index (S3 change-index design
// §4.3-§4.5, discussion #336). Armed only by CARLOS_CHANGE_INDEX=read; unset,
// every pointer below stays nil and every loop reads the bucket exactly as
// before.

import (
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"fmt"
	"log"
	"math/rand/v2"
	"os"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/edgesfile"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// changeIndexArmed reports whether this process reads through the change
// index. Only the exact value "read" arms it (spec §7).
func changeIndexArmed(getenv func(string) string) bool {
	return getenv("CARLOS_CHANGE_INDEX") == "read"
}

// armedChangeWatch is the process's one watch on the change index, and
// armedRecordView the record view that follows it. RunAgent sets both once
// at start-up when armed; nil means today's behaviour everywhere.
var (
	armedChangeWatch atomic.Pointer[release.ChangeWatch]
	armedRecordView  atomic.Pointer[provision.RecordView]
)

// backstopFullDue is the backstop's request for the unrestricted full pass:
// set after each Rebuild, taken by fastTick on the fast converge loop's own
// goroutine (fastDeps' maps are single-goroutine by construction).
var backstopFullDue atomic.Bool

// armChangeIndex arms the change index for this process: a watch on the
// deployment store, and a record view over the same store. A fleet box
// stays disarmed with the variable set: its store door has no conditional
// read, so its watch would be unusable and every Refresh would walk anyway
// (spec §3.4). The view's store must take no agent lock on any read path
// (RecordView.mu is a leaf, spec §4.3): open hands back the plain deployment
// store the other loops use.
func armChangeIndex(ctx context.Context, getenv func(string) string, open func(context.Context) (release.Store, error)) *provision.RecordView {
	if !changeIndexArmed(getenv) {
		return nil
	}
	if shouldSkipPublisherElection(getenv) {
		log.Printf("agent: change index: CARLOS_CHANGE_INDEX=read is ignored on a fleet box; record reads stay on the walk")
		return nil
	}
	store, err := open(ctx)
	if err != nil {
		log.Printf("agent: change index: not armed: %v", err)
		return nil
	}
	watch := release.NewChangeWatch(store)
	v := provision.NewRecordView(store, watch)
	armedChangeWatch.Store(watch)
	armedRecordView.Store(v)
	log.Printf("agent: change index armed: record walks read the record view (backstop %s)", release.ChangeIndexBackstop)
	return v
}

// newChangeGate is a read gate over the armed watch, built by a loop when it
// starts: RunAgent arms the index before it starts any loop, so the watch a
// gate is built over is the one the process keeps. Disarmed, the watch is nil
// and every read is due, as before the index (spec §5).
func newChangeGate() *release.ReadGate { return newChangeGateOver(armedChangeWatch.Load()) }

// newChangeGateOver is the one place the gated reads' maximum age is set: a
// successful read stands in for the next one for at most the backstop.
func newChangeGateOver(w *release.ChangeWatch) *release.ReadGate {
	return release.NewReadGate(w, release.ChangeIndexBackstop)
}

// viewUse remembers whether the last viewStore call used the view, so the
// switch is logged once per change of state, never per tick.
var viewUse struct {
	mu   sync.Mutex
	last int // 0 none yet, 1 using the view, 2 walking
}

func noteViewUse(ok bool, why string) {
	state := 2
	if ok {
		state = 1
	}
	viewUse.mu.Lock()
	changed := viewUse.last != state
	viewUse.last = state
	viewUse.mu.Unlock()
	if !changed {
		return
	}
	if ok {
		log.Printf("agent: record view: record walks read the view")
		return
	}
	log.Printf("agent: record view: %s; record walks read the bucket until it can vouch again", why)
}

// viewStore is the store a record-family walker reads: the armed view's
// current snapshot over live, refreshed first, when the refresh returned nil
// and the snapshot is Complete. Otherwise ok is false and s is live, so the
// caller walks the bucket as before the index (spec §3.4).
//
// Only record-family walkers take it (Reconcile, ReconcileSidecars,
// steering), and the moves pass, which asks it only whether there is any
// move work and acts on its own walk of the bucket (movesIdle). Its ListDirs of
// apps/<a>/ names only apps with record-family state, so app enumerators
// (hostops.Apps, purge and activation sweeps, mail, economics) must never
// get it.
//
// live must take no agent lock on any read path: RecordView.mu is a leaf
// (lock order w.pass -> reconcileMu -> RecordView.mu).
func viewStore(ctx context.Context, live release.Store) (s release.Store, epoch uint64, ok bool) {
	v := armedRecordView.Load()
	if v == nil || live == nil {
		return live, 0, false
	}
	err := v.Refresh(ctx)
	sn := v.Snapshot()
	switch {
	case err != nil:
		noteViewUse(false, fmt.Sprintf("refresh failed (%v)", err))
		return live, 0, false
	case sn == nil || !sn.Complete:
		noteViewUse(false, "the snapshot is incomplete")
		return live, 0, false
	}
	noteViewUse(true, "")
	return sn.Store(live), sn.Epoch, true
}

// recordViewBackstopLoop is the backstop (spec §4.4): Rebuild at start and
// every release.ChangeIndexBackstop, jittered by +-10%, each followed by the
// unrestricted full pass, so every leg is re-run over everything at least
// that often. The full pass runs on the fast converge loop's goroutine
// (backstopFullDue); on a box where that loop is not armed only the
// rebuild runs.
func recordViewBackstopLoop(ctx context.Context, v *provision.RecordView) {
	for {
		recordViewBackstop(ctx, v, &backstopFullDue)
		select {
		case <-ctx.Done():
			return
		case <-time.After(backstopEvery()):
		}
	}
}

// recordViewBackstop is one backstop round: the rebuild, then the request
// for a full pass, whether or not the rebuild succeeded (a failed one
// leaves the view unable to vouch, so that pass walks).
func recordViewBackstop(ctx context.Context, v *provision.RecordView, due *atomic.Bool) {
	if err := v.Rebuild(ctx); err != nil {
		log.Printf("agent: record view backstop: %v", err)
	}
	due.Store(true)
}

// backstopEvery is release.ChangeIndexBackstop jittered by +-10%.
func backstopEvery() time.Duration {
	b := release.ChangeIndexBackstop
	return b - b/10 + rand.N(b/5)
}

// indexScopes is the scopes of obs whose generation differs from acted's,
// sorted, kept to the scopes a pass acts on (apps/<a>/<b> and accounts/<a>).
// The control/ scopes belong to other readers.
func indexScopes(acted, obs map[string]uint64) []string {
	var moved []string
	for scope, gen := range obs {
		if have, ok := acted[scope]; ok && have == gen {
			continue
		}
		if _, _, ok := provision.ScopeApp(scope); ok {
			moved = append(moved, scope)
			continue
		}
		if a, ok := strings.CutPrefix(scope, "accounts/"); ok && a != "" && !strings.Contains(a, "/") {
			moved = append(moved, scope)
		}
	}
	sort.Strings(moved)
	return moved
}

// movedRoutes keeps the routes whose (account, app) has an app scope in
// moved. Account scopes keep nothing here: their effect on a box (an
// account soft-deleted) is Reconcile's, which runs whole.
func movedRoutes(moved []string) func(registry.Route) bool {
	apps := map[[2]string]bool{}
	for _, scope := range moved {
		if a, b, ok := provision.ScopeApp(scope); ok {
			apps[[2]string{a, b}] = true
		}
	}
	return func(rt registry.Route) bool { return apps[[2]string{release.RouteAccount(rt), rt.App}] }
}

// passLegs is the serial-triggered pass set's inputs (fastConvergeLoop),
// shared by the full pass and the scoped one so the two run the same legs
// in the same order and cannot drift.
type passLegs struct {
	store   release.Store
	reg     *registry.Registry
	pub     ed25519.PublicKey
	optDir  string
	restart func(string) error
	opts    []release.PassOption
	priv    *ecdh.PrivateKey // nil: no config leg (no CARLOS_SECRETS_KEY)
	dd      string
	svc     *activator.Service
	writes  hostwrite.Config
	bs      *buildStates
	cfg     func() provision.ReconcileConfig // read per pass, as provisionTickConfig always was
}

// passAfterReconcileHook, when set, runs in every pass right after its
// reconcile leg. Tests only.
var passAfterReconcileHook func()

// full is the serial-triggered pass set, in spec order: mint routes
// (reconcile), then reconcile aliases, then stage them (adopt), then deliver
// their config, over every hosted route.
func (l *passLegs) full(ctx context.Context) { l.run(ctx, "serial pass stages", nil) }

// scoped is the pass an index-only change runs (spec §4.5, codex 10): the
// same legs, with build states, adopt and config sync restricted to the
// hosted routes whose app scope moved, and without the box-wide alias and
// plans legs, so changing one app re-reads that app only. Reconcile and the moves pass
// run whole, over the record view. Telemetry resolves every hosted app (its
// sink takes a whole map), but its reads are gated on their own scopes
// (telemetryReads), so only the moved apps' files are read.
func (l *passLegs) scoped(ctx context.Context, moved []string) {
	l.run(ctx, "scoped pass stages", movedRoutes(moved))
}

// run is one pass set; keep nil is the unrestricted full pass. Each error is
// one log line, never fatal: the hourly floors stand underneath every leg.
func (l *passLegs) run(ctx context.Context, what string, keep func(registry.Route) bool) {
	// Each stage's time, logged together when the set is slow, so the
	// journal says WHICH stage held a promote back.
	var took []string
	t := time.Now()
	mark := func(name string) {
		took = append(took, fmt.Sprintf("%s %s", name, time.Since(t).Round(100*time.Millisecond)))
		t = time.Now()
	}
	defer func(start time.Time) { logSlow(what, start, strings.Join(took, ", ")) }(time.Now())

	if err := reconcileSerialized(ctx, l.store, l.reg, l.cfg()); err != nil {
		log.Printf("agent: fast converge reconcile: %v", err)
	}
	mark("reconcile")
	if hook := passAfterReconcileHook; hook != nil {
		hook()
	}
	// The registry's build-state copies, refreshed at every reconcile
	// (settings spec §4.2), before adopt reads them; refreshAll keeps
	// to its own budget, and the stage is timed apart from reconcile.
	// A scoped pass refreshes only the moved apps' hosts.
	if l.bs != nil {
		l.bs.refreshKept(ctx, keep)
		mark("build states")
	}
	telemetryPass(ctx, l.store, l.reg)
	mark("telemetry")
	// The plans pass is box-wide (the marker, then every served account's
	// limits record, published as one whole snapshot), so a scoped pass
	// leaves it out: the full pass, the instances floor and the backstop's
	// full pass run it.
	if keep == nil {
		plansPass(ctx, l.store, l.reg, os.Getenv)
		mark("plans")
	}
	// The moves pass takes its own snapshot inside w.pass (spec §4.3): it
	// is handed the plain store, never this pass's reconcile snapshot.
	runMovesPass(ctx, l.store)
	mark("moves")
	runRunsPass(ctx)
	mark("runs")
	// A serial move may be a forwarded db read: list the queue now. The
	// scoped pass nudges too: a request written through an indexed store
	// may move only the index. The nudge is a non-blocking channel send.
	nudgeDBReads()
	// Aliases are box-wide (every domain claim, every edges file), so the
	// scoped pass leaves them out: the serial-triggered full pass, the
	// instances floor and the backstop's full pass run them.
	if keep == nil {
		// Folded once per pass and handed in, never re-read inside the
		// loop: this leg runs on every serial bump (a seconds cadence),
		// and the fold is a LoadBoxes plus a List plus one Get per edges
		// file. On a fleet of under a dozen boxes that is a handful of
		// Gets per bump, which is accepted here rather than cached — a
		// cache would need its own staleness rule, and a STALE route set
		// is the one thing a security guard must not have. Aliases read
		// the domain claims, not the record families, so they read the
		// bucket.
		deploymentRoutes := edgesfile.DeploymentRoutes(ctx, l.store, time.Now())
		if err := domains.ReconcileAliases(ctx, l.store, l.reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil {
			log.Printf("agent: fast converge alias pass: %v", err)
		}
		mark("aliases")
	}
	if err := release.PassFilter(ctx, l.reg, l.store, l.pub, l.optDir, l.restart, keep, l.opts...); err != nil {
		log.Printf("agent: fast converge adopt pass: %v", err)
	}
	mark("adopt")
	if l.priv != nil {
		if err := syncExecSerializedFiltered(ctx, keep, l.reg, l.store, l.priv, l.dd, l.svc, l.writes); err != nil {
			log.Printf("agent: fast converge config sync: %v", err)
		}
		mark("config")
	}
}