package console

// The operator doors for platform GeoIP (geoip spec §4.2, §5.5):
//
//	PUT    /api/cli/system/geoip   stage an upload, answer 202 + a job id
//	GET    /api/cli/system/geoip   the pointer, every box, the rollout verdict
//	DELETE /api/cli/system/geoip   off: delete the pointer, sweep the blobs
//
// All three open with apiSystemRoller (an ops owner or CARLOS_SYSTEM_OPERATORS,
// a person, never a token) and are registered with the zero svcrole.Set.

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"sort"
	"strings"
	"time"

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

// geoipCommitDelay is a test seam: a job sleeps this long before Commit, so
// a test can prove the 202 arrives before verification finishes.
var geoipCommitDelay time.Duration

// geoipRetiredAfter is how old a heartbeat may be, for a box with no edges
// record and no fleet registration, before it is left out as retired.
const geoipRetiredAfter = 7 * 24 * time.Hour

func (s *Server) handleAPISystemGeoipPut(w http.ResponseWriter, r *http.Request) {
	addr, _, ok := s.apiSystemRoller(w, r)
	if !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	// The server-wide ReadTimeout is 5 minutes (cmd/console/main.go); a
	// 1 GiB upload gets its own deadline, the way the ship door does it.
	rc := http.NewResponseController(w)
	_ = rc.SetReadDeadline(time.Now().Add(15 * time.Minute))
	defer rc.SetReadDeadline(time.Time{})
	body := http.MaxBytesReader(w, r.Body, geoip.MaxBlobBytes)
	// Stage takes the one publish slot BEFORE reading any of the body
	// (codex R5-1): a second upload is refused here, unread.
	st, err := s.geoip.Stage(r.Context(), body)
	var mbe *http.MaxBytesError
	switch {
	case errors.Is(err, errGeoipBusy):
		jsonError(w, http.StatusConflict, "another geoip publish is in progress. try again when it finishes")
		return
	case errors.As(err, &mbe):
		jsonError(w, http.StatusRequestEntityTooLarge, "the database is larger than 1 GiB")
		return
	case errors.Is(err, errGeoipRefused):
		jsonError(w, http.StatusUnprocessableEntity, err.Error())
		return
	case err != nil:
		log.Printf("console: geoip upload: %v", err)
		jsonError(w, http.StatusBadGateway, "could not receive the database")
		return
	}
	opts := geoipPublishOpts{Origin: "manual", Force: r.URL.Query().Get("force") == "1"}
	id, job := s.newSystemJob()
	job.step("received %d bytes (sha256 %s)", st.size, st.sha[:8])
	// The job owns the slot and the staged directory from here; Commit
	// releases both on every path. A fresh context: the request's dies with
	// the 202, and a backhauled request has only 30s after its body to see
	// headers (codex R4-2).
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
		defer cancel()
		if geoipCommitDelay > 0 {
			time.Sleep(geoipCommitDelay)
		}
		job.step("verifying the database")
		ptr, err := s.geoip.Commit(ctx, st, opts)
		if err == nil {
			job.setResult(ptr)
			job.step("published %s: %s, built %s", ptr.SHA256[:8], ptr.DatabaseType, ptr.Build().Format("2006-01-02"))
			s.auditSystem(addr, "geoip publish", ptr.SHA256[:8]+" "+ptr.DatabaseType)
		} else if errors.Is(err, errGeoipRefused) {
			log.Printf("console: geoip publish refused: %v", err)
		} else {
			log.Printf("console: geoip publish: %v", err)
		}
		job.finish(err)
	}()
	writeJSON(w, http.StatusAccepted, map[string]string{"job": id})
}

