package status

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

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

// FleetApp is one opted-in app's public row — name, count, health, and the
// release it's running. Nothing here exists for an app that hasn't set
// StatusNamed; see BuildFleet's apps grouping.
type FleetApp struct {
	Name      string `json:"name"`
	Instances int    `json:"instances"`
	Healthy   bool   `json:"healthy"`
	Channel   string `json:"channel"`
	Version   string `json:"version"` // short sha as stored on the route
}

// FleetBox is a Box's public row inside the merged feed: everything a Box
// carries except its own Replication (which only ever votes into the
// fleet-wide Replication, never appears per-box) and its own Capacity
// (same — it only ever votes into Fleet.Capacity), plus Stale, which only
// exists at merge time since a lone Box has no "how long ago" to compare
// against.
type FleetBox struct {
	Label  string      `json:"label"`
	Region string      `json:"region,omitempty"`
	Load   *string     `json:"load"`
	Stale  bool        `json:"stale"`
	Routes RouteCounts `json:"routes"`
	// AdoptionRunning DOES appear per-box, unlike Replication and Capacity
	// above, because it is the one reading whose whole value is knowing WHICH
	// box has it wrong: "something in the fleet isn't adopting" sends an
	// operator to ssh into every box in turn, which is barely better than the
	// silence this field exists to end. It carries nothing about a tenant —
	// only whether the platform's own timer is armed on an opaquely-labelled
	// machine.
	AdoptionRunning *bool `json:"adoption_running"`
	// ConfigPathRunning appears per-box for exactly the reason
	// AdoptionRunning does: the value of the reading is knowing WHICH box
	// lost its config-sync trigger, since "somewhere in the fleet config is
	// an hour late" is a search, not an answer. Deliberately NOT rolled up
	// into a fleet-wide field: worst-of aggregates exist here only where the
	// page shows a tile for them, and this one is a per-box operational
	// fact, not a headline about the fleet.
	ConfigPathRunning *bool `json:"config_path_running"`
	// RestartPathRunning appears per-box for exactly the reason
	// ConfigPathRunning does, and is not rolled up for the same reason: it is a
	// per-box operational fact ("this box's restart trigger is not watching"),
	// not a headline about the fleet.
	RestartPathRunning *bool `json:"restart_path_running"`
	// ExpiredCanaries appears per-box for the same reason AdoptionRunning
	// does: the fleet-wide sum says something is expired somewhere, this
	// says on which box, without naming which route. Published in the FEED
	// only for now — the rendered boxes table deliberately has no column
	// for it yet (2026-08-03 review, MINOR 4: this comment used to imply
	// the table showed it; it does not, and adding the column is deferred
	// until an expired canary has actually been hunted via the feed once).
	ExpiredCanaries *int `json:"expired_canaries"`
}

// Fleet is the merged feed: status/fleet.json. Every field here is either a
// fleet-wide aggregate, a band word, or behind an app's own opt-in — see the
// design spec's "rule for new fields".
type Fleet struct {
	GeneratedAt time.Time       `json:"generated_at"`
	Boxes       []FleetBox      `json:"boxes"`
	Instances   map[string]*int `json:"instances"` // total/static/resident/hibernating/awake/asleep
	Capacity    struct {
		Level *string `json:"level"`
	} `json:"capacity"`
	Apps         []FleetApp `json:"apps"`
	AppsUnlisted int        `json:"apps_unlisted"`
	Services     []struct {
		Name string `json:"name"`
		Up   bool   `json:"up"`
	} `json:"services"`
	Replication     Replication `json:"replication"`
	AdoptionRunning *bool       `json:"adoption_running"`
	// ExpiredCanaries is the fleet-wide sum of every box's own count — see
	// sumExpiredCanaries. Null-propagating: one box that couldn't check
	// nulls the whole aggregate, the same rule sumInstances applies to
	// Awake/Asleep, because a partial sum would read as more certain than
	// it is. The status page shows this only when it is a known, positive
	// count — see assets/index.html.
	ExpiredCanaries *int `json:"expired_canaries"`
}

