package host

import (
	"context"
	"log"
	"math"
	"os"
	"sort"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/registry"
)

// Region HA spec 1 §5.4: with CARLOS_REGION_HA on, a relocatable host is
// published as home only by the box its claim's op.exec names, carrying
// the claim's holder.stamp so the edge's home map believes the newest
// installation (HA4). The edge never reads claims; this publisher does, on
// a slow cache, because the safety gate is the activator's fresh read at
// every spawn (claim C4), not this file.

const (
	// edgesClaimTTL is how long one claim read stands: reconcile's sweep
	// (H9). A held row's entry can be stale this long, which can only send
	// a request to a box that then refuses to spawn.
	edgesClaimTTL = 10 * time.Minute
	// edgesClaimReadsPerPass bounds the GETs one pass makes, so a box with
	// thousands of hosts spreads its reads over several passes instead of
	// stalling one. A host not read yet publishes as it did before this
	// file existed.
	edgesClaimReadsPerPass = 256
	// edgesClaimPassEvery is how often the loop runs an extra pass while
	// claim reads are due. The registry timer alone (5 to 6 minutes) would
	// take 40 minutes to refresh 2,000 hosts at 256 a pass; with these
	// passes every host is re-read within about edgesClaimTTL plus four
	// minutes.
	edgesClaimPassEvery = 30 * time.Second
	// edgesClaimPassDeadline bounds one pass's claim reads, so a store that
	// stops answering cannot stall the edges file. A read the deadline cuts
	// short is a failed read: the host keeps its last decision.
	edgesClaimPassDeadline = 20 * time.Second
	// edgesClaimLogEvery: a run of passes with failed claim reads is logged
	// on its first pass and every this many after.
	edgesClaimLogEvery = 60
)

// edgesDecision is what one claim says about this box and one host.
type edgesDecision struct {
	withhold bool  // the claim names another executor, or none
	stamp    int64 // holder.stamp when this box is the executor
}

// edgesClaimEntry is one host's last decision and last read attempt. A
// failed read moves attempted and keeps the decision, so a claim that
// cannot be read goes to the back of the queue instead of holding its
// place at the front (claim C6: an error is never absence).
type edgesClaimEntry struct {
	decision  edgesDecision
	decided   bool // some read of this claim has succeeded
	attempted time.Time
}

// edgesClaims caches one decision per host. It is used only by the edges
// file loop's goroutine.
type edgesClaims struct {
	self    string // this box's reference: the edges file's box_id
	read    func(ctx context.Context, host string) (hostclaim.Observed, error)
	now     func() time.Time
	entries map[string]edgesClaimEntry
	hosts   []string // the relocatable hosts of the last pass
	// passDeadline bounds one pass's claim reads (edgesClaimPassDeadline).
	passDeadline time.Duration
	logf         func(format string, a ...any)
	failingRun   int // consecutive passes with a failed claim read
	// addrs is the box's own addresses, read before every publication
	// (region HA spec 2 §6). It rides here because this cache exists
	// exactly when region HA is on.
	addrs AddressSource
}

func newEdgesClaims(self string, read func(context.Context, string) (hostclaim.Observed, error)) *edgesClaims {
	return &edgesClaims{self: self, read: read, now: time.Now, entries: map[string]edgesClaimEntry{},
		passDeadline: edgesClaimPassDeadline, logf: log.Printf, addrs: boxAddressSource(os.Getenv)}
}

// relocatableRow is spec 1 §2's relocatable host as a registry row shows
// it: a provisioned instance whose process is the activator's own exec
// child. Unit-backed, addr-backed, static and service routes keep their
// box and are published as before; sidecars are never published at all.
func relocatableRow(r registry.Route) bool {
	return r.Provisioned && r.Kind == "instance" && r.Backing == "" && r.Unit == "" &&
		r.Addr == "" && r.Addrs == "" && r.Socket != ""
}

// decideClaim turns one observation into a decision. An absent claim
// leaves the host as it was before claims existed: a region can only be
// given a second box once every host has one (spec §8).
func (c *edgesClaims) decideClaim(obs hostclaim.Observed) edgesDecision {
	if obs.Absent || obs.Claim == nil {
		return edgesDecision{}
	}
	cl := obs.Claim
	if cl.Holder == nil || cl.Op.Exec != c.self {
		return edgesDecision{withhold: true}
	}
	stamp := cl.Holder.Stamp
	if stamp > math.MaxInt64 {
		stamp = 0
	}
	return edgesDecision{stamp: int64(stamp)}
}

func (c *edgesClaims) dueAt(h string, now time.Time) bool {
	e, ok := c.entries[h]
	return !ok || now.Sub(e.attempted) >= edgesClaimTTL
}

// backlog reports whether any host of the last pass has a read due.
func (c *edgesClaims) backlog(now time.Time) bool {
	for _, h := range c.hosts {
		if c.dueAt(h, now) {
			return true
		}
	}
	return false
}