func (s *Server) handleAPISystemGeoipDelete(w http.ResponseWriter, r *http.Request) {
	addr, _, ok := s.apiSystemRoller(w, r)
	if !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	if err := s.geoip.Off(r.Context()); errors.Is(err, errGeoipBusy) {
		jsonError(w, http.StatusConflict, "a geoip publish is in progress. run off again when it finishes")
		return
	} else if err != nil {
		log.Printf("console: geoip off: %v", err)
		jsonError(w, http.StatusBadGateway, "could not turn geoip off")
		return
	}
	s.auditSystem(addr, "geoip off", "")
	writeJSON(w, http.StatusOK, map[string]any{"off": true, "feed_set": os.Getenv("CONSOLE_GEOIP_FEED") != ""})
}

// geoipBoxRow is one box in the status answer.
type geoipBoxRow struct {
	Label    string             `json:"label"`
	Fleet    string             `json:"fleet,omitempty"`
	State    string             `json:"state"` // ok | stale | old binary | no heartbeat | other sha | tunnel | none
	LastSeen *time.Time         `json:"last_seen,omitempty"`
	GeoIP    *status.GeoIPState `json:"geoip,omitempty"`
	Expected bool               `json:"expected"`
	Blocking string             `json:"blocking,omitempty"`
}

func (s *Server) handleAPISystemGeoipGet(w http.ResponseWriter, r *http.Request) {
	if _, _, ok := s.apiSystemRoller(w, r); !ok {
		return
	}
	if !s.systemBucket(w) {
		return
	}
	ctx := r.Context()
	cur, _, present, perr := s.geoip.Current(ctx)
	out := map[string]any{"pointer": nil}
	want := geoipWant{state: geoipOff}
	switch {
	case perr != nil:
		out["pointer_error"] = "the published pointer is unreadable"
		want.state = geoipUnknown
	case present && (cur.Expired || geoip.Expired(cur.DatabaseType, cur.Build(), time.Now())):
		// Expired, marked or not yet marked by the daily job: every edge
		// drops it by the clock, so every mirror must be empty (codex R10
		// on R9-1).
		want.state = geoipOff
	case present:
		want = geoipWant{state: geoipLive, sha: cur.SHA256}
	}
	if perr == nil && present {
		p := map[string]any{"sha256": cur.SHA256, "database_type": cur.DatabaseType, "build": cur.Build(),
			"published": cur.Published, "origin": cur.Origin, "edition": cur.Edition, "size": cur.Size, "expired": cur.Expired}
		if at, ok := geoip.ExpiresAt(cur.DatabaseType, cur.Build()); ok {
			p["expires_at"] = at
		}
		if a, ok := geoip.AttributionFor(cur.DatabaseType); ok {
			p["attribution"] = a.Text
		}
		out["pointer"] = p
	}
	inv, err := s.geoipInventory(ctx)
	if err != nil {
		log.Printf("console: geoip status inventory: %v", err)
		jsonError(w, http.StatusBadGateway, "box heartbeats unreadable")
		return
	}
	ready, rows, blocking := geoipVerdict(want, inv, time.Now())
	out["boxes"], out["ready"], out["blocking"] = rows, ready, blocking
	out["feed_set"] = os.Getenv("CONSOLE_GEOIP_FEED") != ""
	if raw, _, err := release.OpenObject(ctx, s.cfg.Releases, geoip.FeedKey); err == nil {
		b, rerr := io.ReadAll(io.LimitReader(raw, geoip.MaxPointerBytes+1))
		raw.Close()
		if rerr == nil && len(b) <= geoip.MaxPointerBytes {
			if fs, err := geoip.ParseFeedStatus(b); err == nil {
				out["feed"] = map[string]any{"name": fs.Feed, "last_check": fs.LastCheck, "last_success": fs.LastSuccess,
					"edition": fs.LastEdition, "failures": fs.ConsecutiveFailures, "not_found": fs.ConsecutiveNotFound,
					"last_failure": fs.LastFailure, "failing": fs.Failing()}
			}
		}
	}
	writeJSON(w, http.StatusOK, out)
}

