package moves

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// The readiness checks of §7.3. The target box writes T1 to T13 (T13 the cut
// guard's, Paul's ruling of 2026-10-11) into its
// report; the source box writes S1 and S2 into its own. Ready joins them.
const (
	CheckProtocol = "T1"
	CheckSecrets  = "T2"
	CheckEnv      = "T3"
	CheckStorage  = "T4"
	CheckArchive  = "T5"
	CheckDisk     = "T6"
	CheckRelease  = "T7"
	CheckOneOwner = "T8"
	CheckClaim    = "T9"
	CheckLane     = "T10"
	CheckBudget   = "T11"
	CheckQuiet    = "T12"
	// CheckArmed is the cut guard's T13 (Paul's ruling, 2026-10-11): the
	// target box can land the move and roll it back, every hook of both
	// wired, judged when the report is written. It is the report's arming
	// statement: the source refuses to start a cut unless the target's
	// latest report states it passing (TargetReport.Armed).
	CheckArmed = "T13"
)

// AllChecks lists the target's checks in id order. Passing walks it, so the
// first failure it names is the lowest-numbered one.
var AllChecks = []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet, CheckArmed}

// ReportFreshness bounds how old a report may be and still decide ready
// (§7.3); the cut re-checks it at `admitted` (§7.4).
const ReportFreshness = 10 * time.Minute

// ReportSkew is how far ahead of the reader's clock a report may be dated
// (clock skew between boxes). A report dated further ahead is refused, so a
// wrong or stuck clock cannot keep a report fresh. It is the same bound as
// MeasurementSkew.
const ReportSkew = time.Minute

// MeasurementFreshness bounds how old a rate measurement may be and still
// price a cut (§7.3 T11). MeasurementSkew is how far ahead of the reader's
// clock a measurement made on another box may be dated.
const (
	MeasurementFreshness = time.Hour
	MeasurementSkew      = ReportSkew
)

// TargetReport is what the target's box publishes at TargetReportKey, every
// pass (§7.3). Inputs are what its checks were made against; Ready compares
// them with what the source publishes now. Notes are advisories that do not
// fail a check (the `shadowed` names of T3).
type TargetReport struct {
	V                int                  `json:"v"`
	Box              string               `json:"box"`
	Region           string               `json:"region"`
	MoveID           string               `json:"move_id"`
	At               time.Time            `json:"at"`
	Checks           map[string]Check     `json:"checks"`
	Inputs           provision.MoveInputs `json:"inputs"`
	AttemptsAdmitted int                  `json:"attempts_admitted"`
	CutBudgetS       int                  `json:"cut_budget_s"`
	Rates            Rates                `json:"rates"`
	// DBMiB, S3At and DiskAt are the evidence behind Rates, so that Ready
	// can re-price the cut and re-check freshness when it reads the report,
	// not only when the report was written. Rates.Cross is the source's rate
	// as the target used it; Ready prices from the source's current one.
	DBMiB  float64   `json:"db_mib"`
	S3At   time.Time `json:"s3_at"`
	DiskAt time.Time `json:"disk_at"`
	// Entries is the archive's entry count the Archive phase was priced
	// for, and FsyncAt when the per-file sync rate (Rates.FsyncRate) was
	// measured (ruling T7-budget).
	Entries int       `json:"entries"`
	FsyncAt time.Time `json:"fsync_at"`
	Notes   []string  `json:"notes,omitempty"`
}

// RecordBudget copies a BudgetCheck result and the measurements, database
// size and entry count it was priced from into the report, the fields Ready
// re-checks.
func (r *TargetReport) RecordBudget(dbMiB float64, entries int, m Measurements, res BudgetResult) {
	r.DBMiB = dbMiB
	r.Entries = entries
	r.S3At = m.S3.At
	r.DiskAt = m.Disk.At
	r.FsyncAt = m.Fsync.At
	r.Rates = res.Rates
	r.AttemptsAdmitted = res.AttemptsAdmitted
}

