mesh-sdk/go/views/follow.go

package views

import (
	"cmp"
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"slices"
	"sync"
	"time"

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

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

// Follower is a reader's picture of one state of an owner (ADR 0615 §4 to §6): it asks the owner's first-view tool at
// its start, then keeps the picture from the owner's events alone, asking the tool once more only on a gap in revisions
// (said in its log), never on a timer. Started while the owner is down, it keeps asking, never reading the owner's state
// instead, and past Bound raises its condition naming the owner and the tool; the answer clears it.
type Follower struct {
	Owner string
	// Tool is the owner's tool it asks, as the reader's invokes names it (<module>.<tool>).
	Tool  string
	State string
	// Ask calls the tool for the page after after.
	Ask func(ctx context.Context, after string) (Answer, error)
	// Raise and Clear are the reader's condition of this first view (past Bound; cleared by the answer). Nil says
	// nothing.
	Raise func(Status)
	Clear func(Status)
	// Changed is told of each key the picture changes (deleted: the key is gone), after the change is held, with when the
	// owner's bus wrote it: the event's time, or for a key a first view brought the time its answer says (Answer.Written;
	// zero when the owner does not say it, and for a key the first view no longer holds).
	Changed func(key string, value json.RawMessage, deleted bool, at time.Time)
	// Said is told of the status each time an ask starts, is answered or fails, and of a raise and a clear: where a
	// reader keeps it on the bus for its read tool (StatusTools).
	Said func(Status)
	Logf func(string, ...any)
	Now  func() time.Time
	// Bound is how long it asks before it raises; zero is views.Bound. Wait is the first wait between two asks that
	// failed, doubling to a minute; zero is five seconds.
	Bound time.Duration
	Wait  time.Duration

	mu     sync.Mutex
	values map[string]json.RawMessage
	st     Status
	asking bool
	// during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read
	// before it was written.
	during []Event
	// heardRev is the highest revision of the owner's state (made at heardCreated) any event handed to Apply carried,
	// whether it was taken into the picture, passed over or heard while an ask ran: what the picture must reach, so a
	// reader that heard it knows it is behind without a further event.
	heardRev     uint64
	heardCreated time.Time
	// consumerMade is when the bus made the reader's consumer of the owner's events (EventHub.OnBound): an event said
	// before it is not delivered to it, so a picture asked before it is behind.
	consumerMade time.Time
	// catching is set while catchUp runs: one catch-up at a time.
	catching bool
	ready    chan struct{}
}

// Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision
// its picture holds, when it asked and when it was answered, how many asks, the hand acts among them, and whether its
// condition is raised.
type Status struct {
	Owner    string    `json:"owner"`
	Tool     string    `json:"tool"`
	State    string    `json:"state"`
	Revision uint64    `json:"revision"`
	Created  time.Time `json:"created,omitzero"`
	Keys     int       `json:"keys"`
	Asked    time.Time `json:"asked,omitzero"`
	Answered time.Time `json:"answered,omitzero"`
	Asks     int       `json:"asks"`
	ByHand   int       `json:"by-hand"`
	Raised   bool      `json:"raised"`
	// Heard is the highest revision of the owner's state any event heard carried (made at HeardCreated), taken into the
	// picture or not; Behind says why the picture is short of it, or of its consumer's making, and an ask for it runs.
	Heard        uint64    `json:"heard,omitempty"`
	HeardCreated time.Time `json:"heard-created,omitzero"`
	Behind       string    `json:"behind,omitempty"`
	// Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed.
	Why   string `json:"why,omitempty"`
	Error string `json:"error,omitempty"`
}

func (f *Follower) now() time.Time {
	if f.Now != nil {
		return f.Now().UTC()
	}
	return time.Now().UTC()
}

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

func (f *Follower) init() {
	if f.values == nil {
		f.values = map[string]json.RawMessage{}
		f.ready = make(chan struct{})
		f.st = Status{Owner: f.Owner, Tool: f.Tool, State: f.State}
	}
}

func (f *Follower) say(s Status) {
	if f.Said != nil {
		f.mu.Lock()
		s = f.statusLocked(s)
		f.mu.Unlock()
		f.Said(s)
	}
}

// statusLocked is s with what the follower holds beside it: its keys, the highest revision heard and why it is behind.
// The caller holds mu.
func (f *Follower) statusLocked(s Status) Status {
	s.Keys = len(f.values)
	s.Heard, s.HeardCreated, s.Behind = f.heardRev, f.heardCreated, f.behindLocked()
	return s
}

// Status is the follower's status now.
func (f *Follower) Status() Status {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.init()
	return f.statusLocked(f.st)
}

// Ready is closed once a first view was answered.
func (f *Follower) Ready() <-chan struct{} {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.init()
	return f.ready
}

// Values is a copy of the picture.
func (f *Follower) Values() map[string]json.RawMessage {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.init()
	return maps.Clone(f.values)
}

// Get is one key of the picture.
func (f *Follower) Get(key string) (json.RawMessage, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.init()
	v, ok := f.values[key]
	return v, ok
}

// Start asks the first view until it is answered or ctx ends (ADR 0615 §6): each failed ask waits, doubling, and past
// Bound raises the condition. It returns once answered.
func (f *Follower) Start(ctx context.Context) error { return f.askUntil(ctx, "start", false) }

// Again asks the first view once more by hand (a re-read tool, recorded by the caller as a hand act), until answered or
// ctx ends.
func (f *Follower) Again(ctx context.Context, why string) error {
	return f.askUntil(ctx, "by hand: "+why, true)
}

func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error {
	f.mu.Lock()
	f.init()
	if f.asking {
		f.mu.Unlock()
		return nil // one ask at a time; the one running brings the picture to its newest
	}
	f.asking = true
	f.during = nil
	f.st.Why = why
	if byHand {
		f.st.ByHand++
	}
	f.mu.Unlock()
	released := false
	defer func() {
		if !released {
			f.mu.Lock()
			f.asking = false
			f.mu.Unlock()
		}
	}()
	bound, wait := f.Bound, f.Wait
	if bound == 0 {
		bound = Bound
	}
	if wait == 0 {
		wait = 5 * time.Second
	}
	first := f.now()
	for {
		err := f.askOnce(ctx)
		if err == nil {
			// An event heard that the answer does not hold yet (the owner answered before it wrote it, or the event came
			// before the first view and was passed over, or it was a gap heard while this ask ran) leaves the picture
			// behind. No further event is waited for: it is asked again on a goroutine of its own, decided in the same
			// check that ends this ask, so an event heard after the check finds no ask running and asks itself (ADR 0615
			// §5). The ask that returns has brought the picture as far as the owner answered; the module need not wait.
			f.mu.Lock()
			behind := f.behindLocked()
			f.asking, released = false, true
			start := behind != "" && !f.catching
			if start {
				f.catching = true
			}
			f.mu.Unlock()
			if start {
				f.logf("views: %s's %s is behind an event heard (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
				go f.catchUp(ctx)
			}
			return nil
		}
		f.mu.Lock()
		f.st.Error = err.Error()
		raise := !f.st.Raised && f.now().Sub(first) >= bound
		if raise {
			f.st.Raised = true
		}
		s := f.st
		f.mu.Unlock()
		f.logf("the first view %s of %s's %s is not answered (%v): asked again in %s", f.Tool, f.Owner, f.State, err, wait)
		if raise && f.Raise != nil {
			f.Raise(s)
		}
		f.say(s)
		if !f.pause(ctx, wait) {
			return ctx.Err()
		}
		if wait *= 2; wait > time.Minute {
			wait = time.Minute
		}
	}
}

// catchUp asks the first view once more, each time after a pause (the reader's Wait, doubling to a minute: an owner that
// answers from before its last write is not asked in a hot loop), until the picture holds every revision heard. It
// holds no ask while it waits, so a gap, an ask by hand or an event heard meanwhile is not held up by it.
func (f *Follower) catchUp(ctx context.Context) {
	wait := f.Wait
	if wait == 0 {
		wait = 5 * time.Second
	}
	for {
		if !f.pause(ctx, wait) {
			f.mu.Lock()
			f.catching = false
			f.mu.Unlock()
			return
		}
		f.mu.Lock()
		behind := f.behindLocked()
		if behind == "" {
			f.catching = false
			f.mu.Unlock()
			return
		}
		f.mu.Unlock()
		f.logf("views: %s's %s is still behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
		_ = f.askUntil(ctx, "behind: "+behind, false)
		if wait *= 2; wait > time.Minute {
			wait = time.Minute
		}
	}
}

// pause waits a context's bound between two asks, no timer of its own; false once ctx ended.
func (f *Follower) pause(ctx context.Context, wait time.Duration) bool {
	p, cancel := context.WithTimeout(ctx, wait)
	<-p.Done()
	cancel()
	return ctx.Err() == nil
}

// behindLocked says why the picture is behind an event heard, or "" when it holds every revision heard. The caller
// holds mu.
func (f *Follower) behindLocked() string {
	switch {
	case !f.consumerMade.IsZero() && f.consumerMade.After(f.st.Asked) && !f.st.Asked.IsZero():
		return fmt.Sprintf("the consumer of its events was made at %s, after the first view was asked at %s: what was said between is not delivered", f.consumerMade.Format(time.RFC3339), f.st.Asked.Format(time.RFC3339))
	case f.heardCreated.IsZero():
		return ""
	case f.heardCreated.After(f.st.Created):
		return fmt.Sprintf("the state was made again at %s, the picture holds the one made at %s", f.heardCreated.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
	case f.heardCreated.Equal(f.st.Created) && f.heardRev > f.st.Revision:
		return fmt.Sprintf("revision %d heard, the picture holds %d", f.heardRev, f.st.Revision)
	}
	return ""
}

// ConsumerMade tells the follower when the bus made its consumer of the owner's events. A picture asked before that may
// lack an event said between its ask and the consumer, which the consumer never delivers and no later event shows: it
// asks again, by itself (ADR 0443: a consumer is made from now). It does nothing before a first view was asked.
func (f *Follower) ConsumerMade(ctx context.Context, created time.Time) {
	f.mu.Lock()
	f.init()
	if created.After(f.consumerMade) {
		f.consumerMade = created
	}
	behind := f.behindLocked()
	start := behind != "" && !f.asking && !f.catching
	if start {
		f.catching = true
	}
	f.mu.Unlock()
	if start {
		f.logf("views: %s's %s is behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
		go f.catchUp(ctx)
	}
}

// Behind says whether an event heard is not in the picture yet, and why: for a reader's tool and a test.
func (f *Follower) Behind() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.init()
	return f.behindLocked()
}

// askOnce asks every page and, once whole, takes it as the picture.
func (f *Follower) askOnce(ctx context.Context) error {
	f.mu.Lock()
	f.st.Asked, f.st.Asks = f.now(), f.st.Asks+1
	asked := f.st
	f.mu.Unlock()
	f.say(asked)
	var first Answer
	values := map[string]json.RawMessage{}
	written := map[string]time.Time{}
	after := ""
	for page := 0; ; page++ {
		a, err := f.Ask(ctx, after)
		if err != nil {
			return err
		}
		if page == 0 {
			first = a
		}
		maps.Copy(values, a.Values)
		maps.Copy(written, a.Written)
		if a.Next == "" || a.Next == after {
			break
		}
		after = a.Next
	}
	f.mu.Lock()
	old := f.values
	f.values = values
	wasRaised := f.st.Raised
	f.st.Revision, f.st.Created, f.st.Answered, f.st.Raised, f.st.Error = first.Revision, first.Created, f.now(), false, ""
	// The events heard while it asked, after the answer's revision and in order, are taken again.
	during := f.during
	f.during = nil
	slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) })
	for _, ev := range during {
		if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap {
			if ev.Deleted {
				delete(f.values, ev.Key)
			} else {
				f.values[ev.Key] = ev.Value
				written[ev.Key] = ev.At // a change taken again was written when its event says
			}
			f.st.Revision = ev.Revision
		}
	}
	values = maps.Clone(f.values)
	s := f.st
	select {
	case <-f.ready:
	default:
		close(f.ready)
	}
	f.mu.Unlock()
	if wasRaised && f.Clear != nil {
		f.Clear(s)
	}
	f.say(s)
	if f.Changed != nil {
		for k, v := range values {
			if ov, ok := old[k]; !ok || string(ov) != string(v) {
				f.Changed(k, v, false, written[k])
			}
		}
		for k := range old {
			if _, ok := values[k]; !ok {
				f.Changed(k, nil, true, time.Time{})
			}
		}
	}
	return nil
}

// Apply takes one event of the owner's views family into the picture and says whether it was a gap: another Created
// (the owner's state made again), a revision past the one after the picture's, or a value too large to carry. A
// revision at or below the picture's is already in it (a durable consumer handing what it held) and is passed over.
// Before the first view is answered an event is passed over too: the answer holds it.
func (f *Follower) Apply(ev Event) (gap bool, why string) {
	f.mu.Lock()
	f.init()
	switch {
	case ev.Created.After(f.heardCreated):
		f.heardCreated, f.heardRev = ev.Created, ev.Revision
	case ev.Created.Equal(f.heardCreated) && ev.Revision > f.heardRev:
		f.heardRev = ev.Revision
	}
	if f.asking {
		f.during = append(f.during, ev)
	}
	select {
	case <-f.ready:
	default:
		f.mu.Unlock()
		f.logf("views: %s's %s at revision %d was heard before its first view was answered: passed over, the answer holds it or is asked again", f.Owner, f.State, ev.Revision)
		return false, ""
	}
	switch {
	case !ev.Created.Equal(f.st.Created):
		why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
	case ev.Revision <= f.st.Revision:
		held := f.st.Revision
		f.mu.Unlock()
		f.logf("views: %s's %s at revision %d is in the picture already (it holds %d): passed over", f.Owner, f.State, ev.Revision, held)
		return false, ""
	case ev.Revision > f.st.Revision+1:
		why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1)
	case ev.Gap:
		why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision)
	}
	if why != "" {
		f.mu.Unlock()
		return true, why
	}
	if ev.Deleted {
		delete(f.values, ev.Key)
	} else {
		f.values[ev.Key] = ev.Value
	}
	f.st.Revision = ev.Revision
	s := f.statusLocked(f.st)
	f.mu.Unlock()
	f.say(s) // the revision the picture holds follows the events, not only the asks (a read tool says where it stands)
	if f.Changed != nil {
		f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At)
	}
	return false, ""
}

