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"
	"maps"
	"slices"
	"strings"
	"sync"
	"time"

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

	"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")

// 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() WireWriter
	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

	// Other is told of each message of the module's durable consumer on the shared events stream that is not an owner's
	// views event, so a module that hears other events through the same consumer (a module has one) handles them there.
	// Nil says them in the log, acknowledged: the module hears views events alone.
	Other func(jetstream.Msg)
	// Hub is the module's one pull of its durable consumer (process.EventHub), when other parts of the module hear events
	// too: the reader registers its views events on it and does not pull; the host starts the hub once, after every part
	// registered. Nil: the reader pulls the durable itself and takes every message (Other the rest), for a module that
	// hears nothing else.
	Hub *process.EventHub

	mu    sync.Mutex
	pics  map[string]*Picture // by bucket name <owner>_<state>
	order []string
	// adopted are the handlers of the adopted followers' seat events (Adopt), by the event's subject.
	// Several followers adopt one event (two branches' views of the catalog half), each hearing it (design 47 G §532).
	adopted map[string][]adoptedHear

	// After Start and Ask (design 47 G §532): the hub the reader's events come through when it is the module's (Hub) and
	// its handler, the time the bus made the consumer, the asks' life and Said, and of each first view begun, the cancel
	// of its asks, by bucket; status is the state `first-views` each status is kept in.
	hub     *process.EventHub
	handle  func(jetstream.Msg)
	created time.Time
	ctx     context.Context
	said    func(Status)
	asked   bool
	begun   map[string]context.CancelFunc
	status  *process.AwaitedBucket
}

// adoptedHear is one adopted follower's handler of its seat event, by its picture's bucket.
type adoptedHear struct {
	bucket string
	hear   func(context.Context, []byte) error
}

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

// Follow adds the owner's state to follow, as the owner declares it (design 25 § 12.2): tool is its ordinary read tool
// of state as the reader's `invokes` names it (`<owner>.<tool>`, from the owner's manifest's tools). The owner's changes
// are ordinary events (`<owner>.views.<state>`), which the reader's manifest `consumes` one entry per followed state.
// views names no owner: the caller gives it. Called after Start, the view takes its first view at once and then follows
// (design 47 G §532).
func (r *ProcessReader) Follow(owner, state, tool string) (*Picture, error) {
	p, fresh, err := r.follow(owner, state, tool)
	if err == nil && fresh {
		r.begin(p.bucket)
	}
	return p, err
}

func (r *ProcessReader) follow(owner, state, tool string) (*Picture, bool, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if owner == "" || state == "" || !strings.HasPrefix(tool, owner+".") {
		return nil, false, fmt.Errorf("%s follows %s's %s by its read tool <owner>.<tool>, not %q", r.Module, owner, state, tool)
	}
	if r.pics == nil {
		r.pics = map[string]*Picture{}
	}
	bucket := owner + "_" + state
	if p := r.pics[bucket]; p != nil {
		return p, false, nil
	}
	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) {
			if r.holds(p) {
				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, true, nil
}

// holds says whether p is still a picture of the reader: a dropped view writes no finding after its drop cleared them.
func (r *ProcessReader) holds(p *Picture) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.pics[p.bucket] == p
}

// FollowBranch is Follow for a state kept per branch: the view of one branch is named `<owner>.<state>.<branch>` (the
// branch escaped as a key's name is, BranchState) and its picture `<owner>_<state>.<branch>`, so a process follows
// several branches of one state, each its own view (design 47 G §479). Called after Start, it begins at once (§532).
func (r *ProcessReader) FollowBranch(owner, state, branch, tool string) (*Picture, error) {
	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
// process'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 its picture (`<owner>_<state>`) hands each
// change on before the follower's own Changed, set before. Start keeps its status, hears its hand asks, and asks it with
// the others; its events are the seat's event (seatdef's EventSubject of seat and event, on the module's durable
// consumer, as its `consumes` names it), handed to hear, the client's handler, and acknowledged; a state followed already
// is refused. Adopted after Start, as a part adopts a watched branch added while it runs, the view takes its first view
// at once and then follows (design 47 G §532).
func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, error) {
	if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil {
		return nil, errors.New("views: an adopted follower names its owner and state, and its seat's event and handler")
	}
	p, subject, err := r.adopt(f, seat, event, hear)
	if err != nil {
		return nil, err
	}
	r.mu.Lock()
	hub, handle := r.hub, r.handle
	r.mu.Unlock()
	if subject != "" && hub != nil {
		// one subscription per event, with every further part (design 47 G §532)
		if err := hub.Handle(subject+".>", handle); err != nil {
			r.logf("views: %s's event %s is not heard (%v): the view %s is its first view's alone, asked again by hand", r.Module, subject, err, p.bucket)
		}
	}
	r.begin(p.bucket)
	return p, nil
}

