package buildstep

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

	"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 module's steps state (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 Put, so a pause outlives
	// a restart.
	Load func(key string) ([]byte, error)

	mu      sync.Mutex
	inHand  int
	running map[string]RunningAsk
	paused  bool
	wake    chan struct{}
	once    sync.Once
}

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
		}
		if _, err := p.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 {
	capacity := 0
	if p.Capacity != nil {
		capacity = p.Capacity()
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	return !p.paused && p.inHand < capacity
}

// 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 {
		return err // pulling on a pause that cannot be read could build on a node paused on purpose
	}
	wait := p.PullWait
	if wait <= 0 {
		wait = DefaultPullWait
	}
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		if !p.room() {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-p.wake:
			}
			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.
func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
	for _, kind := range delivery.AskKinds {
		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)
	return withKind(msg, delivery.AskDelivery), err
}

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
	}
	step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID)
	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)
		_ = 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()
	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()
				}
			}
		}
	}()
	result, err := protect(ctx, func(ctx context.Context) (any, error) { return p.Run(ctx, ask) })
	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
	}
	finish(p.Put, key, step, result, err)
	if aerr := msg.Ack(); aerr != nil {
		p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
	}
}