mesh-delivery/sdk/views/emit.go

package views

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

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

// Emitter is an owner's process saying each change of its own states on its views family (ADR 0615 §5): it follows
// each state it is given from the revision its mark says it said up to, publishes one Event per change, and writes the
// mark once it has said every change the bus had handed it. A part watches only its own state (ADR 0596 §4): an owner
// follows its own buckets, never another's.
type Emitter struct {
	// Owner is the owner module, said in every event.
	Owner string
	// Subject is where an event of the given local name lands on the owner's own stream (process.Account.StreamSubject).
	Subject func(event string) string
	// Publish sends one message to the owner's stream, its id the bus's dedup id.
	Publish func(ctx context.Context, msg *nats.Msg, id string) error
	// Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end.
	Marks jetstream.KeyValue
	// TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil
	// only logs it.
	TooLarge func(state, key string, size int)
	Logf     func(string, ...any)
	Now      func() time.Time
}

// JetStreamPublish is Emitter.Publish through js, with the bus's dedup id.
func JetStreamPublish(js jetstream.JetStream) func(context.Context, *nats.Msg, string) error {
	return func(ctx context.Context, msg *nats.Msg, id string) error {
		pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
		defer cancel()
		_, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id))
		return err
	}
}

func (e Emitter) now() time.Time {
	if e.Now != nil {
		return e.Now().UTC()
	}
	return time.Now().UTC()
}

func (e Emitter) logf(format string, args ...any) {
	if e.Logf != nil {
		e.Logf(format, args...)
	}
}

// created is when the bucket's stream was made, and its fill now.
func created(ctx context.Context, kv jetstream.KeyValue) (time.Time, uint64, Fill, error) {
	st, err := kv.Status(ctx)
	if err != nil {
		return time.Time{}, 0, Fill{}, err
	}
	b, ok := st.(*jetstream.KeyValueBucketStatus)
	if !ok || b.StreamInfo() == nil {
		return time.Time{}, 0, Fill{}, fmt.Errorf("the bus answered no stream behind %s", st.Bucket())
	}
	info := b.StreamInfo()
	return info.Created.UTC(), info.State.LastSeq, Fill{Keys: uint64(info.State.NumSubjects), Bytes: info.State.Bytes, MaxBytes: info.Config.MaxBytes}, nil
}

// Start follows state's bucket kv: from the revision after its mark when the mark is of the same bucket (the same
// Created), otherwise from the bucket's end, writing a mark there first, so a reader's first view taken now misses
// nothing. It returns once the watch is set; the changes are said on a goroutine until ctx ends, and a watch that ends
// is sent on errs.
func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, errs chan<- error) error {
	made, last, fill, err := created(ctx, kv)
	if err != nil {
		return fmt.Errorf("the state %s cannot be followed for its views: %w", state, err)
	}
	from := last + 1
	if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last {
		from = m.Revision + 1
	} else {
		e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill})
	}
	w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from))
	if err != nil {
		return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err)
	}
	go func() {
		defer w.Stop()
		said := uint64(0)
		for {
			select {
			case <-ctx.Done():
				return
			case u, ok := <-w.Updates():
				if !ok {
					if errs != nil {
						errs <- fmt.Errorf("the watch of %s for its views ended", state)
					}
					return
				}
				if u != nil {
					if err := e.say(ctx, state, made, u); err != nil {
						// Not marked: the next start says it again from the mark, and the bus keeps it once by its id.
						e.logf("views: %s.%s at %d is not said: %v", state, u.Key(), u.Revision(), err)
						continue
					}
					said = u.Revision()
				}
				// Marked once every change handed so far is said, so a busy state writes one mark per burst.
				if said > 0 && len(w.Updates()) == 0 {
					_, _, fill, err := created(ctx, kv)
					if err != nil {
						e.logf("views: the fill of %s is not read: %v", state, err)
					}
					e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill})
					said = 0
				}
			}
		}
	}()
	return nil
}

// Event is the event saying u, a change of state made at made.
func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event {
	ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made}
	if u.Operation() != jetstream.KeyValuePut {
		ev.Deleted = true
		return ev
	}
	if n := len(u.Value()); n > MaxValueBytes {
		ev.Gap = true
		e.logf("views: %s.%s at %d holds %d bytes, more than an event carries: said as a gap, so a reader asks its first view", state, u.Key(), u.Revision(), n)
		if e.TooLarge != nil {
			e.TooLarge(state, u.Key(), n)
		}
		return ev
	}
	ev.Value = json.RawMessage(u.Value())
	if !json.Valid(ev.Value) {
		ev.Value, _ = json.Marshal(string(u.Value()))
	}
	return ev
}

func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error {
	ev := e.Event(state, made, u)
	data, err := ev.Encode()
	if err != nil {
		return err
	}
	msg := nats.NewMsg(e.Subject(EventName(state)))
	msg.Header.Set("x-event-id", ev.ID())
	msg.Header.Set("x-source", e.Owner)
	msg.Header.Set("x-time", e.now().Format(time.RFC3339Nano))
	msg.Header.Set("content-type", "application/json")
	msg.Data = data
	return e.Publish(ctx, msg, ev.ID())
}

func (e Emitter) mark(ctx context.Context, state string) (Mark, bool) {
	if e.Marks == nil {
		return Mark{}, false
	}
	got, err := e.Marks.Get(ctx, state)
	if err != nil {
		if !errors.Is(err, jetstream.ErrKeyNotFound) {
			e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
		}
		return Mark{}, false
	}
	m, err := DecodeMark(got.Value())
	if err != nil {
		e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
		return Mark{}, false
	}
	return m, true
}

func (e Emitter) writeMark(ctx context.Context, state string, m Mark) {
	if e.Marks == nil {
		return
	}
	m.V, m.At = Version, e.now()
	data, err := json.Marshal(m)
	if err != nil {
		e.logf("views: the mark of %s is not written: %v", state, err)
		return
	}
	if _, err := e.Marks.Put(ctx, state, data); err != nil {
		e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err)
	}
}