// AdoptEvent hears a seat's event on the one subscription the followers adopted on that event share, with no picture and no
// first view (design 47 G A-185, A-199): hear is handed each message of the event beside every adopted follower's handler,
// and the message is acknowledged, unless hear fails: then it is left unacknowledged and the bus asks again (A-222). The subscription is made at once when no follower made it, so a part that follows no
// view of the seat still hears the event; adopted before Start, it is made at Start.
func (r *ProcessReader) AdoptEvent(seat, event string, hear func(context.Context, []byte) error) error {
	if seat == "" || event == "" || hear == nil {
		return errors.New("views: an adopted event names its seat, its event and its handler")
	}
	subject := "mesh.seat." + seat + ".event." + event
	r.mu.Lock()
	if r.adopted == nil {
		r.adopted = map[string][]adoptedHear{}
	}
	fresh := len(r.adopted[subject]) == 0
	r.adopted[subject] = append(r.adopted[subject], adoptedHear{hear: hear})
	hub, handle := r.hub, r.handle
	r.mu.Unlock()
	if fresh && hub != nil {
		// one subscription per event, with every further part (design 47 G §532)
		if err := hub.Handle(subject+".>", handle); err != nil {
			r.logf("views: %s's event %s is not heard (%v)", r.Module, subject, err)
		}
	}
	return nil
}

// adopt is Adopt's part under the lock; subject is the event's subject when no follower adopted it before.
func (r *ProcessReader) adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, string, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.pics == nil {
		r.pics = map[string]*Picture{}
	}
	bucket := f.Owner + "_" + f.State
	if r.pics[bucket] != nil {
		return nil, "", fmt.Errorf("views: %s's %s is followed already", f.Owner, f.State)
	}
	p := &Picture{bucket: bucket, ctx: context.Background(), f: f}
	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
	}
	f.Raise, f.Clear = func(s Status) {
		if r.holds(p) {
			r.finding(s, true)
		}
	}, func(s Status) { r.finding(s, false) }
	f.RaiseWaits, f.ClearWaits = func(s Status) {
		if r.holds(p) {
			r.waits(s, true)
		}
	}, func(s Status) { r.waits(s, false) }
	own := f.Changed
	f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) {
		p.changed(key, value, deleted, at)
		if own != nil {
			own(key, value, deleted, at)
		}
	}
	r.pics[bucket] = p
	r.order = append(r.order, bucket)
	if r.adopted == nil {
		r.adopted = map[string][]adoptedHear{}
	}
	subject := "mesh.seat." + seat + ".event." + event
	fresh := len(r.adopted[subject]) == 0
	r.adopted[subject] = append(r.adopted[subject], adoptedHear{bucket: bucket, hear: hear})
	if !fresh {
		subject = ""
	}
	return p, subject, nil
}

// 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 *ProcessReader) 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
	}
	kv := r.Findings()
	if kv == nil {
		return
	}
	at := time.Now()
	if r.Now != nil {
		at = r.Now()
	}
	f := viewWaitsFinding(r.Module, s, at)
	var err error
	if raised {
		err = f.put(kv)
	} else {
		err = f.drop(kv)
	}
	if err != nil {
		r.logf("views: the view-waits finding of %s is not written: %v", s.Tool, err)
	}
}

// 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()
	}
	f := unansweredFinding(r.Module, s, at)
	var err error
	if raised {
		err = f.put(kv)
	} else {
		err = f.drop(kv)
	}
	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 module's durable consumer on the shared events stream (cred.EventConsumer on
// process.EventsStream): an owner's views event (`mesh.mod.<owner>.event.views.<state>`) goes to the follower of that
// owner's state, any other message to Other. The followers are read at each message, so a view followed or adopted
// after Start hears its events and a dropped one no longer does (design 47 G §532).
func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) {
	return func(m jetstream.Msg) {
		r.mu.Lock()
		hears := adoptedHears(r.adopted, m.Subject())
		var owned map[string]*Follower
		if owner, ok := OwnerOfEvent(m.Subject()); ok && hears == nil {
			for _, b := range r.order {
				if f := r.pics[b].f; f.Owner == owner {
					if owned == nil {
						owned = map[string]*Follower{}
					}
					owned[f.State] = f
				}
			}
		}
		r.mu.Unlock()
		if hears != nil {
			failed := false
			for _, h := range hears {
				if err := h.hear(ctx, m.Data()); err != nil {
					r.logf("views: %s is not taken: %v", m.Subject(), err)
					// an event adopted with no picture is asked again by the bus (design 47 G A-222)
					failed = failed || h.bucket == ""
				}
			}
			if !failed {
				_ = m.Ack()
			}
			return
		}
		if owned != nil {
			Hear(ctx, owned, r.logf)(m)
			return
		}
		if r.Other != nil {
			r.Other(m)
			return
		}
		r.logf("views: %s is no views event of an owner followed, acknowledged", m.Subject())
		_ = m.Ack()
	}
}

// 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). A view followed or
// adopted after it is asked at once (design 47 G §532).
func (r *ProcessReader) Ask(ctx context.Context, said func(Status)) {
	r.mu.Lock()
	r.ctx, r.said, r.asked = ctx, said, true
	buckets := slices.Clone(r.order)
	r.mu.Unlock()
	for _, b := range buckets {
		r.begin(b)
	}
}

