package genselect

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

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

var (
	ErrListing        = errors.New("genselect: incomplete or invalid listing")
	ErrDisagreement   = errors.New("genselect: marker and holder disagree")
	ErrIncarnation    = errors.New("genselect: marker incarnation mismatch")
	ErrMarker         = errors.New("genselect: malformed marker")
	ErrVersionID      = errors.New("genselect: immutable version id required")
	ErrDeleteMarker   = errors.New("genselect: selected version is a delete marker")
	ErrPinnedMissing  = errors.New("genselect: pinned snapshot version missing")
	ErrPinnedMismatch = errors.New("genselect: pinned snapshot mismatch")
	ErrGap            = errors.New("genselect: non-contiguous ltx files")
	ErrEmpty          = errors.New("genselect: no restorable data")
	ErrRegion         = errors.New("genselect: baseline belongs to another region")
	ErrCovered        = errors.New("genselect: local baseline is covered")
	ErrInput          = errors.New("genselect: invalid input")
	ErrRoot           = errors.New("genselect: invalid root")
)

type Reader interface {
	ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error)
	GetVersion(context.Context, string, string, string) ([]byte, error)
	OpenVersion(context.Context, string, string, string) (io.ReadCloser, error)
}

var _ Reader = (lease.VersionedStore)(nil)

type Pin struct {
	SnapshotKey     string `json:"snapshot_key"`
	SnapshotVersion string `json:"snapshot_version"`
	SnapshotTXID    uint64 `json:"snapshot_txid"`
	SnapshotCRC64   uint64 `json:"snapshot_crc64"`
}

type Marker struct {
	Op             string    `json:"op"`
	Incarnation    string    `json:"incarnation"`
	Host           string    `json:"host"`
	Box            string    `json:"box"`
	InstalledAt    time.Time `json:"installed_at"`
	ManifestSHA256 string    `json:"manifest_sha256"`
	Pin
}

// PreparedPin is supplied only from an authorized, completed build journal.
// The caller binds this exact generation and pin to the authorized holder,
// operation and destination region before calling SelectPrepared.
type PreparedPin struct {
	Generation      string
	SnapshotKey     string
	SnapshotVersion string
	SnapshotTXID    uint64
	SnapshotCRC64   uint64
}

type Request struct {
	Bucket, Region, Host string
	Holder               hostclaim.Holder
}

type File struct {
	Key       string `json:"key"`
	VersionID string `json:"version_id"`
	Level     int    `json:"level"`
	MinTXID   uint64 `json:"min_txid"`
	MaxTXID   uint64 `json:"max_txid"`
	Size      int64  `json:"size"`
}

type Plan struct {
	Generation    string
	Files         []File
	Root          *File
	Marker        *Marker
	MarkerVersion string
	Pin           *Pin
	MaxTXID       uint64
	MaxLevel0TXID uint64
}

type candidate struct {
	generation, region string
	covered            bool
}

func origin(g string) string {
	if g == "" {
		return hostclaim.RootGeneration
	}
	return g
}

func component(s string) bool {
	return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
}

func generation(s string) bool { return component(s) && s != hostclaim.RootGeneration }

func versionID(s string) bool { return strings.TrimSpace(s) != "" && s != "null" }

// prefix is a generation's prefix under location. Twins of the rule:
// hostbox.GenerationPrefix and moves.ReplicaSource (whose keep is
// coordinates' level-file test); the three change together.
func prefix(location, gen string) string {
	if gen == hostclaim.RootGeneration {
		return location
	}
	return location + "g-" + gen + "/"
}

func coordinates(key, base string) (File, bool) {
	if !strings.HasPrefix(key, base) {
		return File{}, false
	}
	parts := strings.Split(strings.TrimPrefix(key, base), "/")
	if len(parts) != 2 || len(parts[0]) != 4 {
		return File{}, false
	}
	level, err := strconv.ParseUint(parts[0], 16, 16)
	if err != nil || level > 9 || fmt.Sprintf("%04x", level) != parts[0] {
		return File{}, false
	}
	min, max, err := ltx.ParseFilename(parts[1])
	if err != nil || min == 0 || min > max || ltx.FormatFilename(min, max) != parts[1] {
		return File{}, false
	}
	return File{Key: key, Level: int(level), MinTXID: uint64(min), MaxTXID: uint64(max)}, true
}

func exactFile(v lease.ObjectVersion, base string) (File, error) {
	f, ok := coordinates(v.Key, base)
	if !ok || v.Size < 0 {
		return File{}, ErrListing
	}
	if !versionID(v.VersionID) {
		return File{}, ErrVersionID
	}
	if v.DeleteMarker {
		return File{}, ErrDeleteMarker
	}
	f.VersionID, f.Size = v.VersionID, v.Size
	return f, nil
}

