package console

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

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

// disagreeSaid and noPointSaid throttle two diagnostics to once per code.
var disagreeSaid, noPointSaid sync.Map

// regionCandidate is one region an automatic choice may land in (spec
// §6.1). Code is in the record vocabulary: "" is the home.
type regionCandidate struct {
	Code        string
	Home        bool
	AWSRegion   string
	Reachable   bool // some FRESH box publishing the code has a dial address (F1: fresh = a current status snapshot)
	Hosting     bool // ... and its activator configured (unknown counts, as in placements)
	KeyOK       bool // ... and holds the deployment's secrets key (R9)
	Eligible    bool // some fresh box has all of it, every fact POSITIVELY true, and no fresh box lacks any of it (F2)
	NoKeyProven bool // R9: every reachable hosting box, fresh or stale, is a current writer, and none holds the key
}

// unavailableReason is "" for an eligible candidate, else the first bar it
// fails (the C16 vocabulary).
func (c regionCandidate) unavailableReason() string {
	switch {
	case c.Eligible:
		return ""
	case !c.Reachable:
		return "absent"
	case !c.Hosting:
		return "not-hosting"
	case !c.KeyOK:
		return "no-secrets-key"
	}
	return "not-eligible"
}

func isTrue(b *bool) bool { return b != nil && *b }

// isFalse is a field the box PUBLISHED as false; absent is unknown, not no.
func isFalse(b *bool) bool { return b != nil && !*b }