// SourceReport is what the source's box publishes at SourceReportKey.
// Release is the source's live version (release.LiveVersion), the pinned
// release T7 stages. CrossRate is the cross-region copy rate in MiB/s from
// the source's most recent copying round, made through the target bucket's
// region (the path the cut uses), and CrossRateAt is when that round ended.
type SourceReport struct {
	V          int       `json:"v"`
	Box        string    `json:"box"`
	MoveID     string    `json:"move_id"`
	At         time.Time `json:"at"`
	Env        EnvReport `json:"env"`
	S1         Check     `json:"s1"`
	S2         Check     `json:"s2"`
	FilesPlain string    `json:"files_plain"`
	Release    string    `json:"release"`
	// DBBytes is the size of the source's local database file when one is
	// present (a hibernating source may have none). T11 prices the larger
	// of it and the logical size in the newest staged LTX header.
	DBBytes int64 `json:"db_bytes,omitempty"`
	// ReplicaBytes is the database's logical size as the source's replica
	// states it this round: Commit pages of PageSize in the header of the
	// newest LTX file under the claim's generation, the one the lane
	// mirrors. Ready refuses while either size exceeds the size the
	// target's report priced (TargetReport.DBMiB, final review finding 2).
	ReplicaBytes int64       `json:"replica_bytes,omitempty"`
	Archive      *ArchiveRef `json:"archive,omitempty"`
	CrossRate    float64     `json:"cross_rate"`
	CrossRateAt  time.Time   `json:"cross_rate_at"`
}

// ArchiveRef names the archive the source published, by exact version, so
// T5 reads the very objects the source wrote. SealedObjectKey is the sealed
// archive's object key. It is not archive.Manifest's SealedDataKey, which is
// the key that seals the data inside.
type ArchiveRef struct {
	ManifestKey     string `json:"manifest_key"`
	ManifestVersion string `json:"manifest_version"`
	SealedObjectKey string `json:"sealed_object_key"`
	SealedVersion   string `json:"sealed_version"`
	Plain           string `json:"plain"`
}

// Pinned reports whether both objects are named by a version id that pins
// one immutable version (lease.UsableVersionID): not "", and not the "null"
// a versioning-suspended bucket answers, which the next write to the key
// overwrites (final review finding 4). A reference that is not pinned is
// never verified or trusted from a cache.
func (r ArchiveRef) Pinned() bool {
	return lease.UsableVersionID(r.ManifestVersion) && lease.UsableVersionID(r.SealedVersion)
}

// CrossMeasurement is the source's cross-region rate as a measurement. A
// nil report, or one that has measured nothing, gives the zero measurement,
// which Measurements.Rates refuses.
func (s *SourceReport) CrossMeasurement() MeasuredRate {
	if s == nil {
		return MeasuredRate{}
	}
	return MeasuredRate{MiBPerS: s.CrossRate, At: s.CrossRateAt}
}

// Armed reports whether the report states that its box can land the move
// and roll it back (CheckArmed). It fails closed: no report, a report from
// an older box that has no T13, and a failing T13 are all unarmed.
func (r *TargetReport) Armed() bool {
	if r == nil {
		return false
	}
	c, ok := r.Checks[CheckArmed]
	return ok && c.OK
}

// Passing reports whether every check passed, naming the first that did not
// in id order. A check that is missing from the report fails.
func (r TargetReport) Passing() (bool, string) {
	for _, id := range AllChecks {
		c, ok := r.Checks[id]
		if !ok {
			return false, id + " was not reported"
		}
		if !c.OK {
			if c.Reason == "" {
				return false, id + ": did not pass"
			}
			return false, id + ": " + c.Reason
		}
	}
	return true, ""
}

