package host

// The moves retirement evidence provider (moves Plan 5 Task 5; rulings K10,
// M7 and M8 of the Plan 5 ledger). The claim's regional worker
// (hostbox.RegionalWorker) sweeps instances/<host>/ and moves/<host>/ at rest.
// Before it sweeps it asks this provider twice:
//
//   - VerifyRetirement, once per sweep, for the incarnation label of every
//     candidate outside an incarnation directory and for the evidence time of
//     every instances/<host> candidate. Decide labels the moves/<host>
//     candidate with the same incarnation (park/sweep.go), and the engine's
//     proof Check requires the proof to name it, so both answers come from one
//     proof.
//   - VerifyMovesSweep, per moves/<host> candidate (park/move_sweep.go). The
//     engine then stamps the listing's end as that chore's evidence, which is
//     valid only with the precondition proof (K10): the claim shows no move
//     op, and the newest retired move's close is earlier than the listing.
//
// The proof comes from the record's LastMove (M8, fix round 1 ruling 2): the
// durable move id, outcome and close time, bound to the holder's
// incarnation. It holds only when:
//
//   - the claim is serving, so no move op is live on the host;
//   - the record is the holder's own and carries no live move;
//   - LastMove names a move closed at a real time before this clock;
//   - every moves/<host>/<op>/ object is that move's own, or was written
//     before its close.
//
// The claim's released_move journal is not required: any later serving op
// replaces it (CompleteMetadataPlacement, for one). With M4's record-first
// close, LastMove is written before the claim's release, so it is the
// serving claim, with no move op, that proves the release finished.
// LastMove.At follows every park of the move's lanes, and so every staging
// horizon. A zero time is never evidence.
//
// VerifyRetirement answers for less than the proof covers (M7; fix round 1
// ruling 1). The worker binds its one time to every instances/<host>
// candidate, so it refuses unless the close is genuine evidence for the
// residue at the live location:
//
//   - at the live location the close covers only generations the move
//     retired: those superseded at or before the move's own admission. A
//     generation admitted from that admission on was written past the
//     close, and ordinary retirement owns it.
//
// Residue outside the holder's live location is not judged here (Q4). The
// claim's census places it: ClaimRetirementCensus labels a qualified path by
// its own identity, and the claim's view refuses unmapped bare history per
// location (productionClaimRegional.View), so a refusal of it here would
// only repeat theirs and would skip the lane sweep as well.
//
// The provider is registered by RegisterProductionMovesParks, whose view
// stays closed until movesWakeSourceRegistered opens; the claim reaches
// VerifyMovesSweep by type assertion (claimassembly.go).

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
)

// ErrMovesSweepRefused is what every MovesSweepRefusal unwraps to.
var ErrMovesSweepRefused = errors.New("host: the moves sweep is refused")

// MovesSweepRefusal is the provider's refusal of one observation. Each one
// skips that pass's sweep, and the next pass asks again. Anomaly marks a
// refusal that no ordinary move produces: the history contradicts itself, or
// something wrote where no known move could. M8 asks for one page per
// anomaly; the provider sends none itself, and its wiring (Task 9a) pages.
// Reason is for the agent's log, never shown to a user.
type MovesSweepRefusal struct {
	Host, Reason string
	Anomaly      bool
}

func (e *MovesSweepRefusal) Error() string {
	return "host: the moves sweep of " + e.Host + " is refused: " + e.Reason
}

func (e *MovesSweepRefusal) Unwrap() error { return ErrMovesSweepRefused }

// ErrNotMovesEvidence is what every NotMovesEvidence unwraps to.
var ErrNotMovesEvidence = errors.New("host: the move's close is not evidence for this residue")

// NotMovesEvidence is VerifyRetirement's refusal of residue at the live
// location that the move did not retire: a generation written past the
// move's close, or an object no generation owns. The close is not its
// evidence, and ordinary retirement owns it. The worker binds one evidence
// time to every instances/<host> candidate, so the provider cannot abstain
// for one generation; this refusal skips the pass's sweep while such
// residue stands. Reason is for the agent's log, never shown to a user.
type NotMovesEvidence struct {
	Host, Generation, Reason string
}

func (e *NotMovesEvidence) Error() string {
	return "host: the move's close is not evidence for " + e.Host + " generation " + e.Generation + ": " + e.Reason
}

