mesh-sdk/go/views/reader.go

package views

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/handacts"
)

// Reader is a module's pictures of the states of the indexer and the assigner it used to read (novox/hq ADR 0615 §1): it
// asks each owner's read tool for the first view, subscribes the owner's views family, and answers the module's own
// `Keys` and `Get` of a state from the picture, by the name the module's `reads` used to carry (`<owner>.<state>`).
// A module in a stdio tools bundle declares `invokes` for the tools and `consumes` for the events, and names no `reads`
// of these owners any more. Pictures are kept in memory only: the owner's state is the truth, and a restart asks again.
type Reader struct {
	// Module is the reading module, named in the findings it writes.
	Module string
	// Findings writes and clears this module's finding of an unanswered first view (its own state `findings`); nil
	// says it in the log only.
	Findings interface {
		Put(key string, value []byte) error
		Delete(key string) error
	}
	Logf func(string, ...any)
	// Changed is told of each key a followed state changes, after the picture holds it (state is `<owner>.<state>`).
	Changed func(state, key string, value json.RawMessage, deleted bool)
	// Acts records a re-read by hand through the seat mesh-hand-acts before it is asked (the zero Recorder records nothing
	// and the re-read is refused).
	Acts handacts.Recorder
	// 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
	followers map[string]*Follower // by "<owner>.<state>"
	order     []string
	byOwner   map[string]map[string]*Follower // owner -> state -> follower
	watchers  map[string][]func(key string, value []byte, gone bool)
}

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

// Follow adds a state of an owner to follow, asked through the owner's read tool (`<module>.<tool>`, as the reader's
// `invokes` names it), with `values` so the answer is the first view. Call it before Start.
func (r *Reader) Follow(owner, state, tool string) *Follower {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.followers == nil {
		r.followers, r.byOwner = map[string]*Follower{}, map[string]map[string]*Follower{}
	}
	name := owner + "." + state
	f := &Follower{Owner: owner, Tool: tool, State: state, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait}
	f.Ask = func(_ context.Context, after string) (Answer, error) {
		body := map[string]any{"values": true}
		if after != "" {
			body["after"] = after
		}
		raw, err := stdio.Ask(tool, body)
		if err != nil {
			return Answer{}, err
		}
		return DecodeAnswer(raw)
	}
	f.Raise = func(s Status) { r.raise(s) }
	f.Clear = func(s Status) { r.clear(s) }
	f.Changed = func(key string, value json.RawMessage, deleted bool, _ time.Time) {
		r.mu.Lock()
		ws := append([]func(string, []byte, bool){}, r.watchers[name]...)
		r.mu.Unlock()
		for _, w := range ws {
			w(key, value, deleted)
		}
		if r.Changed != nil {
			r.Changed(name, key, value, deleted)
		}
	}
	r.followers[name] = f
	r.order = append(r.order, name)
	if r.byOwner[owner] == nil {
		r.byOwner[owner] = map[string]*Follower{}
	}
	r.byOwner[owner][state] = f
	return f
}

func (r *Reader) findingKey(s Status) (string, []byte, error) {
	return UnansweredFinding(r.Module, s, r.nowTime())
}

func (r *Reader) nowTime() time.Time {
	if r.Now != nil {
		return r.Now().UTC()
	}
	return time.Now().UTC()
}

func (r *Reader) raise(s Status) {
	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
	}
	key, value, err := r.findingKey(s)
	if err == nil {
		err = r.Findings.Put(key, value)
	}
	if err != nil {
		r.logf("the finding of the unanswered first view %s could not be written: %v", s.Tool, err)
	}
}

func (r *Reader) clear(s Status) {
	if r.Findings == nil {
		return
	}
	key, _, err := r.findingKey(s)
	if err == nil {
		err = r.Findings.Delete(key)
	}
	if err != nil {
		r.logf("the finding of the first view %s could not be cleared: %v", s.Tool, err)
	}
}

// Start subscribes each owner's views family (the module's `consumes`: `<owner>.views.<state>`) and asks every first
// view until it is answered, each on a goroutine of its own, so a module starts at once and a reader whose owner is down
// waits, saying so (ADR 0615 §6). It returns when the subscriptions are taken.
func (r *Reader) Start(ctx context.Context) {
	r.mu.Lock()
	names := append([]string(nil), r.order...)
	r.mu.Unlock()
	for _, name := range names {
		f := r.followers[name]
		go func() {
			for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
				err := stdio.Subscribe(f.Owner+"."+EventName(f.State), r.hear(ctx, f))
				if err == nil {
					break
				}
				r.logf("not hearing %s's %s events yet (%v); asking again in %s", f.Owner, f.State, err, wait)
				select {
				case <-ctx.Done():
					return
				case <-time.After(wait):
				}
			}
			if err := f.Start(ctx); err != nil {
				r.logf("the first view %s was not answered: %v", f.Tool, err)
			}
		}()
	}
}

// hear is the handler of one followed state's events: HearEvent routes it, and a gap asks that first view once more as a
// gap, said in the log.
func (r *Reader) hear(ctx context.Context, f *Follower) func(stdio.Envelope) error {
	return func(e stdio.Envelope) error {
		if err := HearEvent(ctx, map[string]*Follower{f.State: f}, e.Body, r.logf); err != nil {
			r.logf("views: %s is not a views event, dropped: %v", e.Key, err)
		}
		return nil
	}
}

