package provision

import (
	"errors"
	"fmt"
	"slices"
	"time"
)

// The offline operator move (docs/superpowers/plans/2026-10-20-offline-
// operator-move.md, Task 1). One hibernating instance moves between two
// regions with planned downtime: the operator stops it, the move copies a
// checked list of everything it depends on, and only then does the target
// open it. The record carries the move as move.kind "offline" through the
// states below, each advanced by exactly one transition in this file.
//
// Kinds never cross. Nothing here touches a legacy or Phase 1 move (every
// transition refuses a move whose kind is not offline), and no legacy or
// Phase 1 transition accepts one (the legacy transitions name the states
// they admit, none of which is an offline state; MoveGuard.check refuses the
// kind outright).
//
// No code in this file Puts a record. Each transition is a mut for
// SetRecordMove, so the write is provision's own read-CAS-write, which
// advances record_seq. Across a whole move the record changes only in
// region, move, last_move, record_seq and move.updated_at.

// MoveKindOffline is MoveState.Kind for an offline operator move. Every other
// move leaves Kind empty.
const MoveKindOffline = "offline"

// The offline move's states, in the order a move takes them:
//
//	[offline-staging ->] offline-stopping -> offline-sealing -> offline-sealed
//	-> offline-landing -> offline-landed -> offline-open -> offline-finishing
//	-> (cleared)
//
// offline-staging is optional: OfflineStage creates the move there instead
// of in stopping, and the one source box that stages (moves.OfflineStager)
// pre-copies the replica into the target's staging prefix while the
// instance keeps serving. Nothing holds, fences or tombstones the instance
// in staging. OfflineStartStaged then takes it to stopping, and the seal's
// own round finds the pre-copied objects and skips them by their receipts.
// offline-reverting is abort's state, reached from staging through landed.
const (
	OfflineStaging   = "offline-staging"
	OfflineStopping  = "offline-stopping"
	OfflineSealing   = "offline-sealing"
	OfflineSealed    = "offline-sealed"
	OfflineLanding   = "offline-landing"
	OfflineLanded    = "offline-landed"
	OfflineOpen      = "offline-open"
	OfflineFinishing = "offline-finishing"
	OfflineReverting = "offline-reverting"
)

// The outcomes OfflineFinish leaves in LastMove.
const (
	OfflineOutcomeDone    = "done"
	OfflineOutcomeAborted = "aborted"
)

// The offline transitions' refusals that can reach an operator. Each is a
// recorded sentence (copy rows MV-O34 to MV-O43); a console door maps an
// offline ErrMoveConflict onto them or onto its own rows (MV-O24, MV-O25,
// MV-O27) and never sends err.Error() through unchecked. The refusals that
// name a value are written in place as one format.
const (
	// refusedOfflineKind is MoveGuard.check's refusal of an offline move
	// (MV-O42).
	refusedOfflineKind = "this is an offline move. use carlos move offline"
	// refusedNotOffline: an offline transition met a legacy or Phase 1 move
	// (MV-O34).
	refusedNotOffline = "this is not an offline move"
	// refusedOtherOfflineMove: the record's offline move has another id
	// (MV-O35).
	refusedOtherOfflineMove = "the record carries a different offline move"
	// refusedOfflineSeal: a seal that names no sealer, no boxes, or a sealer
	// that is not one of them (MV-O37).
	refusedOfflineSeal = "a seal names the box that seals and the boxes that stopped the instance. the box that seals is one of them"
)

// errRecordUnchanged is what a transition returns to say "the record already
// says this": updateRecord then writes nothing and returns the record as read
// (no record_seq advance, no updated_at stamp). Only OfflineAbort from
// reverting uses it. Applied outside updateRecord the mut returns it as an
// error, so a transition that can return it is called only through
// SetRecordMove.
var errRecordUnchanged = errors.New("provision: the record already says this")

// Offline reports whether this is an offline operator move. Nil-safe, so a
// record at rest answers false.
func (m *MoveState) Offline() bool { return m != nil && m.Kind == MoveKindOffline }