// regionCandidates folds an edges snapshot into candidates. expectedFP is
// the console's own secrets fingerprint; "" means it cannot judge, and then
// no coded region is KeyOK. The home is exempt from every eligibility fact.
// A coded candidate's AWS region comes only from its fully eligible boxes
// (codex plan-R2 2); NoKeyProven needs a complete snapshot (codex plan-R2
// 1).
//
// A coded region is Eligible only when BOTH hold (final review F2):
//
//   - at least one fresh box publishing it passes the whole predicate:
//     a dial address, hibernation_configured, serves_shared_pool and
//     accepts_inbound all positively true, and the deployment's key;
//   - no fresh box publishing it that reconcile would ADMIT for the record
//     fails that predicate. Absent and unknown fields fail it.
//
// A record names a region, not a box. internal/provision's reconcile pass
// (inScope, and the CARLOS_INBOUND=none refusal) admits every box whose
// region matches, whose pools serve the shared pool and which takes
// inbound; it never asks whether the box holds the key or has its
// activator wired, so a keyless or activator-less sibling can win the lease
// and wake the instance with nothing. Every box that could serve the record
// must therefore be able to. A box that POSITIVELY publishes
// serves_shared_pool false (a named-pool-only box) or accepts_inbound
// false (CARLOS_INBOUND=none) can never be handed the record, so it neither
// qualifies nor disqualifies the region (Ruling F2, refined). Absent is not
// false: a writer predating the field is on an unknown config, and
// reconcile's defaults (CARLOS_POOLS unset serves the shared pool,
// CARLOS_INBOUND unset accepts) admit it, so it counts as possibly
// admitted and fails. Stale boxes count for neither side (F1): they are
// not evidence that a region works, and not evidence that it is broken.
func regionCandidates(snap edgesSnapshot, homeName, expectedFP string) []regionCandidate {
	type acc struct {
		c                       regionCandidate
		regions                 map[string]int
		hostingBoxes            int
		unknownWriter, keyFound bool
		anyEligible             bool
		allEligible             bool // over fresh boxes; vacuously true
	}
	by := map[string]*acc{}
	get := func(code string, home bool) *acc {
		a, ok := by[code]
		if !ok {
			a = &acc{c: regionCandidate{Code: code, Home: home}, regions: map[string]int{}, allEligible: true}
			by[code] = a
		}
		return a
	}
	h := get("", true)
	h.c.Reachable, h.c.Hosting, h.c.KeyOK, h.c.Eligible = true, true, true, true
	for _, f := range snap.Files {
		home := f.Placement == "" || (homeName != "" && f.Placement == homeName)
		if !home && !provision.ValidRegion(f.Placement) {
			continue // fleet/<name> and malformed codes are never candidates
		}
		code := f.Placement
		if home {
			code = ""
		}
		a := get(code, home)
		if home {
			if f.EIP != "" && f.Region != "" {
				a.regions[f.Region]++
			}
			continue
		}
		reachable := f.EIP != ""
		hosting := isTrue(f.HibernationConfigured) || f.HibernationConfigured == nil
		keyOK := expectedFP != "" && f.SecretsKeyFP == expectedFP
		if reachable && hosting {
			// R9's negative proof counts every box with a dial address,
			// fresh or not: a lingering keyless file still describes that
			// box's configuration, and the explicit door this proof serves
			// admits boxes from the files alone (placements), with no
			// liveness rule of its own. Liveness changes nothing there.
			a.hostingBoxes++
			a.keyFound = a.keyFound || keyOK
			a.unknownWriter = a.unknownWriter || f.AcceptsInbound == nil
		}
		if !f.Fresh {
			// F1: a box with no fresh status snapshot is not evidence of
			// anything but the code's existence (reason "absent"). Its
			// edges file lingers after retirement; see readLiveEdgesSnapshot.
			continue
		}
		// A fresh box with no dial address still fails the predicate (F2):
		// reconcile admits it all the same, so it can take the record and
		// nothing can reach it. It just does not make the region Reachable.
		eligible := reachable && isTrue(f.HibernationConfigured) && isTrue(f.ServesSharedPool) && isTrue(f.AcceptsInbound) && keyOK
		a.anyEligible = a.anyEligible || eligible
		// Only a box reconcile would admit can take the region out: one
		// that does not positively refuse the shared pool or inbound.
		// This mirrors poolServed(cfg.Pools, "") and cfg.Inbound !=
		// InboundNone, the two facts SharedInstanceEligibility publishes.
		if !isFalse(f.ServesSharedPool) && !isFalse(f.AcceptsInbound) {
			a.allEligible = a.allEligible && eligible
		}
		if !reachable {
			continue
		}
		a.c.Reachable = true
		a.c.Hosting = a.c.Hosting || hosting
		if hosting {
			a.c.KeyOK = a.c.KeyOK || keyOK
		}
		if eligible && f.Region != "" {
			a.regions[f.Region]++
		}
	}
	// Complete evidence is the bar for every non-home region (re-review
	// r1): with one edges file unread, or one status snapshot unread, a
	// failing sibling may be the box that is missing, and the keyed box
	// that WAS read would speak for the region alone. Whole-request scope,
	// deliberately: simpler, and stricter than per region.
	evidence := snap.EvidenceComplete()
	if !evidence && len(by) > 1 {
		if _, said := livenessSaid.LoadOrStore("incomplete", true); !said {
			log.Printf("console: an incomplete read of the edges files or box snapshots (edges complete %v, snapshots complete %v), so no region outside the home is automatically eligible this request; said once", snap.Complete, snap.BoxesComplete)
		}
	}
	out := make([]regionCandidate, 0, len(by))
	for _, a := range by {
		if !a.c.Home {
			a.c.Eligible = evidence && a.anyEligible && a.allEligible
		}
		a.c.AWSRegion = majority(a.c.Code, a.regions)
		a.c.NoKeyProven = !a.c.Home && snap.Complete && expectedFP != "" && a.hostingBoxes > 0 && !a.unknownWriter && !a.keyFound
		if a.c.Eligible && !a.c.Home { // a home without a point yields "default" (spec §6.3), not "left out"
			if _, ok := geo.LookupAWSRegion(a.c.AWSRegion); !ok {
				if _, said := noPointSaid.LoadOrStore(a.c.Code, true); !said {
					log.Printf("console: region %q has no point in the shipped AWS region table (AWS region %q), so region guesses leave it out", a.c.Code, a.c.AWSRegion)
				}
			}
		}
		out = append(out, a.c)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Code < out[j].Code })
	return out
}