// Ready is closed once every followed state had its first view.
func (r *Reader) Ready() <-chan struct{} {
	r.mu.Lock()
	fs := make([]*Follower, 0, len(r.followers))
	for _, name := range r.order {
		fs = append(fs, r.followers[name])
	}
	r.mu.Unlock()
	done := make(chan struct{})
	go func() {
		for _, f := range fs {
			<-f.Ready()
		}
		close(done)
	}()
	return done
}

// Keys is the keys of a followed state (`<owner>.<state>`); a state not followed is an error, never a read of it.
func (r *Reader) Keys(state string) ([]string, error) {
	f, err := r.picture(state)
	if err != nil {
		return nil, err
	}
	vals := f.Values()
	keys := make([]string, 0, len(vals))
	for k := range vals {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys, nil
}

// Get is one key of a followed state.
func (r *Reader) Get(state, key string) ([]byte, bool, error) {
	f, err := r.picture(state)
	if err != nil {
		return nil, false, err
	}
	v, ok := f.Get(key)
	return v, ok, nil
}

// Values is every key of a followed state, by key.
func (r *Reader) Values(state string) (map[string][]byte, error) {
	f, err := r.picture(state)
	if err != nil {
		return nil, err
	}
	out := map[string][]byte{}
	for k, v := range f.Values() {
		out[k] = v
	}
	return out, nil
}

func (r *Reader) picture(state string) (*Follower, error) {
	r.mu.Lock()
	f := r.followers[state]
	r.mu.Unlock()
	if f == nil {
		return nil, fmt.Errorf("%s is not a state this module follows", state)
	}
	select {
	case <-f.Ready():
		return f, nil
	default:
		return nil, fmt.Errorf("no first view of %s yet: %s has not answered", state, f.Tool)
	}
}

// Watch is a followed state as a module's store watches one: it returns once the first view was answered and every
// current value was handed to handle, and then hands every change for as long as the process runs (gone is a key
// deleted). A change heard while the current values are handed over may be handed twice; a handler is idempotent.
func (r *Reader) Watch(state string, handle func(key string, value []byte, gone bool)) error {
	r.mu.Lock()
	f := r.followers[state]
	if f == nil {
		r.mu.Unlock()
		return fmt.Errorf("%s is not a state this module follows", state)
	}
	r.mu.Unlock()
	<-f.Ready()
	r.mu.Lock()
	if r.watchers == nil {
		r.watchers = map[string][]func(string, []byte, bool){}
	}
	r.watchers[state] = append(r.watchers[state], handle)
	r.mu.Unlock()
	for k, v := range f.Values() {
		handle(k, v, false)
	}
	return nil
}

// Follows says whether the state is one this reader follows.
func (r *Reader) Follows(state string) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.followers[state] != nil
}

// Statuses is each followed state's status, in the order they were added.
func (r *Reader) Statuses() []Status {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]Status, 0, len(r.order))
	for _, name := range r.order {
		out = append(out, r.followers[name].Status())
	}
	return out
}

// Tools are the reader's two tools (design 25 § 12.2): `<prefix>_first_views` says where each first view stands and
// `<prefix>_first_view_again` asks one again by hand (a hand act, counted in the status).
func (r *Reader) Tools(prefix string) []stdio.Tool {
	return []stdio.Tool{
		{Name: prefix + "_first_views",
			Description: "Where each first view this module keeps stands (novox/hq ADR 0615): the owner and its tool, the revision the picture holds, when it was asked and answered, how many asks (by hand among them) and whether its condition is raised. (r)",
			Run:         func(map[string]any) (any, error) { return map[string]any{"views": r.Statuses()}, nil }},
		{Name: prefix + "_first_view_again",
			Description: "Ask a first view again by hand, for the state named (<owner>.<state>), or every one with none: a hand act, for a picture the operator doubts. Answers where it stands after. (a)",
			Input: map[string]any{"type": "object", "properties": map[string]any{
				"state": map[string]any{"type": "string", "description": "<owner>.<state>; none asks every first view"}}},
			Run: func(args map[string]any) (any, error) {
				want, _ := args["state"].(string)
				want = strings.TrimSpace(want)
				r.mu.Lock()
				var fs []*Follower
				for _, name := range r.order {
					if want == "" || want == name {
						fs = append(fs, r.followers[name])
					}
				}
				r.mu.Unlock()
				if len(fs) == 0 {
					return nil, fmt.Errorf("%q is not a state this module follows", want)
				}
				// A hand act, recorded through the seat mesh-hand-acts before it is done; one that cannot be recorded is
				// said and nothing is asked.
				why := "the operator asked"
				if _, err := r.Acts.Record(handacts.Request{What: "asked a first view again by hand (" + r.Module + ")", Why: why,
					Cause: "first-view-again", Verb: "first-view-again"}); err != nil {
					return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was asked again: %w", err)
				}
				for _, f := range fs {
					if err := f.Again(context.Background(), why); err != nil {
						return nil, err
					}
				}
				return map[string]any{"views": r.Statuses()}, nil
			}},
	}
}

// OwnFindings writes a reader's findings into its own state `findings` through its runtime, where the condition keeper
// raises and clears conditions from them: the Findings of a Reader whose module keeps the state `findings`.
type OwnFindings struct{}

// Put writes one finding.
func (OwnFindings) Put(key string, value []byte) error {
	_, err := stdio.State("findings").Put(key, json.RawMessage(value))
	return err
}

// Delete clears one finding.
func (OwnFindings) Delete(key string) error { return stdio.State("findings").Delete(key) }