package edge

// prober.go — active health probing for multi-upstream addr routes (HA design
// Part II, issue #152). One goroutine per probed host GETs each of its
// upstreams on a fixed cadence and maintains the LIVE SET the request path
// (multiProxy) picks from; the same table is the dashboard's per-upstream
// state.
//
// Two rules shape everything below:
//
//  1. **Optimistic start.** Every upstream of a newly synced host starts LIVE.
//     A reload happens on any registry fingerprint move — an unrelated route
//     changing must never take a healthy fleet off the air for the first
//     fail_after × interval while probes catch up. Probing only ever REMOVES
//     confidence; it never withholds it.
//
//  2. **Probes dial upstreams directly**, never the edge's own public host, so
//     a probe cannot enter the activator and cannot wake a hibernating tenant
//     — the wake hazard that would otherwise make active probing incompatible
//     with hibernation.
//
//     The second half of that guarantee is a MINT-SIDE door, not Reload's:
//     provision.ValidateShape refuses `addrs` together with `hibernate: true`
//     outright (internal/provision/provision.go, "addr with hibernate"), so a
//     probed route is never activator-managed in the first place. Reload's
//     `hib` gate is NOT that door — it reads `Kind == "instance" && Hibernate
//     && !viaTunnel` and says nothing about addrs — so a hand-written registry
//     row could still pair the two, and would then hibernate a process whose
//     upstream the probe keeps marking down. Fixing that means widening the
//     gate, not trusting this comment.

import (
	"context"
	"fmt"
	"io"
	"log"
	"net/http"
	"slices"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/status"
)

// probeTimeout bounds a single probe request end to end (connect through
// body). Held at 2s per the spec: long enough that a loaded-but-alive
// upstream still answers, short enough that a black-holed one cannot stall a
// tick past the default 2s interval.
const probeTimeout = 2 * time.Second

// Defensive fallbacks for a spec that reaches the prober with a non-positive
// cadence or count. Sync runs every incoming spec through
// provision.Health.WithDefaults, which covers the ZEROES; these clamps exist
// for what WithDefaults lets through — NEGATIVES — because mint is not the only
// writer of a route's health JSON (registry.SetUpstreams writes the column
// directly, and the route PATCH endpoint will too), and neither hand-edited
// JSON nor a future writer is obliged to be sane.
//
// The interval clamp is the load-bearing one: time.NewTicker PANICS on a
// non-positive duration, so an unclamped interval_ms of -1 (or 0, on any path
// that reached newHostProber without normalization) would take the whole edge
// process down — every route on the box, not just the misconfigured one. A
// non-positive fail_after must likewise not fail an upstream away on a single
// blip, nor a non-positive rise_after rejoin it on a single pass.
//
// Only NON-POSITIVE values are replaced. A deliberately fast interval is the
// operator's call (and the tests' — they probe at 20ms).
const (
	fallbackProbeInterval = 2000 * time.Millisecond
	fallbackFailAfter     = 3
	fallbackRiseAfter     = 2
)

// UpstreamState is the dashboard-facing view of one upstream: what the probe
// currently believes, when that belief last changed, and why.
//
// An ALIAS, not a declaration: the struct itself lives in internal/status
// (Task 6), because it is a wire type on status/boxes/<label>.json and its
// readers — the console, operators — must be able to decode it without
// importing the edge. See status.UpstreamState's own doc comment for the full
// argument. The alias keeps this package writing plain UpstreamState with no
// conversion where the prober fills it in.
type UpstreamState = status.UpstreamState

// probeSpec is one route's probing contract, parsed from the registry's health
// JSON once per Reload. Health need not have been through WithDefaults: Sync
// normalizes it, so a caller may hand over the raw parse.
type probeSpec struct {
	Host   string // route host — the probe's Host header default
	Addrs  []string
	Health provision.Health
}

