package hostops

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"

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

// moveStartAction is the action of StartClaimMoveRecord's record effect,
// the only effect kind ResolveMoveStart settles.
const moveStartAction = "move-start"

// Outcome is ResolveMoveStart's answer for one move-start effect.
type Outcome string

const (
	// OutcomeComplete: the record-start effect applied. The journal retains
	// the move's own outcome, including a later abort or rollback.
	OutcomeComplete Outcome = "complete"
	// OutcomeNotApplied: it does not, and no record Put the start issued
	// can still land.
	OutcomeNotApplied Outcome = "not-applied"
	// OutcomeUncertain: neither is established; the effect stays uncertain
	// and the error says why.
	OutcomeUncertain Outcome = "uncertain"
)

// ErrMoveStartFenceUnproven retains uncertainty when capture is missing,
// the sequence regressed or repeated CAS conflicts prevented a proof.
var ErrMoveStartFenceUnproven = errors.New("hostops: move-start fence is unproven")

// moveStartPuts binds every attempted conditional Put to its read sequence.
// Only the issuing handle holds it. Missing capture cannot prove a fence.
type moveStartPuts struct {
	id, incarnation, key string
	holder               hostclaim.Holder
	mu                   sync.Mutex
	reads                map[string]uint64
	bases                []uint64
}

func (p *moveStartPuts) highestBase() (uint64, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	var high uint64
	for _, seq := range p.bases {
		if seq > high {
			high = seq
		}
	}
	return high, len(p.bases) > 0
}

// moveStartSlot is a StartHandle's move-start Puts. StartClaimMoveRecord
// fills it from inside its issued effect, which runs at most once per
// handle: the handle's registration id is the move id, and so the effect id.
type moveStartSlot struct{ puts atomic.Pointer[moveStartPuts] }

// moveStartRecorder is the store StartClaimMoveRecord gives the record
// writer. It notes the If-Match ETag of every Put of the move's record
// before passing the Put on. Every layer beneath (data boundary, conditional
// record store, raw bucket) either refuses the Put or issues it with that
// same ETag, so the noted set covers every Put that could still land.
type moveStartRecorder struct {
	s       release.Store
	puts    *moveStartPuts
	capture func(context.Context, uint64) error
}

var _ release.Store = moveStartRecorder{}

func (r moveStartRecorder) Get(ctx context.Context, key string) ([]byte, string, error) {
	body, etag, err := r.s.Get(ctx, key)
	if err == nil && key == r.puts.key {
		_, seq, parseErr := recordSequence(body)
		if parseErr != nil {
			return nil, "", parseErr
		}
		if etag == "" {
			return nil, "", ErrIdentity
		}
		r.puts.mu.Lock()
		if r.puts.reads == nil {
			r.puts.reads = map[string]uint64{}
		}
		r.puts.reads[etag] = seq
		r.puts.mu.Unlock()
	}
	return body, etag, err
}

func (r moveStartRecorder) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
	if key == r.puts.key {
		r.puts.mu.Lock()
		seq, ok := r.puts.reads[etag]
		if ok && etag != "" {
			r.puts.bases = append(r.puts.bases, seq)
		}
		r.puts.mu.Unlock()
		if !ok || etag == "" {
			return "", ErrIdentity
		}
		if r.capture != nil {
			if err := r.capture(ctx, seq); err != nil {
				return "", err
			}
		}
	}
	return r.s.Put(ctx, key, body, etag)
}

func (r moveStartRecorder) List(ctx context.Context, prefix string) ([]string, error) {
	return r.s.List(ctx, prefix)
}

func (r moveStartRecorder) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	return r.s.ListDirs(ctx, prefix)
}

func (r moveStartRecorder) Delete(ctx context.Context, key string) error {
	return r.s.Delete(ctx, key)
}

