package host

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"os"
	"path"
	"time"

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

// FleetBinding is the deployment-bucket object a fleet box publishes to say
// "as of BoundAt, Box (the box's fleet_boxes.label — resolveFleetIdentity,
// NEVER status.Label(machineID)) is the box currently serving Host". Fleet
// boxes are outbound-only — they dial the platform edge over a reverse
// tunnel and can never write the platform box's registry directly
// (customer-fleets design spec, "Routing to N boxes: the lease stamp") — so
// this bucket object is the ONLY channel a fleet box has to tell the
// platform which of its boxes currently holds a given hostname's lease.
// ReconcileFleetBindings is the platform-side reader that turns it back
// into registry.Route.FleetBox.
type FleetBinding struct {
	Host  string `json:"host"`
	Fleet string `json:"fleet"`
	Box   string `json:"box"` // the holding box's fleet_boxes.label
	// BoundAt is unix MILLISECONDS (not seconds) — a wire-shape choice made
	// free at Task 6's fix round, before any binding object has ever shipped
	// to production: two wakes racing on different boxes (or a wake racing
	// the startup pass on the SAME box) can land inside the same second, and
	// millisecond resolution is what lets bindingBlocked's logical-clock
	// guard actually distinguish them instead of comparing equal.
	BoundAt int64 `json:"bound_at"`
}

// fleetBindingKey is the one place the object's key is spelled, shared by
// the writer and the reader below: fleets/<fleet>/bindings/<host>.json,
// ABSOLUTE in the PLATFORM'S deployment bucket — a fleet box has TWO buckets
// (its own fleet/data bucket the activator leases and restores instances
// against, and the platform's shared deployment bucket every box, fleet or
// not, adopts releases from and now publishes/reads bindings in) and this
// key lives in the latter, never the former.
//
// Deliberately not routed through release.NewPrefixStore's
// CARLOS_STATUS_PREFIX rebasing: that wrapper exists so a fleet box's own
// status/metrics/alerts heartbeats land under its private
// "fleets/<name>/" corner instead of the platform's shared paths — but a
// binding is not a heartbeat, it is a cross-box MESSAGE the platform box
// must read back, and it already carries its own fleet name (b.Fleet) in
// the struct. Both WriteFleetBinding and ReconcileFleetBindings take an
// UNPREFIXED release.Store and compute this absolute key themselves, so
// there is exactly one key format to keep in sync, and a fleet box's own
// CARLOS_STATUS_PREFIX (which does apply to its status/metrics stores) has
// no bearing on where its bindings land.
func fleetBindingKey(fleet, host string) string {
	return path.Join("fleets", fleet, "bindings", host+".json")
}

// bindingBlocked reports whether writing b over the binding currently stored
// at its key (existing, nil if none) must be skipped. Two independent
// guards, either one enough to block:
//
//   - Non-stealing: a non-takeover write (the startup pass,
//     publishResidentFleetBindings — a bare "this box has this route
//     resident" observation, not lease-backed) must never overwrite a
//     binding that already names a DIFFERENT box. Only a wake-success write
//     (takeover=true) may take over — it is backed by having just won the
//     per-host S3 lease, which is the actual arbiter of who may serve a
//     host, so it is allowed to restamp regardless of who was there before.
//   - Logical clock, SAME BOX ONLY: a binding already stored FROM THE SAME
//     BOX with a strictly greater BoundAt must never be overwritten by an
//     older write from that same box — this holds regardless of takeover,
//     and fixes a physical-write-race hazard: two concurrent Puts to the
//     same key from one box (a retried wake publish racing its own
//     predecessor, or any other overlap) must not let whichever one's HTTP
//     round-trip happens to land second win just because it arrived later.
//     Scoped to existing.Box == b.Box deliberately (fix round 2, Opus
//     review): BoundAt is each BOX's OWN wall clock, never a shared one —
//     comparing it across two DIFFERENT boxes compares clock skew, not
//     recency. Before this scoping, a dead box A's stale binding stamped
//     with a clock running ahead of surviving box B's would silently and
//     PERMANENTLY block B's lease-backed takeover write (the startup pass
//     can't help either — it's non-stealing by the guard above), stranding
//     the edge on its sorted-fallback channel instead of the real lease
//     holder for as long as the skew persists. A cross-box binding is
//     already fully governed by the non-stealing guard above (blocked
//     unless takeover=true); the clock only ever needs to arbitrate a BOX
//     against its own past self.
func bindingBlocked(existing *FleetBinding, b FleetBinding, takeover bool) bool {
	if existing == nil {
		return false
	}
	if !takeover && existing.Box != b.Box {
		return true
	}
	return existing.Box == b.Box && existing.BoundAt > b.BoundAt
}

