mesh-catalog/modules/nats/cmd/nats-tools/consumer.go

package main

// Where a lost event is located (issue 1966, tools-are-how-a-module-is-spoken-to): one consumer of a stream as the server
// holds it, and the messages a stream keeps for one subject. Both read; neither changes the bus. The consumer is read from
// the server's monitoring document, as nats_streams and nats_backlog are, so it needs no grant; the messages are asked of
// the bus on the module's own credential (STREAM.MSG.GET), as the removal tools are, so what the credential is not
// granted shows as an unanswered request and is said so (unanswered).

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

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

// Consumer is one consumer of one stream: how it is set and where it stands.
func Consumer(jsz []byte, stream, durable string) (map[string]any, error) {
	var doc jszDoc
	if err := json.Unmarshal(jsz, &doc); err != nil {
		return nil, err
	}
	streamFound := false
	for _, a := range doc.Accounts {
		for _, s := range a.Streams {
			if s.Name != stream {
				continue
			}
			streamFound = true
			for _, c := range s.Consumers {
				if c.Name != durable {
					continue
				}
				filters := append([]string(nil), c.Config.FilterSubjects...)
				if c.Config.FilterSubject != "" {
					filters = append(filters, c.Config.FilterSubject)
				}
				kind := "pull"
				if c.Config.DeliverSubject != "" {
					kind = "push"
				}
				return map[string]any{
					"stream": stream, "consumer": c.Name, "kind": kind, "created": c.Created,
					"config": map[string]any{
						"filter_subjects": filters, "deliver_policy": c.Config.DeliverPolicy, "ack_policy": c.Config.AckPolicy,
						"ack_wait_seconds": c.Config.AckWait / int64(time.Second), "max_ack_pending": c.Config.MaxAckPending,
						"max_deliver": c.Config.MaxDeliver, "replay_policy": c.Config.ReplayPolicy,
						"deliver_subject": c.Config.DeliverSubject, "deliver_group": c.Config.DeliverGroup,
					},
					"state": map[string]any{
						"delivered_stream_seq": c.Delivered.Stream, "delivered_consumer_seq": c.Delivered.Consumer,
						"ack_floor_stream_seq": c.AckFloor.Stream, "ack_floor_consumer_seq": c.AckFloor.Consumer,
						"num_pending": c.NumPending, "num_ack_pending": c.NumAckPending, "num_redelivered": c.NumRedelivered,
						"num_waiting": c.NumWaiting, "last_delivered": c.Delivered.LastActive, "last_acked": c.AckFloor.LastActive,
					},
					"stream_state": map[string]any{"first_seq": s.State.FirstSeq, "last_seq": s.State.LastSeq, "messages": s.State.Messages},
				}, nil
			}
		}
	}
	if !streamFound {
		return nil, fmt.Errorf("no stream called %s; nats_streams lists them", stream)
	}
	return nil, fmt.Errorf("stream %s has no consumer called %s; nats_streams with the stream lists its consumers", stream, durable)
}

// StreamRange is a stream's first and last sequence, from the monitoring document.
func StreamRange(jsz []byte, stream string) (first, last uint64, err error) {
	var doc jszDoc
	if err := json.Unmarshal(jsz, &doc); err != nil {
		return 0, 0, err
	}
	for _, a := range doc.Accounts {
		for _, s := range a.Streams {
			if s.Name == stream {
				return uint64(s.State.FirstSeq), uint64(s.State.LastSeq), nil
			}
		}
	}
	return 0, 0, fmt.Errorf("no stream called %s; nats_streams lists them", stream)
}

// Stored is one message a stream keeps.
type Stored struct {
	Seq     uint64
	Subject string
	Time    time.Time
	Data    []byte
}

// errNoMessage is what a MsgGet answers when no message at or after the sequence matches the subject.
var errNoMessage = errors.New("no message found")

// MsgGet asks the stream for the first message at or after seq whose subject matches (any when subject is empty).
type MsgGet func(ctx context.Context, stream string, seq uint64, subject string) (*Stored, error)

const (
	defaultMessages = 20
	maxMessages     = 200
	defaultBytes    = 1 << 20
	// bodyLimit is the largest body an answer carries, and only when asked.
	bodyLimit = 4096
	// block is how many sequences are scanned at once going back from the newest, and maxAsks the most requests one
	// answer makes: a subject rarely kept stops the answer, said as truncated, never a read of the whole stream.
	block   = 4096
	maxAsks = 3000
)

