package buildrun

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

	"github.com/nats-io/nats.go/jetstream"

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

// **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask
// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node
// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under
// mesh.events.dead.<seat stream>.<worker>, with the ask's message id (its ask id), how often it was handed out and when it
// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays
// running and holds a try place, a check's verdict stays running and its pull request waits.
//
// An asker of the build seats (the checker, the builder) hears those copies (DeadLetterHearing: an ordered consumer on
// DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose
// DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its
// reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the
// builder asks its part of a build again).
//
// Belt and braces, where the asker says so in the context (NotTakenBound; the checker's tries): an ask no build node took
// within a bound ends the same way, saying NotTakenWords. An ask waits in the stream as long as the backlog is otherwise
// (ADR 0525 §9): a backlog is no fault.

// LostNode is the node a step ended for a lost ask names: no build node wrote it.
const LostNode = "bus"

// The words of a step ended for its ask.
const (
	LostWords     = "its ask was given up by the build seat"
	NotTakenWords = "no build node took its ask"
)

// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link
// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up.
const (
	DeadHeaderMsgID      = "Mesh-Dead-Msg-Id"
	DeadHeaderDeliveries = "Mesh-Dead-Deliveries"
	DeadHeaderGaveUp     = "Mesh-Dead-Gave-Up"
	DeadHeaderStream     = "Mesh-Dead-Stream"
)

// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat
// stream's copies there (buscompose DeadLetterSubject).
const DeadLettersStream = "DEAD_LETTERS"

// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under.
func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" }

// SeatStream is the stream of a build seat's language (buscompose seatStreamName of node-build-<language>).
func SeatStream(language string) string { return "SEAT_NODE_BUILD_" + strings.ToUpper(language) }

// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it.
type GivenUp struct {
	AskID      string
	Stream     string
	Deliveries int
	At         time.Time
}

// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id.
func GivenUpOf(header map[string][]string) (GivenUp, bool) {
	get := func(k string) string {
		if v := header[k]; len(v) > 0 {
			return strings.TrimSpace(v[0])
		}
		return ""
	}
	g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)}
	if g.AskID == "" {
		return GivenUp{}, false
	}
	g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries))
	g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp))
	return g, true
}

// Words are what a step ended for the given-up ask says.
func (g GivenUp) Words() string {
	at := "at an unknown time"
	if !g.At.IsZero() {
		at = "at " + g.At.UTC().Format(time.RFC3339)
	}
	return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream)
}

// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard.
const keptGivenUp = 1024

// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use.
type DeadLetters struct {
	mu       sync.Mutex
	heard    map[string]GivenUp
	order    []string
	watchers map[string][]chan GivenUp
}

// Heard takes one given-up ask in.
func (d *DeadLetters) Heard(g GivenUp) {
	d.mu.Lock()
	if d.heard == nil {
		d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{}
	}
	if _, ok := d.heard[g.AskID]; !ok {
		d.order = append(d.order, g.AskID)
		if len(d.order) > keptGivenUp {
			delete(d.heard, d.order[0])
			d.order = d.order[1:]
		}
	}
	d.heard[g.AskID] = g
	ws := d.watchers[g.AskID]
	delete(d.watchers, g.AskID)
	d.mu.Unlock()
	for _, w := range ws {
		w <- g // buffered, one each
	}
}

// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel.
func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) {
	ch := make(chan GivenUp, 1)
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.heard == nil {
		d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{}
	}
	if g, ok := d.heard[askID]; ok {
		ch <- g
		return ch, func() {}
	}
	d.watchers[askID] = append(d.watchers[askID], ch)
	return ch, func() {
		d.mu.Lock()
		defer d.mu.Unlock()
		ws := d.watchers[askID]
		for i, w := range ws {
			if w == ch {
				d.watchers[askID] = append(ws[:i], ws[i+1:]...)
				break
			}
		}
		if len(d.watchers[askID]) == 0 {
			delete(d.watchers, askID)
		}
	}
}

type notTakenKey struct{}

// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's).
func NotTakenBound(ctx context.Context, bound time.Duration) context.Context {
	return context.WithValue(ctx, notTakenKey{}, bound)
}

func notTakenBoundOf(ctx context.Context) time.Duration {
	d, _ := ctx.Value(notTakenKey{}).(time.Duration)
	return d
}

// LostAskWaiter is an AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under
// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on
// the node "bus". Every step the build module writes is handed on as it is.
type LostAskWaiter struct {
	Inner AskWaiter
	Dead  *DeadLetters
	// Hearing, when set, is asked to hear DEAD_LETTERS again each time an ask is waited for while it cannot: an ask put is
	// the event that makes hearing matter.
	Hearing interface {
		Deaf() string
		Again()
	}
	Now func() time.Time
}

func (w LostAskWaiter) now() time.Time {
	if w.Now != nil {
		return w.Now()
	}
	return time.Now()
}

