package buildstep

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// **The pull loop the build modules share** (novox/hq ADR 0525 §4-§6, ADR 0536 §4; issue 1887). A build module on a node
// whose build capacity is above zero pulls the asks of its language's seat: from the delivery worker first, then check,
// then recheck, holding at most its capacity at once. Each ask runs as a step under the ask's own id, so its end is the
// key <node>.<ask id> in the module's steps state, which the asker watches; the ask is acknowledged when that end is
// written, kept alive while it runs by an in-progress signal, and redelivered by the stream to another node if this one
// stops. No part keeps a timer for an ask: the pull's own wait and the stream's acknowledgement wait are the bus's.

// Pulled is one ask a source handed over, with what decides its fate on the bus.
type Pulled struct {
	// Kind is the worker it came from, filled by the puller.
	Kind string
	Data []byte
	// Delivered is how many times the bus has handed it over, this time included.
	Delivered uint64
	// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
	// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
	Ack, InProgress, Nak, Term func() error
}

// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
// (FakeAsks).
type AskSource interface {
	// Next takes one ask of kind: with wait 0 only one already waiting, else waiting for one at most wait. It answers
	// nil and no error when none came.
	Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error)
}

// Defaults of a Puller.
var (
	// DefaultPullWait is how long one pull of the delivery worker waits for an ask while every worker is empty: the
	// pull's own wait (as process.Consume's), after which the check and recheck workers are looked at again.
	DefaultPullWait = 5 * time.Second
	// DefaultHeartbeat is how often a running ask tells the bus it is still being worked; under the seat worker's
	// acknowledgement wait of 60 s (mesh-controller internal/broker HolderConsumerFor).
	DefaultHeartbeat = 20 * time.Second
)

// Puller pulls a build seat's asks and runs them.
type Puller struct {
	Source   AskSource
	Language string // the seat's language: delivery.AskGo, AskNode or AskImage
	Node     string // the node this holder runs on, the first part of its step keys
	Put      Put    // writes the module's steps state
	// Capacity is how many asks the node runs at once; read before every pull, so a change takes effect at the next
	// one. 0 pulls nothing.
	Capacity func() int
	// Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result.
	Run func(ctx context.Context, ask delivery.StepAsk) (any, error)
	// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat.
	PullWait  time.Duration
	Heartbeat time.Duration
	Logf      func(format string, args ...any)
	// Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull
	// reads the node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a
	// pause outlives a restart.
	Load func(key string) ([]byte, error)
	// PausePut writes the node's pause key: the module's pauses state (delivery.PausesState), which has no age, so a
	// pause never ends by itself while the steps state drops its old keys. Put when nil.
	PausePut Put
	// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
	// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
	// double. Until it hands a value the puller judges by Capacity.
	WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
	// CoresPerStep is what one ask of this module needs, read at every judgement (NodeCoresPerStep); with it the capacity
	// judged by is bounded by the node's build slice: min(the configured capacity, floor(the quota's cores / CoresPerStep))
	// (EffectiveCapacity). Nil bounds nothing by the CPU. Quota reads the quota (Quota when nil; a test's own).
	CoresPerStep func() float64
	Quota        func() (percent int, from string)
	// Has says whether this node has a need an ask names (delivery.AskNeeds, issue 890). An ask needing what it lacks is
	// declined: its step ends refused as declined (delivery.StepDeclined), the message is dropped for its asker to put
	// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
	// does not take the asker's next copy at once. Nil has nothing.
	Has func(need string) bool

	mu      sync.Mutex
	inHand  int
	running map[string]RunningAsk
	paused  bool
	wake    chan struct{}
	once    sync.Once
	// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
	watched    int
	hasWatched bool
	// judged is the capacity said last (log and capacity.<node>), so a change is said once.
	judged     delivery.Capacity
	hasJudged  bool
	capWritten bool
	capFault   string
	// declined are the kinds whose worker handed an ask this node declined, left out of the pulls without a wait until
	// the next pull's wait ends.
	declined map[string]bool
}

func (p *Puller) init() {
	p.once.Do(func() { p.wake = make(chan struct{}, 1) })
}

func (p *Puller) logf(format string, args ...any) {
	if p.Logf != nil {
		p.Logf(format, args...)
	}
}

// Wake makes a waiting puller look again: after a capacity change, a resume, or a step's end.
func (p *Puller) Wake() {
	p.init()
	select {
	case p.wake <- struct{}{}:
	default:
	}
}

// Pause stops the puller taking asks; the asks it holds run to their end. by names who paused it. With Load set the
// pause is written on the node's pause key first, and an error there leaves the puller as it was.
func (p *Puller) Pause(by string) error { return p.setPaused(true, by) }

// Resume lets the puller take asks again, written as Pause's is.
func (p *Puller) Resume(by string) error { return p.setPaused(false, by) }

