// Package boxvolumes records a platform box's block device mappings while
// it runs (region HA spec 1 §5.9, ruling H19), and decides from those
// records whether a terminated box's park-local chores may be discharged
// without retirement (claim D5's destroyed-local exception, L3).
//
// Nothing here is a termination fence, and nothing about execution,
// admission, relocation or recovery reads it (spec §8, B4-1). A record that
// is wrong or missing can leave a disk unretired for an operator; it can
// never admit a writer.
//
// This package builds, writes and reads records. The loop that describes
// every enrolled box in the region needs the roster's enrollments, and the
// discharge CAS needs L3: both wait for the claim.
package boxvolumes

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

	"github.com/aws/aws-sdk-go-v2/aws"
	awsmiddleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
	"github.com/aws/aws-sdk-go-v2/service/ec2"

	"github.com/carlosframework/platform/internal/boxid"
	"github.com/carlosframework/platform/internal/release"
)

const (
	Prefix = "control/box-volumes/"

	// RefreshEvery is the longest a running box goes without a new record
	// when nothing changes (spec §5.9: "at least once an hour").
	RefreshEvery = time.Hour
	// EnrollmentWindow is how soon after enrollment the first record must
	// be taken for it to stand for the launch-time mappings.
	EnrollmentWindow = time.Hour
	// LastRecordWindow is the most the newest record may predate the
	// termination observation (controller ruling, 2026-10-06). An older one
	// cannot speak for a volume attached after it.
	LastRecordWindow = 2 * time.Hour

	keyTime = "2006-01-02T15-04-05.000000000Z"
)

// Device is one EBS volume in the instance's block device mappings.
type Device struct {
	Device              string `json:"device"`
	VolumeID            string `json:"volume_id"`
	DeleteOnTermination bool   `json:"delete_on_termination"`
	Status              string `json:"status"`
}

// WrittenBy names the regional worker that made the read.
type WrittenBy struct {
	Box        string `json:"box"`
	Enrollment string `json:"enrollment"`
}

// Record is §5.9's volume record,
// control/box-volumes/<enrollment>/<observed_at>.json.
type Record struct {
	V              int         `json:"v"`
	Box            string      `json:"box"`
	Enrollment     string      `json:"enrollment"`
	Cloud          boxid.Cloud `json:"cloud"`
	InstanceState  string      `json:"instance_state"`
	BlockDevices   []Device    `json:"block_devices"`
	RemovedVolumes []string    `json:"removed_volumes"`
	ObservedAt     string      `json:"observed_at"`
	RequestID      string      `json:"request_id"`
	WrittenBy      WrittenBy   `json:"written_by"`
}

// Entry is the roster entry a record describes.
type Entry struct {
	Box        string
	Enrollment string
	Cloud      boxid.Cloud
}

// Observation is one DescribeInstances read of one enrolled box.
type Observation struct {
	State     string
	Devices   []Device
	RequestID string
}

var ErrNotRecordable = errors.New("boxvolumes: not a recordable observation")

// ValidEnrollment reports whether e can be a key segment: 1 to 128 of
// [A-Za-z0-9._-], never "." or "..".
func ValidEnrollment(e string) bool {
	if e == "" || len(e) > 128 || e == "." || e == ".." {
		return false
	}
	for _, r := range e {
		ok := r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '.' || r == '_' || r == '-'
		if !ok {
			return false
		}
	}
	return true
}

// Key is where a record observed at t lives. The time is fixed-width UTC,
// so keys sort in observation order.
func Key(enrollment string, t time.Time) string {
	return Prefix + enrollment + "/" + t.UTC().Format(keyTime) + ".json"
}

// recordable is every state in which EC2 still reports the instance's
// mappings. "terminated" is not one: its BlockDeviceMappings is empty (the
// dev test of 2026-10-06), which would read as every volume removed.
var recordable = map[string]bool{"pending": true, "running": true, "shutting-down": true, "stopping": true, "stopped": true}