// capacityRank orders the capacity band words worst-last, so "worst of a
// set of boxes" is just "the box with the highest rank" — built off
// snapshot.go's capacityWords so the two never drift apart into disagreeing
// about which word beats which.
var capacityRank = map[string]int{
	capacityWords[0]: 0,
	capacityWords[1]: 1,
	capacityWords[2]: 2,
}

// intPtr is the one-line "give me a pointer to this value" a lot of the
// Instances map construction below needs — Go has no address-of-literal
// operator, so every *int field needs a local to take the address of
// without this helper.
func intPtr(v int) *int { return &v }

// SnapshotPeriod is how often every box republishes its own
// status/boxes/<label>.json (the agent's status loop tick, 5 minutes +
// jitter). It lives here rather than at each reader because it is the unit
// BoxStale measures staleness in: the console's private host detail and
// `carlos status -live` both have to answer "is this box dark?" without an
// agent's local const to borrow, and a reader that guessed a different period
// would call a box unreachable at a different moment than the published
// boxes table does.
const SnapshotPeriod = 5 * time.Minute

// staleFactor is how many missed snapshot periods make a box stale. Three
// consecutive misses: past any single hiccup (a slow S3 write, one skipped
// tick), and the point at which a box has plainly stopped self-reporting.
// One constant, one rule — BuildFleet's published FleetBox.Stale and every
// private reader go through BoxStale below, so nothing can drift into calling
// the same box fresh on one page and dark on another.
const staleFactor = 3

// BoxStale is the fleet's one staleness rule: a snapshot older than
// staleFactor×period means the box stopped reporting. Every caller — the
// public merge, the console's private host table, the CLI — asks this
// function, never its own subtraction.
//
// Callers that render must treat true as "this box's numbers are history":
// per the design spec an unreachable host shows how long it has been dark and
// NOTHING else, never its last happy readings.
func BoxStale(b Box, now time.Time, period time.Duration) bool {
	return now.Sub(b.GeneratedAt) > staleFactor*period
}

// LoadBoxes reads every per-box snapshot in the bucket — the same read
// BuildFleet does, exported for the private readers (the console's operator
// fleet page, `carlos status -live`) so there is exactly ONE parser of the
// status/boxes/ shape and one skip-corrupt rule, rather than a second loader
// that could disagree about which snapshots count.
//
// Unlike the merged feed, what comes back here still carries Box.Metrics: raw
// numbers that must reach a private surface only (bucket credentials or a
// session-gated operator page), never fleet.json. See the package comment.
func LoadBoxes(ctx context.Context, s release.Store) ([]Box, error) {
	return loadBoxes(ctx, s)
}

// BuildFleet merges every status/boxes/*.json (stale = GeneratedAt older
// than staleFactor×period — see BoxStale), folds the local registry's routes
// into apps/services, and never invents a number a snapshot didn't carry.
func BuildFleet(ctx context.Context, s release.Store, routes []registry.Route, now time.Time, period time.Duration) (*Fleet, error) {
	boxes, err := loadBoxes(ctx, s)
	if err != nil {
		return nil, err
	}

	f := &Fleet{GeneratedAt: now}

	// stale tracks each box's staleness in lockstep with boxes — computed
	// once here and reused by both the per-box Stale field and the
	// worst-of/replication aggregates below, which must vote only using
	// fresh boxes.
	stale := make([]bool, len(boxes))
	for i, b := range boxes {
		stale[i] = BoxStale(b, now, period)
		f.Boxes = append(f.Boxes, FleetBox{
			Label:              b.Label,
			Region:             b.Region,
			Load:               b.Load,
			Stale:              stale[i],
			Routes:             b.Routes,
			AdoptionRunning:    b.AdoptionRunning,
			ConfigPathRunning:  b.ConfigPathRunning,
			RestartPathRunning: b.RestartPathRunning,
			ExpiredCanaries:    b.ExpiredCanaries,
		})
	}

	f.Instances = sumInstances(boxes)
	f.Capacity.Level = worstCapacity(boxes)
	f.Replication = worstReplication(boxes, stale)
	f.AdoptionRunning = worstAdoption(boxes, stale)
	f.ExpiredCanaries = sumExpiredCanaries(boxes)

	apps, unlisted := buildApps(routes)
	f.Apps = apps
	f.AppsUnlisted = unlisted
	f.Services = buildServices(routes)

	return f, nil
}