func requestLocation(s Reader, req Request) (string, error) {
	h := req.Holder
	location := strings.TrimSuffix(h.Location, "/")
	if s == nil || req.Bucket == "" || req.Region == "" || !component(req.Host) || !component(h.Incarnation) ||
		(location != "instances/"+req.Host && location != "instances/"+req.Host+"/"+h.Incarnation) ||
		(h.Generation != "" && !generation(h.Generation)) {
		return "", ErrInput
	}
	return location + "/", nil
}

func listVersions(ctx context.Context, s Reader, bucket, base string) ([]lease.ObjectVersion, map[string]lease.ObjectVersion, error) {
	// VersionedStore drains all pages before returning. Never use partial data.
	versions, err := s.ListVersions(ctx, bucket, base)
	if err != nil {
		return nil, nil, fmt.Errorf("%w: %w", ErrListing, err)
	}
	latest := make(map[string]lease.ObjectVersion)
	keys := make(map[string]bool)
	for _, v := range versions {
		if !strings.HasPrefix(v.Key, base) {
			return nil, nil, ErrListing
		}
		keys[v.Key] = true
		if v.IsLatest {
			if _, exists := latest[v.Key]; exists {
				return nil, nil, ErrListing
			}
			latest[v.Key] = v
		}
	}
	for k := range keys {
		if _, ok := latest[k]; !ok {
			return nil, nil, ErrListing
		}
	}
	return versions, latest, nil
}

// Select resolves the ordinary baseline using the marker and unfinished run.
// It returns identities only. The consumer must verify every complete file.
func Select(ctx context.Context, s Reader, req Request) (*Plan, error) {
	location, err := requestLocation(s, req)
	if err != nil {
		return nil, err
	}
	versions, latest, err := listVersions(ctx, s, req.Bucket, location)
	if err != nil {
		return nil, err
	}
	h := req.Holder
	p := &Plan{Files: []File{}}
	markerGen := hostclaim.RootGeneration
	if mv, ok := latest[location+"generation.json"]; ok {
		if !versionID(mv.VersionID) {
			return nil, ErrVersionID
		}
		if mv.DeleteMarker {
			return nil, ErrDeleteMarker
		}
		data, err := s.GetVersion(ctx, req.Bucket, mv.Key, mv.VersionID)
		if err != nil {
			return nil, fmt.Errorf("%w: %w", ErrMarker, err)
		}
		var fields map[string]json.RawMessage
		if err := json.Unmarshal(data, &fields); err != nil {
			return nil, ErrMarker
		}
		if raw, ok := fields["snapshot_txid"]; !ok || string(raw) == "null" {
			return nil, ErrMarker
		}
		var m Marker
		dec := json.NewDecoder(bytes.NewReader(data))
		// Unknown additive fields are allowed. Malformed/trailing JSON is not.
		if err := dec.Decode(&m); err != nil {
			return nil, fmt.Errorf("%w: %w", ErrMarker, err)
		}
		if err := dec.Decode(new(any)); err != io.EOF {
			return nil, ErrMarker
		}
		if m.Incarnation != h.Incarnation {
			return nil, ErrIncarnation
		}
		if m.Host != req.Host || !generation(m.Op) || m.Box == "" || m.InstalledAt.IsZero() {
			return nil, ErrMarker
		}
		markerGen, p.Marker, p.MarkerVersion = m.Op, &m, mv.VersionID
	}
	run, err := unfinished(h, markerGen, req.Region)
	if err != nil {
		return nil, err
	}
	selected := candidate{generation: markerGen, region: req.Region}
	for _, c := range run {
		if c.region != "" && c.region != req.Region {
			continue
		}
		if c.covered {
			return nil, ErrCovered
		}
		if c.generation == markerGen || hasObject(latest, location, c.generation) {
			selected = c
			break
		}
	}
	p.Generation = selected.generation
	base := prefix(location, p.Generation)
	var pin *Pin
	if p.Marker != nil && p.Generation == markerGen {
		cp := p.Marker.Pin
		pin = &cp
	}
	return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
}

// SelectPrepared restores only the explicitly authorized prepared generation.
// It never reads the marker or walks admissions, and never chooses a baseline.
func SelectPrepared(ctx context.Context, s Reader, req Request, prepared PreparedPin) (*Plan, error) {
	location, err := requestLocation(s, req)
	if err != nil {
		return nil, err
	}
	if !generation(prepared.Generation) {
		return nil, ErrInput
	}
	// Empty prepared roots require a separate bootstrap protocol, not fallback.
	if prepared.SnapshotTXID == 0 {
		return nil, ErrPinnedMismatch
	}
	base := prefix(location, prepared.Generation)
	versions, latest, err := listVersions(ctx, s, req.Bucket, base)
	if err != nil {
		return nil, err
	}
	pin := &Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion,
		SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
	p := &Plan{Generation: prepared.Generation, Files: []File{}}
	return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
}

