package console

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

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

// The console half of `carlos system` (spec §5). Every door here is for an
// OWNER of the ops account or an address on CARLOS_SYSTEM_OPERATORS, on a
// PERSON session, checked on every call (apiSystemRoller); uploading a local
// build is the owner's alone, and an admitted operator who asks for it gets a
// 403 (systemUploadOwnerOnly). A caller the gate refuses gets a bare 404 like
// the operator doors, so anyone else learns nothing. The console holds the bucket, the signing key and the egress;
// the operator's laptop holds none of them.

const (
	systemUploadMaxBytes = int64(128 << 20)
	systemAssetMaxBytes  = int64(256 << 20)
	systemJobTTL         = time.Hour
)

// systemPlatformRe validates a "<goos>-<goarch>" the operator typed before it
// becomes part of an object key — the same shape and the same reason as
// apicli.go's targetRe, which is why it IS that expression rather than a
// second copy of it that could drift.
var systemPlatformRe = targetRe

// systemJob is one promote in flight. Steps is the narration an operator
// polls; Err is set once, at the end. The mutex guards all three against the
// polling handler.
type systemJob struct {
	mu    sync.Mutex
	Steps []string
	Done  bool
	Err   string
	// Result is a job's own answer when it has one (a geoip publish's
	// pointer, geoip spec §4.2): kept with the job so the poller sees what
	// THIS job published, not whatever the current pointer later says.
	Result any
	at     time.Time
}

func (j *systemJob) setResult(v any) {
	j.mu.Lock()
	j.Result = v
	j.mu.Unlock()
}

func (j *systemJob) step(format string, args ...any) {
	j.mu.Lock()
	j.Steps = append(j.Steps, fmt.Sprintf(format, args...))
	j.mu.Unlock()
}

func (j *systemJob) finish(err error) {
	j.mu.Lock()
	j.Done = true
	if err != nil {
		j.Err = err.Error()
	}
	j.mu.Unlock()
}

func (j *systemJob) done() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.Done
}

// snapshot copies the three published fields under the lock, so the caller
// can serialize them without holding it across a socket write.
func (j *systemJob) snapshot() (steps []string, done bool, errMsg string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	return append([]string(nil), j.Steps...), j.Done, j.Err
}

func (j *systemJob) result() any {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.Result
}

// newSystemJob mints and records a job with no channel slot: for work whose
// own lock does the serialising (the geoip publish slot).
func (s *Server) newSystemJob() (string, *systemJob) {
	s.systemMu.Lock()
	defer s.systemMu.Unlock()
	b := make([]byte, 8)
	rand.Read(b)
	id := hex.EncodeToString(b)
	job := &systemJob{at: time.Now()}
	if s.systemJobs == nil {
		s.systemJobs = map[string]*systemJob{}
	}
	for k, j := range s.systemJobs {
		if j.done() && time.Since(j.at) > systemJobTTL {
			delete(s.systemJobs, k)
		}
	}
	s.systemJobs[id] = job
	return id, job
}

// systemUpload is the one parked local build: `carlos system update --file`
// PUTs the bytes, then names the handle on the update request. There is no
// timestamp on it and no TTL sweep, because there is only ever ONE — the next
// upload evicts it — so the bound on what a forgotten one costs is the upload
// cap, not the age of the slot.
type systemUpload struct {
	version, platform string
	data              []byte
}

// apiSystemRoller is the gate every door in this file opens with: a person
// who is either an OWNER of the ops account or on SystemOperators
// (CARLOS_SYSTEM_OPERATORS, discussion #176). owner reports which, because
// the upload door and an update naming an upload are the owner's alone
// (SystemOperators' doc comment, server.go).
//
// A refusal is a bare 404, the same answer a nonexistent path gives, because
// whether this deployment has a system surface at all is not something a
// member needs to learn. The owner check comes first and a store error is
// not the end: a listed operator must still get in on a console whose role
// lookup failed, and a store error must never be what admits anyone.
func (s *Server) apiSystemRoller(w http.ResponseWriter, r *http.Request) (addr string, owner, ok bool) {
	p, ok := principalFrom(r.Context())
	if !ok || p.Kind != PrincipalUser {
		http.NotFound(w, r)
		return "", false, false
	}
	role, err := s.store.Role(OpsAccountID, p.Address)
	if err != nil {
		log.Printf("console: api system: role: %v", err)
	}
	if err == nil && role == RoleOwner {
		return p.Address, true, true
	}
	if s.systemOperator(p.Address) {
		return p.Address, false, true
	}
	http.NotFound(w, r)
	return "", false, false
}

