package provision

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"time"

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

// RunState is a `carlos run` on the instance record (run spec §5.1). Box
// and BootID name the box and agent boot that picked it; Deadline is the
// pickup deadline while queued and the run's own deadline once picked.
// By is the caller, so a refusal can name who holds the instance (an
// addition to the spec's struct, index item 3).
type RunState struct {
	ID       string    `json:"id"`
	State    string    `json:"state"`
	By       string    `json:"by,omitempty"`
	Box      string    `json:"box,omitempty"`
	BootID   string    `json:"boot,omitempty"`
	Deadline time.Time `json:"deadline"`
}

// The run's states. queued, picked and cancel are open; the rest are
// terminal, and a terminal run never blocks a run, a move or an import.
const (
	RunQueued    = "queued"
	RunPicked    = "picked"
	RunCancel    = "cancel"
	RunDone      = "done"
	RunCancelled = "cancelled"
	RunExpired   = "expired"
)

// Open reports whether r holds the instance: queued, picked or cancel.
func (r *RunState) Open() bool {
	return r != nil && (r.State == RunQueued || r.State == RunPicked || r.State == RunCancel)
}

// ImportUncommitted reports whether the record's seed is an import that is
// not committed yet (seeded spec §5.4): a run is refused then, at the door
// and again at launch (run spec §5.1 step 9, codex R2-2).
func (r Record) ImportUncommitted() bool {
	return r.Seed.BlocksRun()
}

var (
	// ErrRunOpen: a run is open on the instance. A RunOpenError carries it.
	ErrRunOpen = errors.New("provision: a run is open on the instance")
	// ErrRunConflict: the record's run is not in the state the transition
	// needs (another box took it, it was cancelled or expired, or it is
	// another run).
	ErrRunConflict = errors.New("provision: run state conflict")
	// ErrRunMoving: a move is in flight.
	ErrRunMoving = errors.New("provision: the instance is moving")
	// ErrRunImport: the instance is an import that is not committed.
	ErrRunImport = errors.New("provision: the instance is an import that is not committed")
)

// RunOpenError is a refusal because Run holds the instance.
type RunOpenError struct{ Run RunState }

func (e *RunOpenError) Error() string        { return ErrRunOpen.Error() + ": " + e.Run.ID }
func (e *RunOpenError) Is(target error) bool { return target == ErrRunOpen }

// MoveRunOpenError is a move transition's refusal while a run is open: a
// move conflict, so both move doors answer 409. The legacy door answers
// Error(), which carries Sentence (RUN-C43).
type MoveRunOpenError struct{ Run RunState }

func (e *MoveRunOpenError) Error() string { return ErrMoveConflict.Error() + ": " + e.Sentence() }
func (e *MoveRunOpenError) Unwrap() error { return ErrMoveConflict }
func (e *MoveRunOpenError) Sentence() string {
	return fmt.Sprintf("a run is open on this instance: %s. wait for it to finish or cancel it, then move the instance", e.Run.ID)
}

// SetRecordRun applies one run transition under the record's ETag, with
// updateRecord's one retry: a transition re-checks its own state on the
// fresh read, so a lost race refuses rather than overwrites.
func SetRecordRun(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) {
	return updateRecord(ctx, s, acct, app, host, mut)
}

// RunAdmission is what the console decided for a new run. Key is the
// caller's scoped Idempotency-Key (runs.ScopedKey), "" for an unkeyed
// create; Digest is the request's runs.MetaDigest and Caller the caller's
// runs.CallerOf, both set with Key.
type RunAdmission struct {
	ID, By              string
	Deadline            time.Time // the pickup deadline
	Key, Digest, Caller string
}

// RunKey is one keyed admission on the record: the scoped key, the run it
// admitted, the request's digest, the caller (runs.CallerOf, which the
// per-caller cap counts by) and when.
type RunKey struct {
	Key    string    `json:"key"`
	RunID  string    `json:"run_id"`
	Digest string    `json:"digest"`
	Caller string    `json:"caller"`
	At     time.Time `json:"at"`
}

// RunKeyReplay is RunAdmit finding its key already admitted with the same
// digest: RunID is the run that admission made, and this one admitted
// nothing. It is the caller's own earlier write when RunID is the id it
// minted (updateRecord's retry after a lost answer), and a twin's otherwise.
type RunKeyReplay struct{ RunID string }

func (e *RunKeyReplay) Error() string {
	return "provision: the idempotency key already admitted " + e.RunID
}

// ErrRunKeyReused: the key was admitted for a different request.
var ErrRunKeyReused = errors.New("provision: the idempotency key was used for a different request")

// RunKeyFor is the record's entry for the scoped key, when one younger than
// runs.KeyTTL is there.
func (r Record) RunKeyFor(key string, now time.Time) (RunKey, bool) {
	for _, k := range r.RunKeys {
		if k.Key == key && now.Sub(k.At) < runs.KeyTTL {
			return k, true
		}
	}
	return RunKey{}, false
}

// prunedRunKeys is keys without the entries runs.KeyTTL old or older, then
// with only the newest runs.MaxRunKeysPerCaller of each caller, in age order
// (oldest first). The cap is per caller so that no caller's admissions can
// evict another caller's key: a key is honoured for runs.KeyTTL, or until
// that same caller makes runs.MaxRunKeysPerCaller newer keyed admissions on
// this instance, whichever comes first.
func prunedRunKeys(keys []RunKey, now time.Time) []RunKey {
	var fresh []RunKey
	for _, k := range keys {
		if now.Sub(k.At) < runs.KeyTTL {
			fresh = append(fresh, k)
		}
	}
	sort.SliceStable(fresh, func(i, j int) bool { return fresh[i].At.Before(fresh[j].At) })
	kept := make([]bool, len(fresh))
	per := map[string]int{}
	for i := len(fresh) - 1; i >= 0; i-- {
		if per[fresh[i].Caller] < runs.MaxRunKeysPerCaller {
			per[fresh[i].Caller]++
			kept[i] = true
		}
	}
	var out []RunKey
	for i, k := range fresh {
		if kept[i] {
			out = append(out, k)
		}
	}
	return out
}