// Observe reads the entry's instance out of a DescribeInstances reply (see
// termfence.DescribeInstance for the call). It refuses a reply that does
// not name the instance, names it under another owner, or reports a state
// whose mappings are not the instance's.
func Observe(out *ec2.DescribeInstancesOutput, c boxid.Cloud) (Observation, error) {
	if out == nil || !c.Complete() {
		return Observation{}, fmt.Errorf("%w: no reply, or no proven cloud", ErrNotRecordable)
	}
	var o Observation
	if id, ok := awsmiddleware.GetRequestIDMetadata(out.ResultMetadata); ok {
		o.RequestID = id
	}
	for _, r := range out.Reservations {
		for _, inst := range r.Instances {
			if aws.ToString(inst.InstanceId) != c.InstanceID {
				continue
			}
			if aws.ToString(r.OwnerId) != c.Account {
				return Observation{}, fmt.Errorf("%w: %s is owned by %q, the roster says %q", ErrNotRecordable, c.InstanceID, aws.ToString(r.OwnerId), c.Account)
			}
			if inst.State != nil {
				o.State = string(inst.State.Name)
			}
			if !recordable[o.State] {
				return Observation{}, fmt.Errorf("%w: state %q", ErrNotRecordable, o.State)
			}
			for _, m := range inst.BlockDeviceMappings {
				if m.Ebs == nil {
					continue // instance store dies with the instance
				}
				o.Devices = append(o.Devices, Device{Device: aws.ToString(m.DeviceName), VolumeID: aws.ToString(m.Ebs.VolumeId),
					DeleteOnTermination: aws.ToBool(m.Ebs.DeleteOnTermination), Status: string(m.Ebs.Status)})
			}
			sort.Slice(o.Devices, func(i, j int) bool { return o.Devices[i].VolumeID < o.Devices[j].VolumeID })
			return o, nil
		}
	}
	return Observation{}, fmt.Errorf("%w: the reply does not name %s", ErrNotRecordable, c.InstanceID)
}

func volumeIDs(ds []Device) []string {
	out := make([]string, 0, len(ds))
	for _, d := range ds {
		out = append(out, d.VolumeID)
	}
	return out
}

func removed(prev, cur []Device) []string {
	have := volumeIDs(cur)
	var out []string
	for _, id := range volumeIDs(prev) {
		if !slices.Contains(have, id) {
			out = append(out, id)
		}
	}
	sort.Strings(out)
	return out
}

// Next decides whether o, made by writer at now, is a new record after
// prev (nil when the enrollment has none). It is one when there is no
// record yet, when the mapping or the instance state differs from prev,
// and otherwise once RefreshEvery has passed.
func Next(prev *Record, e Entry, writer WrittenBy, o Observation, now time.Time) (Record, bool) {
	rec := Record{V: 1, Box: e.Box, Enrollment: e.Enrollment, Cloud: e.Cloud, InstanceState: o.State,
		BlockDevices: o.Devices, RemovedVolumes: []string{}, ObservedAt: now.UTC().Format(time.RFC3339Nano),
		RequestID: o.RequestID, WrittenBy: writer}
	if rec.BlockDevices == nil {
		rec.BlockDevices = []Device{}
	}
	if prev == nil {
		return rec, true
	}
	if r := removed(prev.BlockDevices, o.Devices); len(r) > 0 {
		rec.RemovedVolumes = r
	}
	if !slices.Equal(prev.BlockDevices, rec.BlockDevices) || prev.InstanceState != o.State {
		return rec, true
	}
	at, err := time.Parse(time.RFC3339Nano, prev.ObservedAt)
	return rec, err != nil || now.Sub(at) >= RefreshEvery
}