func (e *NotMovesEvidence) Unwrap() error { return ErrNotMovesEvidence }

// movesRetirementEvidence implements hostbox.RegionalEvidence and
// park.MoveSweepVerifier over the host's provisioning record.
type movesRetirementEvidence struct {
	Records MovesRecordReader
	Now     func() time.Time
}

// movesRetired is the proof both methods answer from.
type movesRetired struct {
	host, incarnation, location, moveID string
	at                                  time.Time
	holder                              hostclaim.Holder
}

func (e movesRetirementEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
	r, err := e.retired(ctx, o, versions)
	if err != nil {
		return park.MoveSweepProof{}, err
	}
	return park.MoveSweepProof{Host: r.host, ClaimETag: o.ETag, MoveID: r.moveID, Incarnation: r.incarnation, RetiredAt: r.at}, nil
}

func (e movesRetirementEvidence) VerifyRetirement(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
	r, err := e.retired(ctx, o, versions)
	if err != nil {
		return "", time.Time{}, err
	}
	if err := r.retiredGenerations(versions); err != nil {
		return "", time.Time{}, err
	}
	return r.incarnation, r.at, nil
}

// retiredGenerations refuses unless the close is evidence for every object
// at the live location that the engine could make a candidate (fix round 1,
// ruling 1). The marker key and the holder's current generation are the
// live installation's. Any other generation must be one the move retired:
// the claim must still hold the move's own admission (a prepared landing
// admits the generation named after the move, movesprobeowners.go), the
// generation must not be admitted at or after it (admissions are in claim
// order and carry no time), and every version of it must predate the close.
// A rolled-back or aborted move, or a landing that admitted a fresh name,
// leaves no such admission, so no generation is vouched for. Objects are
// classified as the landing's view classifies them (landingGenerationOf),
// after the lock objects the engine itself skips (movesParkLock, fix round
// 2): the writer's lease stands at every live location. The build state
// beside it (movesHolderBuildState, ruling R-PARK) is the location's own,
// the marker's twin, and needs no vouching either.
func (r movesRetired) retiredGenerations(versions []lease.ObjectVersion) error {
	h := r.holder
	refuse := func(gen, format string, args ...any) error {
		return &NotMovesEvidence{Host: r.host, Generation: gen, Reason: fmt.Sprintf(format, args...)}
	}
	moveAdmission := -1
	for i := len(h.Admissions) - 1; i >= 0; i-- {
		if h.Admissions[i].Generation == r.moveID {
			moveAdmission = i
			break
		}
	}
	afterTheMove := map[string]bool{}
	if moveAdmission >= 0 {
		for _, a := range h.Admissions[moveAdmission:] {
			afterTheMove[a.Generation] = true
		}
	}
	for _, v := range versions {
		if loc, ok := movesInstancesLocation(r.host, v.Key); !ok || loc != r.location || movesParkLock(r.host, v.Key) || movesHolderBuildState(r.location, v.Key) {
			continue
		}
		gen, ok := landingGenerationOf(r.location, v.Key)
		switch {
		case !ok:
			return refuse("", "%s belongs to no generation", v.Key)
		case gen == landingMarkerGeneration || gen == h.Generation:
			continue
		case moveAdmission < 0:
			return refuse(gen, "the claim holds no admission of %s to date it by", r.moveID)
		case afterTheMove[gen]:
			return refuse(gen, "it was admitted after %s's admission, so written past the close", r.moveID)
		case v.LastModified.IsZero() || !v.LastModified.Before(r.at):
			return refuse(gen, "%s was written after %s closed", v.Key, r.moveID)
		}
	}
	return nil
}

// movesInstancesLocation is the location park.Decide files an
// instances/<host>/ key under (park's locationOf): a 64-hex first segment
// with a child is its own incarnation location, and everything else is the
// bare instances/<host>. ok is false outside instances/<host>/.
func movesInstancesLocation(host, key string) (string, bool) {
	base := "instances/" + host
	tail, ok := strings.CutPrefix(key, base+"/")
	if !ok {
		return "", false
	}
	first, _, child := strings.Cut(tail, "/")
	if _, err := hex.DecodeString(first); child && err == nil && len(first) == 64 {
		return base + "/" + first, true
	}
	return base, true
}