// WatchAsk is AskWaiter's.
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
	if w.Hearing != nil && w.Hearing.Deaf() != "" {
		w.Hearing.Again()
	}
	inner, err := w.Inner.WatchAsk(ctx, module, askID)
	if err != nil {
		return nil, err
	}
	var dead <-chan GivenUp
	cancel := func() {}
	if w.Dead != nil {
		dead, cancel = w.Dead.watch(askID)
	}
	// The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask.
	var notTaken <-chan struct{}
	bound := notTakenBoundOf(ctx)
	stopBound := func() {}
	if bound > 0 {
		var nt context.Context
		nt, stopBound = context.WithTimeout(context.Background(), bound)
		notTaken = nt.Done()
	}
	out := make(chan delivery.Step)
	go func() {
		defer close(out)
		defer cancel()
		defer func() { stopBound() }()
		send := func(s delivery.Step) bool {
			select {
			case out <- s:
				return true
			case <-ctx.Done():
				return false
			}
		}
		ended := func(reason string) delivery.Step {
			at := w.now().UTC()
			return delivery.Step{ID: askID, Node: LostNode, State: delivery.StepFailed, Reason: reason, Ended: at}
		}
		for {
			select {
			case <-ctx.Done():
				return
			case s, ok := <-inner:
				if !ok {
					return
				}
				if notTaken != nil {
					stopBound() // a node took it: its step is bounded by the asker from its start
					notTaken = nil
				}
				if !send(s) {
					return
				}
			case g := <-dead:
				dead = nil
				send(ended(g.Words()))
				return
			case <-notTaken:
				send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound)))
				return
			}
		}
	}()
	return out, nil
}

// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords).
func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) }

// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound.
func IsNotTaken(s delivery.Step) bool {
	return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords)
}

// **The asker hears DEAD_LETTERS, tries again when it cannot, and says whether it hears** (issue 734). The read is
// granted with the asker's seat traffic (buscompose DeadLetterReads); an asker started before its grant was composed is
// refused, and an asker that only tried once would stay deaf until it starts again, silently. DeadLetterHearing tries in
// the background, never holding the start up, and again on an event (Again): the asker's reconnect to the bus, and every
// ask waited for while it is deaf (LostAskWaiter.Hearing); no timer. Each change of whether it hears is said (Said), so
// the module's health tool says it: a deaf asker is not healthy.

// Hearing is what a DeadLetterHearing says: whether it hears every build seat stream it follows, and why not.
type Hearing struct {
	Heard   bool      `json:"heard"`
	Streams []string  `json:"streams"`
	Why     string    `json:"why,omitempty"`
	At      time.Time `json:"at"`
}

// HearingKey is the key a module keeps its Hearing under, in its own state (the checker's and the builder's hearing).
const HearingKey = "dead-letters"

// DeadLetterFollower starts following one seat stream's copies in DEAD_LETTERS from since, handing each given-up ask to
// heard; it answers the stop, or why it cannot.
type DeadLetterFollower interface {
	Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error)
}

// JetStreamDeadLetters is the bus side of a DeadLetterFollower: an ordered consumer on DEAD_LETTERS filtered to the
// stream's copies, which resumes where it was after a reconnect and, reset before it heard any, starts at since again.
type JetStreamDeadLetters struct{ JS jetstream.JetStream }

// Follow is DeadLetterFollower's.
func (j JetStreamDeadLetters) Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) {
	cons, err := j.JS.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{DeadLetterFilter(stream)}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since})
	if err != nil {
		return nil, err
	}
	cc, err := cons.Consume(func(m jetstream.Msg) {
		if g, ok := GivenUpOf(m.Headers()); ok {
			heard(g)
		}
	})
	if err != nil {
		return nil, err
	}
	return cc.Stop, nil
}

// DeadLetterHearing follows the build seats' given-up asks into Dead. Make it with NewDeadLetterHearing.
type DeadLetterHearing struct {
	follower DeadLetterFollower
	streams  []string
	dead     *DeadLetters
	logf     func(string, ...any)
	said     func(Hearing)
	ctx      context.Context
	since    time.Time

	mu     sync.Mutex
	stops  map[string]func()
	why    map[string]string
	trying bool
	again  bool
	told   *Hearing
	wg     sync.WaitGroup
}

// NewDeadLetterHearing starts hearing the given-up asks of each language's build seat stream, in the background. said
// is told each change of whether it hears (nil tells nothing); ctx ends it.
func NewDeadLetterHearing(ctx context.Context, follower DeadLetterFollower, languages []string, dead *DeadLetters,
	logf func(string, ...any), said func(Hearing)) *DeadLetterHearing {
	h := &DeadLetterHearing{follower: follower, dead: dead, logf: logf, said: said, ctx: ctx, since: time.Now(),
		stops: map[string]func(){}, why: map[string]string{}}
	for _, l := range languages {
		s := SeatStream(l)
		h.streams = append(h.streams, s)
		h.why[s] = "not tried yet"
	}
	if h.logf == nil {
		h.logf = func(string, ...any) {}
	}
	h.Again()
	return h
}