// Write stores rec create-only at its key. Records are never replaced or
// deleted (spec §6); the same bytes written twice are one record.
func Write(ctx context.Context, s release.Store, rec Record) (string, error) {
	if !ValidEnrollment(rec.Enrollment) {
		return "", fmt.Errorf("boxvolumes: %q is not an enrollment", rec.Enrollment)
	}
	at, err := time.Parse(time.RFC3339Nano, rec.ObservedAt)
	if err != nil {
		return "", fmt.Errorf("boxvolumes: observed_at: %w", err)
	}
	body, err := json.Marshal(rec)
	if err != nil {
		return "", err
	}
	key := Key(rec.Enrollment, at)
	_, err = s.Put(ctx, key, body, "")
	if errors.Is(err, release.ErrCASConflict) {
		cur, _, gerr := s.Get(ctx, key)
		if gerr == nil && bytes.Equal(cur, body) {
			return key, nil
		}
		return "", fmt.Errorf("boxvolumes: %s already holds another record: %w", key, err)
	}
	return key, err
}

func decode(body []byte) (Record, error) {
	dec := json.NewDecoder(bytes.NewReader(body))
	dec.DisallowUnknownFields()
	var rec Record
	if err := dec.Decode(&rec); err != nil {
		return Record{}, err
	}
	if _, err := dec.Token(); err != io.EOF {
		return Record{}, errors.New("trailing data")
	}
	return rec, nil
}

// List returns every record stored for an enrollment, in key order. Any
// record that cannot be read or parsed is an error, never a gap: a gap
// could hide the record that shows a removal.
func List(ctx context.Context, s release.Store, enrollment string) ([]Record, error) {
	if !ValidEnrollment(enrollment) {
		return nil, fmt.Errorf("boxvolumes: %q is not an enrollment", enrollment)
	}
	keys, err := s.List(ctx, Prefix+enrollment+"/")
	if err != nil {
		return nil, err
	}
	sort.Strings(keys)
	out := make([]Record, 0, len(keys))
	for _, k := range keys {
		if !strings.HasSuffix(k, ".json") {
			continue
		}
		body, _, err := s.Get(ctx, k)
		if err != nil {
			return nil, fmt.Errorf("boxvolumes: %s: %w", k, err)
		}
		rec, err := decode(body)
		if err != nil {
			return nil, fmt.Errorf("boxvolumes: %s: %w", k, err)
		}
		out = append(out, rec)
	}
	return out, nil
}

// incomplete says why a stored record is not evidence of a running
// instance's mappings, or "" when it is. Write can store such a record
// (Observe records every state in recordable, and a volume's status as
// EC2 reports it), and a stored one may also come from an older writer, a
// hand edit or a bug; any of them keeps the chores open rather than being
// read as "nothing attached".
//
// A volume must be attached or attaching, with one exception: while the
// instance is shutting down EC2 reports its DeleteOnTermination volumes
// as detaching on their way to deletion, so in a shutting-down record a
// detaching volume whose flag is true is still evidence. A detaching
// volume with a false flag, or in any other state, may outlive the box.
func incomplete(r Record) string {
	if !recordable[r.InstanceState] {
		return fmt.Sprintf("has state %q, which carries no mapping", r.InstanceState)
	}
	if len(r.BlockDevices) == 0 {
		return "lists no EBS volume" // a platform box always has its root volume
	}
	for _, d := range r.BlockDevices {
		if d.VolumeID == "" || d.Device == "" {
			return "lists a volume with no id or device"
		}
		dying := r.InstanceState == "shutting-down" && d.DeleteOnTermination && d.Status == "detaching"
		if d.Status != "attached" && d.Status != "attaching" && !dying {
			return fmt.Sprintf("lists volume %s with status %q", d.VolumeID, d.Status)
		}
	}
	return ""
}

// Verdict is whether a terminated box's park-local chores may be
// discharged without retirement, and if not, why.
type Verdict struct {
	Discharge bool
	Reason    string
}