// offlineAt checks what every offline transition checks first: the record
// carries an offline move, it is the move the caller means, and it is in the
// one state the transition leaves. Every failure is ErrMoveConflict.
func offlineAt(rec *Record, id, from string) error {
	if err := offlineIs(rec, id); err != nil {
		return err
	}
	if rec.Move.State != from {
		return fmt.Errorf("%w: %s", ErrMoveConflict, offlineWrongState(rec.Move.State, from))
	}
	return nil
}

// offlineIs is offlineAt without the state.
func offlineIs(rec *Record, id string) error {
	if rec.Move == nil {
		return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNoMove)
	}
	if !rec.Move.Offline() {
		return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNotOffline)
	}
	if rec.Move.ID != id {
		return fmt.Errorf("%w: %s", ErrMoveConflict, refusedOtherOfflineMove)
	}
	return nil
}

// OfflineStart begins an offline move from region `from` to region `to`
// (either may be "", the home placement). It carries MoveRequest's refusals:
// a seeded record, an open run, a move already in flight, the same region and
// a replica. It also refuses a `from` that is not the record's region, and a
// malformed id. It stamps kind, id, from, to, state, requested_by and
// requested_at; it does not touch the region.
func OfflineStart(id, from, to, by string, now time.Time) func(*Record) error {
	return offlineBegin(id, from, to, by, now, OfflineStopping)
}

// StageTarget is what a stage fixes for the whole move: the one box that
// copies (Stager) and where the copy goes (Bucket, the target's data
// bucket, in Region).
type StageTarget struct{ Stager, Bucket, Region string }

// OfflineStage is OfflineStart into offline-staging: the same refusals and
// the same stamps, plus staged_at, the stager and the staging destination
// (a stage that names any of them empty is malformed). The instance keeps
// serving.
func OfflineStage(id, from, to, by string, st StageTarget, now time.Time) func(*Record) error {
	begin := offlineBegin(id, from, to, by, now, OfflineStaging)
	return func(rec *Record) error {
		if st.Stager == "" || st.Bucket == "" {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
		}
		if err := begin(rec); err != nil {
			return err
		}
		rec.Move.Stager, rec.Move.StageBucket, rec.Move.StageRegion = st.Stager, st.Bucket, st.Region
		return nil
	}
}

func offlineBegin(id, from, to, by string, now time.Time, state string) func(*Record) error {
	return func(rec *Record) error {
		if err := seedMoveRefusal(rec); err != nil {
			return err
		}
		if rec.ActiveRun.Open() {
			return &MoveRunOpenError{Run: *rec.ActiveRun}
		}
		if rec.Move != nil {
			return fmt.Errorf("%w: a move to %s is already %s", ErrMoveConflict, rec.Move.To, rec.Move.State)
		}
		if !moveIDShape.MatchString(id) || now.IsZero() {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
		}
		if rec.Region == to {
			return fmt.Errorf("%w: the instance is already in %s", ErrMoveConflict, regionWord(rec.Region))
		}
		if from != rec.Region {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineNotFrom(rec.Region, from))
		}
		if rec.Role != "" {
			return fmt.Errorf("%w: a %s does not move. delete it and create one in the new region", ErrMoveConflict, rec.Role)
		}
		rec.Move = &MoveState{
			Kind:        MoveKindOffline,
			ID:          id,
			From:        from,
			To:          to,
			State:       state,
			RequestedBy: by,
			RequestedAt: now.UTC(),
		}
		if state == OfflineStaging {
			t := now.UTC()
			rec.Move.StagedAt = &t
		}
		return nil
	}
}

