mesh-sdk/go/views/process_reader.go

package views

// A process's pictures of the owners' states it used to read (novox/hq ADR 0615 §1, design 25 § 12.2), for the parts
// that run as processes: each state is asked once through the owner's ordinary read tool with `values` (the first
// view), then kept from the owner's views family, asked again only on a gap or by hand (Follower).
//
// A Picture reads as the key-value bucket the process used to bind (jetstream.KeyValue: Get, Keys, ListKeys, Watch,
// WatchAll), so a reader's code is unchanged; every write is refused, for the owner's state is the owner's alone. The
// picture is kept in memory: the owner's state is the truth and a restart asks again (ADR 0615 §4). ProcessReader is
// the process side of Reader (the stdio side).

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

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

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

// ErrReadOnly is every write to a picture: the state is its owner's.
var ErrReadOnly = errors.New("a first view is read-only: the state is its owner's")

// OwnerReadTool is the owner's ordinary read tool of state, as a reader's `invokes` names it (design 25 § 12.2):
// `<owner>.<short>_read_<state>`, short the owner's name past its last dash-separated `mesh-delivery-` prefix
// (indexer, assigner) and the state's dashes as underscores.
func OwnerReadTool(owner, state string) string {
	short := strings.TrimPrefix(owner, "mesh-delivery-")
	return owner + "." + short + "_read_" + strings.ReplaceAll(state, "-", "_")
}

// ProcessReader is one process's pictures of its owners' states.
type ProcessReader struct {
	// Module is the reading module, named in the finding of a first view unanswered past its bound.
	Module string
	// Call asks a module's tool (process.Conn.Call): the read tools are asked with `values`.
	Call func(key string, body any) (json.RawMessage, error)
	// Findings is the reader's own state `findings`, where the condition keeper raises and clears the condition of a
	// first view unanswered past its bound (ADR 0615 §6); nil says it in the log only.
	Findings func() jetstream.KeyValue
	Logf     func(string, ...any)
	// Now, Bound and Wait are for tests: zero are the clock, Bound and five seconds.
	Now   func() time.Time
	Bound time.Duration
	Wait  time.Duration

	mu    sync.Mutex
	pics  map[string]*Picture // by bucket name <owner>_<state>
	order []string
}

func (r *ProcessReader) logf(format string, args ...any) {
	if r.Logf != nil {
		r.Logf(format, args...)
	}
}

// Follow adds the owner's state to follow through its read tool (OwnerReadTool). Call it before Start.
func (r *ProcessReader) Follow(owner, state string) *Picture {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.pics == nil {
		r.pics = map[string]*Picture{}
	}
	bucket := owner + "_" + state
	if p := r.pics[bucket]; p != nil {
		return p
	}
	tool := OwnerReadTool(owner, state)
	p := &Picture{bucket: bucket, ctx: context.Background()}
	call := r.Call
	ask := AskVia(func(key string, body any) (json.RawMessage, error) {
		m, _ := body.(map[string]any)
		if m == nil {
			m = map[string]any{}
		}
		m["values"] = true
		if call == nil {
			return nil, errors.New("this reader asks no tool")
		}
		return call(key, m)
	}, tool)
	p.f = &Follower{Owner: owner, Tool: tool, State: state, Ask: ask, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait,
		Raise: func(s Status) { r.finding(s, true) }, Clear: func(s Status) { r.finding(s, false) },
		Changed: p.changed}
	r.pics[bucket] = p
	r.order = append(r.order, bucket)
	return p
}

// Picture is the picture of the bucket the process used to bind, by its bus name (`<owner>_<state>`); nil when not
// followed.
func (r *ProcessReader) Picture(bucket string) *Picture {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.pics[bucket]
}

// KeyValue opens a followed state's picture by its bus name, where a process opened the bucket (process.BucketOpener,
// as buildrun.FollowPlacement takes it); any other bucket is refused, for the reader reads its owners' states by their
// first views alone.
func (r *ProcessReader) KeyValue(_ context.Context, bucket string) (jetstream.KeyValue, error) {
	if p := r.Picture(bucket); p != nil {
		return p, nil
	}
	return nil, fmt.Errorf("%s reads %s by its owner's first view alone, and does not follow it", r.Module, bucket)
}

// Followers are the followers, in the order they were added.
func (r *ProcessReader) Followers() []*Follower {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]*Follower, 0, len(r.order))
	for _, b := range r.order {
		out = append(out, r.pics[b].f)
	}
	return out
}