// geoipBoxKey is a box's identity: a label is unique only within its
// namespace, the platform ("") or one fleet (codex R8-1). Two fleets may
// both have node-1, and a fleet label may equal a platform label.
type geoipBoxKey struct{ Fleet, Label string }

func (k geoipBoxKey) String() string {
	if k.Fleet == "" {
		return k.Label
	}
	return k.Fleet + "/" + k.Label
}

// geoipInventoryRead is everything the verdict reconciles (spec §5.5).
// expectedComplete is false when the list of expected boxes itself could
// not be read (a fleet registry or fleet listing failed): that always
// blocks. heartbeatsComplete is false when some heartbeat record could not
// be read: informational only, because an expected box whose record was
// unreadable is missing from boxes and blocks as "no heartbeat" on its own
// (codex R9 on R1-16).
//
// fleetsListed is true only once the fleet registry was opened and its
// fleets listed; fleetsUnread says why not, for the operator log. The GeoIP
// verdict never reads it: it treats an unset or absent registry as a
// platform-only deployment, as it always has. The update settings gate
// (updatesEnforced) does read it, because a fleet list it never saw is not
// evidence that there are no fleet boxes to wait for (spec §4.4).
type geoipInventoryRead struct {
	expected           map[geoipBoxKey]bool
	boxes              map[geoipBoxKey]status.Box
	expectedComplete   bool
	heartbeatsComplete bool
	fleetsListed       bool
	fleetsUnread       string
}

// geoipWantState is what the published pointer asks of every mirror.
type geoipWantState int

const (
	geoipUnknown geoipWantState = iota // the pointer could not be read: never ready
	geoipOff                           // no pointer, or an expired one: mirrors must hold nothing
	geoipLive                          // mirrors must hold sha
)

type geoipWant struct {
	state geoipWantState
	sha   string
}

// geoipInventory reads the expected boxes (platform edges records, active
// fleet registrations) and every heartbeat prefix with LoadBoxesComplete,
// never LoadBoxes, which skips unreadable records silently.
func (s *Server) geoipInventory(ctx context.Context) (geoipInventoryRead, error) {
	inv := geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{}, expectedComplete: true, heartbeatsComplete: true}
	keys, err := s.cfg.Releases.List(ctx, "control/edges/")
	if err != nil {
		return inv, fmt.Errorf("edges records: %w", err)
	}
	for _, k := range keys {
		label := strings.TrimSuffix(strings.TrimPrefix(k, "control/edges/"), ".json")
		if label != "" && !strings.Contains(label, "/") && strings.HasSuffix(k, ".json") {
			inv.expected[geoipBoxKey{Label: label}] = true
		}
	}
	add := func(store release.Store, fleet string) error {
		boxes, complete, err := status.LoadBoxesComplete(ctx, store)
		if err != nil {
			return err
		}
		if !complete {
			inv.heartbeatsComplete = false
		}
		for _, b := range boxes {
			inv.boxes[geoipBoxKey{fleet, b.Label}] = b
		}
		return nil
	}
	if err := add(s.cfg.Releases, ""); err != nil {
		return inv, err
	}
	if s.cfg.Registry == "" {
		inv.fleetsUnread = "no registry path is configured"
		return inv, nil
	}
	reg, err := s.openPassRegistry()
	switch {
	case errors.Is(err, os.ErrNotExist):
		inv.fleetsUnread = err.Error()
		return inv, nil
	case err != nil:
		inv.expectedComplete = false
		inv.fleetsUnread = err.Error()
		return inv, nil
	}
	fleets, err := reg.ListFleets()
	if err != nil {
		inv.expectedComplete = false
		inv.fleetsUnread = "listing fleets: " + err.Error()
		return inv, nil
	}
	inv.fleetsListed = true
	for _, f := range fleets {
		fbs, err := reg.ListFleetBoxes(f.Name)
		if err != nil {
			inv.expectedComplete = false
			continue
		}
		for _, fb := range fbs {
			if fb.Status == "active" {
				inv.expected[geoipBoxKey{f.Name, fb.Label}] = true
			}
		}
		if err := add(release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/"), f.Name); err != nil {
			// That fleet's heartbeats are unknown: its expected boxes are
			// missing from boxes and block as "no heartbeat".
			inv.heartbeatsComplete = false
		}
	}
	return inv, nil
}