// loadBoxes reads every status/boxes/*.json snapshot it can. A snapshot
// that fails to Get or fails to unmarshal is logged and skipped — one box's
// bad write must never take the whole fleet page down, per the plan's
// "corrupt snapshot = logged skip, never fatal" rule.
// maxFleetBoxKeys bounds how many objects loadBoxes will ever process from
// one prefix listing, and maxFleetBoxBodyBytes bounds how large any single
// one of those objects may be before its body is read. Both exist for the
// SAME reason: the prefix loadBoxes reads (status/boxes/, and — via
// release.NewPrefixStore — a customer fleet's own fleets/<name>/status/boxes/)
// is written to by boxes this operator does not control in the fleet case.
// A hostile or badly-misbehaving box could park tens of thousands of keys,
// or one huge body, under its own prefix; without a cap this function would
// list/fetch/unmarshal all of it on every call, feeding both the operator
// page (platform prefix) and every fleet's own page (customer prefix) —
// unbounded memory from data this process does not trust.
const (
	maxFleetBoxKeys      = 200
	maxFleetBoxBodyBytes = 1 << 20 // 1MB
)

func loadBoxes(ctx context.Context, s release.Store) ([]Box, error) {
	boxes, _, err := loadBoxesReport(ctx, s)
	return boxes, err
}

// LoadBoxesComplete is LoadBoxes that also reports whether EVERY listed
// snapshot was read: complete is false when any object's read failed, any
// body was skipped as oversize or corrupt, or the listing was capped. The
// boxes returned are exactly what LoadBoxes would return, and the skips are
// logged the same way; the only addition is the accounting. It exists for
// a reader that must fail CLOSED on partial evidence (the console's
// automatic region placement: a box whose snapshot could not be read has
// UNKNOWN liveness, which is not the same as stale), where LoadBoxes'
// "logged skip, never fatal" posture would quietly drop the one box that
// mattered.
func LoadBoxesComplete(ctx context.Context, s release.Store) (boxes []Box, complete bool, err error) {
	return loadBoxesReport(ctx, s)
}

// statusListingHorizon is how long ago a report's object may last have
// been written and still be listed as a box (region HA plan 2b). Each box
// rewrites its own report every SnapshotPeriod, so an object older than a
// day is a box that has gone: with automatic replacement such reports pile
// up, one per replaced box, and could fill the key cap below, hiding the
// boxes that are running. They are dropped from the listing before any
// read and before the cap, so the cap bounds current reports only. This is
// an optimisation, never authority: every reader still judges each
// report's own GeneratedAt as before (BoxStale, the console's
// VerifyReport), and a day keeps every box an operator would still look
// for as stale. A store that cannot say when an object was written is
// listed whole, as before: one that is not a release.ModLister, and one
// that is but whose ListMod fails. The second is the claim's boundary
// stores (hostops' mutation boundary and conditional record store, the
// mail claim gate): they always have a ListMod method and answer it with
// their own refusal when the store they wrap has none, while their List
// still reads. Because the horizon is only an optimisation, a failed
// ListMod never fails the read: List answers, with main's errors.
const statusListingHorizon = 24 * time.Hour

// listBoxKeys lists status/boxes/, without the reports the horizon drops.
func listBoxKeys(ctx context.Context, s release.Store) ([]string, error) {
	ml, ok := s.(release.ModLister)
	if !ok {
		return s.List(ctx, "status/boxes/")
	}
	mods, err := ml.ListMod(ctx, "status/boxes/")
	if err != nil {
		return s.List(ctx, "status/boxes/")
	}
	cut := time.Now().Add(-statusListingHorizon)
	keys := make([]string, 0, len(mods))
	for _, m := range mods {
		if m.Mod.IsZero() || !m.Mod.Before(cut) {
			keys = append(keys, m.Key)
		}
	}
	return keys, nil
}