// getExistingBinding reads key's current binding + etag for a replace-only
// Put, folding "doesn't exist yet" into (nil, "", nil) — a create, with
// neither guard in bindingBlocked applicable. A corrupt existing object is
// treated as absent (logged) rather than failing the write outright: the
// same "never fatal" spirit ReconcileFleetBindings' reader-side skip uses,
// and it ensures a legitimate new write is never permanently stuck behind a
// broken object at its own key.
func getExistingBinding(ctx context.Context, s release.Store, key string) (*FleetBinding, string, error) {
	body, etag, err := s.Get(ctx, key)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil, "", nil
		}
		return nil, "", err
	}
	var b FleetBinding
	if err := json.Unmarshal(body, &b); err != nil {
		log.Printf("host: fleet binding: existing object at %s is corrupt, treating as absent: %v", key, err)
		return nil, etag, nil
	}
	return &b, etag, nil
}

// WriteFleetBinding publishes b to the PLATFORM'S deployment bucket,
// replacing whatever is currently at its key — UNLESS bindingBlocked says
// the existing binding must win (see its doc comment for the two guards).
// s must be an UNPREFIXED store handle — see fleetBindingKey's doc comment.
//
// takeover distinguishes the two callers: fleetBindingHook (the activator's
// OnWake, a lease-backed wake success) passes true and may claim a binding
// away from any other box; publishResidentFleetBindings (the agent-startup
// pass, a bare "this route is resident here" observation with no lease
// behind it) passes false and must never steal a binding another box
// currently holds.
//
// The read-etag-then-Put dance mirrors status.WriteBox, and for the same
// reason: a box republishes to the SAME key on every wake and once more at
// startup, so a bare create-only Put (ifMatch "") would fail every write
// after the first with ErrCASConflict.
func WriteFleetBinding(ctx context.Context, s release.Store, b FleetBinding, takeover bool) error {
	body, err := json.Marshal(b)
	if err != nil {
		return err
	}
	key := fleetBindingKey(b.Fleet, b.Host)

	existing, etag, err := getExistingBinding(ctx, s, key)
	if err != nil {
		return err
	}
	if bindingBlocked(existing, b, takeover) {
		return nil
	}

	if _, err := s.Put(ctx, key, body, etag); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
		// Lost the race against a concurrent write to the same key —
		// re-read once, re-apply BOTH guards against the fresher state (it
		// may now block a write that was fine against the stale read), and
		// retry — the same single-retry contract status.WriteBox uses.
		existing, etag, err = getExistingBinding(ctx, s, key)
		if err != nil {
			return err
		}
		if bindingBlocked(existing, b, takeover) {
			return nil
		}
		if _, err := s.Put(ctx, key, body, etag); err != nil {
			return err
		}
	}
	return nil
}