// systemOperator reports whether addr is on SystemOperators. It is
// relayOperator's exact-match rule (apirelay.go): the configured entry is
// trimmed and lowercased, the signed-in address is compared as it is, and
// never with EqualFold, whose Unicode folding admits homoglyphs.
func (s *Server) systemOperator(addr string) bool {
	for _, op := range s.cfg.SystemOperators {
		if strings.ToLower(strings.TrimSpace(op)) == addr {
			return true
		}
	}
	return false
}

// auditSystem records an operator's platform action against the OPS account.
//
// It exists instead of s.audit because that one takes an *Account and
// dereferences it: these actions have no tenant account behind them — their
// subject is the deployment itself — so there is no *Account to pass and a
// nil would panic. The ops account is the right home for the row: its owners
// and the deployment's system operators (SystemOperators) are the only
// callers that ever reach these doors, and the row names which person it was.
//
// KNOW WHAT THIS IS AND IS NOT. These rows are keyed (OpsAccountID, app
// "platform"), and NO SURFACE READS THEM TODAY: Store.Events is asked for a
// real app name by the activity tab, and "platform" is not one. That is
// deliberate, not an oversight — the row is the DURABLE record of who moved
// the platform pointer and when, written where a reader can be added later
// (an operator activity view, `carlos system log`) without a migration.
// Until such a reader exists this is a write-only table, and the answer to
// "who promoted this?" comes from the pointer's own Audit list, which
// PromoteSystem signs into the bucket. Do not delete these rows on the
// grounds that nothing reads them; do not present them as an audit trail an
// operator can consult, because they cannot.
func (s *Server) auditSystem(addr, action, detail string) {
	if err := s.store.RecordEvent(OpsAccountID, "platform", addr, action, detail, time.Now()); err != nil {
		log.Printf("console: audit %s platform: %v", action, err)
	}
}

// systemBucket refuses on a console with no deployment bucket. Config.Releases
// == nil is a legal shape (see ensureAccountManifest and bumpSerial, which
// both guard for it), and every door here reads or writes the bucket — the
// pointer, the mirrored artifacts, the box heartbeats — so without this the
// answer to an ops owner on such a console is a nil-interface panic rather
// than a sentence.
func (s *Server) systemBucket(w http.ResponseWriter) bool {
	if s.cfg.Releases == nil {
		jsonError(w, http.StatusServiceUnavailable,
			"this console serves no deployment bucket, so there is no platform pointer to read or move")
		return false
	}
	return true
}

// systemFeed is the release feed this console fetches from: cfg.SystemFeed
// when it is configured, the vendor feed otherwise.
//
// "Configured" means ALL THREE strings, not just one. A feed's three fields
// are a single fact — where to ask, where to download, and the key the
// checksum list must verify under — and any subset of them is not a feed at
// all: a DownloadBase with the vendor's PubKey would fetch one publisher's
// binaries and check them against another's signature, which cannot ever
// succeed and would read as "the release is corrupt" rather than as "this
// console is misconfigured". So a partial override is refused OUT LOUD and
// falls back, rather than being half-honoured.
func (s *Server) systemFeed() publicrelease.Feed {
	f := s.cfg.SystemFeed
	set := 0
	for _, v := range []string{f.LatestURL, f.DownloadBase, f.PubKey} {
		if v != "" {
			set++
		}
	}
	switch set {
	case 0:
		return publicrelease.Default()
	case 3:
		return f
	default:
		log.Printf("console: SystemFeed is only partly configured (latest_url=%t download_base=%t pub_key=%t) — "+
			"a feed is all three or none, so this console is using the vendor feed instead",
			f.LatestURL != "", f.DownloadBase != "", f.PubKey != "")
		return publicrelease.Default()
	}
}

