package moves

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/superfly/ltx"
)

// The seal (§7.3 "seal by version", §7.4 staged): the exact versions of the
// staging lane's replica/, files/, timers/ and state/ objects the cut wrote, the
// required final TXID N, and the two facts a recovery reads, in the claim's
// op journal under hostclaim.JournalSeal. Landing copies these versions and
// nothing else, so a stale round's later write (a newer version, or a
// delete marker) is never installed.
//
// It is written twice under one op: at the end of `stopped` with the replica
// and the timers (the archive does not exist yet), and at the end of
// `staged` with the archive's two objects added. Files is empty until then,
// and the cut never flips on a seal without them (HasFiles).
type Seal struct {
	V       int            `json:"v"`
	N       uint64         `json:"n"`
	Replica []SealedObject `json:"replica"`
	Files   []SealedObject `json:"files,omitempty"`
	Timers  []SealedObject `json:"timers,omitempty"`
	// State is the lane's copy of the instance's build state (settings spec
	// rev 11 §4.3a), at most one object at the lane's state/ key: the
	// version the round that wrote this seal copied. Absent when the source
	// had none; a seal written before this field existed decodes with none.
	State []SealedObject `json:"state,omitempty"`
	// SrcCovered is whether the source's own replica of holder.generation
	// reached the stopped database's position with no pending write (§7.4,
	// rev 7.11): it decides a rollback's clean return (§7.8).
	SrcCovered bool `json:"src_covered"`
	// FromDB is whether the seal was built from the stopped database itself
	// (§5.1 step 9): false for a source that had no local database.
	FromDB   bool      `json:"from_db"`
	SealedAt time.Time `json:"sealed_at"`
	// LeaseUntil is the latest moment a request signed by a staging writer
	// of this lane can still land: the last guarded signature's time plus
	// SignatureLife plus the skew (ruling G7). Bound returns it.
	LeaseUntil time.Time `json:"lease_until"`
}

// SealVersion is the only seal format this code reads or writes.
const SealVersion = 1

// ErrBadSeal is a seal that pins nothing usable: refused on write and on
// read, never repaired.
var ErrBadSeal = errors.New("moves: the seal is not usable")

// Bound is the time after which no staging writer that held the lease can
// still write: the staging park's evidence_at (Plan 5).
func (s Seal) Bound() time.Time { return s.LeaseUntil }

// HasFiles reports whether the seal names the archive: `archived` and the
// flip require it.
func (s Seal) HasFiles() bool { return len(s.Files) > 0 }

// ReachesN reports whether the sealed replica ends exactly at N: its newest
// level file's max TXID is N. A set that stops short cannot restore to N,
// and one that runs past it is not this stopped database's.
func (s Seal) ReachesN() bool {
	var newest ltx.TXID
	for _, o := range s.Replica {
		pos, err := sealedPosition(o.Key)
		if err != nil {
			return false
		}
		newest = max(newest, pos.max)
	}
	return s.N > 0 && uint64(newest) == s.N
}

// Validate refuses a seal that pins nothing: another version, no position,
// no replica, any object without a key or a usable version id (G27: "" and
// "null" pin nothing), a key named twice, a negative size, a replica object
// that is not a level file or is below an LTX header (Task 8's
// NewSealedClient refuses those), no seal time, and no bound or a bound
// before the seal.
func (s Seal) Validate() error {
	if s.V != SealVersion {
		return fmt.Errorf("%w: version %d", ErrBadSeal, s.V)
	}
	if s.N == 0 || len(s.Replica) == 0 {
		return fmt.Errorf("%w: no position or no replica", ErrBadSeal)
	}
	if s.SealedAt.IsZero() || s.LeaseUntil.IsZero() || s.LeaseUntil.Before(s.SealedAt) {
		return fmt.Errorf("%w: no seal time or no writers' bound", ErrBadSeal)
	}
	seen := map[string]bool{}
	if len(s.State) > 1 || (len(s.State) == 1 && !strings.HasSuffix(s.State[0].Key, BuildStateSuffix)) {
		return fmt.Errorf("%w: the state class names %d objects, want one at %s", ErrBadSeal, len(s.State), BuildStateSuffix)
	}
	for _, class := range [][]SealedObject{s.Replica, s.Files, s.Timers, s.State} {
		for _, o := range class {
			if o.Key == "" || !lease.UsableVersionID(o.VersionID) || o.Size < 0 || seen[o.Key] {
				return fmt.Errorf("%w: object %q version %q size %d", ErrBadSeal, o.Key, o.VersionID, o.Size)
			}
			seen[o.Key] = true
		}
	}
	for _, o := range s.Replica {
		if _, err := sealedPosition(o.Key); err != nil {
			return fmt.Errorf("%w: %v", ErrBadSeal, err)
		}
		if o.Size < ltx.HeaderSize {
			return fmt.Errorf("%w: replica object %s is below an LTX header", ErrBadSeal, o.Key)
		}
	}
	return nil
}

func sortObjects(objs []SealedObject) []SealedObject {
	out := append([]SealedObject(nil), objs...)
	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
	return out
}

// Encode is the seal as the claim journal carries it: JSON with every
// object list sorted by key, so one seal always encodes to one string.
func (s Seal) Encode() string {
	c := s
	c.Replica, c.Files, c.Timers, c.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State)
	c.SealedAt, c.LeaseUntil = s.SealedAt.UTC(), s.LeaseUntil.UTC()
	b, err := json.Marshal(c)
	if err != nil {
		// Every field is a plain value: Marshal cannot fail.
		panic(err)
	}
	return string(b)
}

// DecodeSeal reads a journalled seal strictly (no unknown field, no
// trailing data) and validates it. Anything else is ErrBadSeal.
func DecodeSeal(raw string) (Seal, error) {
	dec := json.NewDecoder(bytes.NewReader([]byte(raw)))
	dec.DisallowUnknownFields()
	var s Seal
	if err := dec.Decode(&s); err != nil {
		return Seal{}, fmt.Errorf("%w: %v", ErrBadSeal, err)
	}
	if _, err := dec.Token(); err != io.EOF {
		return Seal{}, fmt.Errorf("%w: trailing data", ErrBadSeal)
	}
	if err := s.Validate(); err != nil {
		return Seal{}, err
	}
	s.Replica, s.Files, s.Timers, s.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State)
	return s, nil
}