func (p *Puller) setPaused(paused bool, by string) error {
	if p.Load != nil {
		key, err := delivery.PauseKey(p.Node)
		if err != nil {
			return err
		}
		raw, err := delivery.EncodePause(delivery.Pause{Paused: paused, By: by, At: now()})
		if err != nil {
			return err
		}
		put := p.PausePut
		if put == nil {
			put = p.Put
		}
		if _, err := put(key, json.RawMessage(raw)); err != nil {
			return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err)
		}
	}
	p.mu.Lock()
	p.paused = paused
	p.mu.Unlock()
	p.Wake()
	return nil
}

// restore reads the node's pause key, when Load is set.
func (p *Puller) restore() error {
	if p.Load == nil {
		return nil
	}
	key, err := delivery.PauseKey(p.Node)
	if err != nil {
		return err
	}
	raw, err := p.Load(key)
	if err != nil {
		return fmt.Errorf("buildstep: reading the pause of %s: %w", p.Node, err)
	}
	if raw == nil {
		return nil
	}
	v, err := delivery.DecodePause(raw)
	if err != nil {
		return fmt.Errorf("buildstep: the pause of %s: %w", p.Node, err)
	}
	p.mu.Lock()
	p.paused = v.Paused
	p.mu.Unlock()
	return nil
}

// Paused says whether the puller takes no asks.
func (p *Puller) Paused() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.paused
}

// InHand is how many asks run now.
func (p *Puller) InHand() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.inHand
}

// RunningAsk is an ask a puller runs now, as its seat's current verb says it.
type RunningAsk struct {
	ID      string    `json:"id"`
	Kind    string    `json:"kind"`
	Module  string    `json:"module,omitempty"`
	Started time.Time `json:"started"`
}

// Running are the asks the puller runs now, oldest first.
func (p *Puller) Running() []RunningAsk {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make([]RunningAsk, 0, len(p.running))
	for _, r := range p.running {
		out = append(out, r)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
	return out
}

// room says whether the puller may take one more ask now.
func (p *Puller) room() bool {
	ok, _, _ := p.roomNow()
	return ok
}

// roomNow is room, with the capacity and the asks in hand it was judged on.
func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
	capacity = p.CapacityDetail().Effective
	p.mu.Lock()
	defer p.mu.Unlock()
	return !p.paused && p.inHand < capacity, capacity, p.inHand
}

// CapacityNow is the capacity the puller judges by (CapacityDetail's Effective).
func (p *Puller) CapacityNow() int {
	return p.CapacityDetail().Effective
}

// CapacityDetail is the capacity the puller judges by now and why: the configured one (the config watch's last value,
// else Capacity), bounded by the node's build slice when CoresPerStep is set (EffectiveCapacity), its quota read now. A
// change from what it said last is logged and written to `capacity.<node>` of the pauses state (PausePut), so the
// delivery view says it too.
func (p *Puller) CapacityDetail() delivery.Capacity {
	p.mu.Lock()
	watched, has := p.watched, p.hasWatched
	p.mu.Unlock()
	configured, from := 0, "the settings file"
	switch {
	case has:
		configured, from = watched, "mesh-config, as the config watch handed it"
	case p.Capacity != nil:
		configured = p.Capacity()
	}
	var c delivery.Capacity
	if p.CoresPerStep == nil {
		c = delivery.Capacity{Configured: configured, Effective: configured, QuotaFrom: "not judged: the module names no cores a step needs",
			Why: fmt.Sprintf("its configured capacity %d", configured)}
	} else {
		quota := p.Quota
		if quota == nil {
			quota = Quota
		}
		percent, qfrom := quota()
		c = EffectiveCapacity(configured, percent, qfrom, p.CoresPerStep())
	}
	c.ConfiguredFrom = from
	p.sayCapacity(c)
	return c
}