// systemBoxes reads every heartbeat: the platform prefix and each fleet's.
//
// strict is what a fleet the console cannot enumerate MEANS, and the two
// callers need opposite answers. The promote (systemPlatforms) decides from
// this list which platforms to mirror, and a fleet it cannot see is a fleet
// whose architecture it would silently omit — so strict refuses rather than
// guesses. The status read only DESCRIBES, and an operator asking what is
// running is better served by the boxes that did answer plus an honest
// "partial" flag than by a blank page. The second return is that flag: true
// means some fleet's boxes are missing from the slice.
//
// NEITHER answer applies to a console with NO REGISTRY. cfg.Registry == ""
// and a registry file that has not been created yet are legal, ordinary
// shapes — a console before its first box, or one that serves no route
// registry at all — and they mean this deployment has no fleets, full stop.
// Treating them as "unreadable" would refuse every promote such a console
// could ever make and pin its status at partial forever, which is what the
// first cut of this function did. Only a registry that EXISTS and still
// cannot be opened, or one whose fleets cannot be listed, is a fleet set this
// console cannot see.
func (s *Server) systemBoxes(ctx context.Context, strict bool) ([]status.Box, bool, error) {
	boxes, err := status.LoadBoxes(ctx, s.cfg.Releases)
	if err != nil {
		return nil, false, err
	}
	if s.cfg.Registry == "" {
		return boxes, false, nil // this console serves no route registry: no fleets
	}
	// openPassRegistry reports the not-yet-created case by wrapping os.Stat's
	// error, so errors.Is reaches it; the unconfigured case is the guard
	// above, since that one is a plain message. Distinguished here rather
	// than in openPassRegistry so its other callers keep the answers they
	// already expect.
	reg, err := s.openPassRegistry()
	switch {
	case errors.Is(err, os.ErrNotExist):
		return boxes, false, nil // no registry file yet: no fleets
	case err != nil:
		return s.systemFleetsUnreadable(boxes, strict, err)
	}
	fleets, err := reg.ListFleets()
	if err != nil {
		return s.systemFleetsUnreadable(boxes, strict, err)
	}
	partial := false
	for _, f := range fleets {
		fb, err := status.LoadBoxes(ctx, release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/"))
		if err != nil {
			if strict {
				return nil, false, fmt.Errorf("fleet %s: cannot read its box heartbeats, refusing to guess which platforms to mirror: %w", f.Name, err)
			}
			log.Printf("console: api system: fleet %s boxes: %v", f.Name, err)
			partial = true
			continue
		}
		boxes = append(boxes, fb...)
	}
	return boxes, partial, nil
}

// systemFleetsUnreadable is the shared answer to "this console cannot say
// what fleets exist": a refusal under strict, a logged degradation otherwise.
func (s *Server) systemFleetsUnreadable(boxes []status.Box, strict bool, err error) ([]status.Box, bool, error) {
	if strict {
		return nil, false, fmt.Errorf("cannot list fleets, refusing to guess which platforms to mirror: %w", err)
	}
	log.Printf("console: api system: list fleets: %v — reporting the platform's own boxes only", err)
	return boxes, true, nil
}

type systemUpdateRequest struct {
	Component string `json:"component"`
	Channel   string `json:"channel"`
	Version   string `json:"version"`
	Upload    string `json:"upload"`
	Spread    string `json:"spread"`
}

// handleAPISystemUpdate starts the promote and answers with a job id. The
// work happens off the request because ingesting a release is minutes of
// downloading and mirroring, and an operator watching a spinner over a
// connection that may drop is not a place to hold it.
func (s *Server) handleAPISystemUpdate(w http.ResponseWriter, r *http.Request) {
	addr, owner, ok := s.apiSystemRoller(w, r)
	if !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	if s.cfg.Key == nil {
		jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only")
		return
	}
	var in systemUpdateRequest
	if !readJSONStrictLimit(w, r, &in, 4<<10) {
		return
	}
	// An upload handle is refused BEFORE a job starts: the job is what
	// consumes the handle, so a check inside it would let an operator spend
	// an owner's parked build on a refused request.
	if in.Upload != "" && !owner {
		systemUploadOwnerOnly(w)
		return
	}
	if in.Component == "" {
		in.Component = release.SystemComponentRouter
	}
	if in.Component != release.SystemComponentRouter {
		jsonError(w, http.StatusBadRequest, "unknown component "+in.Component)
		return
	}
	if in.Channel != release.RungCanary && in.Channel != release.RungStable {
		jsonError(w, http.StatusBadRequest, "channel must be canary or stable")
		return
	}
	spread := 5 * time.Minute
	if in.Spread != "" {
		d, err := time.ParseDuration(in.Spread)
		if err != nil || d < 0 || d > 24*time.Hour {
			jsonError(w, http.StatusBadRequest, "spread must be a duration between 0 and 24h")
			return
		}
		spread = d
	}
	// A named release below the layout floor is refused here, before a
	// job starts, while any box reports a nested, fenced or conflict row
	// or no layout at all (layoutguard.go). Only a named version can be
	// judged: "the latest public release" is resolved inside the job and
	// is never below the floor, since the floor is a past release.
	if in.Version != "" && in.Upload == "" {
		if reason := s.layoutDowngradeBlocked(r.Context(), in.Version); reason != "" {
			jsonError(w, http.StatusConflict, reason)
			return
		}
	}
	id, job, busy := s.startSystemJob(in.Component, in.Channel)
	if busy != "" {
		writeJSON(w, http.StatusConflict, map[string]string{
			"error": "a promote of " + in.Component + "/" + in.Channel + " is already running; watch it or wait for it to finish",
			"job":   busy,
		})
		return
	}
	go s.runSystemUpdate(context.WithoutCancel(r.Context()), id, job, addr, in, spread)
	writeJSON(w, http.StatusOK, map[string]string{"job": id})
}

// startSystemJob claims the ONE in-flight promote slot for a channel and
// mints the job that holds it, or reports the id of the job already there.
//
// One at a time per channel, and the reason is memory as much as
// correctness: a promote holds a whole release asset in the heap while it
// verifies and mirrors it, so N concurrent promotes are N copies, and the
// door is reachable by an ops owner with a script. Serializing per channel
// also removes the CAS race two promotes of different versions into the same
// channel would otherwise run — one of them would lose at PromoteSystem's
// Put and report a conflict the operator did not cause.
//
// It sweeps FINISHED jobs past the TTL on the way past. Keyed on Done rather
// than age: a promote that has legitimately been running for over an hour
// (a slow feed, a large asset) must not have its record swept out from under
// the operator watching it, which is exactly the moment they need it.
func (s *Server) startSystemJob(component, channel string) (id string, job *systemJob, busy string) {
	s.systemMu.Lock()
	defer s.systemMu.Unlock()
	if running, ok := s.systemRunning[component+"/"+channel]; ok {
		return "", nil, running
	}
	b := make([]byte, 8)
	rand.Read(b)
	id = hex.EncodeToString(b)
	job = &systemJob{at: time.Now()}
	if s.systemJobs == nil {
		s.systemJobs = map[string]*systemJob{}
	}
	if s.systemRunning == nil {
		s.systemRunning = map[string]string{}
	}
	for k, j := range s.systemJobs {
		if j.done() && time.Since(j.at) > systemJobTTL {
			delete(s.systemJobs, k)
		}
	}
	s.systemJobs[id] = job
	s.systemRunning[component+"/"+channel] = id
	return id, job, ""
}

// releaseSystemChannel hands the slot back, and only if this job is still the
// one holding it: a compare-and-delete on the id, so a late or repeated
// release can never free a LATER promote's slot and let a third run beside
// it. That makes it idempotent, which is what lets runSystemUpdate both defer
// it (the panic path) and call it explicitly at the point the ordering
// matters.
func (s *Server) releaseSystemChannel(component, channel, id string) {
	key := component + "/" + channel
	s.systemMu.Lock()
	if s.systemRunning[key] == id {
		delete(s.systemRunning, key)
	}
	s.systemMu.Unlock()
}

// runSystemUpdate is the promote: resolve, ingest (or take the upload),
// mirror, sign, bump. Every step is idempotent, so a lost job is re-run.
func (s *Server) runSystemUpdate(ctx context.Context, id string, job *systemJob, actor string, in systemUpdateRequest, spread time.Duration) {
	// Every exit hands the channel back, a panic included — without this a
	// crashed promote would wedge its channel for the life of the process.
	defer s.releaseSystemChannel(in.Component, in.Channel, id)
	err := func() error {
		var rel release.SystemRelease
		if in.Upload != "" {
			s.systemMu.Lock()
			up, ok := s.systemUploads[in.Upload]
			delete(s.systemUploads, in.Upload)
			s.systemMu.Unlock()
			if !ok {
				return errors.New("upload handle unknown or already consumed")
			}
			if in.Version != "" && in.Version != up.version {
				return fmt.Errorf("upload is %s, request says %s", up.version, in.Version)
			}
			job.step("using uploaded %s for %s", up.version, up.platform)
			sum := sha256.Sum256(up.data)
			sha := hex.EncodeToString(sum[:])
			rel = release.SystemRelease{Version: up.version, Source: "file", Artifacts: map[string]release.SystemArtifact{}}
			if err := s.mirrorSystemArtifact(ctx, job, up.version, up.platform, up.data, sha); err != nil {
				return err
			}
			rel.Artifacts[up.platform] = release.SystemArtifact{Key: release.SystemArtifactKey(up.version, up.platform), SHA256: sha}
		} else {
			feed := s.systemFeed()
			tag := in.Version
			if tag == "" {
				var err error
				if tag, err = feed.Latest(ctx); err != nil {
					return err
				}
			}
			if !publicrelease.TagRe.MatchString(tag) {
				return fmt.Errorf("%q is not a release tag; use --file for a local build", tag)
			}
			job.step("resolved %s", tag)
			platforms, err := s.systemPlatforms(ctx)
			if err != nil {
				return err
			}
			rel = release.SystemRelease{Version: tag, Source: "release:" + tag, Artifacts: map[string]release.SystemArtifact{}}
			for _, plat := range platforms {
				data, sha, err := feed.Asset(ctx, tag, publicrelease.AssetName(plat), systemAssetMaxBytes)
				if err != nil {
					return fmt.Errorf("%s: %w", plat, err)
				}
				job.step("verified %s (%d bytes)", publicrelease.AssetName(plat), len(data))
				if err := s.mirrorSystemArtifact(ctx, job, tag, plat, data, sha); err != nil {
					return err
				}
				rel.Artifacts[plat] = release.SystemArtifact{Key: release.SystemArtifactKey(tag, plat), SHA256: sha}
			}
		}
		p, err := s.promoteHostUpdate(ctx, rel, in.Component, in.Channel,
			release.SystemPromoteOptions{Actor: actor, Spread: spread})
		if err != nil {
			return err
		}
		job.step("pointed %s/%s at %s (spread %s)", in.Component, in.Channel, p.Current.Version, p.Spread)
		s.auditSystem(actor, "system-update", "pointed "+in.Component+"/"+in.Channel+" at "+p.Current.Version)
		s.bumpSerial(ctx)
		job.step("serial bumped: boxes on %s start their spread", in.Channel)
		return nil
	}()
	// And hand it back BEFORE marking the job done, so an operator who polls
	// this job to completion and immediately promotes again is never told the
	// channel is busy by the job they just watched finish. The deferred
	// release above then finds nothing to do; it is there for the panic path.
	s.releaseSystemChannel(in.Component, in.Channel, id)
	job.finish(err)
}

// systemPlatforms: the distinct platforms the fleet reports; linux-arm64
// when nothing has reported the field yet. STRICT — a fleet this console
// cannot enumerate stops the promote, because a platform silently left out
// of the mirror is a fleet of boxes that will never adopt.
func (s *Server) systemPlatforms(ctx context.Context) ([]string, error) {
	boxes, _, err := s.systemBoxes(ctx, true)
	if err != nil {
		return nil, err
	}
	set := map[string]bool{}
	for _, b := range boxes {
		if b.Metrics != nil && b.Metrics.Platform != "" {
			set[b.Metrics.Platform] = true
		}
	}
	if len(set) == 0 {
		set["linux-arm64"] = true
	}
	var out []string
	for p := range set {
		out = append(out, p)
	}
	sort.Strings(out)
	return out, nil
}

// mirrorSystemArtifact: present with the same sha → skip; present with a
// different sha → refuse (versions are immutable, spec §3).
func (s *Server) mirrorSystemArtifact(ctx context.Context, job *systemJob, version, platform string, data []byte, sha string) error {
	key := release.SystemArtifactKey(version, platform)
	switch already, err := s.systemArtifactMatches(ctx, key, sha); {
	case err != nil:
		return err
	case already:
		job.step("%s already mirrored", key)
		return nil
	}
	// Create-only: the empty ifMatch IS the CAS, so a concurrent promote of
	// the same version loses rather than overwrites a byte a box may already
	// have fetched. Losing that race is not a failure — the two channels'
	// promotes of one version want the same bytes at the same key — so
	// re-read and let the comparison above decide: identical is done, and
	// different is the immutability refusal it always was.
	if _, err := s.cfg.Releases.Put(ctx, key, data, ""); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
		// The re-read answers every half: identical bytes means the other
		// promote wrote what this one would have, different bytes comes back
		// as the immutability refusal, and absent means the key was taken and
		// released while this promote looked away.
		already, cerr := s.systemArtifactMatches(ctx, key, sha)
		if cerr != nil {
			return cerr
		}
		if !already {
			return fmt.Errorf("%s: another writer holds the key but it is not there now — re-run, every step is idempotent", key)
		}
		job.step("%s mirrored concurrently", key)
		return nil
	}
	job.step("mirrored %s", key)
	return nil
}