// GenerationMax is the loss boundary of generation gen in a restore
// (platform-1e's ruling on #236): the highest TXID of that exact generation
// under location among the restore plan's selected files (a selected
// historical pinned version counts, since the plan holds it). It never
// lists the bucket. A generation the plan holds no file of, which is every
// generation but the plan's own, is 0 and nil: every write in it is lost.
// "" and hostclaim.RootGeneration are the location's own level files, the
// legacy root. A plan file whose recorded coordinates are not its key's is
// ErrInput. The claim writer's loss restore point (hostbox.Writer) and
// moves' loss baseline validator read it alike.
func GenerationMax(p *Plan, location, gen string) (uint64, error) {
	location = strings.TrimSuffix(location, "/")
	if p == nil || location == "" || (gen != "" && gen != hostclaim.RootGeneration && !generation(gen)) {
		return 0, ErrInput
	}
	for _, part := range strings.Split(location, "/") {
		if !component(part) {
			return 0, ErrInput
		}
	}
	base := prefix(location+"/", origin(gen))
	max := uint64(0)
	for _, f := range p.Files {
		c, ok := coordinates(f.Key, base)
		if !ok {
			continue // another generation's file
		}
		if c.Level != f.Level || c.MinTXID != f.MinTXID || c.MaxTXID != f.MaxTXID {
			return 0, ErrInput
		}
		if f.MaxTXID > max {
			max = f.MaxTXID
		}
	}
	return max, nil
}

// currentFile reads one latest version under base: a current level file
// (ok), not one (a retention tombstone, another generation's key, a marker),
// or a key shaped as a level file that does not parse (ErrListing). Select
// and GenerationMax read the coordinates through it alone.
func currentFile(v lease.ObjectVersion, base string) (File, bool, error) {
	if _, ok := coordinates(v.Key, base); !ok {
		if strings.HasPrefix(v.Key, base) {
			parts := strings.Split(strings.TrimPrefix(v.Key, base), "/")
			if len(parts) == 2 && len(parts[0]) == 4 && strings.HasSuffix(parts[1], ".ltx") {
				if level, e := strconv.ParseUint(parts[0], 16, 16); e == nil && level <= 9 {
					return File{}, false, ErrListing
				}
			}
		}
		return File{}, false, nil
	}
	if v.DeleteMarker {
		return File{}, false, nil
	} // Retention tombstones are not current files.
	f, err := exactFile(v, base)
	if err != nil {
		return File{}, false, err
	}
	return f, true, nil
}

// Both entry points share exact-version root selection and continuation.
func selectFiles(ctx context.Context, s Reader, bucket, base string, versions []lease.ObjectVersion,
	latest map[string]lease.ObjectVersion, p *Plan, pin *Pin) (*Plan, error) {
	if pin != nil {
		if pin.SnapshotTXID == 0 {
			if pin.SnapshotKey != "" || pin.SnapshotVersion != "" || pin.SnapshotCRC64 != 0 {
				return nil, ErrPinnedMismatch
			}
		} else {
			f, ok := coordinates(pin.SnapshotKey, base)
			if !ok || f.Level != 9 || f.MinTXID != 1 || f.MaxTXID != pin.SnapshotTXID {
				return nil, ErrPinnedMismatch
			}
			if !versionID(pin.SnapshotVersion) {
				return nil, ErrVersionID
			}
		}
		p.Pin = pin
	}
	var current []File
	for _, v := range latest {
		f, ok, err := currentFile(v, base)
		if err != nil {
			return nil, err
		}
		if !ok {
			continue
		}
		if f.Level == 9 && f.MinTXID != 1 {
			return nil, ErrRoot
		}
		current = append(current, f)
		if f.Level == 0 && f.MaxTXID > p.MaxLevel0TXID {
			p.MaxLevel0TXID = f.MaxTXID
		}
	}
	floor := uint64(0)
	if pin != nil {
		floor = pin.SnapshotTXID
	}
	for _, f := range current {
		// A late duplicate at the pinned key never replaces its acknowledged version.
		if pin != nil && floor > 0 && f.Key == pin.SnapshotKey {
			continue
		}
		if f.Level == 9 && f.MinTXID == 1 && f.MaxTXID >= floor && (p.Root == nil || better(f, *p.Root)) {
			cp := f
			p.Root = &cp
		}
	}
	pinned := false
	if p.Root == nil && floor > 0 {
		var pv *lease.ObjectVersion
		for _, v := range versions {
			if v.Key == pin.SnapshotKey && v.VersionID == pin.SnapshotVersion {
				cp := v
				pv = &cp
				break
			}
		}
		if pv == nil {
			return nil, ErrPinnedMissing
		}
		f, err := exactFile(*pv, base)
		if err != nil {
			return nil, err
		}
		p.Root, pinned = &f, true
	}
	if p.Root != nil {
		if err := checkRoot(ctx, s, bucket, *p.Root, pinned); err != nil {
			return nil, err
		}
		p.Files = append(p.Files, *p.Root)
		p.MaxTXID = p.Root.MaxTXID
	}
	for {
		var best *File
		for _, f := range current {
			if f.Level == 9 || p.MaxTXID == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(p.MaxTXID), ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID)) {
				continue
			}
			if best == nil || better(f, *best) {
				cp := f
				best = &cp
			}
		}
		if best == nil {
			break
		}
		if p.Root == nil {
			if err := checkRoot(ctx, s, bucket, *best, false); err != nil {
				return nil, err
			}
			cp := *best
			p.Root = &cp
		}
		p.Files = append(p.Files, *best)
		p.MaxTXID = best.MaxTXID
	}
	for _, f := range current {
		if f.Level < 9 && f.MaxTXID > p.MaxTXID {
			return nil, ErrGap
		}
	}
	if len(p.Files) == 0 {
		return nil, ErrEmpty
	}
	return p, nil
}