// OfflineStartStaged is staging to stopping: the operator takes a staged
// instance offline. The record's region must still be the move's source.
// It stamps requested_by and requested_at anew, since the instance is
// offline from now, and requested_at is always after staged_at, which is
// how OfflineFinish tells a move that started from one that never did.
func OfflineStartStaged(id, by string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineStaging); err != nil {
			return err
		}
		if rec.Region != rec.Move.From {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
		}
		if now.IsZero() {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
		}
		t := now.UTC()
		if s := rec.Move.StagedAt; s != nil && !t.After(*s) {
			t = s.Add(time.Nanosecond)
		}
		rec.Move.State, rec.Move.RequestedBy, rec.Move.RequestedAt = OfflineStopping, by, t
		return nil
	}
}

// OfflineRestage is staging to staging: the operator asked for another
// pre-copy round. It moves staged_at to now (always forward), which is the
// request the stager answers with exactly one round (the stager report's
// stage_request names the staged_at it served). requested_at stays, so the
// move still reads as never started.
func OfflineRestage(id string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineStaging); err != nil {
			return err
		}
		if rec.Region != rec.Move.From {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
		}
		if now.IsZero() {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
		}
		t := now.UTC()
		if s := rec.Move.StagedAt; s != nil && !t.After(*s) {
			t = s.Add(time.Nanosecond)
		}
		rec.Move.StagedAt = &t
		return nil
	}
}

// StartedOffline reports whether the move ever took its instance offline:
// a move created in stopping did at once, a staged one at OfflineStartStaged.
func (m *MoveState) StartedOffline() bool {
	return m.StagedAt == nil || m.RequestedAt.After(*m.StagedAt)
}

// regionWord is a region as a sentence names it: the home placement's empty
// code reads "home".
func regionWord(r string) string {
	if r == "" {
		return "home"
	}
	return r
}

// OfflineSeal is stopping to sealing: every source box has confirmed the stop
// and the operator asked for the copy. It names the sealer (the one box that
// copies) and the source boxes whose reports the later doors wait for.
func OfflineSeal(id, sealer string, boxes []string) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineStopping); err != nil {
			return err
		}
		if sealer == "" || len(boxes) == 0 || !slices.Contains(boxes, sealer) {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedOfflineSeal)
		}
		rec.Move.State = OfflineSealing
		rec.Move.Sealer = sealer
		rec.Move.Boxes = append([]string(nil), boxes...)
		return nil
	}
}

// OfflineSealedBy is sealing to sealed: the sealer recorded what it copied.
// Only the box the seal named may write it. src is copied, so the journal
// never aliases the caller's value.
func OfflineSealedBy(id, sealer string, src MoveSource) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineSealing); err != nil {
			return err
		}
		if sealer == "" || rec.Move.Sealer != sealer {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineSealedBy(rec.Move.Sealer))
		}
		s := src
		if src.Files != nil {
			f := *src.Files
			s.Files = &f
		}
		rec.Move.Source = &s
		rec.Move.State = OfflineSealed
		return nil
	}
}

// OfflineFlip is sealed to landing, and the one record write that moves the
// region: region = to. The record's region must still be the move's source,
// so a record that was changed under the move is refused rather than flipped.
func OfflineFlip(id string) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineSealed); err != nil {
			return err
		}
		if rec.Region != rec.Move.From {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
		}
		rec.Region = rec.Move.To
		rec.Move.State = OfflineLanding
		return nil
	}
}

// OfflineLandedBy is landing to landed: the target box reports that every
// item is in place. Guarded by the record's region, so a landing that raced an
// abort (which put the region back) is refused.
func OfflineLandedBy(id, to string) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineLanding); err != nil {
			return err
		}
		if rec.Region != to || rec.Move.To != to {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
		}
		rec.Move.State = OfflineLanded
		return nil
	}
}

// OfflineOpenTransition is landed to open: the operator opens the target. The
// region must be the target's. It stamps opened_at from now, which is when the
// instance came back and so what the finished move's offline time is counted
// to.
//
// The brief named this OfflineOpen, which is also the name of the state
// constant; Go has one namespace, so the two transitions that clash with a
// state (this one and OfflineFinishingTransition) carry a suffix.
func OfflineOpenTransition(id string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineLanded); err != nil {
			return err
		}
		if rec.Region != rec.Move.To {
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
		}
		t := now.UTC()
		rec.Move.State, rec.Move.OpenedAt = OfflineOpen, &t
		return nil
	}
}

