mesh-tools/node-tools/internal/bus/ledger.go

package bus

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

	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-tools/node-tools/internal/ledger"
)

// A ledger on the bus (novox/hq ADR 0597, design 25 § 12.4; issue 1966): its history a stream with a subject per
// instance under the binding's subject, its state a key-value bucket, its snapshots a subject per instance under the
// binding's snapshot subject, the latest only (the stream that holds them keeps one per subject, as the bus module
// creates it). **Only the module's ledger connection publishes on its history and its snapshots** (ADR 0597 §9): what
// is read, and the state, go on the runner's own.

// LedgerStores are the stores of one ledger binding: reads and the state on c, the history's appends and the snapshots'
// writes on writer, the module's ledger connection.
func (c *Conn) LedgerStores(b membership.BusBinding, writer *Conn) (ledger.History, ledger.State, ledger.Snapshots) {
	return &busHistory{read: c, write: writer, stream: b.History, prefix: prefixOf(b.Subject)},
		&busState{c: c, bucket: b.State},
		&busSnapshots{read: c, write: writer, prefix: prefixOf(b.Snapshot)}
}

// prefixOf is a binding's subject as the prefix an instance's key is added to: written with or without its `.>`.
func prefixOf(subject string) string {
	return strings.TrimSuffix(strings.TrimSuffix(subject, ">"), ".")
}

