mesh-tools/node-tools/internal/bus/log.go

package bus

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

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

// A module's log on the bus (novox/hq ADR 0297): a stream of operations, one subject per key, that the
// controller creates from what the module declared under `logs` and issues to the module's own instances
// in its membership. It mirrors a bucket (ADR 0201) in how it is reached: what a module may reach is
// decided here, from its membership, and refused with the reason before anything is sent.
//
// **The runtime uses no consumer of a log.** The bus grants it, per issued log, only the publish of an
// entry, the stream's info and the direct get: an append is a publish, a read is a batch direct get
// (NATS 2.11), the last entry of a key is a direct get by its subject.

// LogIssued is one log an assignment may reach (ADR 0297): its owner's alone, so there are no reads.
type LogIssued struct {
	Name    string `json:"name"`
	Stream  string `json:"stream"`
	Subject string `json:"subject"`
	Writes  bool   `json:"writes,omitempty"`
}

// LogEntry is one operation in a log: its sequence is its identity.
type LogEntry struct {
	Sequence uint64          `json:"sequence"`
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	At       time.Time       `json:"at"`
}

// LogPage is one page of a read: Next is the sequence to continue from, 0 when nothing follows; Last
// is the stream's last sequence.
type LogPage struct {
	Entries []LogEntry `json:"entries"`
	Next    uint64     `json:"next"`
	Last    uint64     `json:"last"`
}

// LogInfo is how full a log is.
type LogInfo struct {
	Entries  uint64 `json:"entries"`
	Bytes    uint64 `json:"bytes"`
	MaxBytes uint64 `json:"max_bytes"`
	First    uint64 `json:"first"`
	Last     uint64 `json:"last"`
}

const (
	// LogValueMax is the most one entry's value may hold (ADR 0297 §3); the stream's MaxMsgSize is 4 KiB
	// more, for the headers an append carries.
	LogValueMax = 256 * 1024
	// LogKeyMax is the longest key.
	LogKeyMax = 255
	// LogReadDefault and LogReadMax are a page's entries when none is asked, and at most.
	LogReadDefault = 100
	LogReadMax     = 1000
	// LogPageBytes bounds the bytes of one page.
	LogPageBytes = 4 * 1024 * 1024
)

// logToken is one name of a key: letters, digits, `_` and `-`.
var logToken = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)

func checkLogKey(key string) error {
	if key == "" || len(key) > LogKeyMax {
		return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
			"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
	}
	for _, t := range strings.Split(key, ".") {
		if !logToken.MatchString(t) {
			return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
				"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
		}
	}
	return nil
}

// logFilter is a read's key pattern as a subject filter under the log's subject: `*` for one name, `>`
// for the rest; empty is every key.
func logFilter(l LogIssued, pattern string) (string, error) {
	if pattern == "" {
		return l.Subject + ".>", nil
	}
	parts := strings.Split(pattern, ".")
	for i, p := range parts {
		if p == "**" && i == len(parts)-1 {
			// `**` for the rest, as a state's key pattern and mesh-sdk's log reader say it.
			parts[i], p = ">", ">"
		}
		switch {
		case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p):
		default:
			return "", fmt.Errorf("%q is not a key pattern: names, `*` for one and `>` for the rest, last "+
				"(novox/hq ADR 0297)", pattern)
		}
	}
	if len(pattern) > LogKeyMax {
		return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax)
	}
	return l.Subject + "." + strings.Join(parts, "."), nil
}

// issuedLog is the log a module may reach by a name, or the reason it may not.
func (c *Conn) issuedLog(module, name string) (LogIssued, error) {
	m := c.Membership(module)
	if m == nil {
		return LogIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no log of it is reachable "+
			"until the mesh issues one (novox/hq ADR 0297)", module, c.node)
	}
	var names []string
	for _, l := range m.Logs {
		if l.Name == name {
			return l, nil
		}
		names = append(names, l.Name)
	}
	sort.Strings(names)
	issued := "none"
	if len(names) > 0 {
		issued = strings.Join(names, ", ")
	}
	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, and "+
		"another module's it reads under `reads` as <module>.<log>; it was issued: %s (novox/hq ADR 0297, ADR 0571)",
		module, name, issued)
}

