mesh-delivery/modules/mesh-grants/internal/grants/consumers.go

package grants

import (
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"

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

	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	"mesh-grants/internal/buscompose"
)

// **The durable consumers an account needs** (novox/hq ADR 0561 §6, issue 774): a module's consumer on EVENTS and each
// process's on its work queue. Since the controller ended by its unassignment (ADR 0590, issue 1126) the grants part
// alone creates them, with the configuration the bus composer derives (buscompose.AccountConsumers), only where the bus
// lacks one; one the bus holds only ever gains the filters its module's consumes adds (never loses one or changes another setting, never deleted by a pass), any other difference is only said.

// ConsumerCreator creates one durable consumer the bus does not hold: the part's process alone, whose account is granted
// the create of each account's consumer by stream and name. One the bus holds answers ErrHeld.
type ConsumerCreator interface {
	CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error
	// MatchConsumer brings a held consumer to cfg, keeping where it delivers from: only on a person's ask
	// (grants_consumer_match), never by a pass.
	MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error
	// AddFilters adds filter subjects to a held consumer and nothing else: its filters become the union of what it holds and
	// add, every other setting and its delivery position kept.
	AddFilters(stream, name string, add []string) error
}

// ConsumersState is what one pass found and did of the account consumers.
type ConsumersState struct {
	// On says the part creates what the bus lacks: always, since the controller ended by its unassignment (issue 1126);
	// Why says so.
	On  bool   `json:"on"`
	Why string `json:"why"`
	// Wanted is how many the records compose; Missing those the bus lacks (`<stream>.<name>`), Differ those it holds
	// otherwise, each with how.
	Wanted  int      `json:"wanted"`
	Missing []string `json:"missing,omitempty"`
	Differ  []string `json:"differ,omitempty"`
	// Behind are those held more than BehindBound messages behind their stream's head at this pass and the one before,
	// each with how far.
	Behind  []string `json:"behind,omitempty"`
	Created []string `json:"created,omitempty"`
	// Matched is a consumer brought to the composed configuration on a person's ask, or why not.
	Matched    []string `json:"matched,omitempty"`
	NotCreated []string `json:"not_created,omitempty"`
	// Unread is a stream whose consumers could not be read: nothing on it is compared or created.
	Unread []string `json:"unread,omitempty"`
}

// heldConsumer is one consumer as the bus module's nats_streams reads it for one stream.
type heldConsumer struct {
	Name       string   `json:"name"`
	Filters    []string `json:"filters"`
	MaxDeliver int      `json:"max_deliver"`
	Pending    int64    `json:"pending"`
}

// heldConsumers reads the consumers the bus holds on stream, by name.
func heldConsumers(c Caller, stream string) (map[string]heldConsumer, error) {
	raw, err := c.Call(CallStreams, map[string]any{"stream": stream})
	if err != nil {
		return nil, fmt.Errorf("%s %s: %w", CallStreams, stream, err)
	}
	var got struct {
		Detail []struct {
			Name      string         `json:"name"`
			Consumers []heldConsumer `json:"consumer_detail"`
		} `json:"detail"`
	}
	if err := json.Unmarshal(unwrap(raw), &got); err != nil {
		return nil, fmt.Errorf("%s %s answered no consumers: %w", CallStreams, stream, err)
	}
	out := map[string]heldConsumer{}
	for _, d := range got.Detail {
		if d.Name != stream {
			continue
		}
		for _, c := range d.Consumers {
			out[c.Name] = c
		}
	}
	return out, nil
}

// filtersToAdd are the composed filters of a consumer the bus's one lacks, in the composed order; none when the consumer
// filters the whole stream (no filters) or holds all.
func filtersToAdd(c buscompose.Consumer, h heldConsumer) []string {
	if len(h.Filters) == 0 {
		return nil
	}
	var add []string
	for _, f := range c.Filters {
		if !slices.Contains(h.Filters, f) && !slices.Contains(add, f) {
			add = append(add, f)
		}
	}
	return add
}

// ConsumersOn is why the part creates and matches the account consumers: the controller ended by its unassignment, so
// the part is their only maker (novox/hq ADR 0590, issue 1126).
const ConsumersOn = "the grants part makes and matches the account consumers: the controller ended by its unassignment (issue 1126)"