// sayCapacity logs c when it differs from what was said last, and writes it to the node's capacity key until a write
// lands (a failed write is tried again at the next judgement, its error said once).
func (p *Puller) sayCapacity(c delivery.Capacity) {
	p.mu.Lock()
	same := p.hasJudged && p.judged.Effective == c.Effective && p.judged.Configured == c.Configured && p.judged.ByCPU == c.ByCPU &&
		p.judged.CoresPerStep == c.CoresPerStep && p.judged.QuotaCores == c.QuotaCores
	if !same {
		p.judged, p.hasJudged, p.capWritten = c, true, false
	}
	write := !p.capWritten && p.PausePut != nil
	p.mu.Unlock()
	if !same {
		p.logf("the %s build capacity of %s is %s (configured from %s)", p.Language, p.Node, c.Why, c.ConfiguredFrom)
	}
	if !write {
		return
	}
	key, err := delivery.CapacityKey(p.Node)
	if err != nil {
		return
	}
	c.At = now()
	raw, err := delivery.EncodeCapacity(c)
	if err == nil {
		_, err = p.PausePut(key, json.RawMessage(raw))
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if err == nil {
		p.capWritten, p.capFault = true, ""
		return
	}
	if err.Error() != p.capFault {
		p.capFault = err.Error()
		p.logf("the %s build capacity of %s could not be written to %s (tried again at the next judgement): %v", p.Language, p.Node, key, err)
	}
}

// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
// delivers it again, to another node or to this one restarted.
func (p *Puller) Pull(ctx context.Context) error {
	p.init()
	if p.Source == nil || p.Put == nil || p.Run == nil {
		return errors.New("buildstep: a puller needs its source, its steps' put and its run")
	}
	if err := p.restore(); err != nil {
		// An unreadable pause degrades to not paused, said once: a puller that stops silently is the worse failure, and a
		// node kept from building is held by its capacity (0), not by a pause.
		p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
	}
	go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
	wait := p.PullWait
	if wait <= 0 {
		wait = DefaultPullWait
	}
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		if ok, capacity, inHand := p.roomNow(); !ok {
			// Parked until something wakes it: a capacity change (the config watch), a resume, or a step's end. Every
			// park logs once, so a puller that sits idle says why (issue 1887).
			p.logf("the %s puller on %s parks: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-p.wake:
			}
			_, capacity, inHand = p.roomNow()
			p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
			continue
		}
		msg, err := p.next(ctx, wait)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			p.logf("the %s asks cannot be pulled: %v", p.Language, err)
			select { // the pull's own wait before the next try, as an empty pull's
			case <-ctx.Done():
				return ctx.Err()
			case <-p.wake:
			case <-time.After(wait):
			}
			continue
		}
		if msg == nil {
			continue
		}
		p.take(ctx, msg)
	}
}

// next is one ask from the workers in their order: one already waiting on delivery, check or recheck, and when none
// waits, a pull of the delivery worker that waits for one.
//
// A kind whose worker handed an ask this node declined is left out until that wait ended (issue 890): the pull's own
// wait, no timer of its own.
func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
	p.mu.Lock()
	declined := p.declined
	p.mu.Unlock()
	for _, kind := range delivery.AskKinds {
		if declined[kind] {
			continue
		}
		msg, err := p.Source.Next(ctx, kind, 0)
		if err != nil || msg != nil {
			return withKind(msg, kind), err
		}
	}
	msg, err := p.Source.Next(ctx, delivery.AskDelivery, wait)
	p.mu.Lock()
	p.declined = nil
	p.mu.Unlock()
	return withKind(msg, delivery.AskDelivery), err
}

// decline ends the step of an ask this node lacks a need of, refused as declined with what it lacks, and drops the
// message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed
// back, as one whose start cannot be.
func (p *Puller) decline(msg *Pulled, ask delivery.StepAsk, lacks []string) {
	step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask))
	if err == nil {
		why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID,
			strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", "))
		step.Ended = now().UTC()
		step.State, step.Reason = delivery.StepRefused, why.Error()
		step.Result, err = json.Marshal(delivery.StepDeclined{Declined: true, Needs: ask.Needs, Lacks: lacks})
		if err == nil {
			err = writeEnd(p.Put, key, step)
		}
	}
	if err != nil {
		p.logf("the ask %s cannot be declined on %s, so it is given back: %v", ask.ID, p.Node, err)
		_ = msg.Nak()
		return
	}
	p.logf("the ask %s is declined on %s: it needs %s, and this node lacks %s", ask.ID, p.Node, strings.Join(ask.Needs, ", "), strings.Join(lacks, ", "))
	_ = msg.Term()
	p.mu.Lock()
	if p.declined == nil {
		p.declined = map[string]bool{}
	}
	p.declined[msg.Kind] = true
	p.mu.Unlock()
}

func withKind(msg *Pulled, kind string) *Pulled {
	if msg != nil {
		msg.Kind = kind
	}
	return msg
}