// systemArtifactMatches answers "is sha already the object at key?" —
// absent is (false, nil), present and different is the immutability refusal,
// present and equal is (true, nil).
//
// The comparison STREAMS the object already in the bucket through a hasher
// rather than Getting it. A promote already holds one whole asset in the
// heap (the verified download, waiting to be Put); buffering the existing
// object beside it would make the peak two copies of a platform binary per
// in-flight promote, for no purpose beyond computing one sha256.
func (s *Server) systemArtifactMatches(ctx context.Context, key, sha string) (bool, error) {
	got, err := s.objectSHA256(ctx, key)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return false, nil
	case err != nil:
		return false, err
	}
	if got == sha {
		return true, nil
	}
	return false, fmt.Errorf("%s exists with different bytes — versions are immutable; pick a new version", key)
}

// objectSHA256 streams the object at key through a hasher; absent is
// os.ErrNotExist. A ctx that ended during the read is its error, even on a
// store that read on regardless.
func (s *Server) objectSHA256(ctx context.Context, key string) (string, error) {
	rc, _, err := release.OpenObject(ctx, s.cfg.Releases, key)
	if err != nil {
		return "", err
	}
	defer rc.Close()
	h := sha256.New()
	if _, err := io.Copy(h, rc); err != nil {
		return "", fmt.Errorf("%s: reading the object to compare it: %w", key, err)
	}
	if err := ctx.Err(); err != nil {
		return "", fmt.Errorf("%s: reading the object to compare it: %w", key, err)
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

// bootstrapKey is the binary a new, rebuilt or failover box downloads at
// first boot. `carlos system update` mirrors each release under
// bootstrap/router/ and never writes this key, so it falls behind the fleet
// unless an operator copies stable's build over it (issue #316).
const bootstrapKey = "bootstrap/carlos"

// bootstrapPlatform is the one platform bootstrap/carlos serves.
const bootstrapPlatform = "linux-arm64"

// bootstrapReadBudget bounds the comparison. It reads the whole binary, and
// the status read it rides (also `carlos system update`'s preview) must
// answer inside the CLI's 30-second timeout on slow storage: past the budget
// the row reads unreadable and the rest of the response stands.
var bootstrapReadBudget = 10 * time.Second

// bootstrapState is the status row for bootstrapKey against stable's signed
// linux-arm64 artifact, or nil when there is nothing to compare: no verified
// stable pointer, no linux-arm64 artifact in it, or no bootstrap object (a
// deployment that bootstraps another way). A read failure is reported, never
// fatal to the status read.
func (s *Server) bootstrapState(ctx context.Context, stable *release.SystemPointer) map[string]any {
	if stable == nil {
		return nil
	}
	art, ok := stable.Current.Artifacts[bootstrapPlatform]
	if !ok || art.SHA256 == "" {
		return nil
	}
	row := map[string]any{"version": stable.Current.Version, "key": art.Key}
	ctx, cancel := context.WithTimeout(ctx, bootstrapReadBudget)
	defer cancel()
	got, err := s.objectSHA256(ctx, bootstrapKey)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return nil
	case err != nil:
		log.Printf("console: api system status: %s: %v", bootstrapKey, err)
		row["state"], row["error"] = "unreadable", err.Error()
	case strings.EqualFold(got, art.SHA256):
		row["state"] = "current"
	default:
		row["state"] = "behind"
	}
	return row
}

func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) {
	if _, _, ok := s.apiSystemRoller(w, r); !ok {
		return
	}
	s.systemMu.Lock()
	job, ok := s.systemJobs[r.PathValue("id")]
	s.systemMu.Unlock()
	if !ok {
		jsonError(w, http.StatusNotFound, "no such job — the console may have restarted; re-run, every step is idempotent")
		return
	}
	steps, done, errMsg := job.snapshot()
	out := map[string]any{"steps": steps, "done": done, "error": errMsg}
	if res := job.result(); res != nil {
		out["result"] = res
	}
	writeJSON(w, http.StatusOK, out)
}

