mesh-sdk/go/process/log.go

package process

import (
	"context"
	"fmt"
	"strings"
	"time"

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

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

// **Another module's log, read on this process's own credential** (novox/hq ADR 0571 clause 1; the lead, 2026-10-29): a
// log the module's manifest names in `reads`, `<module>.<log>`, read through an ordered consumer this connection creates
// and lets go of, never a durable one. Newest is the newest entry of each key, Since every entry from a time, and Watch
// those or every entry, then each new one as it is appended: event-driven, nothing reads the log again on a timer. The
// grants are buscompose's readLogGrants, given for a declared read of a log alone.

// LogReader is one module's log this connection reads.
type LogReader struct {
	Module, Name string
	js           func() (jetstream.JetStream, error)
	closed       func() bool
}

// ReadLog is module's log name, read on this connection.
func (c *Conn) ReadLog(module, name string) LogReader {
	return LogReader{Module: module, Name: name, closed: func() bool { return c.nc.IsClosed() },
		js: func() (jetstream.JetStream, error) { return jetstream.New(c.nc) }}
}

func (l LogReader) stream() string  { return "LOG_" + l.Module + "_" + l.Name }
func (l LogReader) subject() string { return "mesh.log." + l.Module + "." + l.Name }

// filter is keys as a subject filter: a key, `*` for one name, `**` for the rest; empty is every key.
func (l LogReader) filter(keys string) string {
	if keys == "" {
		return l.subject() + ".>"
	}
	return l.subject() + "." + watchPattern(keys)
}

func (l LogReader) config(keys string, from stdio.LogFrom) (jetstream.OrderedConsumerConfig, error) {
	cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{l.filter(keys)}}
	switch from.Kind {
	case stdio.LogFromNewestKind:
		cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy
	case stdio.LogFromAllKind:
		cfg.DeliverPolicy = jetstream.DeliverAllPolicy
	case stdio.LogFromSinceKind:
		at := from.Since
		cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at
	case stdio.LogFromNowKind:
		cfg.DeliverPolicy = jetstream.DeliverNewPolicy
	default:
		return cfg, fmt.Errorf("%s's log %s: a read starts from newest, all, since or now, not %q", l.Module, l.Name, from.Kind)
	}
	return cfg, nil
}

func (l LogReader) entry(m jetstream.Msg) (stdio.LogEntry, error) {
	meta, err := m.Metadata()
	if err != nil {
		return stdio.LogEntry{}, err
	}
	return stdio.LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.subject()+"."),
		Value: append([]byte(nil), m.Data()...), At: meta.Timestamp}, nil
}

// Newest is the newest entry of each key keys names, in the log's order.
func (l LogReader) Newest(keys string) ([]stdio.LogEntry, error) {
	return l.collect(keys, stdio.LogFromNewest)
}

// Since is every entry of keys from at on, in the log's order.
func (l LogReader) Since(at time.Time, keys string) ([]stdio.LogEntry, error) {
	return l.collect(keys, stdio.LogFromSince(at))
}

func (l LogReader) collect(keys string, from stdio.LogFrom) ([]stdio.LogEntry, error) {
	var out []stdio.LogEntry
	stop, err := l.watch(keys, from, func(e stdio.LogEntry) error { out = append(out, e); return nil }, false)
	if stop != nil {
		stop()
	}
	return out, err
}

// Watch hands handle each entry from where from says and then every new one, in the log's order, and returns once the
// entries there at its start have been handled. The watch lasts as long as the connection; an error from handle ends
// it, as the connection closing does. Before the entries at its start are handled, either is returned as the error.
func (l LogReader) Watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error) error {
	_, err := l.watch(keys, from, handle, true)
	return err
}

func (l LogReader) watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error, follow bool) (func(), error) {
	cfg, err := l.config(keys, from)
	if err != nil {
		return nil, err
	}
	js, err := l.js()
	if err != nil {
		return nil, err
	}
	bind, stopBind := context.WithTimeout(context.Background(), stateTimeout)
	defer stopBind()
	cons, err := js.OrderedConsumer(bind, l.stream(), cfg)
	if err != nil {
		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
	}
	info, err := cons.Info(bind)
	if err != nil {
		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
	}
	pending := info.NumPending
	it, err := cons.Messages()
	if err != nil {
		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
	}
	for ; pending > 0; pending-- {
		m, err := it.Next()
		if err == nil {
			var e stdio.LogEntry
			if e, err = l.entry(m); err == nil {
				err = handle(e)
			}
		}
		if err != nil {
			it.Stop()
			return nil, fmt.Errorf("%s's log %s: %w", l.Module, l.Name, err)
		}
	}
	if !follow {
		return it.Stop, nil
	}
	go func() {
		defer it.Stop()
		for {
			m, err := it.Next()
			if err != nil || (l.closed != nil && l.closed()) {
				return
			}
			e, err := l.entry(m)
			if err != nil || handle(e) != nil {
				return
			}
		}
	}()
	return it.Stop, nil
}