// retired proves the newest retired move from the claim observation, the
// holder's record and the complete census, or refuses.
func (e movesRetirementEvidence) retired(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (movesRetired, error) {
	if ctx == nil || e.Records == nil || e.Now == nil {
		return movesRetired{}, hostwrite.ErrUnavailable
	}
	if o.Absent || o.Claim == nil || o.ETag == "" || o.Host == "" || o.Claim.Host != o.Host {
		return movesRetired{}, hostwrite.ErrUnavailable
	}
	c, host := o.Claim, o.Host
	refuse := func(anomaly bool, format string, args ...any) (movesRetired, error) {
		return movesRetired{}, &MovesSweepRefusal{Host: host, Reason: fmt.Sprintf(format, args...), Anomaly: anomaly}
	}
	if c.Holder == nil {
		// A free or joining claim names no record, so no move history. A free
		// claim's lanes are the claim's own proof (claimFreeMoveProof), which
		// the claim asks first; a joining parking claim reaches here. It is an
		// ordinary state, so it is a refusal, quiet in the agent's log, and
		// not an anomaly (Q4). Once the claim has a holder again the next
		// pass proves the move.
		return refuse(false, "the claim has no holder, so no record to prove a closed move from")
	}
	if c.Op.Kind != hostclaim.KindServing {
		return refuse(false, "the claim is under a %s op", c.Op.Kind)
	}
	h := c.Holder
	if h.Family != "instance" || h.Key != provision.RecordKey(h.Account, h.App, host) {
		return refuse(false, "the holder is not an instance record")
	}
	// A moved holder is at the bare location, the activator's own lease
	// (ruling T5-lease, movescut.go writerLease); a scoped holder is at its
	// incarnation's. Anything else is a location the census cannot place.
	location := strings.TrimSuffix(h.Location, "/")
	if location != "instances/"+host && location != "instances/"+host+"/"+h.Incarnation {
		return refuse(true, "the holder is at %s, neither the bare nor the incarnation location", h.Location)
	}
	rec, err := e.Records(ctx, h.Account, h.App, host)
	if err != nil {
		return movesRetired{}, err
	}
	if rec == nil {
		return refuse(true, "the holder's record is missing")
	}
	inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	if err != nil || rec.Account != h.Account || rec.App != h.App || rec.Host != host || inc != h.Incarnation {
		return refuse(true, "the record is not the holder's incarnation")
	}
	if rec.Move != nil {
		return refuse(false, "the record carries the live move %s", rec.Move.ID)
	}
	last := rec.LastMove
	if last == nil || last.ID == "" {
		return refuse(false, "the record has closed no move")
	}
	if last.At.IsZero() || !last.At.Before(e.Now()) {
		return refuse(true, "the close of %s has no time before now", last.ID)
	}
	movesRoot := "moves/" + host + "/"
	for _, v := range versions {
		if strings.HasPrefix(v.Key, "instances/"+host+"/") {
			continue
		}
		tail, ok := strings.CutPrefix(v.Key, movesRoot)
		if !ok {
			return refuse(true, "the census holds %s, outside this host", v.Key)
		}
		if movesParkLock(host, v.Key) {
			continue // the moves/<host> park's own lease
		}
		op, _, inLane := strings.Cut(tail, "/")
		if !inLane {
			return refuse(true, "%s is outside every lane", v.Key)
		}
		if op == last.ID {
			continue
		}
		if v.LastModified.IsZero() || !v.LastModified.Before(last.At) {
			return refuse(true, "lane %s was written after %s closed", op, last.ID)
		}
	}
	return movesRetired{host: host, incarnation: h.Incarnation, location: location, moveID: last.ID, at: last.At.UTC(), holder: *h}, nil
}

// movesParkLock mirrors the park engine's lock predicate (park's isLock over
// locationOf, both unexported): a key is a lock exactly when it is the
// lock.json at the root of its own park location, instances/<host>,
// instances/<host>/<incarnation> or moves/<host>. That is the writer's
// lease at a holder location (lease.Acquire) and a park's own lease.
// Decide never makes it a candidate, so the close need not vouch for it.
// TestMovesParkLockMatchesTheEngine keeps the two in step through
// park.Decide.
func movesParkLock(host, key string) bool {
	loc, ok := movesInstancesLocation(host, key)
	if !ok {
		if !strings.HasPrefix(key, "moves/"+host+"/") {
			return false
		}
		loc = "moves/" + host
	}
	return key == loc+"/lock.json"
}