func (r *ProcessReader) finding(s Status, raised bool) {
	if raised {
		r.logf("the first view %s of %s's %s is unanswered past its bound: %s", s.Tool, s.Owner, s.State, s.Error)
	}
	if r.Findings == nil {
		return
	}
	kv := r.Findings()
	if kv == nil {
		return
	}
	at := time.Now()
	if r.Now != nil {
		at = r.Now()
	}
	key, value, err := UnansweredFinding(r.Module, s, at)
	if err == nil && raised {
		_, err = kv.Put(context.Background(), key, value)
	} else if err == nil {
		err = kv.Delete(context.Background(), key)
	}
	if err != nil {
		r.logf("views: the finding of the first view %s is not written: %v", s.Tool, err)
	}
}

// Hear is the handler of the owner's views family on the reader's durable consumer (cred.EventConsumer on the owner's
// views stream): each event to the follower of its state.
func (r *ProcessReader) Hear(ctx context.Context, owner string) func(jetstream.Msg) {
	fs := map[string]*Follower{}
	for _, f := range r.Followers() {
		if f.Owner == owner {
			fs[f.State] = f
		}
	}
	return Hear(ctx, fs, r.logf)
}

// Owners are the owners followed, sorted.
func (r *ProcessReader) Owners() []string {
	var out []string
	for _, f := range r.Followers() {
		if !slices.Contains(out, f.Owner) {
			out = append(out, f.Owner)
		}
	}
	slices.Sort(out)
	return out
}

// Ask asks every first view until answered, each on a goroutine of its own, so the process starts at once and waits for
// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (KeepStatus).
func (r *ProcessReader) Ask(ctx context.Context, said func(Status)) {
	for _, f := range r.Followers() {
		if said != nil {
			f.Said = said
		}
		go func() {
			if err := f.Start(ctx); err != nil {
				r.logf("the first view %s was not answered: %v", f.Tool, err)
			}
		}()
	}
}

// Ready is closed once every followed state had its first view.
func (r *ProcessReader) Ready() <-chan struct{} {
	fs := r.Followers()
	done := make(chan struct{})
	go func() {
		for _, f := range fs {
			<-f.Ready()
		}
		close(done)
	}()
	return done
}

// Picture is one owner's state as the process reads it: a read-only jetstream.KeyValue over a Follower. Its
// revisions are the picture's own, one per change it took, so a watcher sees them rise; they are not the owner's. An
// entry's Created is when the owner's bus wrote the key (Follower.Changed), zero when the owner did not say it.
type Picture struct {
	jetstream.KeyValue // never set: every method a reader may call is the picture's own

	bucket string
	ctx    context.Context
	f      *Follower

	mu       sync.Mutex
	seq      uint64
	revs     map[string]uint64
	ats      map[string]time.Time // when the owner's bus wrote each key, as the follower handed it (zero unsaid)
	watchers []*watcher
}

// Follower is the picture's follower (its status, a gap or a hand ask).
func (p *Picture) Follower() *Follower { return p.f }

func (p *Picture) changed(key string, value json.RawMessage, deleted bool, at time.Time) {
	p.mu.Lock()
	p.seq++
	if p.revs == nil {
		p.revs = map[string]uint64{}
	}
	p.revs[key] = p.seq
	if p.ats == nil {
		p.ats = map[string]time.Time{}
	}
	p.ats[key] = at
	op := jetstream.KeyValuePut
	if deleted {
		op, value = jetstream.KeyValueDelete, nil
	}
	e := entry{bucket: p.bucket, key: key, value: value, rev: p.seq, at: at, op: op}
	ws := slices.Clone(p.watchers)
	p.mu.Unlock()
	for _, w := range ws {
		w.send(e)
	}
}

// answered is nil once the first view is answered, else the error a read before it gets (never a wait).
func (p *Picture) answered() error {
	select {
	case <-p.f.Ready():
		return nil
	default:
		return fmt.Errorf("no first view of %s yet: %s has not answered", p.bucket, p.f.Tool)
	}
}

func (p *Picture) ready(ctx context.Context) error {
	select {
	case <-p.f.Ready():
		return nil
	case <-ctx.Done():
		return fmt.Errorf("no first view of %s yet (%s has not answered): %w", p.bucket, p.f.Tool, ctx.Err())
	}
}

// Bucket is the bus name of the owner's state the picture stands for.
func (p *Picture) Bucket() string { return p.bucket }

