mesh-sdk/go/process/consumers.go

package process

import (
	"context"
	"encoding/json"
	"sync"
	"time"

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

// ConsumersState is the name of the state a module keeps its durable consumers' facts in, one key per consumer by its name,
// history 1 (design 47 G A-166, A-176): what a module's tool reads of a consumer, since a tool runs in a runtime of its own and
// cannot read the consumer's information.
const ConsumersState = "consumers"

// ConsumerFacts is one durable consumer as ConsumerLedger keeps it.
type ConsumerFacts struct {
	// Stream and Name are the consumer's stream and its name.
	Stream string `json:"stream"`
	Name   string `json:"name"`
	// Acknowledged is the stream sequence of the last message applied.
	Acknowledged uint64 `json:"acknowledged"`
	// Pending is the messages the consumer has not been given yet, as the last message said.
	Pending uint64 `json:"pending"`
	// LastEvent is the subject of the last event applied and LastAt its own `at`, or when the bus stored it where it has none.
	LastEvent string    `json:"last-event,omitempty"`
	LastAt    time.Time `json:"last-at,omitempty"`
	// OldestUnacknowledged is when the oldest message delivered and not yet applied was delivered; zero when none is.
	OldestUnacknowledged time.Time `json:"oldest-unacknowledged,omitempty"`
}

// ConsumerSink stores one key of the module's state consumers.
type ConsumerSink func(ctx context.Context, key string, value []byte) error

// ConsumerLedger writes a durable consumer's facts into the module's state consumers after each message delivered and each message
// applied, so a tool reads them and nothing is read on a timer (design 47 G A-166, A-176). One ledger serves one consumer.
type ConsumerLedger struct {
	stream, name string
	sink         ConsumerSink
	now          func() time.Time
	logf         func(string, ...any)

	mu     sync.Mutex
	facts  ConsumerFacts
	unacks map[uint64]time.Time // delivered and not applied, by stream sequence
}

// NewConsumerLedger is the ledger of the consumer name on stream, writing through sink; a nil now is the clock and a nil logf says
// nothing.
func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.Time, logf func(string, ...any)) *ConsumerLedger {
	if now == nil {
		now = time.Now
	}
	if logf == nil {
		logf = func(string, ...any) {}
	}
	return &ConsumerLedger{stream: stream, name: name, sink: sink, now: now, logf: logf, unacks: map[uint64]time.Time{},
		facts: ConsumerFacts{Stream: stream, Name: name}}
}

// Delivered notes that m was given to the consumer, and writes the facts; a message given again keeps its first delivery's time.
func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
	md, err := m.Metadata()
	if err != nil {
		l.logf("consumers: %s's message has no metadata, its delivery is not kept: %v", l.name, err)
		return
	}
	l.mu.Lock()
	// A message delivered again keeps the time it was first delivered, so one that never settles grows old (A-187): asked again
	// in this run, the time kept; after a restart, which kept none, the time the bus stored it, never later than its first
	// delivery.
	if _, known := l.unacks[md.Sequence.Stream]; !known {
		first := l.now().UTC()
		if md.NumDelivered > 1 && !md.Timestamp.IsZero() && md.Timestamp.Before(first) {
			first = md.Timestamp.UTC()
		}
		l.unacks[md.Sequence.Stream] = first
	}
	l.facts.Pending = md.NumPending
	l.mu.Unlock()
	l.write()
}

// Applied notes that m was applied and acknowledged, at the event's own `at` (zero: when the bus stored it), and writes the facts.
func (l *ConsumerLedger) Applied(m jetstream.Msg, at time.Time) {
	md, err := m.Metadata()
	if err != nil {
		l.logf("consumers: %s's message has no metadata, its application is not kept: %v", l.name, err)
		return
	}
	if at.IsZero() {
		at = md.Timestamp
	}
	l.mu.Lock()
	delete(l.unacks, md.Sequence.Stream)
	if md.Sequence.Stream > l.facts.Acknowledged {
		l.facts.Acknowledged = md.Sequence.Stream
	}
	l.facts.Pending = md.NumPending
	l.facts.LastEvent, l.facts.LastAt = m.Subject(), at.UTC()
	l.mu.Unlock()
	l.write()
}

// Facts are the facts now.
func (l *ConsumerLedger) Facts() ConsumerFacts {
	l.mu.Lock()
	defer l.mu.Unlock()
	f := l.facts
	f.OldestUnacknowledged = time.Time{}
	for _, t := range l.unacks {
		if f.OldestUnacknowledged.IsZero() || t.Before(f.OldestUnacknowledged) {
			f.OldestUnacknowledged = t
		}
	}
	return f
}

func (l *ConsumerLedger) write() {
	if l.sink == nil {
		return
	}
	data, err := json.Marshal(l.Facts())
	if err == nil {
		err = l.sink(context.Background(), l.name, data)
	}
	if err != nil {
		l.logf("consumers: the facts of %s are not written to the state %s: %v", l.name, ConsumersState, err)
	}
}