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"

	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
)

// **The durable consumers an account needs** (novox/hq ADR 0561 §6, issue 774). The controller creates a module's
// consumer on EVENTS and each process's on its work queue when it issues the account. From the write that hands its
// lease away the grants part creates them, with the configuration the bus composer derives (buscompose.AccountConsumers),
// only where the bus lacks one; one the bus holds is never changed or deleted, a difference is only said. Before that
// write they are compared in shadow alone, so two parts never create one consumer.

// ReadLease is the controller's lease as the part reads it (buscompose.NamedReaders names the vault and the part), and
// LeaseKey its one key.
const (
	ReadLease = buscompose.ControllerLease
	LeaseKey  = buscompose.ControllerLeaseKey
)

// 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
}

// ConsumersState is what one pass found and did of the account consumers.
type ConsumersState struct {
	// On says the lease is handed away, so the part creates what the bus lacks; Why says why it is on or off.
	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"`
	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"`
}

// 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
}

// leaseHandedAway reads the lease as the part's reads give it.
func leaseHandedAway(r Reader) (bool, string) {
	raw, found, err := r.Get(ReadLease, LeaseKey)
	return buscompose.LeaseHandedAway(raw, found, err)
}

// consumers compares the account consumers the records compose with those the bus holds, and creates the missing ones
// once the lease is handed away and 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 = leaseHandedAway(p.Read)
	res.Consumers = st
	wanted := buscompose.AccountConsumers(users)
	st.Wanted = len(wanted)
	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 {
				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)
}

// sayDiffer raises the warning condition mesh-grants.consumer-differs while an account consumer the part owns (from the
// lease write) 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. Before the lease write the consumers are the
// controller's, and a difference is only said in the pass.
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 := asks.Condition{Key: ConsumerDiffersCondition, Subject: asks.Scope{Scope: "module", ID: "mesh-grants"}}
	if why == "" {
		c.State = asks.StateCleared
	} else {
		c.Severity, c.State = asks.SeverityWarning, asks.StateRaised
		c.Summary = fmt.Sprintf("%d durable consumers differ from their composed configuration", len(st.Differ))
		c.Headline, c.Needs = "A module's consumer is not as composed", "a look, and grants_consumer_match for each"
		c.Explanation = asks.NeedsYou + "a look. 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 while the controller holds
// the lease (the consumers are the controller's then), and 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")
	}
	if on, why := leaseHandedAway(p.Read); !on {
		return nil, fmt.Errorf("not asked: %s; the controller makes and matches the consumers while it holds the lease", why)
	}
	in := ReadInputs(p.Read, &p.cache)
	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, "; ")
}