// Discharge decides from records (as List returns them) whether every EBS
// volume e's instance had attached, from the first record taken within
// EnrollmentWindow of enrolledAt to the last, taken within LastRecordWindow
// of terminatedAt (the fence's observed_at), was DeleteOnTermination, with
// none ever removed (spec §5.9, H19). Every record is checked, none
// skipped: one with another version, enrollment or cloud identity, one
// that is incomplete or carries no mapping, a volume not attached or
// attaching (bar a DeleteOnTermination one detaching as the instance
// shuts down), a record outside enrolledAt to terminatedAt, or a gap between
// records over LastRecordWindow keeps the chores open, as do missing or
// misordered times. Anything short of that keeps the chores open for an
// operator, which costs cleanup and never recovery.
func Discharge(records []Record, e Entry, enrolledAt, terminatedAt time.Time) Verdict {
	no := func(format string, a ...any) Verdict { return Verdict{Reason: fmt.Sprintf(format, a...)} }
	if !e.Cloud.Complete() || e.Enrollment == "" {
		return no("the roster entry has no proven cloud")
	}
	if enrolledAt.IsZero() || terminatedAt.IsZero() || terminatedAt.Before(enrolledAt) {
		return no("the enrollment and termination times are missing or out of order")
	}
	type stamped struct {
		at  time.Time
		rec Record
	}
	var mine []stamped
	for _, r := range records {
		// Every record listed under this enrollment's prefix is checked,
		// none skipped: one that is not this enrollment's, this instance's
		// or a version this binary reads could be the adverse record.
		if r.V != 1 {
			return no("a record for %s has version %d", e.Enrollment, r.V)
		}
		if r.Enrollment != e.Enrollment {
			return no("a record listed for %s names enrollment %q", e.Enrollment, r.Enrollment)
		}
		if r.Cloud != e.Cloud {
			return no("a record for %s names another cloud identity", e.Enrollment)
		}
		if reason := incomplete(r); reason != "" {
			return no("the record observed at %q %s", r.ObservedAt, reason)
		}
		at, err := time.Parse(time.RFC3339Nano, r.ObservedAt)
		if err != nil {
			return no("record with unreadable observed_at %q", r.ObservedAt)
		}
		mine = append(mine, stamped{at, r})
	}
	if len(mine) == 0 {
		return no("no volume record for %s", e.Enrollment)
	}
	sort.SliceStable(mine, func(i, j int) bool { return mine[i].at.Before(mine[j].at) })
	if first := mine[0].at; first.After(enrolledAt.Add(EnrollmentWindow)) {
		return no("the first record (%s) is more than %s after enrollment", first.Format(time.RFC3339), EnrollmentWindow)
	}
	if last := mine[len(mine)-1].at; terminatedAt.Sub(last) > LastRecordWindow {
		return no("the newest record (%s) is more than %s before the termination", last.Format(time.RFC3339), LastRecordWindow)
	}
	for i, s := range mine {
		if s.at.Before(enrolledAt) || s.at.After(terminatedAt) {
			return no("the record observed at %s lies outside enrollment to termination", s.rec.ObservedAt)
		}
		if i > 0 && s.at.Sub(mine[i-1].at) > LastRecordWindow {
			return no("no record between %s and %s, more than %s", mine[i-1].rec.ObservedAt, s.rec.ObservedAt, LastRecordWindow)
		}
		for _, d := range s.rec.BlockDevices {
			if !d.DeleteOnTermination {
				return no("volume %s was not DeleteOnTermination at %s", d.VolumeID, s.rec.ObservedAt)
			}
		}
		if len(s.rec.RemovedVolumes) > 0 {
			return no("volumes %v were removed by %s", s.rec.RemovedVolumes, s.rec.ObservedAt)
		}
		if i > 0 {
			if r := removed(mine[i-1].rec.BlockDevices, s.rec.BlockDevices); len(r) > 0 {
				return no("volumes %v left the mapping by %s", r, s.rec.ObservedAt)
			}
		}
	}
	return Verdict{Discharge: true}
}