// consumers compares the account consumers the records compose with those the bus holds, and creates the missing ones
// where the part holds a creator.
func (p *Part) consumers(res *Result, recs buscompose.Records, match string) {
	users, err := buscompose.Users(recs)
	if err != nil {
		res.Errors = append(res.Errors, "the account consumers could not be composed: "+err.Error())
		return
	}
	st := &ConsumersState{}
	st.On, st.Why = true, ConsumersOn
	res.Consumers = st
	wanted := buscompose.AccountConsumers(users)
	st.Wanted = len(wanted)
	seen := map[string]bool{}
	defer func() { p.behindSeen = seen }()
	byStream := map[string][]buscompose.Consumer{}
	for _, mc := range wanted {
		byStream[mc.Consumer.Stream] = append(byStream[mc.Consumer.Stream], mc.Consumer)
	}
	for _, stream := range sortedKeys(byStream) {
		held, err := heldConsumers(p.Call, stream)
		if err != nil {
			st.Unread = append(st.Unread, stream+": "+err.Error())
			continue
		}
		for _, c := range byStream[stream] {
			key := stream + "." + c.Name
			h, ok := held[c.Name]
			if ok {
				if h.Pending > BehindBound {
					if p.behindSeen[key] {
						st.Behind = append(st.Behind, fmt.Sprintf("%s: %d pending", key, h.Pending))
					}
					seen[key] = true
				}
				// A consumer whose module now consumes more events of the stream gains those filters by itself, only ever added:
				// a filter it holds that is no longer composed stays, and nothing else about it changes.
				if add := filtersToAdd(c, h); len(add) > 0 && st.On && p.Consumers != nil {
					if err := p.Consumers.AddFilters(stream, c.Name, add); err != nil {
						// A refusal that stands (an overlap with another filter, say) is said once, not at every pass.
						why := key + ": filters " + strings.Join(add, ", ") + " not added: " + err.Error()
						if p.filterRefused[key] != why {
							st.Matched = append(st.Matched, why)
							if p.Logf != nil {
								p.Logf("%s", why)
							}
						}
						if p.filterRefused == nil {
							p.filterRefused = map[string]string{}
						}
						p.filterRefused[key] = why
					} else {
						delete(p.filterRefused, key)
						st.Matched = append(st.Matched, key+": filters added: "+strings.Join(add, ", "))
						if p.Logf != nil {
							p.Logf("consumer %s gained the filters %s", key, strings.Join(add, ", "))
						}
						p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream, "filters-added": add})
						h.Filters = append(slices.Clone(h.Filters), add...)
					}
				}
				why := consumerDiffers(c, h)
				if why != "" && key == match && st.On && p.Consumers != nil {
					if err := p.Consumers.MatchConsumer(stream, buscompose.ConsumerConfig(c)); err != nil {
						st.Matched = append(st.Matched, key+": not matched: "+err.Error())
					} else {
						st.Matched = append(st.Matched, key+": matched, asked by hand")
						continue
					}
				}
				if why != "" {
					st.Differ = append(st.Differ, key+": "+why)
				}
				continue
			}
			if !st.On || p.Consumers == nil {
				st.Missing = append(st.Missing, key)
				continue
			}
			switch err := p.Consumers.CreateConsumer(stream, buscompose.ConsumerConfig(c)); {
			case err == nil:
				st.Created = append(st.Created, key)
				p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream})
			case errors.Is(err, ErrHeld):
			default:
				st.Missing = append(st.Missing, key)
				st.NotCreated = append(st.NotCreated, key+": "+err.Error())
			}
		}
	}
	p.sayDiffer(st)
	p.sayBehind(st)
}

// sayDiffer raises the warning condition mesh-grants.consumer-differs while an account consumer the part owns (from the
// controller's end) differs from its composed configuration, naming each with the fields that differ, and clears it when
// they match. Said once per set of differences, again when it changes.
func (p *Part) sayDiffer(st *ConsumersState) {
	if p.Say == nil {
		return
	}
	why := ""
	if st.On && len(st.Differ) > 0 {
		why = strings.Join(st.Differ, "; ")
	}
	p.holdMu.Lock()
	defer p.holdMu.Unlock()
	if p.differKnown && why == p.differSaid {
		return
	}
	c := oc.Condition{Key: ConsumerDiffersCondition, Subject: oc.ConditionSubject{Scope: "module", ID: "mesh-grants"}}
	if why == "" {
		c.State = occ.StateCleared
	} else {
		c.Severity, c.State = occ.SeverityWarning, occ.StateRaised
		c.Summary = fmt.Sprintf("%d durable consumers differ from their composed configuration", len(st.Differ))
		c.Headline = "A module's consumer is not as composed"
		needsYou(&c, "a look, and grants_consumer_match for each", "The bus holds these consumers otherwise than the bus composer "+
			"composes them, and the grants part never changes a held one by itself (its delivery position would be lost): "+why)
		c.Resolved = "every account consumer matches, by grants_consumer_match or otherwise"
	}
	if p.Say(c) == nil {
		p.differSaid, p.differKnown = why, true
	}
}