// OfflineFinishingTransition is open to finishing: the operator asked the
// source to clean up. After open the move only goes forward.
func OfflineFinishingTransition(id string) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineAt(rec, id, OfflineOpen); err != nil {
			return err
		}
		rec.Move.State = OfflineFinishing
		return nil
	}
}

// OfflineAbort puts the move on its way back: any of staging, stopping,
// sealing, sealed, landing or landed becomes reverting. From landing and landed the
// region goes back to the source in the same write (the flip is undone with
// the state, never separately). From open and finishing it refuses, because
// the target already serves. From reverting it changes nothing and writes
// nothing (updateRecord sees errRecordUnchanged and returns the record as
// read), so a second abort is a success that leaves record_seq and updated_at
// alone. Call it through SetRecordMove.
func OfflineAbort(id string) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineIs(rec, id); err != nil {
			return err
		}
		switch rec.Move.State {
		case OfflineReverting:
			return errRecordUnchanged
		case OfflineStaging, OfflineStopping, OfflineSealing, OfflineSealed:
		case OfflineLanding, OfflineLanded:
			rec.Region = rec.Move.From
		default:
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineAfterOpen(rec.Move.State))
		}
		rec.Move.State = OfflineReverting
		return nil
	}
}

// OfflineFinish clears the move and leaves its outcome in last_move. It is
// legal from finishing (outcome done, region must be the target's) and from
// reverting (outcome aborted, region must be the source's); anything else is
// refused. The outcome's offline time runs from the request to the open for a
// done move and to now for an aborted one; a move aborted while it was still
// staging was never offline, and records none.
func OfflineFinish(id string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := offlineIs(rec, id); err != nil {
			return err
		}
		m := rec.Move
		out := &MoveOutcome{ID: id, At: now.UTC()}
		end := now
		switch m.State {
		case OfflineFinishing:
			if rec.Region != m.To {
				return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
			}
			out.Outcome = OfflineOutcomeDone
			if m.OpenedAt != nil {
				end = *m.OpenedAt
			}
		case OfflineReverting:
			if rec.Region != m.From {
				return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
			}
			out.Outcome = OfflineOutcomeAborted
		default:
			return fmt.Errorf("%w: %s", ErrMoveConflict, offlineFinishState(m.State))
		}
		if s := end.Sub(m.RequestedAt).Seconds(); s > 0 && m.StartedOffline() {
			out.OfflineS = s
		}
		rec.LastMove, rec.Move = out, nil
		return nil
	}
}

// offlineWrongState is MV-O36.
func offlineWrongState(state, from string) string {
	return fmt.Sprintf("this offline move is %s. this step follows %s", state, from)
}

// offlineSealedBy is MV-O38.
func offlineSealedBy(box string) string {
	return fmt.Sprintf("this offline move is sealed by %s", box)
}

// offlineAfterOpen is MV-O40.
func offlineAfterOpen(state string) string {
	return fmt.Sprintf("this offline move is %s. it cannot be aborted once it is open", state)
}

// offlineFinishState is MV-O41.
func offlineFinishState(state string) string {
	return fmt.Sprintf("this offline move is %s. it finishes from finishing or reverting", state)
}

// offlineNotFrom is MV-O43.
func offlineNotFrom(region, from string) string {
	return fmt.Sprintf("the instance is in %s, not %s", regionWord(region), regionWord(from))
}

// offlineRegionMoved is the refusal for a record whose region is not where
// the step expects it (MV-O39): something wrote the region under the move.
func offlineRegionMoved(region string) string {
	return fmt.Sprintf("the record's region is %s, which this step does not expect. the instance changed under the move", regionWord(region))
}