// publishExpecting publishes body on subject on the connection, written only when the subject's last message is still
// at expect (0: it has none); ledger.ErrRaced when it is not.
func publishExpecting(c *Conn, subject string, body []byte, expect uint64) (uint64, error) {
	msg := nats.NewMsg(subject)
	msg.Data = body
	msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(expect, 10))
	reply, err := c.nc.RequestMsg(msg, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return 0, fmt.Errorf("no stream on the bus keeps %s", subject)
	}
	if err != nil {
		return 0, err
	}
	var ack struct {
		Seq   uint64 `json:"seq"`
		Error *struct {
			ErrCode     uint16 `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
	}
	if err := json.Unmarshal(reply.Data, &ack); err != nil {
		return 0, fmt.Errorf("the bus answered the write to %s with what is not an acknowledgement: %q", subject, reply.Data)
	}
	if ack.Error != nil {
		if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence ||
			code == jetstream.JSErrCodeStreamWrongLastSequenceConstant {
			return 0, ledger.ErrRaced
		}
		return 0, fmt.Errorf("the bus refused the write to %s: %s (%d)", subject, ack.Error.Description, ack.Error.ErrCode)
	}
	return ack.Seq, nil
}

// lastOn is the last message on a subject of the stream that keeps it: its bytes and sequence, nil and 0 for none.
func lastOn(c *Conn, stream, subject string) ([]byte, uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	js, err := c.jetStream()
	if err != nil {
		return nil, 0, err
	}
	if stream == "" {
		if stream, err = js.StreamNameBySubject(ctx, subject); err != nil {
			return nil, 0, fmt.Errorf("no stream on the bus keeps %s: %w", subject, err)
		}
	}
	s, err := js.Stream(ctx, stream)
	if err != nil {
		return nil, 0, fmt.Errorf("the stream %s: %w", stream, err)
	}
	m, err := s.GetLastMsgForSubject(ctx, subject)
	if errors.Is(err, jetstream.ErrMsgNotFound) {
		return nil, 0, nil
	}
	if err != nil {
		return nil, 0, err
	}
	return m.Data, m.Sequence, nil
}

type busHistory struct {
	read, write    *Conn
	stream, prefix string
}

func (h *busHistory) Last(key string) (uint64, error) {
	_, seq, err := lastOn(h.read, h.stream, h.prefix+"."+key)
	return seq, err
}

func (h *busHistory) Append(key string, op []byte, expect uint64) (uint64, error) {
	if h.write == nil {
		return 0, errors.New("the ledger's history is published only on its module's ledger connection, and none is held")
	}
	return publishExpecting(h.write, h.prefix+"."+key, op, expect)
}

func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, error) {
	var out []ledger.Kept
	for {
		want := StreamReadMost
		if limit > 0 && limit-len(out) < want {
			want = limit - len(out)
		}
		page, err := h.read.readStream(h.stream, h.prefix+"."+keys, from, want)
		if err != nil {
			return out, err
		}
		for _, m := range page.Messages {
			out = append(out, ledger.Kept{Sequence: m.Sequence, Data: m.Body})
		}
		if page.Next == 0 || (limit > 0 && len(out) >= limit) {
			return out, nil
		}
		from = page.Next
	}
}

type busState struct {
	c      *Conn
	bucket string
}

func (s *busState) kv(ctx context.Context) (jetstream.KeyValue, error) {
	js, err := s.c.jetStream()
	if err != nil {
		return nil, err
	}
	kv, err := js.KeyValue(ctx, s.bucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return nil, fmt.Errorf("the ledger's state %s is not on the bus", s.bucket)
	}
	return kv, err
}

func (s *busState) Get(key string) ([]byte, uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := s.kv(ctx)
	if err != nil {
		return nil, 0, err
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, 0, nil
	}
	if err != nil {
		return nil, 0, err
	}
	return e.Value(), e.Revision(), nil
}

// isFull is a write a bucket or stream refused because it holds as much as it may.
func isFull(err error) bool {
	msg := err.Error()
	return strings.Contains(msg, "maximum bytes exceeded") || strings.Contains(msg, "maximum messages exceeded") ||
		strings.Contains(msg, "insufficient resources")
}

func (s *busState) Write(key string, value []byte, revision uint64) (uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := s.kv(ctx)
	if err != nil {
		return 0, err
	}
	var rev uint64
	if revision == 0 {
		rev, err = kv.Create(ctx, key, value)
	} else {
		rev, err = kv.Update(ctx, key, value, revision)
	}
	switch {
	case err == nil:
		return rev, nil
	case errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) || isWrongSequence(err):
		return 0, ledger.ErrChanged
	case isFull(err):
		return 0, fmt.Errorf("%w: %v", ledger.ErrFull, err)
	}
	return 0, err
}

func (s *busState) Keys() ([]string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := s.kv(ctx)
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

type busSnapshots struct {
	read, write *Conn
	prefix      string
}

func (s *busSnapshots) Get(key string) ([]byte, uint64, error) {
	return lastOn(s.read, "", s.prefix+"."+key)
}

func (s *busSnapshots) Put(key string, value []byte, expect uint64) error {
	if s.write == nil {
		return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held")
	}
	_, err := publishExpecting(s.write, s.prefix+"."+key, value, expect)
	return err
}

// Watch hands deliver every instance's record under a key pattern as it is, then each change; a removed key is handed as
// nil.
func (s *busState) Watch(pattern string, deliver func(key string, value []byte, current bool) error) (func(), error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	kv, err := s.kv(ctx)
	cancel()
	if err != nil {
		return nil, err
	}
	w, err := kv.Watch(context.Background(), pattern)
	if err != nil {
		return nil, err
	}
	hand := func(e jetstream.KeyValueEntry, current bool) {
		var value []byte
		if e.Operation() == jetstream.KeyValuePut {
			value = e.Value()
		}
		if err := deliver(e.Key(), value, current); err != nil {
			s.c.Logf("[mesh-tools] a ledger watch of %s was not taken: %v", s.bucket, err)
		}
	}
	timer := time.NewTimer(StateTimeout)
	defer timer.Stop()
	for current := true; current; {
		select {
		case e, ok := <-w.Updates():
			if !ok {
				return nil, errors.New("the ledger's state watch ended before its current states were handed over")
			}
			if e == nil {
				current = false
				continue
			}
			hand(e, true)
		case <-timer.C:
			_ = w.Stop()
			return nil, fmt.Errorf("the ledger's state %s did not hand its current states in %s", s.bucket, StateTimeout)
		}
	}
	go func() {
		for e := range w.Updates() {
			if e != nil {
				hand(e, false)
			}
		}
	}()
	return func() { _ = w.Stop() }, nil
}

// LedgerIndexes are the buckets of a ledger binding's declared indexes, by index name, on the runner's own connection.
func (c *Conn) LedgerIndexes(b membership.BusBinding) map[string]ledger.Index {
	out := map[string]ledger.Index{}
	for name, bucket := range b.Indexes {
		out[name] = &busState{c: c, bucket: bucket}
	}
	return out
}

func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) {
	if pattern == "" {
		pattern = ">"
	}
	keys, err := s.Keys()
	if err != nil {
		return nil, err
	}
	sort.Strings(keys)
	var out []buscalls.IndexEntry
	for _, k := range keys {
		if limit > 0 && len(out) >= limit {
			break
		}
		if !SubjectMatches(pattern, k) {
			continue
		}
		v, _, err := s.Get(k)
		if err != nil {
			return out, err
		}
		if v != nil {
			out = append(out, buscalls.IndexEntry{Key: k, Value: valueOf(v)})
		}
	}
	return out, nil
}

func (s *busState) Put(key string, value []byte) error {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := s.kv(ctx)
	if err != nil {
		return err
	}
	_, err = kv.Put(ctx, key, value)
	return err
}

func (s *busState) Delete(key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := s.kv(ctx)
	if err != nil {
		return err
	}
	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
		return err
	}
	return nil
}