// ConsumerDiffersCondition is the key of the warning raised while an account consumer differs (ADR 0561 §6).
const ConsumerDiffersCondition = "mesh-grants.consumer-differs"

// MatchConsumer is the tool consumer-match (ADR 0561 §6, §9): it asks the part's process to bring one account consumer
// the records compose to its composed configuration, keeping where it delivers from. Refused for a
// consumer the records do not compose.
func (p *Part) MatchConsumer(stream, name string) (any, error) {
	if stream == "" || name == "" {
		return nil, errors.New("consumer-match names a stream and a consumer")
	}
	in := p.inputs()
	recs, _ := Records(in)
	users, err := buscompose.Users(recs)
	if err != nil {
		return nil, err
	}
	found := false
	for _, mc := range buscompose.AccountConsumers(users) {
		if mc.Consumer.Stream == stream && mc.Consumer.Name == name {
			found = true
		}
	}
	if !found {
		return nil, fmt.Errorf("the records compose no account consumer %s on %s: the part brings only those to their configuration", name, stream)
	}
	ask, _ := json.Marshal(Ask{By: "grants_consumer_match", At: p.now(), Consumer: stream + "." + name})
	rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
	if err != nil {
		return nil, fmt.Errorf("the ask could not be written: %w", err)
	}
	return map[string]any{"asked": rev, "says": "the part's process brings " + stream + "." + name + " to its composed " +
		"configuration at this ask, keeping where it delivers from; the last pass's consumers.matched says what became of it"}, nil
}

// consumerDiffers says how a held consumer differs from the composed one in what nats_streams reads back, or "".
func consumerDiffers(c buscompose.Consumer, h heldConsumer) string {
	var why []string
	want, have := slices.Sorted(slices.Values(c.Filters)), slices.Sorted(slices.Values(h.Filters))
	if !slices.Equal(want, have) {
		why = append(why, fmt.Sprintf("filters %v, composed %v", have, want))
	}
	maxDeliver := c.MaxDeliver
	if maxDeliver == 0 {
		maxDeliver = -1
	}
	if h.MaxDeliver != maxDeliver {
		why = append(why, fmt.Sprintf("max deliver %d, composed %d", h.MaxDeliver, maxDeliver))
	}
	sort.Strings(why)
	return strings.Join(why, "; ")
}

// BehindBound is how many messages behind its stream's head a consumer may be before it is said far behind: the bound
// mesh-self-check's bus-consumer probe judged by, which ended with the controller (issue 1126).
const BehindBound = 1000

// ConsumerBehindCondition is the key of the warning raised while an account consumer stays far behind its stream's head.
const ConsumerBehindCondition = "mesh-grants.consumer-behind"

// sayBehind raises the warning condition mesh-grants.consumer-behind while an account consumer the part owns is more
// than BehindBound messages behind its stream's head at two passes running (one look at a burst is not a consumer that
// stays behind, novox/hq issue 277), naming each with how far, and clears it when none is. Said once per set, again
// when it changes.
func (p *Part) sayBehind(st *ConsumersState) {
	if p.Say == nil {
		return
	}
	// The consumers alone decide whether it is said again, never how far each is behind, which changes every pass: the
	// counts stay in the last pass's consumers.behind (its state user-list, key pass), and the condition is not said again
	// for them.
	var names []string
	for _, b := range st.Behind {
		name, _, _ := strings.Cut(b, ": ")
		names = append(names, name)
	}
	why := strings.Join(names, ", ")
	p.holdMu.Lock()
	defer p.holdMu.Unlock()
	if p.behindKnown && why == p.behindSaid {
		return
	}
	c := oc.Condition{Key: ConsumerBehindCondition, Subject: oc.ConditionSubject{Scope: "module", ID: "mesh-grants"}}
	if why == "" {
		c.State = occ.StateCleared
	} else {
		c.Severity, c.State = occ.SeverityWarning, occ.StateRaised
		c.Summary = fmt.Sprintf("%d durable consumers are more than %d messages behind their stream's head", len(st.Behind), BehindBound)
		c.Headline = "Messages pile up for a listener"
		needsYou(&c, "a look at the module that reads them", "These consumers stayed far behind their stream's head at two passes "+
			"running, so what is sent to their modules waits unread (how far each is, in the grants part's last pass): "+why)
		c.Resolved = "every account consumer is back within the bound"
	}
	if p.Say(c) == nil {
		p.behindSaid, p.behindKnown = why, true
	}
}