// Deaf says why a stream is not heard, or "" when every one is.
func (h *DeadLetterHearing) Deaf() string {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.deafLocked()
}

func (h *DeadLetterHearing) deafLocked() string {
	var out []string
	for _, s := range h.streams {
		if why, deaf := h.why[s]; deaf {
			out = append(out, s+": "+why)
		}
	}
	if len(out) == 0 {
		return ""
	}
	return DeadLettersStream + " is not heard, so an ask a build seat gives up on is waited for until its bound: " +
		strings.Join(out, "; ")
}

// Again tries, in the background, every stream not heard yet; a call while it tries makes it try once more after.
func (h *DeadLetterHearing) Again() {
	h.mu.Lock()
	if h.trying {
		h.again = true
		h.mu.Unlock()
		return
	}
	h.trying = true
	h.mu.Unlock()
	h.wg.Add(1)
	go func() {
		defer h.wg.Done()
		for {
			h.try()
			h.mu.Lock()
			if !h.again || h.ctx.Err() != nil {
				h.trying = false
				h.mu.Unlock()
				return
			}
			h.again = false
			h.mu.Unlock()
		}
	}()
}

// Wait returns when no try runs.
func (h *DeadLetterHearing) Wait() { h.wg.Wait() }

// Stop ends every stream's following.
func (h *DeadLetterHearing) Stop() {
	h.mu.Lock()
	defer h.mu.Unlock()
	for s, stop := range h.stops {
		stop()
		delete(h.stops, s)
	}
}

func (h *DeadLetterHearing) try() {
	for _, s := range h.streams {
		h.mu.Lock()
		_, deaf := h.why[s]
		h.mu.Unlock()
		if !deaf || h.ctx.Err() != nil {
			continue
		}
		stream := s
		stop, err := h.follower.Follow(h.ctx, stream, h.since, func(g GivenUp) {
			h.logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream,
				g.Deliveries, g.At.Format(time.RFC3339))
			h.dead.Heard(g)
		})
		h.mu.Lock()
		if err != nil {
			h.why[stream] = err.Error()
			h.logf("the asks %s gives up on are not heard (%s of %s); tried again at the next reconnect or ask: %v",
				stream, DeadLetterFilter(stream), DeadLettersStream, err)
		} else {
			delete(h.why, stream)
			h.stops[stream] = stop
			h.logf("the asks %s gives up on are heard from %s", stream, DeadLettersStream)
		}
		h.mu.Unlock()
	}
	h.tell()
}

// tell says the hearing when it changed.
func (h *DeadLetterHearing) tell() {
	h.mu.Lock()
	why := h.deafLocked()
	now := Hearing{Heard: why == "", Streams: append([]string(nil), h.streams...), Why: why, At: time.Now().UTC()}
	same := h.told != nil && h.told.Heard == now.Heard && h.told.Why == now.Why
	if !same {
		h.told = &now
	}
	h.mu.Unlock()
	if !same && h.said != nil {
		h.said(now)
	}
}

// HearingState is the name of the state an asker keeps its Hearing in (`<module>_hearing`, key HearingKey), which its
// health tool reads.
const HearingState = "hearing"

// HearingKeeper is where a Hearing is kept: a jetstream.KeyValue, or a process.AwaitedBucket for a state that may not be
// on the bus at the part's start (issue 1790), which keeps the latest until it appears and answers process.ErrNotYet.
type HearingKeeper interface {
	Put(ctx context.Context, key string, value []byte) (uint64, error)
}

// SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and
// says so once. A write kv keeps for later (process.ErrNotYet) is not an error: the awaited bucket said it waits.
func SayHearing(ctx context.Context, kv HearingKeeper, logf func(string, ...any)) func(Hearing) {
	if kv == nil {
		if logf != nil {
			logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState)
		}
		return nil
	}
	return func(h Hearing) {
		data, err := json.Marshal(h)
		if err == nil {
			_, err = kv.Put(ctx, HearingKey, data)
		}
		if err != nil && !errors.Is(err, process.ErrNotYet) && logf != nil {
			logf("whether %s is heard could not be kept: %v", DeadLettersStream, err)
		}
	}
}

// HearingHealth is the health answer of an asker's kept Hearing (raw, empty when none is kept): healthy while it hears
// DEAD_LETTERS, unhealthy saying why while it does not or has not said.
func HearingHealth(raw []byte) health.Answer {
	if len(raw) == 0 {
		return health.Unhealthy("it has not said whether it hears " + DeadLettersStream + " (issue 734): it is starting, or its state " + HearingState + " is not there")
	}
	var h Hearing
	if err := json.Unmarshal(raw, &h); err != nil {
		return health.Unhealthy("what it keeps of hearing " + DeadLettersStream + " is not a hearing: " + err.Error())
	}
	if !h.Heard {
		return health.Unhealthy(h.Why + " (since " + h.At.Format(time.RFC3339) + "; tried again at its next reconnect or ask)")
	}
	return health.Healthy("it hears " + DeadLettersStream + " for " + strings.Join(h.Streams, ", "))
}