mesh-sdk/go/views/reader.go

package views

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

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

// Reader is a module's pictures of the owners' states 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 FindingWriter
	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 before it is asked (nil records nothing
	// and the re-read is refused).
	Acts HandActs
	// 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)
	// adopted are the followers a seat's client made (Adopt), by "<owner>.<state>": their events are a seat's, heard by
	// the client's own handler, not an owner's views family.
	adopted map[string]adoption
}

// adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler.
type adoption struct {
	seat, event string
	hear        func(context.Context, []byte) error
}

// local is an adopted follower's event as the stdio runtime subscribes it, `<seat>.<event>`.
func (a adoption) local() string { return a.seat + "." + a.event }

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)
	}
	r.wireLocked(name, f, nil)
	return f
}

// BranchState is the name of a per-branch state's view: `<state>.<branch>`, the branch escaped as a key's name is (design
// 47 G §479), so a reader names the view `<owner>.<state>.<branch>`. A state that is not per branch keeps its name.
func BranchState(state, branch string) string { return state + "." + keys.EscapeName(branch) }

// FollowBranch is Follow for a state kept per branch: the view of one branch is named `<owner>.<state>.<branch>`
// (BranchState), so a reader follows several branches of one state, each its own view and its own row of
// `<module>_first_views` (design 47 G §479). Call it before Start.
func (r *Reader) FollowBranch(owner, state, branch, tool string) *Follower {
	return r.Follow(owner, BranchState(state, branch), tool)
}

// Adopt takes a follower a seat's client made (its Ask a seat's verb, such as the catalog half's view) as one of the
// reader's first views, so the client's callers write no glue: the reader sets its conditions (the unanswered first view
// and view-waits, in the reader's findings), its log and clock, and hands each change to Watch, Changed and the
// follower's own Changed, set before; Keys, Get, Values, Watch, Statuses and the two tools name it `<owner>.<state>`.
// Its events are the seat's event (`<seat>.<event>`, as the reader's `consumes` names it), which Start subscribes and
// hands to hear, the client's handler. Call it before Start; a state followed already is refused.
func (r *Reader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) error {
	if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil {
		return errors.New("views: an adopted follower names its owner and state, and its seat's event and handler")
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.followers == nil {
		r.followers, r.byOwner = map[string]*Follower{}, map[string]map[string]*Follower{}
	}
	name := f.Owner + "." + f.State
	if r.followers[name] != nil {
		return fmt.Errorf("views: %s is followed already", name)
	}
	if f.Logf == nil {
		f.Logf = r.Logf
	}
	if f.Now == nil {
		f.Now = r.Now
	}
	if f.Bound == 0 {
		f.Bound = r.Bound
	}
	if f.Wait == 0 {
		f.Wait = r.Wait
	}
	r.wireLocked(name, f, f.Changed)
	if r.adopted == nil {
		r.adopted = map[string]adoption{}
	}
	r.adopted[name] = adoption{seat: seat, event: event, hear: hear}
	return nil
}

// wireLocked makes f one of the reader's first views under name: its conditions are the reader's, and each change goes
// to the watchers, Changed and own (the follower's own Changed, or nil). The caller holds mu.
func (r *Reader) wireLocked(name string, f *Follower, own func(string, json.RawMessage, bool, time.Time)) {
	f.Raise = func(s Status) { r.raise(s) }
	f.Clear = func(s Status) { r.clear(s) }
	f.RaiseWaits = func(s Status) { r.waits(s, true) }
	f.ClearWaits = func(s Status) { r.waits(s, false) }
	f.Changed = func(key string, value json.RawMessage, deleted bool, at 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)
		}
		if own != nil {
			own(key, value, deleted, at)
		}
	}
	r.followers[name] = f
	r.order = append(r.order, name)
	if r.byOwner[f.Owner] == nil {
		r.byOwner[f.Owner] = map[string]*Follower{}
	}
	r.byOwner[f.Owner][f.State] = f
}

// waits writes (raised) or clears the reader's view-waits finding of s (design 47 *A part with no view says it waits*).
func (r *Reader) waits(s Status, raised bool) {
	if raised {
		r.logf("%s has waited for %s's view %s since %s", r.Module, s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339))
	}
	if r.Findings == nil {
		return
	}
	f := ViewWaitsFinding(r.Module, s, r.nowTime())
	var err error
	if raised {
		err = r.Findings.Put(f)
	} else {
		err = r.Findings.Delete(f)
	}
	if err != nil {
		r.logf("the view-waits finding of %s is not written: %v", s.Tool, err)
	}
}

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
	}
	if err := r.Findings.Put(UnansweredFinding(r.Module, s, r.nowTime())); 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
	}
	if err := r.Findings.Delete(UnansweredFinding(r.Module, s, r.nowTime())); 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...)
	adopted := maps.Clone(r.adopted)
	r.mu.Unlock()
	for _, name := range names {
		f := r.followers[name]
		pattern, handle := f.Owner+"."+EventName(f.State), r.hear(ctx, f)
		if a, ok := adopted[name]; ok {
			pattern, handle = a.local(), r.hearAdopted(ctx, a)
		}
		go func() {
			for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
				err := stdio.Subscribe(pattern, handle)
				if err == nil {
					break
				}
				r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, 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
	}
}

// hearAdopted is the handler of an adopted follower's seat event: the client's handler takes it, and what it cannot read is
// said in the log.
func (r *Reader) hearAdopted(ctx context.Context, a adoption) func(stdio.Envelope) error {
	return func(e stdio.Envelope) error {
		if err := a.hear(ctx, e.Body); err != nil {
			r.logf("views: %s is not taken: %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 before it is done; one that cannot be recorded is
				// said and nothing is asked.
				why := "the operator asked"
				if err := recordHandAct(r.Acts, HandAct{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, 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`. Encode is
// the module's: it turns a Finding into the state's key and value as the seat mesh-conditions' contract lays them out.
type OwnFindings struct {
	Encode func(Finding) (key string, value []byte, err error)
}

// Put writes one finding.
func (o OwnFindings) Put(f Finding) error {
	key, value, err := o.Encode(f)
	if err != nil {
		return err
	}
	_, err = stdio.State("findings").Put(key, json.RawMessage(value))
	return err
}

// Delete clears one finding.
func (o OwnFindings) Delete(f Finding) error {
	key, _, err := o.Encode(f)
	if err != nil {
		return err
	}
	return stdio.State("findings").Delete(key)
}