package edgesfile

import (
	"context"
	"encoding/json"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/boxchoice"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// The deployment-wide placement read: every box's edges file joined with
// its liveness. The console's automatic-eligibility doors and the box
// agent's relocation chooser (region HA stage B Task 10) read through it,
// so a relocation's target and a create's executor come from the same
// evidence. Moved here unchanged from the console (instancesplacement.go,
// boxchoice.go).

// PlacementView is the reader's half of the edges-file schema this
// join needs: version, box, region, and the host set. Roles are not
// consulted — a host a box names under ANY role is served there in some
// capacity, and "where is this instance" is the question, not "who is its
// write home" (that is the home map's question, internal/edge).
type PlacementView struct {
	V         int    `json:"v"`
	Box       string `json:"box"`
	EIP       string `json:"eip"`
	Region    string `json:"region"`
	Placement string `json:"placement"`
	// DataBucket is the box's CARLOS_HIBERNATE_BUCKET, verbatim
	// (internal/host/edgesfile.go writes it from that env and omits the key
	// entirely when it is unset). It carries the move lane's actual
	// coordinates, and doubles as positive hosting evidence from a writer
	// too old to publish HibernationConfigured.
	DataBucket string `json:"data_bucket"`
	// DataBucketRegion is that bucket's AWS region, as the box publishes it
	// beside DataBucket. An offline move's stage fixes both on the record.
	DataBucketRegion string `json:"data_bucket_region"`
	// HibernationConfigured is the box's own answer to "is my hibernation
	// activator wired" (internal/host/edgesfile.go). A POINTER because
	// absent and false are different facts and the difference is the whole
	// point: absent is a writer that predates the field, whose capability is
	// UNKNOWN, while false is a box that published the field and said no.
	// Reading absence as no is what made an earlier version withdraw healthy
	// regions mid-roll — see placements.
	HibernationConfigured *bool `json:"hibernation_configured"`
	// ServesSharedPool and AcceptsInbound: POINTERS, absent = a writer
	// predating them (unknown). SecretsKeyFP: absent = no key, or a writer
	// predating it; AcceptsInbound's presence tells the two apart
	// (nearest-region spec §6.1, R9).
	ServesSharedPool *bool  `json:"serves_shared_pool"`
	AcceptsInbound   *bool  `json:"accepts_inbound"`
	SecretsKeyFP     string `json:"secrets_key_fp"`
	// Box evidence (region HA spec 1 §5.1). All absent on a writer that
	// predates it: BoxID "", Capabilities nil, Draining false, Headroom nil.
	BoxID        string   `json:"box_id"`
	Capabilities []string `json:"capabilities"`
	Draining     bool     `json:"draining"`
	Headroom     *struct {
		MemFreeMiB   int64 `json:"mem_free_mib"`
		DiskFreeMiB  int64 `json:"disk_free_mib"`
		MemTotalMiB  int64 `json:"mem_total_mib"`
		DiskTotalMiB int64 `json:"disk_total_mib"`
		Awake        int   `json:"awake"`
	} `json:"headroom"`
	Hosts map[string]json.RawMessage `json:"hosts"`

	// Generation is when the box last wrote the file (RFC 3339), which
	// region HA spec 2's placement policy reads as headroom's age.
	Generation string `json:"generation"`

	// The moves protocol (moves spec §11), read only by the Phase 1 move
	// door's floor check (selectMoveBoxes); Task 13 adds the writer's side.
	// MovesProtocol is the protocol the box speaks (absent: 0).
	// MovesFloorMinRelease is the published floor's min_release the box
	// has armed locally. PrevRelease is its carlos.prev's release, "none"
	// when it keeps none, and PrevAllowedFor the floor's min_release that
	// carlos.prev was judged at or above.
	MovesProtocol        int    `json:"moves_protocol"`
	MovesFloorMinRelease string `json:"moves_floor_min_release"`
	MovesFloorState      string `json:"moves_floor_state,omitempty"`
	MovesFloorGeneration uint64 `json:"moves_floor_generation,omitempty"`
	PrevRelease          string `json:"prev_release"`
	PrevAllowedFor       string `json:"prev_allowed_for"`
	// The seeds floor (seeded-instances spec §3.3), read only by the
	// console's seedRegionReady.
	SeedsFloorState      string `json:"seeds_floor_state,omitempty"`
	SeedsFloorGeneration uint64 `json:"seeds_floor_generation,omitempty"`
	SeedsProtocol        int    `json:"seeds_protocol"`
	SeedsFloorMinRelease string `json:"seeds_floor_min_release"`
	PrevSeedsAllowedFor  string `json:"prev_seeds_allowed_for"`
	// Fresh is not on the wire: ReadLiveSnapshot sets it when the
	// box's own status snapshot is still fresh (status.BoxStale). An edges
	// file is not liveness, see ReadLiveSnapshot.
	Fresh bool `json:"-"`
}

// livenessSaid throttles ReadLiveSnapshot's two diagnostics to once
// per process, the way noPointSaid does for a code with no point.
var livenessSaid sync.Map

// ReadLiveSnapshot is ReadSnapshot joined with box liveness, for
// AUTOMATIC eligibility (regionCandidates). An edges file is not evidence
// that its box is alive: nothing deletes a retired box's file, and the
// file's own Generation moves only when the file is rewritten, so a
// healthy box's is arbitrarily old (internal/edgesfile.EdgeIPs's doc
// comment has the whole argument). The status snapshot is written every
// tick, so status.BoxStale is a real liveness signal, and this is the same
// join EdgeIPs makes, over the same period.
//
// A box with no fresh snapshot, and every box when the snapshots cannot be
// loaded, is simply not Fresh. The fold then counts it for nothing, so a
// deployment whose status store is split from its deployment bucket
// (CARLOS_STATUS_BUCKET, which the console does not read) degrades to
// home-only guesses. That is safe but invisible, so both cases are said
// once: the load failure, and files found with no snapshot matching any.
// The explicit door's placements stay on the generous file-only read; a
// person asked for that region and can see it refused later.
func ReadLiveSnapshot(ctx context.Context, s release.Store, now time.Time) (Snapshot, error) {
	snap, err := ReadSnapshot(ctx, s)
	if err != nil {
		return snap, err
	}
	// No edges file at all is not complete evidence by itself (Task 10
	// review M5): the join still runs, so a fresh status snapshot with no
	// edges file (ruling (A) below) makes the evidence incomplete, as it
	// does beside any other file. With no file and no fresh snapshot the
	// evidence is complete and empty.
	// Completeness is carried separately from staleness: LoadBoxesComplete
	// returns what LoadBoxes would, plus whether every listed snapshot was
	// actually read. A skipped read leaves that box's liveness UNKNOWN,
	// and unknown is never "stale".
	boxes, boxesComplete, err := status.LoadBoxesComplete(ctx, s)
	if err != nil {
		if _, said := livenessSaid.LoadOrStore("load", true); !said {
			log.Printf("edges: could not load box snapshots (%v), so no box outside the home counts for automatic region placement until they load", err)
		}
		return snap, nil
	}
	snap.BoxesComplete = boxesComplete
	// known is every box with a status object at all; live is the fresh
	// subset. The join key is the status snapshot's Label against the edges
	// file's own box field, the pair edgesfile.EdgeIPs joins on; the object
	// key names play no part.
	known, live := map[string]bool{}, map[string]bool{}
	for _, b := range boxes {
		known[b.Label] = true
		if !status.BoxStale(b, now, status.SnapshotPeriod) {
			live[b.Label] = true
		}
	}
	published := make(map[string]bool, len(snap.Files))
	joined := 0
	for i := range snap.Files {
		published[snap.Files[i].Box] = true
		snap.Files[i].Fresh = live[snap.Files[i].Box]
		if snap.Files[i].Fresh {
			joined++
		}
		// Ruling (B) reversed, re-review r2, on production evidence: every
		// box writes both objects and a retired box leaves both behind, so
		// an ABSENT snapshot is unknown liveness (incomplete evidence,
		// whole-request) and a STALE one is the retired box (not fresh,
		// counts for neither side, evidence complete).
		if !known[snap.Files[i].Box] {
			snap.BoxesComplete = false
			if _, said := livenessSaid.LoadOrStore("nostatus", true); !said {
				log.Printf("edges: box %q has an edges file under %s but no status snapshot at all, so no region outside the home is automatically eligible until it reports; said once", snap.Files[i].Box, Prefix)
			}
		}
	}
	// Ruling (A), re-review r2: a FRESH snapshot whose box has no edges
	// file is incomplete evidence, whole-request, like a failed read.
	// Reconcile admits a box without its edges file (its publication may
	// simply have failed), so a live box the fold cannot see could still
	// take the record. A stale snapshot with no file is a retired box and
	// is ignored.
	for label := range live {
		if !published[label] {
			snap.BoxesComplete = false
			if _, said := livenessSaid.LoadOrStore("unmatched", true); !said {
				log.Printf("edges: box %q has a fresh status snapshot but no edges file under %s, so no region outside the home is automatically eligible until it publishes one; said once", label, Prefix)
			}
		}
	}
	if joined == 0 && len(snap.Files) > 0 {
		if _, said := livenessSaid.LoadOrStore("none", true); !said {
			log.Printf("edges: %d edges file(s) under %s but no fresh box snapshot matched any of them, so automatic region placement sees the home alone (are edges and status in different buckets? see CARLOS_STATUS_BUCKET/CARLOS_STATUS_PREFIX)", len(snap.Files), Prefix)
		}
	}
	return snap, nil
}

// Snapshot is one read of control/edges/. Complete is false when any
// listed file could not be read or parsed: positive facts about the boxes
// that WERE read still hold, but nothing may be concluded about "every
// box" (R9's NoKeyProven, codex plan-R2 1).
type Snapshot struct {
	Files    []PlacementView
	Complete bool
	// BoxesComplete: every listed status snapshot was read, and every box
	// on either side has its counterpart, so each file's Fresh is a KNOWN
	// answer. Set only by ReadLiveSnapshot; a plain ReadSnapshot
	// carries no liveness evidence at all. Unknown liveness is not stale
	// (re-review r1 hole 2, r2 rulings (A) and (B) reversed): a snapshot
	// that could not be read, a fresh snapshot with no edges file, or an
	// edges file with no snapshot at all each make the evidence incomplete.
	BoxesComplete bool
}

// EvidenceComplete is the evidence bar for AUTOMATIC non-home eligibility: every
// listed edges file read and parsed, and every listed status snapshot
// read. Short of it a failing sibling may simply be missing from the fold,
// so nothing non-home is eligible for the request (ruling, re-review r1).
func (s Snapshot) EvidenceComplete() bool { return s.Complete && s.BoxesComplete }

// ReadSnapshot REPORTS a failed listing instead of answering empty:
// the door must tell "could not read" (503, C28) from "no box publishes
// this code" (codex R2-2). A single unreadable or malformed file is
// skipped, as before, and marks the snapshot incomplete.
func ReadSnapshot(ctx context.Context, s release.Store) (Snapshot, error) {
	keys, err := s.List(ctx, Prefix)
	if err != nil {
		return Snapshot{}, err
	}
	snap := Snapshot{Complete: true}
	for _, key := range keys {
		name, ok := strings.CutPrefix(key, Prefix)
		if !ok || !strings.HasSuffix(name, ".json") || strings.Contains(name, "/") {
			continue
		}
		body, _, err := s.Get(ctx, key)
		if err != nil {
			log.Printf("edges: could not read %s for the placement join: %v", key, err)
			snap.Complete = false
			continue
		}
		var f PlacementView
		if err := json.Unmarshal(body, &f); err != nil || f.V < 1 || f.Box == "" {
			log.Printf("edges: skipping %s in the placement join: not a versioned edges file", key)
			snap.Complete = false
			continue
		}
		snap.Files = append(snap.Files, f)
	}
	return snap, nil
}

// BoxChoice is the snapshot as boxchoice's predicate reads it. Every field
// the predicate reads is carried: a dropped field would silently read as
// absent, so the console's eligibility tests run through this conversion.
func (s Snapshot) BoxChoice() boxchoice.Snapshot {
	out := boxchoice.Snapshot{Complete: s.EvidenceComplete(), Boxes: make([]boxchoice.Box, 0, len(s.Files))}
	for _, f := range s.Files {
		b := boxchoice.Box{
			Box: f.Box, BoxID: f.BoxID, EIP: f.EIP, Region: f.Region, Placement: f.Placement,
			HibernationConfigured: f.HibernationConfigured, ServesSharedPool: f.ServesSharedPool,
			AcceptsInbound: f.AcceptsInbound, SecretsKeyFP: f.SecretsKeyFP, Capabilities: f.Capabilities,
			Draining: f.Draining, Generation: f.Generation, Fresh: f.Fresh,
		}
		if f.Headroom != nil {
			b.Headroom = &boxchoice.Headroom{MemFreeMiB: f.Headroom.MemFreeMiB, DiskFreeMiB: f.Headroom.DiskFreeMiB,
				MemTotalMiB: f.Headroom.MemTotalMiB, DiskTotalMiB: f.Headroom.DiskTotalMiB, Awake: f.Headroom.Awake}
		}
		out.Boxes = append(out.Boxes, b)
	}
	return out
}

// LiveSnapshot is ReadLiveSnapshot in boxchoice's form: the read
// hostops.ExecutorChooser's Edges takes.
func LiveSnapshot(ctx context.Context, s release.Store, now time.Time) (boxchoice.Snapshot, error) {
	snap, err := ReadLiveSnapshot(ctx, s, now)
	if err != nil {
		return boxchoice.Snapshot{}, err
	}
	return snap.BoxChoice(), nil
}