// Ready is §7.3's promotion rule: a round at most the bar, a passing target
// report and source readiness, each under ReportFreshness old, and at least
// one probe attempt admitted by the budget (R3-7). It is a pure reading and
// only ever gates promotion. A later stale report reads false, and nothing
// here, or in MovePromoteReady, which is legal from preparing only, takes a
// move back out of ready (Review Focus 5).
//
// The target's checks must also have been made against what the source
// publishes now (round-5 finding 9), its release included: a fresh passing report for archive A
// must not promote after the source has published archive B. And the budget
// must have been priced for a final mirror round as large as the bar allows
// (F28(e)), so raising CARLOS_MOVE_READY_DELTA_BYTES cannot promote on an
// estimate made under a smaller one. Nor can a database that has grown past
// the size the target priced (sizeCovered, final review finding 2).
func Ready(roundDelta, bar int64, target *TargetReport, source *SourceReport, now time.Time) (bool, string) {
	if roundDelta > bar {
		return false, fmt.Sprintf("the last round moved %d bytes and the bar is %d", roundDelta, bar)
	}
	if target == nil {
		return false, "the target has not published a readiness report yet"
	}
	if age := now.Sub(target.At); age > ReportFreshness {
		return false, "the target's readiness report is stale"
	} else if age < -ReportSkew {
		return false, "the target's readiness report is dated ahead of this clock"
	}
	if ok, why := target.Passing(); !ok {
		return false, why
	}
	if target.AttemptsAdmitted < 1 {
		return false, "T11: the cut budget admits no probe attempt"
	}
	priced := target.Rates.MirrorBar
	if priced == 0 {
		priced = MirrorBarMiB
	}
	if want := BarMiB(bar); !finite(priced) || !(priced >= want) { // written so NaN fails
		return false, fmt.Sprintf("T11: the cut budget was priced for a final round of %g MiB and the bar is %g MiB", priced, want)
	}
	if source == nil {
		return false, "the source has not published a readiness report yet"
	}
	if age := now.Sub(source.At); age > ReportFreshness {
		return false, "the source's readiness report is stale"
	} else if age < -ReportSkew {
		return false, "the source's readiness report is dated ahead of this clock"
	}
	if !source.S1.OK {
		return false, "S1: " + reasonOr(source.S1, "the source did not answer")
	}
	if !source.S2.OK {
		return false, "S2: " + reasonOr(source.S2, "the source could not confirm its spawn state")
	}
	if why := budgetEvidence(target, source, bar, now); why != "" {
		return false, why
	}
	if why := sizeCovered(target, source); why != "" {
		return false, why
	}
	if target.Inputs.FilesPlain != source.FilesPlain {
		return false, fmt.Sprintf("the target checked files %s and the source now publishes %s", target.Inputs.FilesPlain, source.FilesPlain)
	}
	if target.Inputs.EnvReport != source.Env.Digest {
		return false, fmt.Sprintf("the target checked launch environment %s and the source now reports %s", target.Inputs.EnvReport, source.Env.Digest)
	}
	// The release promotion pins is the one T7 established on the target,
	// never one the source moved to since (Task 14 review finding 3).
	if target.Inputs.Release == "" {
		return false, "the target has not checked a release yet"
	}
	if target.Inputs.Release != source.Release {
		return false, fmt.Sprintf("the target checked release %s and the source now runs %s", target.Inputs.Release, source.Release)
	}
	return true, ""
}

// budgetEvidence re-prices the cut from the evidence the reports carry, at
// `now`, and returns why it cannot be trusted, or "". Every rate must be a
// positive finite number measured within MeasurementFreshness (the target's
// S3 and disk rates by their report times, the cross-region rate by the
// source's CrossRateAt), the cut window and database size must be usable,
// and the recomputed budget must admit an attempt. The stored
// AttemptsAdmitted may be lower than the recomputation (the reports are
// refreshed on their own schedules) but never higher: a report cannot claim
// more attempts than its own evidence buys.
func budgetEvidence(target *TargetReport, source *SourceReport, bar int64, now time.Time) string {
	if !usable(target.Rates.Cross) || !usable(target.Rates.S3) || !usable(target.Rates.Disk) || !usable(target.Rates.FsyncRate) {
		return "T11: the target's report carries a rate that is not a usable number"
	}
	m := Measurements{
		Cross: source.CrossMeasurement(),
		S3:    MeasuredRate{MiBPerS: target.Rates.S3, At: target.S3At},
		Disk:  MeasuredRate{MiBPerS: target.Rates.Disk, At: target.DiskAt},
		Fsync: FilesRate{PerS: target.Rates.FsyncRate, At: target.FsyncAt},
	}
	res := BudgetCheck(target.DBMiB, target.Entries, m, bar, float64(target.CutBudgetS), now)
	if !res.Check.OK {
		return "T11: " + res.Check.Reason
	}
	if target.AttemptsAdmitted > res.AttemptsAdmitted {
		return fmt.Sprintf("T11: the target's report claims %d probe attempts and its measurements admit %d", target.AttemptsAdmitted, res.AttemptsAdmitted)
	}
	return ""
}