// equal reports whether two specs describe the same probing job. Sync uses it
// to decide restart-or-leave-alone, so it must cover every field the running
// goroutine captured: host (the default Host header), the addr list INCLUDING
// order (round-robin walks it), and the whole health object. Sync compares
// NORMALIZED specs, so Health{} and Health{}.WithDefaults() are equal here.
func (s probeSpec) equal(o probeSpec) bool {
	return s.Host == o.Host && s.Health == o.Health && slices.Equal(s.Addrs, o.Addrs)
}

// proberSet owns one prober goroutine per probed host plus the live-state
// table both the request path and the status snapshot read.
//
// Lifecycle mirrors Activator.RetireExcept: Reload computes the desired specs
// and calls Sync, which starts probers for new hosts, stops those whose route
// vanished, and replaces those whose spec changed. It is the only other thing
// in this package that owns per-route goroutines across a reload.
//
// A *proberSet must come from newProberSet; the zero value is not usable and a
// nil receiver panics rather than quietly reporting "no live upstreams" — a
// silent nil would turn a wiring mistake into fleet-wide 503s instead of a
// loud failure in the first test that hits it.
type proberSet struct {
	// mu guards hosts only. The per-host state table lives on hostProber
	// under its own lock, so a probe tick writing state never contends with
	// Sync, and Sync never has to hold this lock while waiting for a
	// goroutine to exit (which would stall Live on the request hot path).
	mu     sync.Mutex
	hosts  map[string]*hostProber
	client *http.Client
}

func newProberSet() *proberSet {
	return &proberSet{
		hosts: make(map[string]*hostProber),
		client: &http.Client{
			Timeout: probeTimeout,
			// A redirect is not health: an upstream answering 302 has not
			// served the configured status, and following the hop would
			// probe something the operator never named.
			CheckRedirect: func(*http.Request, []*http.Request) error {
				return http.ErrUseLastResponse
			},
			Transport: &http.Transport{
				// Probes are the same tiny request to the same handful of
				// addrs forever, so idle connections are worth keeping —
				// bounded tightly, since every probed upstream holds one.
				MaxIdleConnsPerHost:   1,
				IdleConnTimeout:       90 * time.Second,
				ResponseHeaderTimeout: probeTimeout,
			},
		},
	}
}

// hostProber is one host's probing job: its spec, the clamped cadence derived
// from it, its goroutine's stop handle, and the per-addr state that goroutine
// maintains.
//
// The state table is OWNED BY THIS STRUCT, not by the set — that is what makes
// Sync's replace safe. A replaced prober's goroutine can still be mid-probe
// when the new one is installed, and whatever it writes lands in its own
// retired table, which nothing reads any more. (Sync additionally waits for
// the old goroutine to observe its stop before starting the new one, so even
// the log lines stay ordered.)
type hostProber struct {
	spec     probeSpec
	client   *http.Client
	path     string
	wantCode int
	hostHdr  string
	interval time.Duration
	failAt   int
	riseAt   int

	cancel context.CancelFunc
	ctx    context.Context
	done   chan struct{}

	mu    sync.Mutex
	order []string // addrs in spec order — Live/State iterate this, not the map
	state map[string]*upstreamState
}

// upstreamState is the mutable half of UpstreamState plus the pass counter,
// which is bookkeeping the dashboard has no use for.
type upstreamState struct {
	up        bool
	since     time.Time
	fails     int
	passes    int
	lastError string
}

