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
)

// 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 and no larger than 4 KiB. 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.",
			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")
				}
				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)
				ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
				defer cancel()
				get, closeBus, err := open(ctx)
				if err != nil {
					return nil, err
				}
				defer closeBus()
				return NewestMessages(ctx, get, stream, subject, first, last, int(count), int(bytes), body)
			}},
	}
}