// bodyStream is the one stream that shows a body: the shared EVENTS, which holds every event, the seats' kept event streams
// (SEAT_<seat>_<event>) being sourced from it. Every other stream holds calls and replies, asks, mail, dead letters, records
// or logs, whose bodies are not an event's to show: it answers sequence, subject, time and size only.
const bodyStream = "EVENTS"

// MessageRow is one message of the answer.
type MessageRow struct {
	Seq         uint64    `json:"seq"`
	Subject     string    `json:"subject"`
	Time        time.Time `json:"time"`
	Size        int       `json:"size"`
	Body        string    `json:"body,omitempty"`
	BodyOmitted string    `json:"body_omitted,omitempty"`
}

// NewestMessages is the messages of the stream on the subject (a filter may hold wildcards), newest first, at most count
// and at most maxBytes of their sizes; first and last are the stream's own range. Said: whether it stopped before the
// oldest the stream keeps.
func NewestMessages(ctx context.Context, get MsgGet, stream, subject string, first, last uint64, count, maxBytes int, body bool) (map[string]any, error) {
	if count <= 0 {
		count = defaultMessages
	}
	if count > maxMessages {
		count = maxMessages
	}
	if maxBytes <= 0 {
		maxBytes = defaultBytes
	}
	var found []*Stored
	asks, bytes, stopped := 0, 0, ""
	end := last + 1 // exclusive
scan:
	for end > first && len(found) < count {
		start := first
		if end-first > block {
			start = end - block
		}
		var chunk []*Stored
		for seq := start; seq < end; {
			if asks++; asks > maxAsks {
				stopped = fmt.Sprintf("stopped after %d requests", maxAsks)
				break scan
			}
			m, err := get(ctx, stream, seq, subject)
			if errors.Is(err, errNoMessage) {
				break
			}
			if err != nil {
				return nil, unanswered(err, fmt.Sprintf("the messages of %s could not be read", stream))
			}
			if m.Seq >= end {
				break
			}
			chunk = append(chunk, m)
			seq = m.Seq + 1
		}
		for i := len(chunk) - 1; i >= 0; i-- {
			if len(found) >= count {
				stopped = fmt.Sprintf("stopped at the %d messages asked for", count)
				break scan
			}
			if bytes+len(chunk[i].Data) > maxBytes && len(found) > 0 {
				stopped = fmt.Sprintf("stopped at %d bytes", maxBytes)
				break scan
			}
			bytes += len(chunk[i].Data)
			found = append(found, chunk[i])
		}
		end = start
	}
	rows := make([]MessageRow, 0, len(found))
	for _, m := range found {
		r := MessageRow{Seq: m.Seq, Subject: m.Subject, Time: m.Time, Size: len(m.Data)}
		switch {
		case !body:
		case len(m.Data) > bodyLimit:
			r.BodyOmitted = fmt.Sprintf("larger than %d bytes", bodyLimit)
		default:
			r.Body = string(m.Data)
		}
		rows = append(rows, r)
	}
	out := map[string]any{"stream": stream, "subject": subject, "first_seq": first, "last_seq": last, "returned": len(rows), "messages": rows}
	if stopped != "" {
		out["stopped"] = stopped
	}
	return out, nil
}

// MsgOpen opens the bus for one answer and gives the reader of messages on it; close ends the connection.
type MsgOpen func(ctx context.Context) (get MsgGet, close func(), err error)

// LiveMsgGet reads messages by STREAM.MSG.GET on the module's own credential, one connection for the whole answer.
func LiveMsgGet(bus Bus) MsgOpen {
	return func(ctx context.Context) (MsgGet, func(), error) {
		js, closeBus, err := bus(ctx)
		if err != nil {
			return nil, nil, err
		}
		return func(ctx context.Context, stream string, seq uint64, subject string) (*Stored, error) {
			sctx, cancel := step(ctx)
			defer cancel()
			req := map[string]any{"seq": seq}
			if subject != "" {
				req["next_by_subj"] = subject
			}
			raw, _ := json.Marshal(req)
			rep, err := js.Conn().RequestWithContext(sctx, "$JS.API.STREAM.MSG.GET."+stream, raw)
			if err != nil {
				return nil, err
			}
			return decodeStored(rep.Data)
		}, closeBus, nil
	}
}