// Get is one key; before the first view is answered it is an error, never a wait.
func (p *Picture) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
	if err := p.answered(); err != nil {
		return nil, err
	}
	v, ok := p.f.Get(key)
	if !ok {
		return nil, jetstream.ErrKeyNotFound
	}
	p.mu.Lock()
	rev, at := p.revs[key], p.ats[key]
	p.mu.Unlock()
	return entry{bucket: p.bucket, key: key, value: v, rev: rev, at: at, op: jetstream.KeyValuePut}, nil
}

// Keys is every key; none is jetstream.ErrNoKeysFound, as the bus says it; before the first view is answered it is an
// error, never a wait.
func (p *Picture) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) {
	if err := p.answered(); err != nil {
		return nil, err
	}
	var keys []string
	for k := range p.f.Values() {
		keys = append(keys, k)
	}
	if len(keys) == 0 {
		return nil, jetstream.ErrNoKeysFound
	}
	slices.Sort(keys)
	return keys, nil
}

// ListKeys is Keys on a channel.
func (p *Picture) ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error) {
	keys, err := p.Keys(ctx, opts...)
	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, err
	}
	ch := make(chan string, len(keys))
	for _, k := range keys {
		ch <- k
	}
	close(ch)
	return lister{ch}, nil
}

// WatchAll is Watch of every key.
func (p *Picture) WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
	return p.Watch(ctx, ">", opts...)
}

// Watch hands, once the first view is answered, every current value of the keys matching (a subject pattern: `*` one
// token, `>` the rest), then nil, then each change, as the bus's watch does. Options are not read: a picture holds no
// history, and deletes are always handed.
func (p *Picture) Watch(ctx context.Context, keys string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
	w := &watcher{ch: make(chan jetstream.KeyValueEntry, 64), match: keys, wake: make(chan struct{}, 1), stop: make(chan struct{})}
	go func() {
		defer close(w.ch)
		if p.ready(ctx) != nil {
			return
		}
		p.mu.Lock()
		p.watchers = append(p.watchers, w)
		p.mu.Unlock()
		defer p.unwatch(w)
		vals := p.f.Values()
		names := make([]string, 0, len(vals))
		for k := range vals {
			names = append(names, k)
		}
		slices.Sort(names)
		p.mu.Lock()
		first := make([]jetstream.KeyValueEntry, 0, len(names)+1)
		for _, k := range names {
			if matches(keys, k) {
				first = append(first, entry{bucket: p.bucket, key: k, value: vals[k], rev: p.revs[k], at: p.ats[k], op: jetstream.KeyValuePut})
			}
		}
		p.mu.Unlock()
		first = append(first, nil)
		w.run(ctx, first)
	}()
	return w, nil
}

func (p *Picture) unwatch(w *watcher) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.watchers = slices.DeleteFunc(p.watchers, func(x *watcher) bool { return x == w })
}

// The writes and the reads a picture cannot answer are refused, never handed to a bucket.
func (p *Picture) Put(context.Context, string, []byte) (uint64, error)       { return 0, ErrReadOnly }
func (p *Picture) PutString(context.Context, string, string) (uint64, error) { return 0, ErrReadOnly }
func (p *Picture) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) {
	return 0, ErrReadOnly
}
func (p *Picture) Update(context.Context, string, []byte, uint64) (uint64, error) {
	return 0, ErrReadOnly
}
func (p *Picture) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly }
func (p *Picture) Purge(context.Context, string, ...jetstream.KVDeleteOpt) error  { return ErrReadOnly }
func (p *Picture) PurgeDeletes(context.Context, ...jetstream.KVPurgeOpt) error    { return ErrReadOnly }
func (p *Picture) GetRevision(context.Context, string, uint64) (jetstream.KeyValueEntry, error) {
	return nil, errors.New("a first view holds no history")
}
func (p *Picture) History(context.Context, string, ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) {
	return nil, errors.New("a first view holds no history")
}
func (p *Picture) WatchFiltered(ctx context.Context, keys []string, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
	if len(keys) != 1 {
		return nil, errors.New("a first view watches one pattern")
	}
	return p.Watch(ctx, keys[0], opts...)
}
func (p *Picture) ListKeysFiltered(ctx context.Context, filters ...string) (jetstream.KeyLister, error) {
	keys, err := p.Keys(ctx)
	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, err
	}
	ch := make(chan string, len(keys))
	for _, k := range keys {
		if len(filters) == 0 || slices.ContainsFunc(filters, func(f string) bool { return matches(f, k) }) {
			ch <- k
		}
	}
	close(ch)
	return lister{ch}, nil
}
func (p *Picture) Status(context.Context) (jetstream.KeyValueStatus, error) {
	return nil, errors.New("a first view has no bucket status: the owner's read tool says its fill")
}