// Hear is a reader's consumer of one owner's views family, routing each event to the follower of its state; a gap asks
// that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated.
func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) {
	return func(m jetstream.Msg) {
		if err := HearEvent(ctx, followers, m.Data(), logf); err != nil {
			logf("views: %s is not a views event, dropped: %v", m.Subject(), err)
			_ = m.Term()
			return
		}
		_ = m.Ack()
	}
}

// HearEvent is Hear for one event given as its data, for a reader that takes the owner's events by another path than
// a consumer of its own (the tool runner's): routed to the follower of its state, an event of a state it does not
// follow passed over, and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is
// not a views event is answered as an error, and nothing is asked.
func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, logf func(string, ...any)) error {
	ev, err := DecodeEvent(data)
	if err != nil {
		return err
	}
	f := followers[ev.State]
	if f == nil {
		logf("views: %s's %s is a state this reader does not follow: event at revision %d passed over", ev.Owner, ev.State, ev.Revision)
		return nil
	}
	if gap, why := f.Apply(ev); gap {
		logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool)
		go func() { _ = f.Gap(ctx, why) }()
	}
	return nil
}

// Gap asks the first view once more for a gap the reader saw, until answered or ctx ends: counted among its asks,
// never among its asks by hand.
func (f *Follower) Gap(ctx context.Context, why string) error {
	return f.askUntil(ctx, "a gap: "+why, false)
}

// AskVia is a Follower's Ask through a call of the owner's tool (process.Conn.Call, or stdio.Ask through the tool
// runner), decoding its answer whatever wraps it (buscalls.Unwrap): an error the tool says is the ask's.
func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) {
	return func(_ context.Context, after string) (Answer, error) {
		body := map[string]any{}
		if after != "" {
			body["after"] = after
		}
		raw, err := call(tool, body)
		if err != nil {
			return Answer{}, err
		}
		answer, said, ok := buscalls.Unwrap(raw)
		if !ok {
			return Answer{}, fmt.Errorf("%s answered: %s", tool, said)
		}
		raw = answer
		return DecodeAnswer(raw)
	}
}