// take starts one ask: it decodes it, refuses one that is not of this seat, and runs it as a step under its own id.
func (p *Puller) take(ctx context.Context, msg *Pulled) {
	kind := msg.Kind
	ask, err := delivery.DecodeStepAsk(msg.Data)
	if err == nil && (ask.Language != p.Language || (kind != "" && ask.Kind != kind)) {
		err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
	}
	if err != nil {
		// An ask that is not one runs nowhere: delivering it again would only fail again.
		p.logf("an ask dropped: %v", err)
		_ = msg.Term()
		return
	}
	if lacks := delivery.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
		p.decline(msg, ask, lacks)
		return
	}
	step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask))
	if err != nil {
		p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
		_ = msg.Nak()
		return
	}
	if msg.Delivered > delivery.AskRuns {
		// Run AskRuns times and never ended: each run took its node down or died with it. It is run no more, and said
		// (issue 1907): the asker's key fails with why.
		why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1)
		p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why)
		_ = finish(p.Put, key, step, nil, why, p.Logf) // run no more either way: the asker's bound says it then
		_ = msg.Term()
		return
	}
	module := ""
	if ask.Build != nil {
		module = ask.Build.Module
	}
	p.mu.Lock()
	p.inHand++
	if p.running == nil {
		p.running = map[string]RunningAsk{}
	}
	p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started}
	p.mu.Unlock()
	p.logf("the %s ask %s starts on %s: %s", ask.Kind, ask.ID, p.Node, step.What)
	go p.run(ctx, msg, ask, key, step)
}

func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key string, step delivery.Step) {
	defer func() {
		p.mu.Lock()
		p.inHand--
		delete(p.running, ask.ID)
		p.mu.Unlock()
		p.Wake()
	}()
	beat := p.Heartbeat
	if beat <= 0 {
		beat = DefaultHeartbeat
	}
	done := make(chan struct{})
	go func() {
		t := time.NewTicker(beat)
		defer t.Stop()
		for {
			select {
			case <-done:
				return
			case <-t.C:
				if msg.InProgress != nil {
					_ = msg.InProgress()
				}
			}
		}
	}()
	ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)}
	result, err := protect(ctx, func(ctx context.Context) (any, error) {
		return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
	})
	step = ph.end() // its end keeps the phase it ended in
	close(done)
	if ctx.Err() != nil {
		// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
		return
	}
	if ferr := finish(p.Put, key, step, result, err, p.Logf); ferr != nil {
		// No end was written: the ask is not acknowledged as done, so the stream gives it again and a run that can
		// write its end (or the poison rule, issue 1907) ends it where its asker sees it (issue 1930).
		p.logf("the ask %s ran on %s and no end of it could be written, so it is given back: %v", ask.ID, p.Node, ferr)
		_ = msg.Nak()
		return
	}
	outcome := "passed"
	if err != nil {
		outcome = "did not pass: " + err.Error()
		if len(outcome) > 300 {
			outcome = outcome[:300] + "..."
		}
	}
	p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node,
		now().Sub(step.Started).Round(time.Second), step.Phase, outcome)
	if aerr := msg.Ack(); aerr != nil {
		p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
	}
}

// watchCapacity follows the module's config value `capacity` when WatchConfig is set (issue 1887, 2026-10-25: g14's
// puller started at capacity 0 sat parked with capacity 2 set, because nothing woke it to read its settings file
// again). The value comes as mesh-config resolved it for this node, the event of the change itself: the runtime hands
// the value now and then each change its machine's provisioner client says moved, so nothing polls, and the puller does
// not wait for the node-engine to fill its settings file again. Each value handed is kept, used in place of Capacity,
// and wakes the puller. A watch that cannot start leaves the puller on Capacity, said once.
func (p *Puller) watchCapacity() {
	if p.WatchConfig == nil {
		return
	}
	err := p.WatchConfig(CapacitySetting, func(c stdio.ProvisionChange) error {
		n, why := p.capacityFrom(c)
		p.mu.Lock()
		before, had := p.watched, p.hasWatched
		p.watched, p.hasWatched = n, true
		p.mu.Unlock()
		switch {
		case why != "":
			p.logf("the %s puller on %s pulls nothing: %s", p.Language, p.Node, why)
		case !had || before != n:
			p.logf("the %s build capacity of %s is %d, as mesh-config resolved it", p.Language, p.Node, n)
		}
		p.Wake()
		return nil
	})
	if err != nil {
		p.logf("the %s build capacity of %s cannot be followed on the bus, so it is read from the settings file when the "+
			"puller wakes: %v", p.Language, p.Node, err)
	}
}

// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
// pull before it serves, and a watch asked before that would be refused.
func WatchConfigServed(key string, handle func(stdio.ProvisionChange) error) error {
	<-stdio.Served()
	return stdio.WatchConfig(key, handle)
}

// capacityFrom is the node's capacity in one value the watch handed, and why it is 0 when it cannot be read.
func (p *Puller) capacityFrom(c stdio.ProvisionChange) (int, string) {
	if c.Refused != "" {
		return 0, "its capacity is no longer answered: " + c.Refused
	}
	setting := ""
	if len(c.Value) > 0 && string(c.Value) != "null" {
		if err := json.Unmarshal(c.Value, &setting); err != nil {
			return 0, fmt.Sprintf("its capacity %s is not a string", c.Value)
		}
	}
	n, err := capacityOf(setting, p.Node)
	if err != nil {
		return 0, err.Error()
	}
	return n, ""
}