func unfinished(h hostclaim.Holder, marker, region string) ([]candidate, error) {
	gen := origin(h.Generation)
	if gen == marker {
		return []candidate{{generation: marker, region: region}}, nil
	}
	var out []candidate
	covered := false
	seen := map[string]bool{}
	for i := len(h.Admissions) - 1; i >= 0; i-- {
		a := h.Admissions[i]
		if origin(a.Generation) != gen || a.Region != region {
			return nil, ErrDisagreement
		}
		if a.Mode == "warm" {
			if origin(a.Predecessor) != gen || a.PredecessorRegion != region {
				return nil, ErrDisagreement
			}
			continue
		}
		if a.Mode != "cold" || !generation(a.Generation) || seen[gen] {
			return nil, ErrDisagreement
		}
		seen[gen] = true
		out = append(out, candidate{generation: gen, region: region, covered: covered})
		gen, region, covered = origin(a.Predecessor), a.PredecessorRegion, a.Covered
		if gen == marker {
			return append(out, candidate{generation: gen, region: region, covered: covered}), nil
		}
		if !generation(gen) || region == "" {
			return nil, ErrDisagreement
		}
	}
	return nil, ErrDisagreement
}

func hasObject(latest map[string]lease.ObjectVersion, location, gen string) bool {
	base := prefix(location, gen)
	for k := range latest {
		if !strings.HasPrefix(k, base) {
			continue
		}
		if gen == hostclaim.RootGeneration {
			if _, ok := coordinates(k, base); !ok {
				continue
			}
		}
		return true
	}
	return false
}

func better(a, b File) bool {
	if a.MaxTXID != b.MaxTXID {
		return a.MaxTXID > b.MaxTXID
	}
	if a.MinTXID != b.MinTXID {
		return a.MinTXID < b.MinTXID
	}
	return a.Level > b.Level
}

func checkRoot(ctx context.Context, s Reader, bucket string, f File, pinned bool) error {
	category := ErrRoot
	if pinned {
		category = ErrPinnedMismatch
	}
	rc, err := s.OpenVersion(ctx, bucket, f.Key, f.VersionID)
	if err != nil {
		if errors.Is(err, lease.ErrIsDeleteMarker) {
			return fmt.Errorf("%w: %w", ErrDeleteMarker, err)
		}
		if pinned && errors.Is(err, lease.ErrMirrorNotFound) {
			return fmt.Errorf("%w: %w", ErrPinnedMissing, err)
		}
		return fmt.Errorf("%w: %w", category, err)
	}
	dec := ltx.NewDecoder(rc)
	err = dec.DecodeHeader()
	closeErr := rc.Close()
	if err != nil {
		return fmt.Errorf("%w: %w", category, err)
	}
	if closeErr != nil {
		return fmt.Errorf("%w: %w", category, closeErr)
	}
	h := dec.Header()
	if uint64(h.MinTXID) != f.MinTXID || uint64(h.MaxTXID) != f.MaxTXID {
		return category
	}
	return nil
}