// regionCands is regionCandidates with this console's home and expected
// secrets fingerprint. The snapshot must come from liveEdges: a plain
// readEdgesSnapshot marks no box fresh, so every coded region would read
// as absent.
func (s *Server) regionCands(snap edgesSnapshot) []regionCandidate {
	return regionCandidates(snap, s.cfg.HomeRegion, s.expectedSecretsFP())
}

// liveEdges is the one read every automatic-eligibility caller makes: the
// edges files joined with box liveness as of now (F1).
func (s *Server) liveEdges(ctx context.Context) (edgesSnapshot, error) {
	return readLiveEdgesSnapshot(ctx, s.cfg.Releases, time.Now())
}

// candsOrHome is the candidates from a fresh snapshot, or the home alone
// when the listing fails (or there is no bucket): for displays and guesses
// that must not fail.
func (s *Server) candsOrHome(ctx context.Context) []regionCandidate {
	if s.cfg.Releases == nil {
		return homeOnlyCandidates()
	}
	snap, err := s.liveEdges(ctx)
	if err != nil {
		return homeOnlyCandidates()
	}
	return s.regionCands(snap)
}

// majority picks the most common AWS region, ties to the lexically
// smallest, and logs a disagreement once per code.
func majority(code string, regions map[string]int) string {
	best, n := "", 0
	for r, c := range regions {
		if c > n || (c == n && r < best) {
			best, n = r, c
		}
	}
	if len(regions) > 1 {
		if _, said := disagreeSaid.LoadOrStore(code, true); !said {
			log.Printf("console: boxes for region %q disagree on their AWS region %v; using %s", code, regions, best)
		}
	}
	return best
}

func homeOnlyCandidates() []regionCandidate {
	return []regionCandidate{{Code: "", Home: true, Reachable: true, Hosting: true, KeyOK: true, Eligible: true}}
}

func findCandidate(cands []regionCandidate, code string) (regionCandidate, bool) {
	for _, c := range cands {
		if c.Code == code {
			return c, true
		}
	}
	return regionCandidate{}, false
}

func eligibleCount(cands []regionCandidate) int {
	n := 0
	for _, c := range cands {
		if c.Eligible {
			n++
		}
	}
	return n
}

type policyState int

const (
	policyAbsent policyState = iota
	policyKnown
	policyUnreadable
)

// regionPolicy is what the manifest and the fleet placement say about
// where an instance may land (spec §6.1).
type regionPolicy struct {
	State  policyState
	M      *provision.Manifest
	Placed string
}

// appPolicy reads the manifest directly (never instanceManifest, which
// answers the caller itself) and the fleet placement. Any failure is
// policyUnreadable: an unrestricted policy is never invented.
func (s *Server) appPolicy(ctx context.Context, acct *Account, app string) regionPolicy {
	var p regionPolicy
	if s.cfg.Releases == nil {
		return regionPolicy{State: policyUnreadable} // no bucket: nothing to read, so nothing is allowed
	}
	m, err := provision.GetManifest(ctx, s.cfg.Releases, acct.Sqid, app)
	switch {
	case err == nil:
		p.State, p.M = policyKnown, m
	case errors.Is(err, os.ErrNotExist):
		p.State = policyAbsent
	default:
		log.Printf("console: region policy %s/%s: read manifest: %v", acct.Sqid, app, err)
		return regionPolicy{State: policyUnreadable}
	}
	reg, err := openRegistry(s.cfg.Registry)
	if err != nil {
		log.Printf("console: region policy %s/%s: open registry: %v", acct.Sqid, app, err)
		return regionPolicy{State: policyUnreadable}
	}
	placed, err := s.appPlacement(reg, acct, app)
	if err != nil {
		log.Printf("console: region policy %s/%s: placement: %v", acct.Sqid, app, err)
		return regionPolicy{State: policyUnreadable}
	}
	p.Placed = placed
	return p
}

func (p regionPolicy) pooled() bool { return p.State == policyKnown && p.M.Pool != "" }

// allows is the door's own allow-list predicate on a record-vocabulary code.
func (p regionPolicy) allows(code string) bool {
	switch p.State {
	case policyKnown:
		return instanceRegionAllowed(p.M.Regions, code)
	case policyAbsent:
		return true
	}
	return false
}