// handleAPISystemUpload parks one local build in memory for the job that
// names its handle. It is a streamed multipart read rather than a JSON body
// because the payload is a binary; the meta part comes FIRST so the version
// and platform are known — and refusable — before any bytes are buffered.
func (s *Server) handleAPISystemUpload(w http.ResponseWriter, r *http.Request) {
	_, owner, ok := s.apiSystemRoller(w, r)
	if !ok {
		return
	}
	if !owner {
		systemUploadOwnerOnly(w)
		return
	}
	if !s.systemBucket(w) {
		return // refuse before buffering bytes no promote could ever consume
	}
	r.Body = http.MaxBytesReader(w, r.Body, systemUploadMaxBytes+64<<10)
	mr, err := r.MultipartReader()
	if err != nil {
		jsonError(w, http.StatusBadRequest, "multipart body required")
		return
	}
	var meta struct {
		Version  string `json:"version"`
		Platform string `json:"platform"`
	}
	first, err := mr.NextPart()
	if err != nil || first.FormName() != "meta" {
		jsonError(w, http.StatusBadRequest, "first part must be meta")
		return
	}
	if err := json.NewDecoder(io.LimitReader(first, 4<<10)).Decode(&meta); err != nil {
		jsonError(w, http.StatusBadRequest, "meta: "+err.Error())
		return
	}
	if !release.ValidVersion(meta.Version) || meta.Version == "dev" {
		jsonError(w, http.StatusBadRequest, "meta.version is required and must be a version")
		return
	}
	if !systemPlatformRe.MatchString(meta.Platform) {
		jsonError(w, http.StatusBadRequest, "meta.platform is required (e.g. linux-arm64): the console cannot run a foreign binary to ask it")
		return
	}
	part, err := mr.NextPart()
	if err != nil || part.FormName() != "file" {
		jsonError(w, http.StatusBadRequest, "second part must be file")
		return
	}
	data, err := io.ReadAll(io.LimitReader(part, systemUploadMaxBytes+1))
	if err != nil {
		jsonError(w, http.StatusBadRequest, "file: "+err.Error())
		return
	}
	if int64(len(data)) > systemUploadMaxBytes {
		jsonError(w, http.StatusRequestEntityTooLarge, "file exceeds 128 MiB")
		return
	}
	b := make([]byte, 8)
	rand.Read(b)
	handle := hex.EncodeToString(b)
	s.systemMu.Lock()
	if s.systemUploads == nil {
		s.systemUploads = map[string]systemUpload{}
	}
	for k := range s.systemUploads { // one slot: a forgotten upload is not a memory leak
		delete(s.systemUploads, k)
	}
	s.systemUploads[handle] = systemUpload{version: meta.Version, platform: meta.Platform, data: data}
	s.systemMu.Unlock()
	writeJSON(w, http.StatusOK, map[string]any{"upload": handle, "size": len(data)})
}