// decide returns a decision for each host that has one. Hosts whose last
// attempt is older than edgesClaimTTL are read, never-attempted first and
// then oldest attempt first, up to edgesClaimReadsPerPass. A read that
// fails keeps the previous decision; a host with no successful read has
// none, and publishes as before. Once ctx ends no further read is made:
// the hosts not reached stay due, in their place, for a later pass.
func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edgesDecision {
	now := c.now()
	c.hosts = append(c.hosts[:0], hosts...)
	want := make(map[string]bool, len(hosts))
	var due []string
	for _, h := range hosts {
		want[h] = true
		if c.dueAt(h, now) {
			due = append(due, h)
		}
	}
	for h := range c.entries {
		if !want[h] {
			delete(c.entries, h)
		}
	}
	sort.Slice(due, func(i, j int) bool {
		ei, iok := c.entries[due[i]]
		ej, jok := c.entries[due[j]]
		if iok != jok {
			return !iok // never attempted first
		}
		if !ei.attempted.Equal(ej.attempted) {
			return ei.attempted.Before(ej.attempted)
		}
		return due[i] < due[j]
	})
	if len(due) > edgesClaimReadsPerPass {
		due = due[:edgesClaimReadsPerPass]
	}
	var failed, made int
	var firstErr error
	var firstHost string
	for _, h := range due {
		if ctx.Err() != nil {
			break
		}
		e := c.entries[h]
		e.attempted = now
		made++
		if obs, err := c.read(ctx, h); err == nil {
			e.decision, e.decided = c.decideClaim(obs), true
		} else {
			if failed == 0 {
				firstErr, firstHost = err, h
			}
			failed++
		}
		c.entries[h] = e
	}
	c.logReads(made, failed, firstHost, firstErr)
	out := make(map[string]edgesDecision, len(c.entries))
	for h, e := range c.entries {
		if e.decided {
			out[h] = e.decision
		}
	}
	return out
}

// logReads logs a pass's failed claim reads, throttled: the first pass of
// a failing run and every edgesClaimLogEvery-th after it, then once when a
// pass's reads all succeed again.
func (c *edgesClaims) logReads(made, failed int, host string, err error) {
	if failed > 0 {
		c.failingRun++
		if c.failingRun == 1 || c.failingRun%edgesClaimLogEvery == 0 {
			c.logf("agent: edges file: %d of %d claim reads failed (failing pass #%d), first %s: %v; those hosts keep their last decision", failed, made, c.failingRun, host, err)
		}
		return
	}
	if made > 0 && c.failingRun > 0 {
		c.logf("agent: edges file: claim reads recovered after %d failing passes", c.failingRun)
		c.failingRun = 0
	}
}

// applyEdgesClaims withholds or stamps every relocatable row in hosts
// (edgesFileWrite's served map, before aliases join it, so an alias of a
// withheld host is withheld with it and an alias of a stamped host takes
// its stamp). claims nil leaves hosts untouched. A host hostclaim.Key
// rejects has no claim to read and publishes as before. The pass's reads
// share one deadline, claims.passDeadline.
func applyEdgesClaims(ctx context.Context, claims *edgesClaims, routes []registry.Route, hosts map[string]edgeHost) {
	if claims == nil {
		return
	}
	var reloc []string
	for _, r := range routes {
		if _, served := hosts[r.Host]; served && relocatableRow(r) {
			if _, err := hostclaim.Key(r.Host); err != nil {
				continue
			}
			reloc = append(reloc, r.Host)
		}
	}
	ctx, cancel := context.WithTimeout(ctx, claims.passDeadline)
	defer cancel()
	for host, d := range claims.decide(ctx, reloc) {
		h, ok := hosts[host]
		if !ok {
			continue
		}
		if d.withhold {
			delete(hosts, host)
			continue
		}
		h.Stamp = d.stamp
		hosts[host] = h
	}
}

// edgesLoopState is edgesFileLoop's per-tick decision: the registry and
// timer schedule, plus, with region HA on, an extra pass every
// edgesClaimPassEvery while claim reads are due. A claim pass is a
// publication like any other: it waits out the schedule's retry backoff
// after a failed write, and its own outcome moves that backoff, so a bucket
// that refuses every write is not asked every 30 seconds (Astra plan R2-1).
type edgesLoopState struct {
	sched      *edgesFileSchedule
	claims     *edgesClaims
	claimsNext time.Time
	schedDue   bool
	claimDue   bool
}

// due reports whether to run a pass on this tick.
func (s *edgesLoopState) due(now time.Time, fp string, fpOK bool) bool {
	s.schedDue = s.sched.due(now, fp, fpOK)
	s.claimDue = s.claims != nil && !now.Before(s.claimsNext) && !now.Before(s.sched.retryAt) && s.claims.backlog(now)
	return s.schedDue || s.claimDue
}

// timed reports whether the pass due just allowed is the timer's.
func (s *edgesLoopState) timed() bool { return s.schedDue && s.sched.timed }

// done records the pass's outcome. Every pass, claim passes included, goes
// through the schedule: a failure advances its backoff, and a success
// clears it and records the fingerprint the pass published. A claim pass
// is never the timer's (due cleared timed), so the timer does not move.
func (s *edgesLoopState) done(now time.Time, ok bool) {
	if s.schedDue || s.claimDue {
		s.sched.done(now, ok)
	}
	if s.claims != nil {
		s.claimsNext = now.Add(edgesClaimPassEvery)
	}
}