// ResolveMoveStart resolves the exact issuing handle's uncertain move-start
// effect from its exact journal or a newer record sequence that fences every
// attempted Put. Only the runner records settlement; read errors and failed
// fence writes retain uncertainty.
// Previously settled results are returned without issuing any new write.
func ResolveMoveStart(ctx context.Context, h *StartHandle, o hostclaim.Observed, moveID string) (Outcome, error) {
	if h == nil || h.w.ready() != nil || moveID == "" || moveID != h.id || o.Absent || o.Claim == nil || o.Claim.Holder == nil {
		return OutcomeUncertain, ErrIdentity
	}
	c, holder := o.Claim, o.Claim.Holder
	if c.Op.Kind != hostclaim.KindMoving || c.Op.ID != moveID || c.Op.Intent["admission_id"] != h.id || c.Op.Intent["app_id"] != h.appID ||
		holder.Family != "instance" || holder.Account != h.account || holder.App != h.app {
		return OutcomeUncertain, ErrIdentity
	}
	puts := h.moveStart.puts.Load()
	if puts == nil || puts.id != moveID || puts.incarnation != holder.Incarnation || puts.key != holder.Key {
		return OutcomeUncertain, ErrIdentity
	}
	a, err := h.w.read(ctx, h.account, h.app, h.appID)
	if err != nil {
		return OutcomeUncertain, err
	}
	e, ok := a.Value.Effects[moveID]
	if !ok || e.OperationID != h.id || e.AppID != h.appID || e.Owner != h.owner || e.Action != moveStartAction || e.Resource != holder.Key ||
		e.Inputs["move_id"] != moveID || e.Inputs["incarnation"] != holder.Incarnation {
		return OutcomeUncertain, ErrIdentity
	}
	switch e.State {
	case "complete":
		return settledMoveStart(e, moveID)
	case "uncertain":
	default:
		// Prepared or issued: the issuer has not joined and settled its call.
		return OutcomeUncertain, ErrPending
	}
	result, err := decideMoveStart(ctx, h.w, puts, o)
	if err != nil {
		return OutcomeUncertain, err
	}
	if err = h.Runner(h.w.now).resolveUncertain(ctx, e, result); err != nil {
		// A concurrent resolution may have recorded a decision first.
		if a, rerr := h.w.read(ctx, h.account, h.app, h.appID); rerr == nil {
			if cur, ok := a.Value.Effects[moveID]; ok && cur.State == "complete" {
				return settledMoveStart(cur, moveID)
			}
		}
		return OutcomeUncertain, err
	}
	if result == moveID {
		return OutcomeComplete, nil
	}
	return OutcomeNotApplied, nil
}

// settledMoveStart maps a complete move-start effect to its outcome. Any
// other recorded result is not a decision this function made or can read.
func settledMoveStart(e Effect, moveID string) (Outcome, error) {
	switch e.ResultReference {
	case moveID:
		return OutcomeComplete, nil
	case "not-applied":
		return OutcomeNotApplied, nil
	}
	return OutcomeUncertain, ErrIdentity
}

// decideMoveStart never uses ETag inequality as proof. A lost fence answer
// stays uncertain; a later call may prove that fence from its sequence.
func decideMoveStart(ctx context.Context, w *ControlWriter, puts *moveStartPuts, o hostclaim.Observed) (string, error) {
	base, captured := puts.highestBase()
	if !captured {
		return "", ErrMoveStartFenceUnproven
	}
	for attempt := 0; attempt < 3; attempt++ {
		current, err := ReadRecord(ctx, w.raw, puts.key)
		if err != nil {
			return "", err
		}
		if !matches(current, &puts.holder) {
			return "", ErrIdentity
		}
		_, seq, err := recordSequence(current.Body)
		if err != nil {
			return "", err
		}
		rec, err := provision.DecodeRecord(current.Body)
		if err != nil {
			return "", err
		}
		if rec.Move != nil && rec.Move.ID == puts.id && rec.Move.Incarnation == puts.incarnation {
			return puts.id, nil
		}
		// A box may finish the move while the console's start result is
		// uncertain. The terminal journal on this verified incarnation is
		// positive evidence too, even after Move has been cleared.
		if rec.LastMove != nil && rec.LastMove.ID == puts.id {
			return puts.id, nil
		}
		if seq > base {
			return "not-applied", nil
		}
		// A lower counter is a violated migration/write invariant, not authority
		// to repair it and settle. Equal requires a new conditional sequence.
		if seq < base {
			return "", ErrMoveStartFenceUnproven
		}
		if err := CheckCreateFreeze(ctx, w.raw, w.pub); err != nil {
			return "", err
		}
		fresh, err := hostclaim.Read(ctx, w.raw, o.Host)
		if err != nil {
			return "", err
		}
		if fresh.Absent || fresh.ETag != o.ETag {
			return "", hostclaim.ErrConflict
		}
		if _, _, err = provision.WriteRecord(provision.WithRecordSequence(ctx), w.raw, puts.key, current.Body, current.Body, current.ETag); err == nil {
			return "not-applied", nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return "", err
		}
		// The delayed Put may have won: reread its exact journal before deciding.
	}
	return "", ErrMoveStartFenceUnproven
}