// systemUploadOwnerOnly is the one refusal a system operator meets: a local
// build is the ops owner's (SystemOperators' doc comment, server.go). A 403
// rather than the gate's 404, because apiSystemRoller has already admitted
// the caller to these doors, so it teaches them only what THEY lack.
func systemUploadOwnerOnly(w http.ResponseWriter) {
	jsonError(w, http.StatusForbidden,
		"shipping a local build needs an owner of the ops account. as a system operator you can update to a signed release")
}

// handleAPISystemRollback points a channel back at what it ran before. It
// answers inline rather than as a job: nothing is fetched or mirrored,
// because the previous release travels whole inside the pointer's history.
func (s *Server) handleAPISystemRollback(w http.ResponseWriter, r *http.Request) {
	addr, _, ok := s.apiSystemRoller(w, r)
	if !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	if s.cfg.Key == nil {
		jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only")
		return
	}
	var in struct {
		Component string `json:"component"`
		Channel   string `json:"channel"`
	}
	if !readJSONStrictLimit(w, r, &in, 1<<10) {
		return
	}
	// The same two checks the update door makes, and for the same reason:
	// both arguments become part of an object key, and a door that validates
	// what it writes but not what it rewinds is a door with two standards.
	if in.Component == "" {
		in.Component = release.SystemComponentRouter
	}
	if in.Component != release.SystemComponentRouter {
		jsonError(w, http.StatusBadRequest, "unknown component "+in.Component)
		return
	}
	if in.Channel != release.RungCanary && in.Channel != release.RungStable {
		jsonError(w, http.StatusBadRequest, "channel must be canary or stable")
		return
	}
	// The rollback target is the pointer's History[0], exactly what
	// RollbackSystem will move to; it is read first so the layout guard
	// can judge it before the pointer moves (design §8.2, "The console's
	// rollback door"). A pointer that cannot be read here is left for
	// RollbackSystem to report in its own words below.
	if cur, _, err := release.GetSystemPointer(r.Context(), s.cfg.Releases, s.cfg.ReleasePub, in.Component, in.Channel); err == nil && len(cur.History) > 0 {
		if reason := s.layoutDowngradeBlocked(r.Context(), cur.History[0].Version); reason != "" {
			jsonError(w, http.StatusConflict, reason)
			return
		}
	}
	p, err := s.rollbackHostUpdate(r.Context(), in.Component, in.Channel, addr)
	var held activationHeldError
	switch {
	case errors.Is(err, release.ErrCASConflict):
		// M-6: the pointer moved between the read and the conditional
		// write. No retry: a rollback pops history, so repeating it against
		// a pointer the operator never saw could go back one release too far.
		jsonError(w, http.StatusConflict, "the "+in.Channel+" pointer moved while this rollback was being prepared, so nothing was rolled back. Check carlos system status, then run the rollback again if it is still wanted")
		return
	case errors.As(err, &held):
		jsonError(w, http.StatusConflict, held.Error())
		return
	case errors.Is(err, release.ErrNoSystemHistory):
		jsonError(w, http.StatusConflict, err.Error())
		return
	case errors.Is(err, os.ErrNotExist):
		jsonError(w, http.StatusNotFound, "no "+in.Channel+" pointer to roll back")
		return
	case err != nil:
		jsonError(w, http.StatusInternalServerError, err.Error())
		return
	}
	s.auditSystem(addr, "system-rollback", "pointed "+in.Component+"/"+in.Channel+" back at "+p.Current.Version)
	s.bumpSerial(r.Context())
	writeJSON(w, http.StatusOK, map[string]string{"channel": in.Channel, "version": p.Current.Version})
}