// RunAdmit admits a run: refused while a move is in flight, while a run is
// open, or while the record is an uncommitted import. A terminal run is
// replaced.
//
// A keyed admission checks its key first, before every refusal, so a retry
// is answered even while its own run is open: the key already admitted with
// the same digest is a *RunKeyReplay, with another digest ErrRunKeyReused.
// A keyed admission that succeeds records its key in the same write.
// run_keys is pruned by age on every admission and by each caller's count
// when one is added; nothing else changes it.
func RunAdmit(p RunAdmission) func(*Record) error {
	return func(rec *Record) error {
		now := time.Now().UTC()
		keys := prunedRunKeys(rec.RunKeys, now)
		if p.Key != "" {
			if p.Digest == "" || p.Caller == "" {
				return fmt.Errorf("%w: malformed admission", ErrRunConflict)
			}
			if k, ok := (Record{RunKeys: keys}).RunKeyFor(p.Key, now); ok {
				if k.Digest != p.Digest {
					return ErrRunKeyReused
				}
				return &RunKeyReplay{RunID: k.RunID}
			}
		}
		switch {
		case p.ID == "" || p.Deadline.IsZero():
			return fmt.Errorf("%w: malformed admission", ErrRunConflict)
		case rec.Move != nil:
			return ErrRunMoving
		case rec.ActiveRun.Open():
			return &RunOpenError{Run: *rec.ActiveRun}
		case rec.ActiveRun != nil && rec.ActiveRun.ID == p.ID:
			// A terminal run's id is never reused: a box still retrying
			// RunPick for it would launch a run its caller cancelled or that
			// expired (codex R2-6).
			return fmt.Errorf("%w: run id already used", ErrRunConflict)
		case rec.ImportUncommitted():
			return ErrRunImport
		}
		rec.ActiveRun = &RunState{ID: p.ID, State: RunQueued, By: p.By, Deadline: p.Deadline.UTC()}
		if p.Key != "" {
			keys = prunedRunKeys(append(keys, RunKey{Key: p.Key, RunID: p.ID, Digest: p.Digest, Caller: p.Caller, At: now}), now)
		}
		if len(keys) == 0 {
			keys = nil // an empty list stays absent from the record
		}
		rec.RunKeys = keys
		return nil
	}
}

// runIn returns a copy of the record's run when it is id in one of states.
func runIn(rec *Record, id string, states ...string) (RunState, error) {
	r := rec.ActiveRun
	if r == nil || r.ID != id {
		return RunState{}, ErrRunConflict
	}
	for _, s := range states {
		if r.State == s {
			return *r, nil
		}
	}
	return RunState{}, ErrRunConflict
}

// RunPick moves a queued run to picked by box under boot, with the run's
// own deadline (run spec §5.2). Only the box whose CAS lands launches.
func RunPick(id, box, boot string, deadline time.Time) func(*Record) error {
	return func(rec *Record) error {
		if box == "" || boot == "" || deadline.IsZero() {
			// A zero deadline would make the run lost the moment it is picked.
			return fmt.Errorf("%w: malformed pick", ErrRunConflict)
		}
		r, err := runIn(rec, id, RunQueued)
		if err != nil {
			return err
		}
		r.State, r.Box, r.BootID, r.Deadline = RunPicked, box, boot, deadline.UTC()
		rec.ActiveRun = &r
		return nil
	}
}

// RunCancelTransition cancels: queued becomes cancelled, which is terminal,
// so no box ever launches it; picked becomes cancel, and the owning box
// stops the unit (run spec §5.7). cancel stays cancel.
func RunCancelTransition(id string) func(*Record) error {
	return func(rec *Record) error {
		r, err := runIn(rec, id, RunQueued, RunPicked, RunCancel)
		if err != nil {
			return err
		}
		if r.State == RunQueued {
			r.State = RunCancelled
		} else {
			r.State = RunCancel
		}
		rec.ActiveRun = &r
		return nil
	}
}

// RunExpire expires a queued run (run spec §5.2, §5.8): terminal, so a box
// that turns up late can never pick it. A picked run is not expired: the
// pickup won.
func RunExpire(id string) func(*Record) error {
	return func(rec *Record) error {
		r, err := runIn(rec, id, RunQueued)
		if err != nil {
			return err
		}
		r.State = RunExpired
		rec.ActiveRun = &r
		return nil
	}
}

// RunFinish marks a picked or cancelling run done: the box after it wrote
// result.json, or the console's sweep for a lost run.
func RunFinish(id string) func(*Record) error {
	return func(rec *Record) error {
		r, err := runIn(rec, id, RunPicked, RunCancel)
		if err != nil {
			return err
		}
		r.State = RunDone
		rec.ActiveRun = &r
		return nil
	}
}

// RunClear removes a terminal run from the record, only while the record
// still names this run (run spec §5.8).
func RunClear(id string) func(*Record) error {
	return func(rec *Record) error {
		if _, err := runIn(rec, id, RunDone, RunCancelled, RunExpired); err != nil {
			return err
		}
		rec.ActiveRun = nil
		return nil
	}
}