// geoipVerdict is spec §5.5's reconciliation. ready only when every listed
// box is fresh, readable and GeoIP-aware, every mirroring box holds the
// live pointer's sha, and every read was complete.
func geoipVerdict(want geoipWant, inv geoipInventoryRead, now time.Time) (bool, []geoipBoxRow, []string) {
	var rows []geoipBoxRow
	var blocking []string
	keys := map[geoipBoxKey]bool{}
	for k := range inv.expected {
		keys[k] = true
	}
	for k, b := range inv.boxes {
		if inv.expected[k] || now.Sub(b.GeneratedAt) <= geoipRetiredAfter {
			keys[k] = true
		}
	}
	sorted := make([]geoipBoxKey, 0, len(keys))
	for k := range keys {
		sorted = append(sorted, k)
	}
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].String() < sorted[j].String() })
	for _, k := range sorted {
		name, expected := k.String(), inv.expected[k]
		row := geoipBoxRow{Label: k.Label, Fleet: k.Fleet, Expected: expected}
		b, seen := inv.boxes[k]
		retire := ". if this box is retired, delete its edges record (control/edges/" + k.Label + ".json)"
		if k.Fleet != "" {
			retire = ". if this box is retired, detach it from fleet " + k.Fleet
		}
		switch {
		case !seen:
			row.State = "no heartbeat"
			row.Blocking = name + ": no heartbeat" + retire
		case status.BoxStale(b, now, status.SnapshotPeriod):
			t := b.GeneratedAt
			row.LastSeen = &t
			row.State = "stale"
			row.Blocking = name + ": no heartbeat since " + t.UTC().Format(time.RFC3339) + retire
		case b.Metrics == nil || b.Metrics.GeoIP == nil:
			t := b.GeneratedAt
			row.LastSeen = &t
			row.State = "old binary"
			row.Blocking = name + ": runs a binary that does not strip forged geo headers"
		default:
			t := b.GeneratedAt
			row.LastSeen, row.GeoIP = &t, b.Metrics.GeoIP
			switch g := b.Metrics.GeoIP; {
			case g.Mode == "tunnel", g.Mode == "none":
				row.State = g.Mode
			case want.state == geoipLive && g.SHA256 != want.sha:
				row.State = "other sha"
				row.Blocking = name + ": has " + short8(g.SHA256) + ", not the published " + short8(want.sha)
			case want.state == geoipOff && g.SHA256 != "":
				// Off or expired: a mirror still serving a generation has
				// not caught up (codex R9-1).
				row.State = "still loaded"
				row.Blocking = name + ": still has " + short8(g.SHA256) + " loaded"
			default:
				row.State = "ok"
			}
		}
		// Only an EXPECTED box (an edges record, an active fleet
		// registration) can hold the verdict up: deleting a retired box's
		// edges record unblocks at once, though its last heartbeat stays
		// listed for information (codex R8 on R1-16).
		if !expected {
			row.Blocking = ""
		}
		if row.Blocking != "" {
			blocking = append(blocking, row.Blocking)
		}
		rows = append(rows, row)
	}
	if want.state == geoipUnknown {
		blocking = append(blocking, "database: the published pointer could not be read")
	}
	if !inv.expectedComplete {
		blocking = append(blocking, "boxes: the list of fleet boxes could not be read")
	}
	if rows == nil {
		rows = []geoipBoxRow{}
	}
	if blocking == nil {
		blocking = []string{}
	}
	return len(blocking) == 0, rows, blocking
}

func short8(sha string) string {
	if sha == "" {
		return "nothing"
	}
	if len(sha) > 8 {
		return sha[:8]
	}
	return sha
}