// ReconcileFleetBindings is the platform-side half of fleet routing: for
// every fleets/<fleet>/bindings/<host>.json object published under any
// fleet, if the route registered for that host is actually PLACED on that
// SAME fleet (registry.FleetPlacement(rt.Region) == the binding's own
// Fleet) and its FleetBox doesn't already match, stamp it via SetFleetBox.
//
// The fleet-placement check is a v1 trust-boundary requirement, not a
// nicety: fleet boxes are outbound-only, but nothing at the bucket layer
// stops fleet A's box from publishing a binding that names a host actually
// placed on fleet B (a same-named host collision, or a stale/rewritten
// object) — without this check that binding could silently override fleet
// B's own routing for a host fleet A's box has no business serving. There
// is deliberately no "newest wins" tie-break for a cross-fleet collision:
// two bindings for the exact same (fleet, host) pair cannot exist (the key
// IS fleet+host), and a binding whose fleet doesn't match the route's own
// placement is rejected outright, never preferred by timestamp — an earlier
// version of this function used a cross-fleet "newest BoundAt" reduction
// before this check existed; it's gone because the placement check makes it
// both wrong (a malicious/stale binding could out-race a legitimate one by
// BoundAt alone) and unnecessary (same-fleet duplicates are structurally
// impossible).
//
// s must be an UNPREFIXED store handle (see fleetBindingKey) — this walks
// the same absolute "fleets/" keys WriteFleetBinding writes, and is only
// ever called on platform boxes, which never set CARLOS_STATUS_PREFIX in
// the first place.
//
// Corrupt objects and per-key read failures are logged and skipped, never
// fatal — the loadBoxes precedent (internal/status/fleet.go:197-216): one
// bad binding must not blank out routing for every other fleet host. A
// missing route, or one not placed on the binding's own fleet, is silently
// skipped too — neither is this binding's to stamp.
func ReconcileFleetBindings(ctx context.Context, s release.Store, reg *registry.Registry) error {
	fleetDirs, err := s.ListDirs(ctx, "fleets/")
	if err != nil {
		return err
	}
	for _, dir := range fleetDirs {
		keys, err := s.List(ctx, dir+"bindings/")
		if err != nil {
			log.Printf("host: fleet bindings: list %s: %v", dir, err)
			continue
		}
		for _, key := range keys {
			body, _, err := s.Get(ctx, key)
			if err != nil {
				log.Printf("host: fleet bindings: skipping %s: read failed: %v", key, err)
				continue
			}
			var b FleetBinding
			if err := json.Unmarshal(body, &b); err != nil {
				log.Printf("host: fleet bindings: skipping %s: corrupt binding: %v", key, err)
				continue
			}
			rt, err := reg.Get(b.Host)
			if err != nil {
				log.Printf("host: fleet bindings: read route %s: %v", b.Host, err)
				continue
			}
			if rt == nil {
				continue // no matching route on this platform box
			}
			// Never stamp a sidecar. SetFleetBox on a fleet-placed route is
			// precisely what turns the edge's fleet dispatch arm into a live
			// proxy, so this reader is the last gate on the chain and it
			// refuses on kind rather than trusting its input.
			//
			// Redundant with residentBindingHosts on the publishing side, and
			// deliberately so: the two halves run on DIFFERENT machines and
			// upgrade independently. A fleet box still on an older binary
			// keeps publishing sidecar bindings into the bucket for the whole
			// of a mixed-version roll, and this platform box is what reads
			// them. A binding is an untrusted claim written by another host;
			// the route row is local truth.
			if rt.IsSidecar() {
				continue
			}
			if fleet, ok := registry.FleetPlacement(rt.Region); !ok || fleet != b.Fleet {
				// The route isn't placed on THIS binding's fleet at all —
				// v1 trust boundary: refuse to let it stamp (see doc
				// comment above).
				continue
			}
			if rt.FleetBox == b.Box {
				continue
			}
			if err := reg.SetFleetBox(b.Host, b.Box); err != nil {
				log.Printf("host: fleet bindings: stamp %s -> %s: %v", b.Host, b.Box, err)
			}
		}
	}
	return nil
}