func newHostProber(spec probeSpec, client *http.Client) *hostProber {
	h := spec.Health
	hp := &hostProber{
		spec:     spec,
		client:   client,
		path:     h.Path,
		wantCode: h.Status,
		hostHdr:  h.HostHeader,
		interval: time.Duration(h.IntervalMS) * time.Millisecond,
		failAt:   h.FailAfter,
		riseAt:   h.RiseAfter,
		done:     make(chan struct{}),
		state:    make(map[string]*upstreamState, len(spec.Addrs)),
	}
	if hp.path == "" {
		hp.path = "/"
	}
	if hp.path[0] != '/' {
		hp.path = "/" + hp.path
	}
	if hp.wantCode == 0 {
		hp.wantCode = http.StatusOK
	}
	if hp.hostHdr == "" {
		// WithDefaults deliberately leaves HostHeader empty so this
		// substitution can see "unset" rather than a value it invented.
		hp.hostHdr = spec.Host
	}
	if h.IntervalMS <= 0 {
		hp.interval = fallbackProbeInterval
	}
	if hp.failAt <= 0 {
		hp.failAt = fallbackFailAfter
	}
	if hp.riseAt <= 0 {
		hp.riseAt = fallbackRiseAfter
	}
	now := time.Now()
	for _, addr := range spec.Addrs {
		if addr == "" || hp.state[addr] != nil {
			continue // validation refuses both; be inert if one slips through
		}
		hp.order = append(hp.order, addr)
		// Rule 1: everything starts live.
		hp.state[addr] = &upstreamState{up: true, since: now}
	}
	hp.ctx, hp.cancel = context.WithCancel(context.Background())
	return hp
}

// stop cancels the goroutine's context — aborting any in-flight probe at once,
// so this does not wait out the 2s client timeout — and blocks until it has
// exited.
func (hp *hostProber) stop() {
	hp.cancel()
	<-hp.done
}

// Sync starts, stops and replaces probers so the running set matches want.
// Called from Reload after the upstream map swap, beside RetireExcept.
//
// Specs with no addrs are not probed (nothing to probe); such a host holds no
// entry, and Live/State report it as absent.
//
// Concurrent callers are safe (the set lock serializes the diff), but be aware
// that a call can BLOCK on stop()ing a retired prober — including one a peer
// call has installed but not yet started, since the goroutine launch happens
// after the lock is released. Reload is the single caller today, so this is a
// note for whoever adds the second one, not a live hazard.
func (p *proberSet) Sync(want map[string]probeSpec) {
	// Normalize BEFORE both the compare and the construction. A caller that
	// hands over a raw Health{} on one Reload and a WithDefaults()ed one on the
	// next is describing the same probing job; comparing them un-normalized
	// would call that a change and restart every prober on the box — resetting
	// every fail/rise run — on each Reload a mixed-path writer triggers.
	norm := make(map[string]probeSpec, len(want))
	for host, w := range want {
		w.Health = w.Health.WithDefaults()
		norm[host] = w
	}
	want = norm

	var retired, started []*hostProber

	p.mu.Lock()
	for host, hp := range p.hosts {
		w, ok := want[host]
		if ok && len(w.Addrs) > 0 && hp.spec.equal(w) {
			continue // unchanged: leave the goroutine and its counters alone
		}
		retired = append(retired, hp)
		delete(p.hosts, host)
	}
	for host, w := range want {
		if len(w.Addrs) == 0 {
			continue
		}
		if _, ok := p.hosts[host]; ok {
			continue
		}
		// Built (and therefore all-live) INSIDE the same critical section
		// that removed the outgoing prober, so a concurrent Live never
		// observes a replaced host as having no upstreams — a gap there
		// would be a 503 for a route that is perfectly healthy.
		hp := newHostProber(w, p.client)
		p.hosts[host] = hp
		started = append(started, hp)
	}
	p.mu.Unlock()

	// Outside the lock, and before the replacements run: retiring waits for
	// each old goroutine to exit, and blocking Live (request hot path) for
	// that wait would be exactly the stall this whole feature exists to
	// avoid.
	for _, hp := range retired {
		hp.stop()
	}
	for _, hp := range started {
		go hp.run()
	}
}

// Live reports the currently-live upstreams for host, in the route's addr
// order, as a fresh slice the caller may keep. Unknown host (or one with no
// live upstream) → empty: the request path answers 503 + Retry-After rather
// than dialing something the probe says is dead.
func (p *proberSet) Live(host string) []string {
	p.mu.Lock()
	hp := p.hosts[host]
	p.mu.Unlock()
	if hp == nil {
		return nil
	}
	hp.mu.Lock()
	defer hp.mu.Unlock()
	out := make([]string, 0, len(hp.order))
	for _, addr := range hp.order {
		if hp.state[addr].up {
			out = append(out, addr)
		}
	}
	return out
}

