package moves

import (
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"path/filepath"
	"time"

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

// The offline operator move's contract (docs/superpowers/plans/2026-10-20-
// offline-operator-move.md, Task 1): the dependency manifest, the report each
// box writes at each step, and the keys, owners and paths every task shares.
// Nothing here writes anything. The record side is provision/offline_move.go.

// OfflineItem is one row of the dependency list: a thing the instance needs
// that the move must carry, rebuild or consciously leave behind. ID names the
// item in a report's Items. Disposition says what the move does with it.
//
// Owner is the step that proves the item where it must end up, and the
// door that step gates requires that step's OK:
//
//	door     a door refuses the move up front (nothing to check later)
//	seal     the sealer's report proves it (the source's own state)
//	landing  the target's landing report proves it; the open door requires it
//	none     no report proves it; the Disposition says why
//
// Source is the step that reports the expected value on the source side: the
// sealer's report ("seal"), which the flip door requires, or "" when the item
// has no source-side value. A carried item is Owner landing and Source seal:
// the sealer records what it copied, the target proves it arrived equal.
// Exactly one Owner per row; Source is independent of it.
type OfflineItem struct{ ID, Item, Disposition, Owner, Source string }

// The owners and sources.
const (
	OfflineOwnerDoor    = "door"
	OfflineOwnerSeal    = "seal"
	OfflineOwnerLanding = "landing"
	OfflineOwnerNone    = "none"

	OfflineSourceSeal = "seal"
)

// OfflineManifest is the plan's dependency list, one row per item. It is the
// checklist: flip needs every Source "seal" item OK in the sealer's report,
// open needs every Owner "landing" item OK in the target's, a report may
// carry only items listed here, and TestOfflineManifestIsCovered fails if a
// row has no owner. A Disposition begins with its kind (carried, rebuilt,
// refused, parked, not carried) and a reason after the first colon.
var OfflineManifest = []OfflineItem{
	{"db", "SQLite DB at its final position", "carried: instances/<host>/ is mirrored to the target's moves/<host>/ and promoted by the target's first wake after open", OfflineOwnerLanding, OfflineSourceSeal},
	{"db-tail", "writes the source's stop failed to checkpoint", "carried: the sealer checkpoints first when a source report is ahead of the replica", OfflineOwnerSeal, OfflineSourceSeal},
	{"build-state", "instances/<host>/build-state.json", "carried: inside the replica prefix, after CarryRequest", OfflineOwnerSeal, OfflineSourceSeal},
	{"lease", "instances/<host>/lock.json", "not carried: the source's lock becomes the tombstone until finish; a stale tombstone of another move in the target bucket is lifted at landing", OfflineOwnerLanding, OfflineSourceSeal},
	{"files", "flat <DataDir>/<host>/ and the nested instance directory minus platform artifacts", "carried: sealed archive at moves/@offline/<host>/<id>/files.sealed in the target bucket", OfflineOwnerLanding, OfflineSourceSeal},
	{"env", "<host>.env: instance key, admin token and overrides", "carried: byte for byte inside the archive", OfflineOwnerLanding, OfflineSourceSeal},
	{"key-adoption", "<host>.instance-key-adoption.json", "not carried: written only for claim launches; a copy on the target is parked", OfflineOwnerLanding, ""},
	{"config", "<host>.config.env: the app's settings, store binding and secrets", "rebuilt: rendered on the target from the bucket bundle", OfflineOwnerLanding, ""},
	{"env-pin", "the composed launch environment", "rebuilt: app settings must be equal on both rows; box settings are not compared", OfflineOwnerLanding, OfflineSourceSeal},
	{"release", "the build the source runs", "rebuilt: not copied, ordinary adoption stages the channel's build for the ready row after open", OfflineOwnerLanding, OfflineSourceSeal},
	{"store", "object store bucket and its credentials", "not carried: the bucket stays where it is and its binding travels in config, so env-pin covers it", OfflineOwnerNone, ""},
	{"timers", "timer rows, once-timers the app armed included", "carried: a timers snapshot beside the archive, replaced on the held row", OfflineOwnerLanding, OfflineSourceSeal},
	{"next-due", "the row's NextDue", "carried: in the journal's source baselines", OfflineOwnerLanding, OfflineSourceSeal},
	{"replication", "the row's replication rung", "carried: in the journal's source baselines", OfflineOwnerLanding, OfflineSourceSeal},
	{"row", "the registry row", "rebuilt: reconcile mints it from the record, held at flat paths", OfflineOwnerLanding, ""},
	{"routing", "edges file, backhaul, certificates, derived aliases", "not carried: produced by the existing passes once the row is ready, and the runbook confirms a certificate path before the move", OfflineOwnerNone, ""},
	{"source-retire", "everything the source box keeps for the host after open", "parked: renamed, never deleted, and listed path by path in the source's parked report", OfflineOwnerNone, ""},
	{"sidecar", "<host>-sidecar.env", "refused: a source report that sees one fails the seal", OfflineOwnerSeal, OfflineSourceSeal},
	{"restore-verify", "control/restore-verify.json entry", "not carried: the next drill writes a fresh verdict, and until then the old one describes the source copy", OfflineOwnerNone, ""},
	{"warm-cache", "local DB, meta dirs, files and env a box holds for the host", "parked: renamed under .offline-moves on the source after open and on a target that already has them", OfflineOwnerLanding, ""},
	{"receipts", "move-receipts/<host>.json in the source bucket", "not carried: cleared at finish, and self-checking", OfflineOwnerNone, ""},
	{"class", "kind, addr, fleet, pool, hibernate, replica, replicas, steering, unit, seed and open run", "refused: the start door and the start transition refuse every class the move cannot carry", OfflineOwnerDoor, ""},
}

// OfflineItemOwner is the owner of manifest item id.
func OfflineItemOwner(id string) (owner string, ok bool) {
	for _, it := range OfflineManifest {
		if it.ID == id {
			return it.Owner, true
		}
	}
	return "", false
}

func offlineItem(id string) (OfflineItem, bool) {
	for _, it := range OfflineManifest {
		if it.ID == id {
			return it, true
		}
	}
	return OfflineItem{}, false
}

// OfflineOwned lists the manifest items an owner decides, in manifest order.
func OfflineOwned(owner string) []OfflineItem {
	var out []OfflineItem
	for _, it := range OfflineManifest {
		if it.Owner == owner {
			out = append(out, it)
		}
	}
	return out
}

// OfflineSourced lists the manifest items whose expected value the source
// side reports (Source "seal"), in manifest order.
func OfflineSourced() []OfflineItem {
	var out []OfflineItem
	for _, it := range OfflineManifest {
		if it.Source == OfflineSourceSeal {
			out = append(out, it)
		}
	}
	return out
}

// The roles a report names: a source-region box that stopped the host, the
// one source box that seals, the target box that lands, and the one source
// box that pre-copies the replica while the move is offline-staging (the
// stager, which the stage door picks with OfflineStager and pins on the
// record as move.stager).
const (
	OfflineRoleSource = "source"
	OfflineRoleSealer = "sealer"
	OfflineRoleTarget = "target"
	OfflineRoleStager = "stager"
)

// The phases a report names.
const (
	OfflinePhaseStopped  = "stopped"
	OfflinePhaseSealed   = "sealed"
	OfflinePhaseLanded   = "landed"
	OfflinePhaseOpened   = "opened"
	OfflinePhaseParked   = "parked"
	OfflinePhaseGone     = "gone"
	OfflinePhaseLifted   = "lifted"
	OfflinePhaseReverted = "reverted"
	OfflinePhaseFailed   = "failed"
	// OfflinePhaseStaged is a stager's report after a pre-copy round.
	OfflinePhaseStaged = "staged"
)

// OfflineReportVersion is OfflineReport.V.
const OfflineReportVersion = 1

// StagedObject is one object of the staged replica prefix: its size and ETag,
// keyed in OfflineReport.Staged by the key relative to the prefix.
type StagedObject struct {
	Size int64  `json:"size"`
	ETag string `json:"etag"`
}

// RetiredPath is one line of a source's parked report: a path the box kept
// for the host, what was done to it (parked, absent or left) and why.
type RetiredPath struct {
	Path   string `json:"path"`
	Action string `json:"action"`
	Reason string `json:"reason,omitempty"`
}

// The actions a RetiredPath may name.
const (
	RetiredParked = "parked"
	RetiredAbsent = "absent"
	RetiredLeft   = "left"
)

// OfflineReport is what a box writes at control/offline-moves/ at each step
// of an offline move (OfflineReportKey). It is evidence, never a decision: the
// doors read it and decide. One struct serves every role, and a role fills
// only the fields it observed.
type OfflineReport struct {
	V    int    `json:"v"`
	Op   string `json:"op"`
	Host string `json:"host"`
	Box  string `json:"box"`
	// Role is source, sealer or target; Phase is stopped, sealed, landed,
	// opened, parked, gone, lifted, reverted or failed.
	Role  string           `json:"role"`
	Phase string           `json:"phase"`
	Items map[string]Check `json:"items,omitempty"`
	At    time.Time        `json:"at"`

	Held     bool `json:"held"`
	Fenced   bool `json:"fenced"`
	Quiesced bool `json:"quiesced"`
	// LocalDB: the DB file or the activator's litestream meta dir is present.
	LocalDB    bool   `json:"local_db"`
	Files      bool   `json:"files"`
	Env        bool   `json:"env"`
	Sidecar    bool   `json:"sidecar"`
	RowPresent bool   `json:"row_present"`
	LocalTXID  uint64 `json:"local_txid,omitempty"`

	FinalTXID    uint64                  `json:"final_txid,omitempty"`
	Staged       map[string]StagedObject `json:"staged,omitempty"`
	FilesPlain   string                  `json:"files_plain,omitempty"`
	FilesSealed  string                  `json:"files_sealed,omitempty"`
	FilesBytes   int64                   `json:"files_bytes,omitempty"`
	FilesEntries int                     `json:"files_entries,omitempty"`

	EnvSHA256    string `json:"env_sha256,omitempty"`
	KeySHA256    string `json:"key_sha256,omitempty"`
	EnvDigest    string `json:"env_digest,omitempty"`
	Release      string `json:"release,omitempty"`
	TimersDigest string `json:"timers_digest,omitempty"`
	Replication  string `json:"replication,omitempty"`
	NextDue      int64  `json:"next_due,omitempty"`

	TombstoneOwner string    `json:"tombstone_owner,omitempty"`
	TombstoneUntil time.Time `json:"tombstone_until,omitzero"`

	Procs         int  `json:"procs,omitempty"`
	LayoutJournal bool `json:"layout_journal,omitempty"`

	// Retired is the parked report's inventory of what the source kept.
	Retired []RetiredPath `json:"retired,omitempty"`

	// The stager's progress, rewritten after every pre-copy round while
	// the move is offline-staging (At is when the round ended): the
	// replica as the round's own listing after it found it, lock.json
	// aside; how much of it the staged copy holds, key and size for key
	// and size; and what the round itself copied. A failed round keeps
	// the last progress it had and says why in StageError, a box
	// diagnostic and never copy.
	ReplicaObjects int    `json:"replica_objects,omitempty"`
	ReplicaBytes   int64  `json:"replica_bytes,omitempty"`
	StagedObjects  int    `json:"staged_objects,omitempty"`
	StagedBytes    int64  `json:"staged_bytes,omitempty"`
	RoundObjects   int    `json:"round_objects,omitempty"`
	RoundBytes     int64  `json:"round_bytes,omitempty"`
	StageError     string `json:"stage_error,omitempty"`
	// StageRequest is the record's staged_at the round answered: one round
	// per stage request (provision.OfflineStage, OfflineRestage), and a
	// stager runs none while its report already answers the record's.
	StageRequest time.Time `json:"stage_request,omitzero"`
}

var (
	offlineRoles  = map[string]bool{OfflineRoleSource: true, OfflineRoleSealer: true, OfflineRoleTarget: true, OfflineRoleStager: true}
	offlinePhases = map[string]bool{OfflinePhaseStopped: true, OfflinePhaseSealed: true, OfflinePhaseLanded: true, OfflinePhaseOpened: true,
		OfflinePhaseParked: true, OfflinePhaseGone: true, OfflinePhaseLifted: true, OfflinePhaseReverted: true, OfflinePhaseFailed: true,
		OfflinePhaseStaged: true}
)

// offlineCarries reports whether a role's report may carry the item. A
// stager's report carries none: the pre-copy decides nothing the doors
// check, and the seal's own round proves the db item.
func offlineCarries(role string, it OfflineItem) bool {
	switch role {
	case OfflineRoleSource, OfflineRoleSealer:
		return it.Source == OfflineSourceSeal
	case OfflineRoleTarget:
		return it.Owner == OfflineOwnerLanding
	}
	return false
}

// ValidateOfflineReport reports why a report cannot be trusted as a report: a
// wrong version, an unknown role or phase, an unusable key part, or an item
// the manifest does not list or that this role does not carry. A reader
// calls it before it reads anything else.
func ValidateOfflineReport(r OfflineReport) error {
	if r.V != OfflineReportVersion {
		return fmt.Errorf("offline report: version %d, want %d", r.V, OfflineReportVersion)
	}
	if OfflineReportKey(r.Host, r.Op, r.Role, r.Box) == "" {
		return fmt.Errorf("offline report: host, op, role or box is not usable")
	}
	if !offlinePhases[r.Phase] {
		return fmt.Errorf("offline report: unknown phase %q", r.Phase)
	}
	for id := range r.Items {
		it, ok := offlineItem(id)
		if !ok {
			return fmt.Errorf("offline report: item %q is not in the manifest", id)
		}
		if !offlineCarries(r.Role, it) {
			return fmt.Errorf("offline report: item %q (owner %s, source %q) is not carried by a %s report", id, it.Owner, it.Source, r.Role)
		}
	}
	return nil
}

// ItemsOK says whether the report holds an OK Check for every manifest
// item the owner decides, and names the ones that are missing or not OK, in
// manifest order. The open door uses it for owner landing; flip checks the
// sealer's source values with SourceItemsOK.
func (r OfflineReport) ItemsOK(owner string) (ok bool, bad []string) {
	for _, it := range OfflineOwned(owner) {
		if c, found := r.Items[it.ID]; !found || !c.OK {
			bad = append(bad, it.ID)
		}
	}
	return len(bad) == 0, bad
}

// SourceItemsOK is ItemsOK for the flip door: whether the report holds an OK
// Check for every item the source side reports (Source "seal"), and the ones
// that are missing or not OK.
func (r OfflineReport) SourceItemsOK() (ok bool, bad []string) {
	for _, it := range OfflineSourced() {
		if c, found := r.Items[it.ID]; !found || !c.OK {
			bad = append(bad, it.ID)
		}
	}
	return len(bad) == 0, bad
}

// offlineSegment reports a string fit to be one path or key segment: not
// empty, not a dot name, and made only of letters, digits and _ . - so it can
// never escape the directory or prefix it is joined under.
func offlineSegment(s string) bool {
	if s == "" || s == "." || s == ".." {
		return false
	}
	for _, c := range s {
		switch {
		case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '_', c == '.', c == '-':
		default:
			return false
		}
	}
	return true
}

// OfflineReportKey is where a box writes its report:
// control/offline-moves/<host>/<op>/<role>/<box>.json. It is "" when host
// fails provision.ValidHostShape, role is not source, sealer, target or
// stager, or op
// or box is not a plain name; a caller that gets "" has a bug and writes
// nothing. The prefix is outside every create-only and no-delete prefix, so
// a box may rewrite its own report.
func OfflineReportKey(host, op, role, box string) string {
	if !provision.ValidHostShape(host) || !offlineSegment(op) || !offlineRoles[role] || !offlineSegment(box) {
		return ""
	}
	return "control/offline-moves/" + host + "/" + op + "/" + role + "/" + box + ".json"
}

// OfflineReportPrefix is the prefix a door lists to read every report of one
// move, "" for a host that fails provision.ValidHostShape or an unusable op.
func OfflineReportPrefix(host, op string) string {
	if !provision.ValidHostShape(host) || !offlineSegment(op) {
		return ""
	}
	return "control/offline-moves/" + host + "/" + op + "/"
}

// OfflineSealKey is where the sealer writes the seal, once and create-only:
// control/offline-moves/<host>/<op>/seal.json. Unlike a report it is never
// rewritten. It is an OfflineReport (role sealer, phase sealed) holding what
// the sealer decided and recorded at the seal: the sealed position
// (FinalTXID, Staged), the files and env digests, the env pin, the release,
// the timers digest, the replication rung and next due, TombstoneOwner
// (moved:<op>), and Items, an OK Check for every manifest item with Source
// "seal". At is when it was sealed; TombstoneUntil is zero (the live
// expiry is the sealer's per-pass report). The record's source.evidence is
// this key and the sha256 of its bytes (OfflineSealEvidence). "" for a host
// that fails provision.ValidHostShape or an unusable op.
func OfflineSealKey(host, op string) string {
	if !provision.ValidHostShape(host) || !offlineSegment(op) {
		return ""
	}
	return "control/offline-moves/" + host + "/" + op + "/seal.json"
}

// OfflineSealEvidence is source.evidence for a seal written as raw: the seal
// key, "@", and the hex sha256 of the bytes. "" for a bad host or op.
func OfflineSealEvidence(host, op string, raw []byte) string {
	key := OfflineSealKey(host, op)
	if key == "" {
		return ""
	}
	sum := sha256.Sum256(raw)
	return key + "@" + hex.EncodeToString(sum[:])
}

// ValidateOfflineSeal reports why a seal cannot be trusted as the seal of op
// on host: it is a valid sealer report of that move, phase sealed, that names
// the move's tombstone owner and carries an OK Check for every item the
// source side reports.
func ValidateOfflineSeal(r OfflineReport, host, op string) error {
	if err := ValidateOfflineReport(r); err != nil {
		return err
	}
	if r.Host != host || r.Op != op || r.Role != OfflineRoleSealer || r.Phase != OfflinePhaseSealed {
		return fmt.Errorf("offline seal: not the sealed report of this move")
	}
	if r.TombstoneOwner != TombstoneOwner(op) {
		return fmt.Errorf("offline seal: tombstone owner %q, want %q", r.TombstoneOwner, TombstoneOwner(op))
	}
	if ok, bad := r.SourceItemsOK(); !ok {
		return fmt.Errorf("offline seal: items not OK: %v", bad)
	}
	return nil
}

// OfflineStager is the stage door's choice of the box that pre-copies the
// replica while a move is offline-staging: the lexically first of boxes,
// the source region's eligible boxes as the fleet's edges files list them,
// "" for none. The door pins it on the record (MoveState.Stager) and it
// never changes for the move: a box that joins later never stages or
// cleans for it, so exactly one box ever writes the staging prefix (Astra
// round 1, finding 2).
func OfflineStager(boxes []string) string {
	stager := ""
	for _, b := range boxes {
		if b != "" && (stager == "" || b < stager) {
			stager = b
		}
	}
	return stager
}

// OfflineActiveKey is the one lock of the offline move: while it names a
// host whose record still carries that move, no other offline move starts.
const OfflineActiveKey = "control/offline-moves/active.json"

// OfflineActive is the lock's body.
type OfflineActive struct {
	Host    string    `json:"host"`
	Op      string    `json:"op"`
	Account string    `json:"account"`
	App     string    `json:"app"`
	At      time.Time `json:"at"`
}

// OfflineArchivePrefix is where the move's sealed files archive and its
// timers snapshot live in the target bucket: moves/@offline/<host>/<op>/. The
// "@" cannot start a host, so it never collides with a host's own staging
// prefix. "" for a host that fails provision.ValidHostShape or an unusable op.
func OfflineArchivePrefix(host, op string) string {
	if !provision.ValidHostShape(host) || !offlineSegment(op) {
		return ""
	}
	return "moves/@offline/" + host + "/" + op + "/"
}

// TombstoneOwner is the owner string of the source lease's tombstone for a
// move: "moved:" + op. The lease package acts only on this owner.
func TombstoneOwner(op string) string { return "moved:" + op }

// TombstoneTTL is the longest a tombstone is written for. The sealer
// re-asserts it every pass and finish lifts it.
const TombstoneTTL = 7 * 24 * time.Hour

// The roles OfflineParkDir accepts besides the report roles: the target parks
// what it finds before landing under "target-before", and the landing's
// verify scratch under "verify" (Task 5). Nothing else is accepted; a numbered
// suffix ("target-before-2") is the caller's, built by renaming within the
// directory this returns, never passed through the helper.
const (
	OfflineParkTargetBefore = "target-before"
	OfflineParkVerify       = "verify"
)

var offlineParkRoles = map[string]bool{
	OfflineRoleSource: true, OfflineRoleSealer: true, OfflineRoleTarget: true,
	OfflineParkTargetBefore: true, OfflineParkVerify: true,
}

// OfflineParkDir is where a box parks what it renames for the host:
// <dataDir>/.offline-moves/<host>/<op>/<role>/. role is source, sealer,
// target, OfflineParkTargetBefore or OfflineParkVerify. "" when dataDir is
// empty, host fails provision.ValidHostShape, role is none of those, or op is
// not a plain name; a caller must not rename into "".
func OfflineParkDir(dataDir, host, op, role string) string {
	if dataDir == "" || !provision.ValidHostShape(host) || !offlineSegment(op) || !offlineParkRoles[role] {
		return ""
	}
	return filepath.Join(dataDir, ".offline-moves", host, op, role)
}

// OfflineNamesOK is the validity the key helpers share, for a caller that
// wants to ask before it builds anything.
func OfflineNamesOK(host, op, role, box string) bool {
	return OfflineReportKey(host, op, role, box) != ""
}