package console

import (
	"context"
	"log"
	"sort"
	"sync"
	"time"

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

// livenessSaid throttles the console's own incomplete-evidence diagnostic
// to once per process (the liveness join's are edgesfile's).
var livenessSaid sync.Map

// edgesFilePrefix is where every box publishes what it serves — one object
// per box, control/edges/<box>.json, written by internal/host's edges-file
// pass and read by every edge's home map. This is the third reader of that
// prefix (internal/edge, internal/steering) and, like them, declares its own
// view of the wire shape rather than sharing a type: the writer's schema
// grows additively and each reader names only the fields it needs.
const edgesFilePrefix = "control/edges/"

// servedBy is one box that serves a host, as the instances API reports it
// (issue #337). Region is the box's AWS region as the edges file carries it
// (e.g. "eu-west-1") — the vocabulary a consumer fleet surface already shows
// from its own IMDS read — and omitted when the file predates the field.
type servedBy struct {
	Box    string `json:"box"`
	Region string `json:"region,omitempty"`
}

// edgesFilePlacementView is the edges-file placement view, shared with
// the box agent's relocation chooser through internal/edgesfile (region HA
// stage B Task 10): one read for every automatic-eligibility caller.
type edgesFilePlacementView = edgesfile.PlacementView

// loadEdgesJoin folds every readable edges file into host → the boxes that
// serve it, box-sorted so two consoles answer identically. Best-effort in
// the same spirit as loadBoxJoin: a failed list, an unreadable object, a
// parse failure or an unversioned file each cost only the answer that object
// would have contributed — the instances themselves are the record store's
// story and must never fail to list because a placement hint is missing.
// Keys that are not <box>.json directly under the prefix are not read at
// all, the same shape filter the other readers apply.
func loadEdgesJoin(ctx context.Context, s release.Store) map[string][]servedBy {
	out := map[string][]servedBy{}
	for _, f := range readEdgesFiles(ctx, s) {
		for host := range f.Hosts {
			out[host] = append(out[host], servedBy{Box: f.Box, Region: f.Region})
		}
	}
	for host := range out {
		sort.Slice(out[host], func(i, j int) bool { return out[host][i].Box < out[host][j].Box })
	}
	return out
}

// readEdgesFiles is the shared best-effort read under both consumers:
// list, shape-filter, parse, skip anything unreadable or unversioned. A
// failed list is an empty answer, logged — the callers each degrade to
// their pre-#337/#329 behaviour rather than failing the request. It is
// readEdgesSnapshot with the listing error folded into that empty answer.
func readEdgesFiles(ctx context.Context, s release.Store) []edgesFilePlacementView {
	if s == nil {
		return nil
	}
	snap, err := readEdgesSnapshot(ctx, s)
	if err != nil {
		log.Printf("console: api instances: could not list %s for the placement join: %v", edgesFilePrefix, err)
		return nil
	}
	return snap.Files
}

// edgesSnapshot is one read of control/edges/ (edgesfile.Snapshot).
type edgesSnapshot = edgesfile.Snapshot

// readEdgesSnapshot is edgesfile.ReadSnapshot.
func readEdgesSnapshot(ctx context.Context, s release.Store) (edgesSnapshot, error) {
	return edgesfile.ReadSnapshot(ctx, s)
}

// readLiveEdgesSnapshot is edgesfile.ReadLiveSnapshot: the edges files
// joined with box liveness, for automatic eligibility.
func readLiveEdgesSnapshot(ctx context.Context, s release.Store, now time.Time) (edgesSnapshot, error) {
	return edgesfile.ReadLiveSnapshot(ctx, s, now)
}

// placementCapability is what one placement code's boxes can do, folded
// across every box publishing that code. Three separate facts, because the
// three doors that consult them are asking three different questions, and
// conflating those is how a static site came to be refused for want of an
// activator it never needed.
type placementCapability struct {
	// Reachable: some box publishing this code has a dial address. A box
	// nothing can reach is not a landing target for anything.
	Reachable bool
	// Hosting: some reachable box publishing this code has its hibernation
	// activator configured. Only a SOCKET-BACKED INSTANCE needs this — a
	// static route is served from bucket bytes by every edge, and an
	// addr-backed service proxies to a process the platform never spawns.
	Hosting bool
	// Staging: some reachable box publishing this code publishes actual
	// data-bucket COORDINATES, which is the move door's requirement and is
	// strictly stronger than Hosting. internal/host's moveTargets drops any
	// target whose file carries no data_bucket, and runMoveRound then
	// reports "target region %s publishes no data bucket yet — waiting" for
	// as long as the record lives. A door admitting such a move promises
	// something the worker refuses forever.
	Staging bool
}

// placements folds the fleet's edges files into each published placement
// code's capability (issue #329, discussion #110). The second return reports
// whether ANY file carried a placement code at all: false means the fleet
// predates the field (or the read failed) and the create gate stays DORMANT,
// accepting what it always accepted, so rolling boxes and console in either
// order is safe.
//
// The hosting question is answered PER BOX, from that box's own
// hibernation_configured field, and is never inferred across boxes. That
// distinction is what discussion #110's review paid for. An earlier version
// keyed hosting on data_bucket alone — and because an absent data_bucket is
// exactly what an OLD WRITER and a bucket-less new box both produce, it had
// to decide when silence meant "no". It decided by asking whether any OTHER
// box had spoken, so the first box in a fleet to adopt a bucket-publishing
// build withdrew every region that had not yet rolled, all of them healthy:
// minutes for an uninterrupted fleet-wide roll, hours at one box an hour,
// unbounded if a roll stalled. One writer's upgrade cannot establish another
// writer's capabilities.
//
// So each file answers only for itself, and UNKNOWN is not NO:
//
//	hibernation_configured present → that box's own answer, believed
//	absent                         → a writer predating the field: unknown,
//	                                 and it keeps the eligibility it had
//	                                 before the field existed
//
// That last line is what makes a box roll safe in either direction, and it
// is deliberately generous. A genuinely unconfigured box on an old build
// stays eligible until it rolls and says so, so a create into it can still
// strand — but that is the bug we already had, whereas refusing creates into
// regions that host perfectly well would be a new one, caused BY the fix.
//
// None of these three is a promise that a wake will SUCCEED. A box whose IAM
// role cannot reach the bucket it names, or whose CARLOS_HIBERNATE_REGION is
// wrong, publishes hibernation_configured true and fails every lease. This
// is configuration, which is all an edges file can honestly carry.
func placements(ctx context.Context, s release.Store) (caps map[string]placementCapability, published bool) {
	caps = map[string]placementCapability{}
	for _, f := range readEdgesFiles(ctx, s) {
		if f.Placement == "" {
			continue
		}
		published = true
		caps[f.Placement] = foldCapability(caps[f.Placement], f)
	}
	return caps, published
}

// foldCapability adds one box's edges file to its code's capability.
func foldCapability(c placementCapability, f edgesFilePlacementView) placementCapability {
	// Every capability below is gated on reachability: a box with no dial
	// address contributes nothing, however well configured it is. The code
	// is still RECORDED, so an unreachable region reads as known-and-unarmed
	// rather than as absent.
	if f.EIP == "" {
		return c
	}
	// Unknown is not no. An absent hibernation_configured with no
	// data_bucket is a writer that predates the field, and it keeps the
	// eligibility it had before the field existed.
	hosting := true
	if f.HibernationConfigured != nil {
		hosting = *f.HibernationConfigured
	}
	c.Reachable = true
	c.Hosting = c.Hosting || hosting
	c.Staging = c.Staging || f.DataBucket != ""
	return c
}

// movePlacements is the move door's reading of the same files: placements
// plus the home, folded under the record's "" from every edges file that
// carries no placement (a home box sets no CARLOS_REGION). The rules are
// placements' own: no dial address, no capability, and staging is a
// published data bucket. homeNamed is the sorted boxes whose file names
// home as its placement (only when home is named): the console reads such a
// box as home (homeRegionIn), but the box does not, so it cannot take a
// move into home (MV-N511). published keeps placements' meaning, so a fleet
// whose only boxes are the home's still publishes no armed regions.
//
// A separate reader, not a flag on placements (Review Focus 4):
// placements feeds the create gate's published flag and armedPlacements,
// and a home folded in there would change create.
func movePlacements(ctx context.Context, s release.Store, home string) (caps map[string]placementCapability, homeNamed []string, published bool) {
	caps = map[string]placementCapability{}
	for _, f := range readEdgesFiles(ctx, s) {
		if f.Placement != "" {
			published = true
			if homePlacementCode(home, f.Placement) == "" {
				homeNamed = append(homeNamed, f.Box)
			}
		}
		caps[f.Placement] = foldCapability(caps[f.Placement], f)
	}
	sort.Strings(homeNamed)
	return caps, homeNamed, published
}

// armedPlacements is the picker answer in the shape its callers have always
// taken: code -> "can take an instance right now", which is Reachable AND
// Hosting.
func armedPlacements(ctx context.Context, s release.Store) (set map[string]bool, published bool) {
	caps, published := placements(ctx, s)
	return hostingSet(caps), published
}

// hostingSet and reachableSet project a capability map onto the one axis a
// particular refusal is about, so the list a caller is handed names regions
// that would actually satisfy the bar they just failed.
func hostingSet(caps map[string]placementCapability) map[string]bool {
	set := make(map[string]bool, len(caps))
	for code, c := range caps {
		set[code] = c.Reachable && c.Hosting
	}
	return set
}

func reachableSet(caps map[string]placementCapability) map[string]bool {
	set := make(map[string]bool, len(caps))
	for code, c := range caps {
		set[code] = c.Reachable
	}
	return set
}