// Eject drops addr from host's live set immediately — the in-request
// dial-failure fast path, so a box that died between probe ticks costs one
// retry rather than up to fail_after × interval of 502s. Fails is set to
// failAfter so the upstream is exactly as down as a probe-failed one; probe
// passes rejoin it after rise_after, the same as any other recovery.
//
// A host or addr this set does not know is a no-op: the request path can
// legitimately race a Reload that just dropped either.
func (p *proberSet) Eject(host, addr string) {
	p.mu.Lock()
	hp := p.hosts[host]
	p.mu.Unlock()
	if hp == nil {
		return
	}
	hp.mu.Lock()
	st := hp.state[addr]
	if st == nil {
		hp.mu.Unlock()
		return
	}
	const reason = "dial failed on a live request"
	was := st.up
	st.up = false
	st.passes = 0
	st.fails = hp.failAt
	st.lastError = reason
	if was {
		st.since = time.Now()
	}
	hp.mu.Unlock()
	if was {
		log.Printf("edge: probe %s %s down (%s)", host, addr, reason)
	}
}

// State returns the probe's view of every probed host, keyed by host, each
// upstream in the route's addr order. Everything returned is a copy: Task 6's
// snapshot serializes this while probes keep running.
func (p *proberSet) State() map[string][]UpstreamState {
	p.mu.Lock()
	hps := make(map[string]*hostProber, len(p.hosts))
	for host, hp := range p.hosts {
		hps[host] = hp
	}
	p.mu.Unlock()

	out := make(map[string][]UpstreamState, len(hps))
	for host, hp := range hps {
		hp.mu.Lock()
		states := make([]UpstreamState, 0, len(hp.order))
		for _, addr := range hp.order {
			st := hp.state[addr]
			states = append(states, UpstreamState{
				Addr:      addr,
				Up:        st.up,
				Since:     st.since,
				Fails:     st.fails,
				LastError: st.lastError,
			})
		}
		hp.mu.Unlock()
		out[host] = states
	}
	return out
}

// Stop stops every prober and empties the set (tests and shutdown). Sync may
// be called again afterwards; it simply starts fresh probers.
func (p *proberSet) Stop() {
	p.mu.Lock()
	hps := make([]*hostProber, 0, len(p.hosts))
	for host, hp := range p.hosts {
		hps = append(hps, hp)
		delete(p.hosts, host)
	}
	p.mu.Unlock()
	for _, hp := range hps {
		hp.stop()
	}
	// Every probed upstream held an idle connection (MaxIdleConnsPerHost 1);
	// with no prober left to reuse them they are just sockets on a box that was
	// told to stop. A later Sync gets fresh ones.
	p.client.CloseIdleConnections()
}

// run is one host's probe loop: probe every upstream, wait out the interval,
// repeat. The first pass runs immediately so a dead upstream is caught within
// fail_after × interval of the reload rather than one interval later.
func (hp *hostProber) run() {
	defer close(hp.done)
	t := time.NewTicker(hp.interval)
	defer t.Stop()
	for {
		hp.tick()
		select {
		case <-hp.ctx.Done():
			return
		case <-t.C:
		}
	}
}