// matches says whether key matches a subject pattern of the bus's watch.
func matches(pattern, key string) bool {
	if pattern == "" || pattern == ">" {
		return true
	}
	pt, kt := strings.Split(pattern, "."), strings.Split(key, ".")
	for i, t := range pt {
		if t == ">" {
			return len(kt) > i
		}
		if i >= len(kt) || (t != "*" && t != kt[i]) {
			return false
		}
	}
	return len(pt) == len(kt)
}

// watcher hands a watch's entries in order: the current values first, then each change heard, never blocking the
// follower (the changes wait in a queue of its own).
type watcher struct {
	ch    chan jetstream.KeyValueEntry
	match string
	wake  chan struct{}
	stop  chan struct{}
	once  sync.Once

	mu    sync.Mutex
	queue []jetstream.KeyValueEntry
}

func (w *watcher) send(e entry) {
	if !matches(w.match, e.key) {
		return
	}
	w.mu.Lock()
	w.queue = append(w.queue, e)
	w.mu.Unlock()
	select {
	case w.wake <- struct{}{}:
	default:
	}
}

func (w *watcher) run(ctx context.Context, first []jetstream.KeyValueEntry) {
	out := func(e jetstream.KeyValueEntry) bool {
		select {
		case w.ch <- e:
			return true
		case <-ctx.Done():
			return false
		case <-w.stop:
			return false
		}
	}
	for _, e := range first {
		if !out(e) {
			return
		}
	}
	for {
		w.mu.Lock()
		q := w.queue
		w.queue = nil
		w.mu.Unlock()
		for _, e := range q {
			if !out(e) {
				return
			}
		}
		select {
		case <-w.wake:
		case <-ctx.Done():
			return
		case <-w.stop:
			return
		}
	}
}

func (w *watcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch }

func (w *watcher) Stop() error {
	w.once.Do(func() { close(w.stop) })
	return nil
}

type lister struct{ ch chan string }

func (l lister) Keys() <-chan string { return l.ch }
func (l lister) Stop() error         { return nil }

type entry struct {
	bucket, key string
	value       []byte
	rev         uint64
	at          time.Time
	op          jetstream.KeyValueOp
}

func (e entry) Bucket() string                  { return e.bucket }
func (e entry) Key() string                     { return e.key }
func (e entry) Value() []byte                   { return e.value }
func (e entry) Revision() uint64                { return e.rev }
func (e entry) Created() time.Time              { return e.at }
func (e entry) Delta() uint64                   { return 0 }
func (e entry) Operation() jetstream.KeyValueOp { return e.op }

// StreamOf is the bus name of owner's views family (IndexerStream, AssignerStream): `<short>-views`
// as the owner's old `streams` section names it.
func StreamOf(owner string) string {
	return catalogue.StreamBusName(strings.TrimPrefix(owner, "mesh-delivery-") + "-views")
}

// Start hears each owner's views family on the process's durable consumer of it (its manifest's `consumes`
// `<owner>.views.<state>`), keeps each status in the process's state `first-views` and hears the hand asks the tools
// bundle's `<reader>_first_view_again` writes there (StatusTools), then asks every first view (Ask). A consumer
// the bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand.
func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) {
	for _, owner := range r.Owners() {
		stream := StreamOf(owner)
		cons, err := cred.EventConsumer(ctx, js, stream)
		if err == nil {
			_, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf)
		}
		if err != nil {
			r.logf("the consumer %s on %s is not heard (%v): %s's states are the first view's alone, asked again by hand, until the %s starts again",
				cred.EventConsumerName(), stream, err, owner, r.Module)
		}
	}
	var said func(Status)
	if status != nil {
		status.Ready(func(kv jetstream.KeyValue) {
			if err := HearAgain(ctx, kv, r.Followers(), r.logf); err != nil {
				r.logf("views: the hand asks of %s's first views are not heard (%v) until it starts again", r.Module, err)
			}
		})
		said = func(s Status) {
			if kv := status.KV(); kv != nil {
				KeepStatus(ctx, kv, r.logf)(s)
			}
		}
	}
	r.Ask(ctx, said)
}