// SPDX-License-Identifier: MIT

// Package executionrequest runs the effects of one executionstatus model.
// It owns no authority, provider selection, cross-request cache or retry policy.
package executionrequest

import (
	"context"
	"sync"
	"time"

	model "github.com/mudler/LocalAI/pkg/addons/model/executionstatus"
)

// IO must preserve the admitted binding and observation ID through the owner
// carrier. Read supplies a payload only after strict contract decoding. Validate
// checks current owner authority, selection and runtime identity, never reported
// readiness. Implementations must honor context cancellation.
type IO struct {
	Read     func(context.Context, model.Binding, model.ObservationID, time.Time) (model.Payload, error)
	Validate func(context.Context, model.Binding) error
}

type Timer interface{ Stop() bool }

// Clock preserves monotonic time; fake clocks let tests deliver the deadline
// without sleeps. AfterFunc must schedule callbacks, not invoke them inline.
type Clock interface {
	Now() time.Time
	AfterFunc(time.Duration, func()) Timer
}
type realClock struct{}

func (realClock) Now() time.Time                             { return time.Now() }
func (realClock) AfterFunc(d time.Duration, fn func()) Timer { return time.AfterFunc(d, fn) }

// Request is an explicitly owned observation lifetime. Call Close at the real
// enclosing request/turn boundary. Do not retain it in a service or chat cache.
// Events are serialized under mu; only Reduce decides transitions and effects.
type Request struct {
	mu               sync.Mutex
	ctx              context.Context
	b                model.Binding
	q                model.ObservationID
	io               IO
	clock            Clock
	origin, deadline time.Time
	state            model.State
	trace            []model.Transition
	changed          chan struct{}
	cancel           context.CancelFunc
	timer            Timer
	stopParent       func() bool
}

// New consumes an already authoritative admission, not a frontend selection
// claim. Q is generated by the host caller once per admitted request. A nil
// clock uses real monotonic time. Missing IO fails closed without retry.
func New(ctx context.Context, b model.Binding, q model.ObservationID, eligibility model.Eligibility, io IO, clock Clock) *Request {
	if clock == nil {
		clock = realClock{}
	}
	now := clock.Now()
	deadline := now.Add(5 * time.Second)
	if d, ok := ctx.Deadline(); ok && d.Before(deadline) {
		deadline = d
	}
	r := &Request{ctx: ctx, b: b, q: q, io: io, clock: clock, origin: now, deadline: deadline, state: model.Unbound{}, changed: make(chan struct{})}
	r.mu.Lock()
	r.step(model.Bind{B: b, Q: q, Eligibility: eligibility})
	r.mu.Unlock()
	stop := context.AfterFunc(ctx, r.Close)
	r.mu.Lock()
	r.stopParent = stop
	if _, closed := r.state.(model.Closed); closed {
		stop()
	}
	r.mu.Unlock()
	return r
}
func (r *Request) now() int64 { return r.clock.Now().Sub(r.origin).Nanoseconds() }

// step runs reducer effects while holding mu. External reads run outside mu;
// their completion feeds an event, including after cancellation or withdrawal.
func (r *Request) step(e model.Event) {
	tr := model.Reduce(r.state, e)
	r.state = tr.State
	// Copy slices so Trace callers cannot alter the shell's history.
	tr.Effects = append([]model.Effect(nil), tr.Effects...)
	r.trace = append(r.trace, tr)
	for _, effect := range tr.Effects {
		switch effect.(type) {
		case model.Read:
			ctx, cancel := context.WithCancel(r.ctx)
			r.cancel = cancel
			r.timer = r.clock.AfterFunc(r.deadline.Sub(r.clock.Now()), func() { r.mu.Lock(); defer r.mu.Unlock(); r.step(model.Tick{Now: r.now()}) })
			go r.read(ctx, cancel)
		case model.Cancel:
			if r.cancel != nil {
				r.cancel()
			}
		case model.Publish:
			close(r.changed)
			r.changed = make(chan struct{})
		}
	}
	if _, pending := r.state.(model.Pending); !pending && r.timer != nil {
		r.timer.Stop()
		r.timer = nil
	}
}
func (r *Request) read(ctx context.Context, cancel context.CancelFunc) {
	defer cancel()
	payload := model.InvalidPayload()
	var err error
	if r.io.Read != nil {
		payload, err = r.io.Read(ctx, r.b, r.q, r.deadline)
	}
	// The carrier's return is not freshness proof. Check before any positive
	// publication; the consuming caller checks again before using it.
	fresh := r.io.Validate != nil && r.io.Validate(ctx, r.b) == nil
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.ctx.Err() != nil {
		r.step(model.Close{})
	}
	if !fresh {
		r.step(model.Invalidate{Cause: "stale-binding"})
	}
	if err != nil {
		r.step(model.Failure{B: r.b, Q: r.q, Cause: "read-failed", Now: r.now()})
		return
	}
	r.step(model.Reply{B: r.b, Q: r.q, Payload: payload, Now: r.now()})
}

// Observe coalesces consumers of the same request. A consumer's cancellation
// stops only its wait; parent cancellation closes the entire observation.
// A settled error never causes another read: the reducer enforces R7.
func (r *Request) Observe(ctx context.Context) model.Value {
	r.mu.Lock()
	if r.ctx.Err() != nil {
		r.step(model.Close{})
	}
	r.step(model.Probe{B: r.b, Q: r.q, Now: r.now(), Deadline: r.deadline.Sub(r.origin).Nanoseconds()})
	for {
		if ctx.Err() != nil {
			r.mu.Unlock()
			return model.Unknown()
		}
		if _, pending := r.state.(model.Pending); !pending {
			if _, observed := r.state.(model.Observed); !observed {
				v := model.Project(r.state)
				r.mu.Unlock()
				return v
			}
			r.mu.Unlock()
			fresh := r.io.Validate != nil && r.io.Validate(ctx, r.b) == nil
			r.mu.Lock()
			if r.ctx.Err() != nil {
				r.step(model.Close{})
			}
			// A canceled consumer cannot establish freshness failure for peers.
			// Parent cancellation above still closes the shared observation.
			if ctx.Err() != nil {
				r.mu.Unlock()
				return model.Unknown()
			}
			if !fresh {
				r.step(model.Invalidate{Cause: "stale-binding"})
			}
			v := model.Project(r.state)
			if ctx.Err() != nil {
				v = model.Unknown()
			}
			r.mu.Unlock()
			return v
		}
		changed := r.changed
		r.mu.Unlock()
		select {
		case <-changed:
		case <-ctx.Done():
			return model.Unknown()
		}
		r.mu.Lock()
	}
}

func (r *Request) Invalidate(cause model.Cause) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.step(model.Invalidate{Cause: cause})
}
func (r *Request) Close() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.step(model.Close{})
	if r.stopParent != nil {
		r.stopParent()
	}
}

// Trace returns detached reducer transitions for model/runtime correspondence
// checks. It includes ignored events, never credentials or wire payloads.
func (r *Request) Trace() []model.Transition {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := append([]model.Transition(nil), r.trace...)
	for i := range out {
		out[i].Effects = append([]model.Effect(nil), out[i].Effects...)
	}
	return out
}

// Binding returns the immutable admitted identity for presentation only.
func (r *Request) Binding() model.Binding { return r.b }