func loadBoxesReport(ctx context.Context, s release.Store) (boxes []Box, complete bool, err error) {
	keys, err := listBoxKeys(ctx, s)
	if err != nil {
		return nil, false, err
	}
	complete = true
	if len(keys) > maxFleetBoxKeys {
		log.Printf("status: fleet: %d keys under status/boxes/, capping at %d", len(keys), maxFleetBoxKeys)
		keys = keys[:maxFleetBoxKeys]
		complete = false
	}
	for _, key := range keys {
		body, _, err := s.Get(ctx, key)
		if err != nil {
			log.Printf("status: fleet: skipping %s: read failed: %v", key, err)
			complete = false
			continue
		}
		if len(body) > maxFleetBoxBodyBytes {
			log.Printf("status: fleet: skipping %s: body %d bytes exceeds %d byte cap", key, len(body), maxFleetBoxBodyBytes)
			complete = false
			continue
		}
		var b Box
		if err := json.Unmarshal(body, &b); err != nil {
			log.Printf("status: fleet: skipping %s: corrupt snapshot: %v", key, err)
			complete = false
			continue
		}
		boxes = append(boxes, b)
	}
	return boxes, complete, nil
}

// sumInstances adds each RouteCounts component across every box — stale
// boxes included, since staleness is a "trust this less" signal rendered
// loudly per-box, not a reason to silently drop a box's last-known counts
// from the fleet total. Static/Resident/Hibernating are plain ints on Box
// (always known), so their sums are always known too. Awake/Asleep are
// *int on Box (null when a box's socket table was unreadable), so per the
// plan's null-propagation rule, ANY box contributing null nulls the whole
// summed field — a partial sum would read as more certain than it is.
func sumInstances(boxes []Box) map[string]*int {
	var totalStatic, totalResident, totalHibernating, awakeSum, asleepSum int
	awakeKnown, asleepKnown := true, true
	for _, b := range boxes {
		totalStatic += b.Routes.Static
		totalResident += b.Routes.Resident
		totalHibernating += b.Routes.Hibernating
		if b.Routes.Awake == nil {
			awakeKnown = false
		} else {
			awakeSum += *b.Routes.Awake
		}
		if b.Routes.Asleep == nil {
			asleepKnown = false
		} else {
			asleepSum += *b.Routes.Asleep
		}
	}
	instances := map[string]*int{
		"total":       intPtr(totalStatic + totalResident + totalHibernating),
		"static":      intPtr(totalStatic),
		"resident":    intPtr(totalResident),
		"hibernating": intPtr(totalHibernating),
	}
	if awakeKnown {
		instances["awake"] = intPtr(awakeSum)
	} else {
		instances["awake"] = nil
	}
	if asleepKnown {
		instances["asleep"] = intPtr(asleepSum)
	} else {
		instances["asleep"] = nil
	}
	return instances
}

// sumExpiredCanaries adds each box's own ExpiredCanaries count — stale boxes
// included, same reasoning as sumInstances: a box's last-known count is a
// "trust this less" signal rendered loudly per-box (FleetBox.Stale), not a
// reason to silently drop it from the fleet total. Per the null-propagation
// rule every *int aggregate here follows, ANY box contributing nil (it
// couldn't check) nils the whole sum — a partial count would read as more
// certain than it is. No boxes at all sums to 0, not nil: there is nothing
// unknown about an empty fleet.
func sumExpiredCanaries(boxes []Box) *int {
	sum := 0
	for _, b := range boxes {
		if b.ExpiredCanaries == nil {
			return nil
		}
		sum += *b.ExpiredCanaries
	}
	return intPtr(sum)
}

// worstCapacity picks the worst capacity band among ALL boxes, stale ones
// included: a box that stopped reporting still had a last-known fill level,
// and dropping it from the vote just because its snapshot aged out would
// hide a real "getting full" or "at capacity" reading behind the newer,
// possibly-rosier boxes — exactly the silent-omission failure the plan's
// privacy rules refuse to allow, just pointed at capacity instead of a
// hostname. (Contrast worstReplication below, which the plan explicitly
// scopes to fresh boxes — staleness there means "we don't know", not "we
// know it was bad".) A box with an unknown (nil) band doesn't vote — nil
// never invents a word. All-nil input yields nil.
func worstCapacity(boxes []Box) *string {
	var worst *string
	worstRank := -1
	for _, b := range boxes {
		if b.Capacity == nil {
			continue
		}
		rank, ok := capacityRank[*b.Capacity]
		if !ok {
			continue // defensive: an unrecognized band word doesn't vote
		}
		if rank > worstRank {
			worstRank = rank
			v := *b.Capacity
			worst = &v
		}
	}
	return worst
}