// sizeCovered refuses a database that has grown past the size the target's
// report priced (final review finding 2). TargetReport.DBMiB is the size
// both T11's budget and T6's disk check were made for, and the source's
// report carries the size now: its local file (DBBytes) and its replica's
// logical size (ReplicaBytes). A larger one leaves both checks stale even
// while the report is fresh, so promotion waits for the target's next
// report, which prices the new size. Equal or smaller is covered.
func sizeCovered(target *TargetReport, source *SourceReport) string {
	size := max(source.DBBytes, source.ReplicaBytes)
	if now := float64(size) / (1 << 20); now > target.DBMiB {
		return fmt.Sprintf("T11: the database has grown to %g MiB since the target priced %g MiB. waiting for the target's next report", now, target.DBMiB)
	}
	return ""
}

func reasonOr(c Check, fallback string) string {
	if c.Reason == "" {
		return fallback
	}
	return c.Reason
}

// ---- measured rates and T11 ----

// MeasuredRate is one throughput measurement, in MiB/s, with when and over
// how many bytes it was taken.
type MeasuredRate struct {
	MiBPerS float64   `json:"mib_per_s"`
	At      time.Time `json:"at"`
	Bytes   int64     `json:"bytes"`
}

// RateOf is bytes over d as a measurement made at `at`. No bytes or no
// elapsed time is no measurement (a zero rate), never an infinite one.
func RateOf(bytes int64, d time.Duration, at time.Time) MeasuredRate {
	m := MeasuredRate{At: at, Bytes: bytes}
	if bytes > 0 && d > 0 {
		m.MiBPerS = float64(bytes) / (1 << 20) / d.Seconds()
	}
	return m
}

// FilesRate is a per-file sync measurement: files synced per second, with
// when and over how many files it was taken (ruling T7-budget).
type FilesRate struct {
	PerS  float64   `json:"per_s"`
	At    time.Time `json:"at"`
	Files int       `json:"files"`
}

// Measurements are the rates T11 prices a cut from. The target measures
// regional S3, its disk and its per-file sync rate itself. Cross comes from
// the source's report (SourceReport.CrossMeasurement), as only the source's
// copying rounds cross the regions.
type Measurements struct {
	Cross, S3, Disk MeasuredRate
	Fsync           FilesRate
}

// BarMiB is a ready-delta bar in bytes as the MiB that Rates.MirrorBar
// counts. A bar of zero or less is zero, which Estimate prices at the
// default, never below it.
func BarMiB(bytes int64) float64 {
	if bytes <= 0 {
		return 0
	}
	return float64(bytes) / (1 << 20)
}

// Rates turns the measurements into the rates Estimate takes, with the
// operator's ready-delta bar (bytes) as the mirror bar. Every rate must be
// a positive finite number taken in the last MeasurementFreshness; any other
// is refused, naming which. The worked example's rates are never a fallback.
func (m Measurements) Rates(now time.Time, barBytes int64) (Rates, error) {
	for _, c := range []struct {
		which string
		m     MeasuredRate
	}{
		{"the cross-region copy path", m.Cross},
		{"regional S3", m.S3},
		{"the disk", m.Disk},
	} {
		if !fresh(c.m.MiBPerS, c.m.At, now) {
			return Rates{}, &NoFreshRateError{Which: c.which}
		}
	}
	if !fresh(m.Fsync.PerS, m.Fsync.At, now) {
		return Rates{}, &NoFreshRateError{Which: "the disk's per-file sync"}
	}
	return Rates{Cross: m.Cross.MiBPerS, S3: m.S3.MiBPerS, Disk: m.Disk.MiBPerS, MirrorBar: BarMiB(barBytes), FsyncRate: m.Fsync.PerS}, nil
}

// NoFreshRateError is Measurements.Rates refusing a rate: Which names it.
type NoFreshRateError struct{ Which string }