// tick probes each upstream once, sequentially. Sequential is deliberate at
// these counts (a route has a handful of upstreams, not hundreds): it bounds
// the goroutine's own concurrency at one in-flight request and keeps the log
// lines of a whole-fleet outage in addr order. The ctx checks are what keep
// stop() prompt on a route with several dead addrs.
//
// Under a TOTAL outage a tick can outlast the interval: N black-holed addrs ×
// the 2s probe timeout against, say, a 2s cadence. time.Ticker has a one-slot
// buffer, so the missed ticks collapse into one and the loop simply runs
// back-to-back rather than queueing a backlog. That is the intended
// degradation — the probe rate self-limits to "as fast as the timeouts allow"
// and returns to cadence the moment upstreams answer again.
//
// hp.order is read without the lock on purpose: it is fixed by
// newHostProber before this goroutine is started (an addr set change is a new
// hostProber, never a mutation), so the `go hp.run()` in Sync is the
// happens-before edge. Live/State still take the lock because they share it
// with the state map.
func (hp *hostProber) tick() {
	for _, addr := range hp.order {
		if hp.ctx.Err() != nil {
			return
		}
		err := hp.probe(addr)
		// A probe cancelled by our own stop says nothing about the upstream.
		// Recording it would let a retiring prober log a "down" transition
		// (and stamp a LastError) for a host that was merely reloaded.
		if hp.ctx.Err() != nil {
			return
		}
		hp.record(addr, err)
	}
}

// probe performs one health request. A nil error means the upstream answered
// the configured status; anything else is the reason string that reaches both
// the transition log and the dashboard's LastError.
func (hp *hostProber) probe(addr string) error {
	req, err := http.NewRequestWithContext(hp.ctx, http.MethodGet, "http://"+addr+hp.path, nil)
	if err != nil {
		return fmt.Errorf("bad probe target: %v", err)
	}
	// Host, not the header map: net/http reads Request.Host for the wire
	// Host line, and an upstream that vhosts (or redirects on an unexpected
	// name) must see the name the operator configured.
	req.Host = hp.hostHdr
	resp, err := hp.client.Do(req)
	if err != nil {
		// Connect refused, timeout, reset, TLS-less port answering garbage:
		// from the edge's side these are one thing — no usable response
		// arrived — and the spec names that class "dial".
		return fmt.Errorf("dial: %v", err)
	}
	// Drain a bounded amount so the connection can be reused; a probe body is
	// expected to be tiny and is never inspected.
	io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
	resp.Body.Close()
	if resp.StatusCode != hp.wantCode {
		return fmt.Errorf("status %d (want %d)", resp.StatusCode, hp.wantCode)
	}
	return nil
}

// record folds one probe result into an upstream's state and logs transitions
// with their reason. Consecutive counts only: a single pass clears the failure
// run and a single failure clears the recovery run, which is what makes
// fail_after/rise_after mean "sustained", not "cumulative".
//
// Fails and LastError are the DOWN upstream's explanation, so they survive the
// passing probes of a recovery and are cleared only on the up transition. Zero
// them on the first pass instead and the dashboard shows the mid-recovery
// upstream as `Up:false, Fails:0, LastError:""` — down for no stated reason,
// which is the one thing an operator staring at a failover must not see.
func (hp *hostProber) record(addr string, probeErr error) {
	hp.mu.Lock()
	st := hp.state[addr]
	if st == nil { // addr vanished from this prober — impossible today
		hp.mu.Unlock()
		return
	}
	var transition, reason string
	switch {
	case probeErr == nil:
		if st.up {
			// Healthy and staying healthy: the failure run ends here.
			st.fails = 0
			st.lastError = ""
			break
		}
		st.passes++
		if st.passes >= hp.riseAt {
			st.up = true
			st.passes = 0
			st.fails = 0
			st.lastError = ""
			st.since = time.Now()
			transition = "up"
		}
	default:
		st.passes = 0
		st.fails++
		st.lastError = probeErr.Error()
		if st.up && st.fails >= hp.failAt {
			st.up = false
			st.since = time.Now()
			transition, reason = "down", st.lastError
		}
	}
	hp.mu.Unlock()

	switch transition {
	case "down":
		log.Printf("edge: probe %s %s down (%s)", hp.spec.Host, addr, reason)
	case "up":
		log.Printf("edge: probe %s %s up", hp.spec.Host, addr)
	}
}