func decodeStored(b []byte) (*Stored, error) {
	var doc struct {
		Error *struct {
			Code        int
			ErrCode     int    `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
		Message *struct {
			Seq     uint64    `json:"seq"`
			Subject string    `json:"subject"`
			Data    []byte    `json:"data"`
			Time    time.Time `json:"time"`
		} `json:"message"`
	}
	if err := json.Unmarshal(b, &doc); err != nil {
		return nil, err
	}
	if doc.Error != nil {
		if doc.Error.Code == 404 {
			return nil, errNoMessage
		}
		return nil, fmt.Errorf("the bus refused: %s (%d)", doc.Error.Description, doc.Error.ErrCode)
	}
	if doc.Message == nil {
		return nil, errNoMessage
	}
	return &Stored{Seq: doc.Message.Seq, Subject: doc.Message.Subject, Time: doc.Message.Time, Data: doc.Message.Data}, nil
}

// readTools are the two reads that locate a lost event.
func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool {
	return []stdio.Tool{
		{Name: "nats_consumer",
			Description: "One consumer of one stream, in full: how it is set (filters, deliver policy, ack policy, ack wait, max ack pending, max deliver) and where it stands (delivered and acknowledged-up-to sequence of the stream and of the consumer, pending, unacknowledged, redelivered, waiting pulls, last delivery), with the stream's own first and last sequence beside it — the question behind an event a module did not hear.",
			Input:       map[string]any{"stream": str("the stream, e.g. EVENTS"), "consumer": str("the consumer's durable name, e.g. novox_mesh-module-assigner")},
			Run: func(a map[string]any) (any, error) {
				if !name.MatchString(arg(a, "stream")) || !name.MatchString(arg(a, "consumer")) {
					return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _)")
				}
				doc, err := jsz()
				if err != nil {
					return nil, err
				}
				return Consumer(doc, arg(a, "stream"), arg(a, "consumer"))
			}},
		{Name: "nats_stream_messages",
			Description: "The messages a stream keeps for one subject (a filter with * or > is allowed), newest first: sequence, subject, time and size of each, and the body only when asked for, no larger than 4 KiB, and only on the stream EVENTS (every kept event is there); on every other stream (kept events, asks, calls and replies, mail, dead letters, records, logs) only sequence, subject, time and size. Bounded by count (default 20, at most 200) and bytes (default 1 MiB); says when it stopped before the oldest the stream keeps. Read on the bus module's own credential. Refuses the streams of a module's kept state (KV_*, OBJ_*): their values are read through their owners' tools; this tool is for lost events.",
			Input: map[string]any{"stream": str("the stream, e.g. EVENTS"), "subject": str("a subject or filter, e.g. mesh.mod.mesh-delivery-indexer.event.view-changed"),
				"count": num("at most this many (default 20, at most 200)"), "bytes": num("at most this many bytes of messages (default 1048576)"),
				"body": map[string]any{"type": "boolean", "description": "carry the body of each message of at most 4 KiB"}},
			Run: func(a map[string]any) (any, error) {
				stream, subject := arg(a, "stream"), strings.TrimSpace(arg(a, "subject"))
				if !name.MatchString(stream) || subject == "" || strings.ContainsAny(subject, " \t") {
					return nil, errors.New("stream (one name) and subject (a subject or filter, no space) are required")
				}
				if strings.HasPrefix(stream, "KV_") || strings.HasPrefix(stream, "OBJ_") {
					return nil, fmt.Errorf("%s is a module's kept state (a bucket or an object store), which can hold secrets: it is read through its owner's tools, never here; this tool reads the events of a stream", stream)
				}
				doc, err := jsz()
				if err != nil {
					return nil, err
				}
				first, last, err := StreamRange(doc, stream)
				if err != nil {
					return nil, err
				}
				count, _ := a["count"].(float64)
				bytes, _ := a["bytes"].(float64)
				body, _ := a["body"].(bool)
				bodyHidden := ""
				if body && stream != bodyStream {
					body, bodyHidden = false, stream+" is not the stream EVENTS: its bodies are not shown, only sequence, subject, time and size"
				}
				ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
				defer cancel()
				get, closeBus, err := open(ctx)
				if err != nil {
					return nil, err
				}
				defer closeBus()
				out, err := NewestMessages(ctx, get, stream, subject, first, last, int(count), int(bytes), body)
				if err == nil && bodyHidden != "" {
					out["body_not_shown"] = bodyHidden
				}
				return out, err
			}},
	}
}