func (e *NoFreshRateError) Error() string { return NoFreshRate(e.Which) }

// NoFreshRate is T11's sentence for a missing or stale rate (MV-N68).
func NoFreshRate(which string) string {
	return fmt.Sprintf("no fresh rate measurement for %s", which)
}

// fresh is a usable rate measured within MeasurementFreshness of now and no
// more than MeasurementSkew ahead of it.
func fresh(rate float64, at, now time.Time) bool {
	age := now.Sub(at)
	return !at.IsZero() && usable(rate) && age <= MeasurementFreshness && age >= -MeasurementSkew
}

// BudgetResult is T11 decided: the check, the attempts the window admits,
// the rates and the price, for the report to carry.
type BudgetResult struct {
	Check            Check
	AttemptsAdmitted int
	Rates            Rates
	Seconds          float64 // the cut priced with one probe attempt
}

// BudgetCheck is T11 (§7.3): price a cut of a dbMiB database whose archive
// holds entries files at the measured rates, with barBytes as the final
// mirror bar, and admit the probe attempts that fit windowSeconds. It fails
// closed: a missing, stale or unusable measurement, a negative entry count,
// a database size or window that is not a usable number, or a window that
// admits no attempt, fails the check with zero attempts admitted.
func BudgetCheck(dbMiB float64, entries int, m Measurements, barBytes int64, windowSeconds float64, now time.Time) BudgetResult {
	fail := func(why string) BudgetResult { return BudgetResult{Check: Check{Reason: why}} }
	if !finite(dbMiB) || dbMiB < 0 {
		return fail("the database size could not be measured")
	}
	if entries < 0 {
		return fail("the archive's file count could not be measured")
	}
	if !finite(windowSeconds) || windowSeconds <= 0 {
		return fail("the cut window is not a usable number of seconds")
	}
	rates, err := m.Rates(now, barBytes)
	var nf *NoFreshRateError
	if errors.As(err, &nf) {
		return fail(NoFreshRate(nf.Which))
	}
	if err != nil {
		return fail("the cut budget could not be priced")
	}
	e := Estimate(dbMiB, entries, rates)
	n := e.AttemptsAdmitted(windowSeconds)
	need := e.Total(1)
	res := BudgetResult{AttemptsAdmitted: n, Rates: rates, Seconds: need}
	if n < 1 {
		res.Check.Reason = fmt.Sprintf("the cut needs about %.0f seconds with one probe attempt and the window is %.0f seconds", need, windowSeconds)
		return res
	}
	res.Check.OK = true
	return res
}

// ---- writing a recurring object ----

// PutReplace writes key whether or not it exists. release.Store.Put with an
// empty ifMatch is create-only, so a recurring object (a report, the
// console record) is written by Get for the current etag (absent is "") and
// then Put with it, retrying once when another writer got in between. A Get
// that fails for any reason but absence is returned, never written over.
// Every mutable object the series writes to the deployment bucket goes
// through here.
//
// It has twins: putCAS in internal/edge/certsync.go (three attempts) and
// status.WriteBox in internal/status/snapshot.go (one retry, a second
// conflict is returned loudly). Both are the same read-etag-then-If-Match
// loop with different policies and error wording. Neither fits here:
// putCAS is unexported in internal/edge, and WriteBox writes a box's own
// status snapshot to its fixed key, not an arbitrary key and body.
// This one retries ONCE on conflict, like WriteBox: a report has a single
// writer per pass, so two conflicts in a row mean something else is writing
// the key, and the caller sees ErrCASConflict rather than a silent third try.
func PutReplace(ctx context.Context, s release.Store, key string, body []byte) error {
	var err error
	for attempt := 0; attempt < 2; attempt++ {
		_, etag, gerr := s.Get(ctx, key)
		switch {
		case gerr == nil:
		case errors.Is(gerr, os.ErrNotExist):
			etag = ""
		default:
			return fmt.Errorf("read %s before replacing it: %w", key, gerr)
		}
		if _, err = s.Put(ctx, key, body, etag); err == nil {
			return nil
		}
		if !errors.Is(err, release.ErrCASConflict) {
			break
		}
	}
	return fmt.Errorf("write %s: %w", key, err)
}