// notOnTheBus says a log is issued and its stream is not there yet.
func notOnTheBus(l LogIssued) error {
	return fmt.Errorf("the log %q is issued and its stream %s is not on the bus yet: the controller creates "+
		"it from the catalogue on its next raise (novox/hq ADR 0297)", l.Name, l.Stream)
}

// LogAppend writes one entry for a key, as the module, and answers its sequence. With expect, it is
// written only when the key's last entry is still at that sequence (0: the key has none); otherwise it
// is refused as a lost compare-and-set, the same error a lost state update answers (ADR 0259).
func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect *uint64) (uint64, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return 0, err
	}
	if !l.Writes {
		return 0, fmt.Errorf("%s was issued the log %s without writing it (novox/hq ADR 0297)", module, name)
	}
	if err := checkLogKey(key); err != nil {
		return 0, err
	}
	if len(value) == 0 || !json.Valid(value) {
		return 0, fmt.Errorf("a log entry's value is JSON (novox/hq ADR 0297)")
	}
	if len(value) > LogValueMax {
		return 0, fmt.Errorf("a log entry's value is at most %d bytes (256 KiB), and this one is %d (novox/hq ADR 0297)",
			LogValueMax, len(value))
	}
	if field := credentialField(value); field != "" {
		return 0, fmt.Errorf("%s's %s.%s carries a field %q, which names a credential: no secret is kept in a log, "+
			"which keeps it for good. Name the secret and fetch it on request/reply (novox/hq ADR 0201 §13, ADR 0297)",
			module, name, key, field)
	}
	msg := nats.NewMsg(l.Subject + "." + key)
	msg.Data = value
	if expect != nil {
		msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(*expect, 10))
	}
	reply, err := c.nc.RequestMsg(msg, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return 0, notOnTheBus(l)
	}
	if err != nil {
		return 0, err
	}
	var ack struct {
		Seq   uint64 `json:"seq"`
		Error *struct {
			Code        int    `json:"code"`
			ErrCode     uint16 `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
	}
	if err := json.Unmarshal(reply.Data, &ack); err != nil {
		return 0, fmt.Errorf("the bus answered the append to %s with what is not an acknowledgement: %q", name, reply.Data)
	}
	if ack.Error != nil {
		// The key's last entry is no longer the one expected: a lost compare-and-set, answered as a lost
		// state update is (10164 is the same refusal from a replicated stream).
		if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence ||
			code == jetstream.JSErrCodeStreamWrongLastSequenceConstant {
			return 0, changed{name + "." + key}
		}
		return 0, fmt.Errorf("the bus refused the append to %s.%s: %s (%d)", name, key, ack.Error.Description, ack.Error.ErrCode)
	}
	if ack.Seq == 0 {
		return 0, fmt.Errorf("the bus answered the append to %s.%s with no sequence", name, key)
	}
	return ack.Seq, nil
}

// LogRead is the entries from a sequence on, or from a time on (since, when not zero), of the keys matching a pattern, at
// most limit of them and about 4 MiB, by one batch direct get. A log another module keeps is read as its own is, by the
// name its reads declare, `<module>.<log>` (novox/hq ADR 0571).
func (c *Conn) LogRead(module, name string, from uint64, since time.Time, keys string, limit int) (*LogPage, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	switch {
	case limit == 0:
		limit = LogReadDefault
	case limit < 0 || limit > LogReadMax:
		return nil, fmt.Errorf("a read's limit is 1 to %d entries, %d when none is given, not %d (novox/hq ADR 0297)",
			LogReadMax, LogReadDefault, limit)
	}
	filter, err := logFilter(l, keys)
	if err != nil {
		return nil, err
	}
	if from != 0 && !since.IsZero() {
		return nil, fmt.Errorf("a read of %s starts from a sequence or from a time, not both", name)
	}
	if from == 0 {
		from = 1
	}
	req := map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}
	if !since.IsZero() {
		delete(req, "seq")
		req["start_time"] = since.UTC()
	}
	entries, more, err := c.logBatch(l, req)
	if err != nil {
		return nil, err
	}
	page := &LogPage{Entries: entries}
	if more && len(entries) > 0 {
		page.Next = entries[len(entries)-1].Sequence + 1
	}
	info, err := c.logInfo(l)
	if err != nil {
		return nil, err
	}
	page.Last = info.Last
	return page, nil
}

// LogNewest is the newest entry of each key matching a pattern, in the log's order, by one batch direct get (mesh-sdk
// process.LogReader.Newest).
func (c *Conn) LogNewest(module, name, keys string) ([]LogEntry, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	filter, err := logFilter(l, keys)
	if err != nil {
		return nil, err
	}
	entries, more, err := c.logBatch(l, map[string]any{"multi_last": []string{filter}, "batch": LogReadMax})
	if err != nil {
		return nil, err
	}
	if more {
		return nil, fmt.Errorf("%s holds more than %d keys matching %q: name fewer", name, LogReadMax, keys)
	}
	return entries, nil
}

// logBatch is one batch direct get of a log: its entries, and whether more are pending after them.
func (c *Conn) logBatch(l LogIssued, request map[string]any) ([]LogEntry, bool, error) {
	req, _ := json.Marshal(request)
	inbox := c.nc.NewInbox()
	sub, err := c.nc.SubscribeSync(inbox)
	if err != nil {
		return nil, false, err
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
		return nil, false, err
	}
	entries := []LogEntry{}
	deadline := time.Now().Add(StateTimeout)
	for {
		msg, err := sub.NextMsg(time.Until(deadline))
		if err != nil {
			if errors.Is(err, nats.ErrNoResponders) {
				return nil, false, notOnTheBus(l)
			}
			if errors.Is(err, nats.ErrTimeout) {
				return nil, false, fmt.Errorf("the read of %s did not end in %s", l.Name, StateTimeout)
			}
			return nil, false, err
		}
		switch status := msg.Header.Get("Status"); status {
		case "":
			e, err := logEntryOf(l, msg)
			if err != nil {
				return nil, false, err
			}
			entries = append(entries, *e)
		case "204": // the end of the batch: what is still pending after the last entry sent
			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
			return entries, pending > 0, nil
		case "404": // nothing from there on
			return entries, false, nil
		case "503":
			return nil, false, notOnTheBus(l)
		default:
			return nil, false, fmt.Errorf("the read of %s was refused by the bus: %s %s", l.Name, status, msg.Header.Get("Description"))
		}
	}
}

// LogWatch hands deliver every entry of the keys matching a pattern from a sequence on (from), from a time on (since),
// or, with neither, each new one, in the log's order, one at a time, through an ordered consumer it lets go of when
// stopped (mesh-sdk process.LogReader.Watch). It returns once the entries there at its start are handed over; an error
// from deliver is said, and the watch goes on with the next entry.
func (c *Conn) LogWatch(module, name, keys string, from uint64, since time.Time, deliver func(LogEntry) error) (func(), error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	filter, err := logFilter(l, keys)
	if err != nil {
		return nil, err
	}
	cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverNewPolicy}
	switch {
	case from != 0 && !since.IsZero():
		return nil, fmt.Errorf("a watch of %s starts from a sequence or from a time, not both", name)
	case from != 0:
		cfg.DeliverPolicy, cfg.OptStartSeq = jetstream.DeliverByStartSequencePolicy, from
	case !since.IsZero():
		at := since.UTC()
		cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at
	}
	js, err := jetstream.New(c.nc)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	cons, err := js.OrderedConsumer(ctx, l.Stream, cfg)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, notOnTheBus(l)
	}
	if err != nil {
		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
	}
	info, err := cons.Info(ctx)
	if err != nil {
		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
	}
	it, err := cons.Messages()
	if err != nil {
		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
	}
	hand := func(m jetstream.Msg) error {
		meta, err := m.Metadata()
		if err != nil {
			return err
		}
		e := LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.Subject+"."),
			Value: valueOf(m.Data()), At: meta.Timestamp.UTC()}
		return deliver(e)
	}
	for pending := info.NumPending; pending > 0; pending-- {
		m, err := it.Next()
		if err == nil {
			err = hand(m)
		}
		if err != nil {
			it.Stop()
			return nil, fmt.Errorf("the watch of %s: %w", name, err)
		}
	}
	var stopped atomic.Bool
	go func() {
		for {
			m, err := it.Next()
			if err != nil || stopped.Load() {
				return // stopped, or the connection closed; a message fetched as it stopped is not handed over
			}
			if err := hand(m); err != nil {
				c.Logf("[mesh-tools] %s did not take entry of %s: %v; it is in the log, a read finds it", module, name, err)
			}
		}
	}()
	return func() { stopped.Store(true); it.Stop() }, nil
}

// LogLast is a key's last entry, or nil when it has none.
func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	if err := checkLogKey(key); err != nil {
		return nil, err
	}
	reply, err := c.nc.Request("$JS.API.DIRECT.GET."+l.Stream+"."+l.Subject+"."+key, nil, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return nil, notOnTheBus(l)
	}
	if err != nil {
		return nil, err
	}
	switch status := reply.Header.Get("Status"); status {
	case "":
		return logEntryOf(l, reply)
	case "404":
		return nil, nil
	default:
		return nil, fmt.Errorf("the last entry of %s.%s could not be read: %s %s", name, key, status, reply.Header.Get("Description"))
	}
}

// LogInfo is a log's entries, bytes, cap and first and last sequences.
func (c *Conn) LogInfo(module, name string) (*LogInfo, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	return c.logInfo(l)
}

func (c *Conn) logInfo(l LogIssued) (*LogInfo, error) {
	reply, err := c.nc.Request("$JS.API.STREAM.INFO."+l.Stream, nil, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return nil, notOnTheBus(l)
	}
	if err != nil {
		return nil, err
	}
	var info struct {
		Config struct {
			MaxBytes int64 `json:"max_bytes"`
		} `json:"config"`
		State struct {
			Msgs     uint64 `json:"messages"`
			Bytes    uint64 `json:"bytes"`
			FirstSeq uint64 `json:"first_seq"`
			LastSeq  uint64 `json:"last_seq"`
		} `json:"state"`
		Error *struct {
			ErrCode     uint16 `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
	}
	if err := json.Unmarshal(reply.Data, &info); err != nil {
		return nil, fmt.Errorf("the bus answered the info of %s with what is not one: %q", l.Name, reply.Data)
	}
	if info.Error != nil {
		if info.Error.ErrCode == uint16(jetstream.JSErrCodeStreamNotFound) {
			return nil, notOnTheBus(l)
		}
		return nil, fmt.Errorf("the info of the log %s was refused: %s (%d)", l.Name, info.Error.Description, info.Error.ErrCode)
	}
	var maxBytes uint64
	if info.Config.MaxBytes > 0 {
		maxBytes = uint64(info.Config.MaxBytes)
	}
	return &LogInfo{Entries: info.State.Msgs, Bytes: info.State.Bytes, MaxBytes: maxBytes,
		First: info.State.FirstSeq, Last: info.State.LastSeq}, nil
}

// logEntryOf is an entry a direct get answered: its key is its subject under the log's.
func logEntryOf(l LogIssued, msg *nats.Msg) (*LogEntry, error) {
	subject := msg.Header.Get("Nats-Subject")
	key, ok := strings.CutPrefix(subject, l.Subject+".")
	if !ok {
		return nil, fmt.Errorf("the bus answered an entry of %s on %q, which is not under %s", l.Name, subject, l.Subject)
	}
	seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
	if err != nil {
		return nil, fmt.Errorf("the bus answered an entry of %s with no sequence", l.Name)
	}
	at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
	if err != nil {
		return nil, fmt.Errorf("the bus answered entry %d of %s with no time: %v", seq, l.Name, err)
	}
	return &LogEntry{Sequence: seq, Key: key, Value: valueOf(msg.Data), At: at.UTC()}, nil
}