// fleetBindingHook builds the activator's OnWake hook for a fleet box: on
// every successful wake, publish a FleetBinding naming this box (identified
// by fleet/box, resolveFleetIdentity's result — the channel label, never a
// hashed pseudonym) as the current holder, with takeover=true since a
// wake's lease acquisition is exactly the authority that may claim a
// binding away from whoever held it before. activator.Service already
// invokes OnWake in its own goroutine (see activator.go's wake), so this
// function runs off the wake path entirely — a slow or failing bucket Put
// can never delay or fail a request. Errors are logged only: a binding is
// eventually-consistent by design (the design spec's failure sequence — box
// goes dark, heartbeat ages, lease expires, the next request wakes a
// surviving box, which restamps — is exactly the fallback that covers a
// dropped write here).
//
// The hook takes the whole ROUTE, which is what the activator's wake path has
// in hand anyway, and refuses on its Kind: a sidecar host must never have a
// binding published for it, and this is the write path that carries TAKEOVER
// semantics, so an errant one here does not merely appear — it claims the host
// away from whoever legitimately held it. The filter lives here rather than
// resting on "the activator never wakes a sidecar": that is true of the
// activator TODAY, and it is a fact a later task in this same branch removes
// (the sidecar worker goes through the same wake() that fires OnWake). A guard
// that depends on a caller's current behaviour is not a guard.
//
// Taking the route also removes the registry read this refusal used to need.
// That read could fail, and a security-relevant check whose input can fail
// must then decide which way to fail: it published anyway (fail OPEN), because
// a dropped binding is a routing outage and the fallback that covers it is
// slow. The route the wake already holds is the same local truth the read was
// fetching, so the refusal is now total and depends on no I/O at all.
func fleetBindingHook(store release.Store, fleet, box string) func(rt registry.Route) {
	return func(rt registry.Route) {
		if rt.IsSidecar() {
			return
		}
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		b := FleetBinding{Host: rt.Host, Fleet: fleet, Box: box, BoundAt: time.Now().UnixMilli()}
		if err := WriteFleetBinding(ctx, store, b, true); err != nil {
			log.Printf("agent: fleet binding for %s: %v", rt.Host, err)
		}
	}
}

// publishResidentFleetBindings runs once at fleet-box agent startup: every
// route already present in this box's local registry gets a binding
// written immediately, without waiting for an activator wake, with
// takeover=false — this is a bare "this route is resident here" observation
// with no lease behind it, so it must never steal a binding a wake-success
// write already holds for a DIFFERENT box (see WriteFleetBinding/
// bindingBlocked). Two gaps this closes that a wake-only publish would
// miss:
//   - an agent restart discards activator.Service's in-memory entries, but
//     an already-running exec child (or a unit-backed tenant) keeps serving
//     — its host would otherwise have no fresh binding until its NEXT idle
//     wake, which may be a long time coming, or never for a host that is
//     always awake;
//   - a route that isn't activator-managed at all (Hibernate unset, or a
//     unit-backed Backing) never goes through activator.wake in the first
//     place, so it would never get a binding published any other way.
//
// Best-effort and non-fatal per host: a failed List aborts the whole pass
// (logged; the next restart tries again), but one host's write failure does
// not stop the rest of the pass from covering everyone else.
func publishResidentFleetBindings(store release.Store, fleet, box string, reg *registry.Registry) {
	routes, err := reg.List()
	if err != nil {
		log.Printf("agent: fleet bindings startup pass: list routes: %v", err)
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	now := time.Now().UnixMilli()
	for _, host := range residentBindingHosts(routes) {
		b := FleetBinding{Host: host, Fleet: fleet, Box: box, BoundAt: now}
		if err := WriteFleetBinding(ctx, store, b, false); err != nil {
			log.Printf("agent: fleet binding for %s: %v", host, err)
		}
	}
}

// residentBindingHosts is the set of hosts this box advertises as served —
// the pure host-selection half of publishResidentFleetBindings, split out so
// the refusal below is directly testable.
//
// Sidecar routes (Kind "sidecar") are excluded. Nothing routes to a sidecar,
// so a binding claiming a box serves it is a lie the platform side then acts
// on: ReconcileFleetBindings stamps SetFleetBox from it, and a fleet-placed
// route is exactly what the edge's fleet dispatch arm turns into a live
// internet-facing proxy. Filtered at the publisher rather than at the
// reconciler because this is the only place that decides what a box CLAIMS —
// the claim is what must never exist.
func residentBindingHosts(routes []registry.Route) []string {
	var out []string
	for _, rt := range routes {
		if rt.Host == "" || rt.IsSidecar() {
			continue
		}
		out = append(out, rt.Host)
	}
	return out
}