// begin asks bucket's first view once Ask ran, on a goroutine of its own, until answered or the view is dropped.
func (r *ProcessReader) begin(bucket string) {
	r.mu.Lock()
	p := r.pics[bucket]
	if !r.asked || p == nil || r.begun[bucket] != nil {
		r.mu.Unlock()
		return
	}
	if r.begun == nil {
		r.begun = map[string]context.CancelFunc{}
	}
	ctx, cancel := context.WithCancel(r.ctx)
	r.begun[bucket] = cancel
	f, said, created := p.f, r.said, r.created
	r.mu.Unlock()
	if said != nil {
		f.Said = func(s Status) {
			if r.holds(p) {
				said(s)
			}
		}
	}
	if !created.IsZero() {
		f.ConsumerMade(ctx, created)
	}
	go func() {
		if err := f.Start(ctx); err != nil && ctx.Err() == nil {
			r.logf("the first view %s was not answered: %v", f.Tool, err)
		}
	}()
}

// Drop ends the view of owner's state, as a part drops a watched branch removed while it runs (design 47 G §532, §614):
// it asks it no more and hears none of its events, its status leaves the state `first-views` and so its row leaves
// `<module>_first_views`, and every finding it raised (its unanswered first view, its view-waits) is cleared through
// the reader's findings, the drop and each cleared finding said in the log. A state not followed is an error.
func (r *ProcessReader) Drop(owner, state string) error {
	bucket := owner + "_" + state
	r.mu.Lock()
	p := r.pics[bucket]
	if p == nil {
		r.mu.Unlock()
		return fmt.Errorf("%s does not follow %s's %s", r.Module, owner, state)
	}
	cancel := r.begun[bucket]
	delete(r.pics, bucket)
	delete(r.begun, bucket)
	r.order = slices.DeleteFunc(r.order, func(b string) bool { return b == bucket })
	for subject, hs := range r.adopted {
		if hs = slices.DeleteFunc(hs, func(h adoptedHear) bool { return h.bucket == bucket }); len(hs) == 0 {
			delete(r.adopted, subject)
		} else {
			r.adopted[subject] = hs
		}
	}
	status := r.status
	r.mu.Unlock()
	if cancel != nil {
		cancel()
	}
	r.logf("views: %s drops its view %s.%s", r.Module, owner, state)
	if status != nil {
		if kv := status.KV(); kv != nil {
			if err := kv.Delete(context.Background(), StatusKey(owner, state)); err != nil {
				r.logf("views: the status of the dropped view %s.%s is not removed from %s: %v", owner, state, StatusState, err)
			}
		}
	}
	s := p.f.Status()
	if s.Raised {
		r.finding(s, false)
		r.logf("views: %s clears its finding %s of the dropped view %s.%s", r.Module, UnansweredKind, owner, state)
	}
	if s.WaitsRaised {
		r.waits(s, false)
		r.logf("views: %s clears its finding %s of the dropped view %s.%s", r.Module, ViewWaitsKind, owner, state)
	}
	return nil
}

// 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 }

// EventsPattern is the subjects of every owner's views events on the shared events stream.
const EventsPattern = "mesh.mod.*.event." + EventPrefix + "*"

// Start hears the owners' views events on the process's durable consumer of the shared events stream (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) {
	hub, pattern, own := r.Hub, EventsPattern, false
	if hub == nil {
		hub, pattern, own = process.NewEventHub(r.logf), ">", true
	}
	hub.OnBound(func(created time.Time) {
		r.mu.Lock()
		r.created = created
		r.mu.Unlock()
		for _, f := range r.Followers() {
			f.ConsumerMade(ctx, created)
		}
	})
	hear := r.Hear(ctx)
	err := hub.Handle(pattern, hear)
	r.mu.Lock()
	r.status = status
	if !own {
		// an event adopted after the start is handled on the module's hub (Adopt); the reader's own hub takes every one
		r.hub, r.handle = hub, hear
	}
	r.mu.Unlock()
	if !own {
		r.mu.Lock()
		subjects := slices.Sorted(maps.Keys(r.adopted))
		r.mu.Unlock()
		for _, subject := range subjects {
			// one subscription per event, with every further part (design 47 G §532)
			if err == nil {
				err = hub.Handle(subject+".>", hear)
			}
		}
	}
	if err == nil && own {
		err = hub.Start(ctx, js, cred)
	}
	if err != nil {
		r.logf("the consumer %s on %s is not heard (%v): the states followed are the first views' alone, asked again by hand, until the %s starts again",
			cred.EventConsumerName(), process.EventsStream, err, 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)
}

// adoptedHears are the handlers of every follower adopted on the event whose subject is subject or has further parts
// after it (`<event>.<module>.<branch>.<commit or group>`, design 47 G §521, §532), or nil.
func adoptedHears(adopted map[string][]adoptedHear, subject string) []adoptedHear {
	if h := adopted[subject]; h != nil {
		return h
	}
	for i := strings.LastIndex(subject, "."); i > 0; i = strings.LastIndex(subject, ".") {
		subject = subject[:i]
		if h := adopted[subject]; h != nil {
			return h
		}
	}
	return nil
}