// handleAPISystemStatus is the whole picture in one read: what each channel
// points at, and what every box actually runs against it.
func (s *Server) handleAPISystemStatus(w http.ResponseWriter, r *http.Request) {
	if _, _, ok := s.apiSystemRoller(w, r); !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	ctx := r.Context()
	pointers := map[string]any{}
	targets := map[string]string{}
	var stable *release.SystemPointer
	for _, ch := range []string{release.RungCanary, release.RungStable} {
		p, _, err := release.GetSystemPointer(ctx, s.cfg.Releases, s.cfg.ReleasePub, release.SystemComponentRouter, ch)
		switch {
		case errors.Is(err, os.ErrNotExist):
			// null means exactly one thing: this channel has never been
			// promoted. It must not also mean "there IS a pointer and this
			// console cannot verify it" — a tampered or wrongly-signed
			// pointer reading as "never promoted" is the one failure an
			// operator most needs to see, shown as the one thing they would
			// ignore.
			pointers[ch] = nil
		case err != nil:
			log.Printf("console: api system status: %s pointer: %v", ch, err)
			pointers[ch] = map[string]any{"error": err.Error()}
		default:
			pointers[ch] = map[string]any{"version": p.Current.Version, "promoted_at": p.PromotedAt, "spread": p.Spread, "source": p.Current.Source,
				"history": len(p.History)}
			targets[ch] = p.Current.Version
			if ch == release.RungStable {
				stable = p
			}
		}
	}
	boxes, partial, err := s.systemBoxes(ctx, false)
	if err != nil {
		jsonError(w, http.StatusBadGateway, "box heartbeats unreadable: "+err.Error())
		return
	}
	now := time.Now()
	rows := make([]map[string]any, 0, len(boxes))
	for _, b := range boxes {
		row := map[string]any{"label": b.Label, "region": b.Region, "stale": status.BoxStale(b, now, status.SnapshotPeriod),
			"platform": "", "channel": "", "release": "", "target": "", "system": nil}
		if b.Metrics != nil {
			ch := b.Metrics.SystemChannel
			if ch == "" {
				ch = release.RungStable
			}
			row["platform"], row["channel"], row["release"] = b.Metrics.Platform, ch, b.Metrics.Release
			row["target"] = targets[ch]
			if b.Metrics.System != nil {
				row["system"] = b.Metrics.System
			}
		}
		rows = append(rows, row)
	}
	sort.Slice(rows, func(i, j int) bool { return rows[i]["label"].(string) < rows[j]["label"].(string) })
	// partial says the box list is INCOMPLETE — some fleet's heartbeats
	// could not be read. Always present, so a reader never has to tell a
	// missing key from a false one, and so "no boxes on that fleet" and
	// "could not ask that fleet" are different answers here.
	//
	// bootstrap is null when there is nothing to compare (bootstrapState).
	writeJSON(w, http.StatusOK, map[string]any{"pointers": pointers, "boxes": rows, "partial": partial,
		"bootstrap": s.bootstrapState(ctx, stable)})
}