// worstReplication picks the worst replication reading among non-stale
// boxes: unhealthy beats healthy beats unknown, and ties within a band
// break toward the larger lag (still the worse reading). Nothing fresh
// with a known reading yields the zero Replication (both fields nil) —
// unknown, not a guessed "healthy".
func worstReplication(boxes []Box, stale []bool) Replication {
	var worst Replication
	haveWorst := false
	for i, b := range boxes {
		if stale[i] || b.Replication.Healthy == nil {
			continue
		}
		switch {
		case !haveWorst:
			worst = b.Replication
			haveWorst = true
		case !*b.Replication.Healthy && *worst.Healthy:
			// This box is unhealthy and the current worst isn't — a real
			// outage always outranks "just some lag".
			worst = b.Replication
		case *b.Replication.Healthy == *worst.Healthy &&
			b.Replication.LagSecs != nil && worst.LagSecs != nil &&
			*b.Replication.LagSecs > *worst.LagSecs:
			// Same health band; the larger lag is the worse reading.
			worst = b.Replication
		}
	}
	return worst
}

// worstAdoption folds the boxes' adoption readings into one fleet-wide
// answer, worst-first: a single box that is KNOWN not to be adopting makes the
// fleet false, because a real defect must never be averaged away by the
// healthy boxes around it. Failing that, an unknown box nulls the aggregate —
// "all the boxes we could ask said yes" is not the same claim as "the fleet is
// adopting", and this package does not make the stronger claim from the weaker
// evidence. Only when every voting box is known-true is the fleet true.
//
// Stale boxes don't vote at all (same rule as worstReplication): a snapshot
// that stopped arriving means "we no longer know what that box is doing", and
// a reading from twenty minutes ago is not evidence about now. That is a real
// blind spot rather than a solved problem — but staleness is itself rendered
// loudly per-box, so it is a visible blind spot, which is the whole point of
// this field.
// The order of the checks matters and is not the order the boxes happen to
// arrive in: a known false is decided BEFORE any unknown is allowed to null
// the result, so one unreachable box can never mask a box that reported a dead
// timer.
func worstAdoption(boxes []Box, stale []bool) *bool {
	anyUnknown, anyTrue := false, false
	for i, b := range boxes {
		if stale[i] {
			continue
		}
		switch {
		case b.AdoptionRunning == nil:
			anyUnknown = true
		case !*b.AdoptionRunning:
			no := false
			return &no
		default:
			anyTrue = true
		}
	}
	if anyUnknown || !anyTrue {
		return nil
	}
	yes := true
	return &yes
}

// routeGroup collects one App name's routes in the order they appeared in
// the input slice — used for both the apps and services groupings below,
// which differ only in which Kind they filter on and how they reduce a
// group down to one published row.
type routeGroup struct {
	name   string
	routes []registry.Route
}

// groupRoutesByApp buckets routes matching kind into per-app groups,
// preserving first-seen order so output is deterministic without needing a
// separate sort pass.
//
// The group key is (ACCOUNT, app), not app: app names are unique within an
// account, never globally, so two tenants can both run a "blog". Keying on
// the name alone would merge them — publishing one tenant's outage under the
// other's name, and letting either tenant's StatusNamed opt-in decide whether
// the other appears at all. The published NAME stays the app name; the
// account is only ever the grouping key, never anything the public page
// reveals.
func groupRoutesByApp(routes []registry.Route, kind string) []routeGroup {
	type key struct{ acct, app string }
	index := map[key]int{}
	var groups []routeGroup
	for _, r := range routes {
		if r.Kind != kind {
			continue
		}
		k := key{release.RouteAccount(r), r.App}
		i, ok := index[k]
		if !ok {
			i = len(groups)
			index[k] = i
			groups = append(groups, routeGroup{name: r.App})
		}
		groups[i].routes = append(groups[i].routes, r)
	}
	return groups
}

