mesh-delivery/modules/mesh-grants/cmd/mesh-grants/bus.go

package main

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

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

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

// requestWithin bounds one request to the bus: a read the bus refuses is answered by no reply, so the bound is how a
// refusal is told from a slow answer. It is no clock of the part's work: nothing is ever started by it.
const requestWithin = 10 * time.Second

// bucketOf is the bucket a read names: `<module>.<state>` is `<module>_<state>`.
func bucketOf(state string) string {
	at := strings.LastIndex(state, ".")
	if at <= 0 {
		return state
	}
	return state[:at] + "_" + state[at+1:]
}

// busReader reads other parts' state on the bus, and keeps every value of a state it watches, so a composition reads
// those from memory. A bucket it cannot open (refused, or not there yet) is an error of the call, never the end.
type busReader struct {
	ctx context.Context
	js  jetstream.JetStream

	mu      sync.Mutex
	kvs     map[string]jetstream.KeyValue
	world   map[string]map[string][]byte
	settled map[string]bool
}

func newBusReader(ctx context.Context, js jetstream.JetStream) *busReader {
	return &busReader{ctx: ctx, js: js, kvs: map[string]jetstream.KeyValue{}, world: map[string]map[string][]byte{},
		settled: map[string]bool{}}
}

func (b *busReader) kv(state string) (jetstream.KeyValue, error) {
	b.mu.Lock()
	kv, ok := b.kvs[state]
	b.mu.Unlock()
	if ok {
		return kv, nil
	}
	ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
	defer cancel()
	kv, err := b.js.KeyValue(ctx, bucketOf(state))
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			err = fmt.Errorf("no answer in %s: a read the bus refuses is answered by none (%w)", requestWithin, err)
		}
		return nil, fmt.Errorf("the bucket %s cannot be opened: %w", bucketOf(state), err)
	}
	b.mu.Lock()
	b.kvs[state] = kv
	b.mu.Unlock()
	return kv, nil
}

func (b *busReader) Keys(state string) ([]string, error) {
	b.mu.Lock()
	if b.settled[state] {
		out := make([]string, 0, len(b.world[state]))
		for k := range b.world[state] {
			out = append(out, k)
		}
		b.mu.Unlock()
		sort.Strings(out)
		return out, nil
	}
	b.mu.Unlock()
	kv, err := b.kv(state)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
	defer cancel()
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return []string{}, nil
	}
	return keys, err
}

func (b *busReader) Get(state, key string) ([]byte, bool, error) {
	b.mu.Lock()
	if b.settled[state] {
		v, ok := b.world[state][key]
		b.mu.Unlock()
		return v, ok, nil
	}
	b.mu.Unlock()
	kv, err := b.kv(state)
	if err != nil {
		return nil, false, err
	}
	ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
	defer cancel()
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	return e.Value(), true, nil
}

// Watch keeps every value of a state from now on and hands each change after what stands to changed; it returns once
// what stands is held. The watch ending is handed to ended, and the state is read from the bus again until watched.
func (b *busReader) Watch(state string, changed func(string, []byte), ended func(error)) error {
	kv, err := b.kv(state)
	if err != nil {
		return err
	}
	w, err := kv.WatchAll(b.ctx)
	if err != nil {
		return fmt.Errorf("the bucket %s cannot be watched: %w", bucketOf(state), err)
	}
	put := func(e jetstream.KeyValueEntry) {
		b.mu.Lock()
		defer b.mu.Unlock()
		if b.world[state] == nil {
			b.world[state] = map[string][]byte{}
		}
		if e.Operation() == jetstream.KeyValuePut {
			b.world[state][e.Key()] = e.Value()
		} else {
			delete(b.world[state], e.Key())
		}
	}
	lost := func(err error) {
		w.Stop()
		b.mu.Lock()
		b.settled[state] = false
		delete(b.world, state)
		delete(b.kvs, state)
		b.mu.Unlock()
		ended(err)
	}
	settling, stop := context.WithTimeout(b.ctx, requestWithin)
	defer stop()
settle:
	for {
		select {
		case <-settling.Done():
			w.Stop()
			if b.ctx.Err() != nil {
				return b.ctx.Err()
			}
			return fmt.Errorf("the watch of %s handed nothing in %s: a watch the bus refuses is answered by none", bucketOf(state), requestWithin)
		case e, ok := <-w.Updates():
			if !ok {
				w.Stop()
				return fmt.Errorf("the watch of %s ended before it held what stands", bucketOf(state))
			}
			if e == nil {
				break settle
			}
			put(e)
		}
	}
	b.mu.Lock()
	b.settled[state] = true
	b.mu.Unlock()
	go func() {
		for {
			select {
			case <-b.ctx.Done():
				w.Stop()
				return
			case e, ok := <-w.Updates():
				if !ok {
					lost(errors.New("the server ended it"))
					return
				}
				if e != nil {
					put(e)
					var value []byte
					if e.Operation() == jetstream.KeyValuePut {
						value = e.Value()
					}
					changed(e.Key(), value)
				}
			}
		}
	}()
	return nil
}

// ownStore is the part's own state on its process connection: a state it did not announce, or one the bus refuses,
// fails the call, named.
type ownStore struct{ c *process.Conn }

func (s ownStore) Get(state, key string) ([]byte, uint64, bool, error) {
	e, err := s.c.State(state).Get(key)
	if err != nil || e == nil {
		return nil, 0, false, err
	}
	return e.Value, e.Revision, true, nil
}

func (s ownStore) Put(state, key string, value []byte) (uint64, error) {
	return s.c.State(state).Put(key, json.RawMessage(value))
}

// emitting is the process connection as the part's emitter: its events, with no options.
type emitting struct{ c *process.Conn }

func (e emitting) Emit(event string, body any) error { return e.c.Emit(event, body) }