// buildApps groups the registry's "instance" routes by App and publishes
// only groups where EVERY route opted in via StatusNamed — one un-opted
// route keeps its whole app dark, never a partial reveal. Everything else
// folds into the unlisted count so the aggregate is still honest about how
// many apps exist without naming the ones that didn't ask to be named.
func buildApps(routes []registry.Route) ([]FleetApp, int) {
	groups := groupRoutesByApp(routes, "instance")

	var apps []FleetApp
	for _, g := range groups {
		allNamed := true
		healthy := true
		for _, r := range g.routes {
			if !r.StatusNamed {
				allNamed = false
			}
			if r.Status != "ready" {
				healthy = false
			}
		}
		if !allNamed {
			continue
		}
		first := g.routes[0]
		for _, r := range g.routes[1:] {
			if r.Channel != first.Channel || r.Version != first.Version {
				log.Printf("status: fleet: app %q routes disagree on channel/version (using first route's %q/%q)",
					g.name, first.Channel, first.Version)
				break
			}
		}
		apps = append(apps, FleetApp{
			Name:      g.name,
			Instances: len(g.routes),
			Healthy:   healthy,
			Channel:   first.Channel,
			Version:   first.Version,
		})
	}
	return apps, len(groups) - len(apps)
}

// buildServices groups the registry's "service" routes by App and reports
// each as up only if every route in the group is ready. Service names are
// platform-reserved (svc/*), not tenant hostnames, so — unlike apps —
// there's no opt-in gate here.
func buildServices(routes []registry.Route) []struct {
	Name string `json:"name"`
	Up   bool   `json:"up"`
} {
	groups := groupRoutesByApp(routes, "service")

	var services []struct {
		Name string `json:"name"`
		Up   bool   `json:"up"`
	}
	for _, g := range groups {
		up := true
		for _, r := range g.routes {
			if r.Status != "ready" {
				up = false
			}
		}
		services = append(services, struct {
			Name string `json:"name"`
			Up   bool   `json:"up"`
		}{Name: g.name, Up: up})
	}
	return services
}

// pageAssets is the fixed publish order for assetsFS's contents: css before
// html, both before fleet.json (see PublishFleet). It is NOT fs.ReadDir
// order — "index.html" sorts alphabetically before "status.css", which
// would flip the required css-before-html order — so the sequence is
// spelled out explicitly rather than derived from a directory walk.
var pageAssets = []string{"status.css", "index.html"}

// PublishFleet writes the whole viewer: the embedded page assets first,
// then fleet.json last. Order mirrors Eleven's upload-styles-first rule —
// fleet.json is the object readers poll most often, so a reader who fetches
// it the instant after a publish must never be able to load a page that
// references a stylesheet or markup that hasn't landed yet. Every object
// goes through putOverwrite: PublishFleet republishes all three keys on
// every pass (every 5 minutes, per the plan), so a create-only Put would
// fail from the second pass onward.
func PublishFleet(ctx context.Context, s release.Store, f *Fleet) error {
	for _, name := range pageAssets {
		body, err := assetsFS.ReadFile("assets/" + name)
		if err != nil {
			return err
		}
		if err := putOverwrite(ctx, s, "status/"+name, body); err != nil {
			return err
		}
	}

	body, err := json.MarshalIndent(f, "", "  ")
	if err != nil {
		return err
	}
	return putOverwrite(ctx, s, "status/fleet.json", body)
}

// putOverwrite lands body at key with last-writer-wins semantics — the same
// read-etag-then-Put dance WriteBox uses, for the same reason: every
// status/ object PublishFleet writes gets republished on the same key each
// pass (fleet.json every 5 minutes, alongside the boxes), and
// release.Store's Put is CAS-only, so a bare create-only Put would fail
// every tick after the first with ErrCASConflict. One retry on a lost race;
// a second conflict is surfaced rather than retried again, since the status
// lease is supposed to make PublishFleet the only writer touching these
// keys — two losses in a row past the lease means something else is wrong.
func putOverwrite(ctx context.Context, s release.Store, key string, body []byte) error {
	etag, err := currentEtag(ctx, s, key)
	if err != nil {
		return err
	}
	if _, err := s.Put(ctx, key, body, etag); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
		etag, err = currentEtag(ctx, s, key)
		if err != nil {
			return err
		}
		if _, err := s.Put(ctx, key, body, etag); err != nil {
			return err
		}
	}
	return nil
}