mesh-sdk/go/log.go

package stdio

import (
	"encoding/json"
	"fmt"
	"time"
)

// Log — a module's record of operations on the bus, kept as long as the log is (novox/hq ADR 0297). Not the
// current value: that is state, and a log's state is its fold. Each entry is one operation on one key, its
// sequence in the log is its identity, and the order of the log is the order of the operations. A module
// declares each log it owns in its manifest (`logs`) and names it here as it named it there; only its owner appends.
// Another module's log is read by naming it `<module>.<log>` in the manifest's `reads` and here (novox/hq ADR 0571
// clause 1): read, newest per key, from a time, and followed. The runtime refuses, with the reason, one it did not issue.

// LogEntry is one operation on one key, as the log holds it.
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: its entries, in order, the sequence to read on from (0 when nothing
// follows) and the log's last sequence when it was read.
type LogPage struct {
	Entries []LogEntry `json:"entries"`
	Next    uint64     `json:"next"`
	Last    uint64     `json:"last"`
}

// LogInfo is how much a log holds: its entries and bytes, its cap in bytes, and its first and last sequence.
type LogInfo struct {
	Entries  uint64 `json:"entries"`
	Bytes    uint64 `json:"bytes"`
	MaxBytes uint64 `json:"max_bytes"`
	First    uint64 `json:"first"`
	Last     uint64 `json:"last"`
}

// KeptLog is one log this module owns.
type KeptLog struct{ name string }

// Log is a log this module owns, by its local name, or reads, `<module>.<log>`.
func Log(name string) KeptLog { return KeptLog{name: name} }

const logTimeout = 30 * time.Second

func (l KeptLog) ask(verb string, params map[string]any) (json.RawMessage, error) {
	c, err := live()
	if err != nil {
		return nil, err
	}
	params["log"] = l.name
	return c.ask("mesh/log."+verb, params, logTimeout)
}

func (l KeptLog) append(params map[string]any) (uint64, error) {
	raw, err := l.ask("append", params)
	if err != nil {
		return 0, err
	}
	var appended struct {
		Sequence uint64 `json:"sequence"`
	}
	return appended.Sequence, json.Unmarshal(raw, &appended)
}

// Append writes an entry for key, whatever came before it, and returns its sequence. A key is one or more
// names of letters, digits, `-` and `_`, joined by `.`; a value is JSON of at most 256 KiB. A full log
// refuses the entry: it never drops an older one to make room.
func (l KeptLog) Append(key string, value any) (uint64, error) {
	return l.append(map[string]any{"key": key, "value": value})
}

// AppendAt writes an entry for key only if the key's last entry is still at expect — the sequence the caller
// read, 0 for a key with no entry yet — and returns its sequence. When another entry for the key came in
// between, nothing is written and the error is ErrChanged, as for a lost State update: read again and
// decide again.
func (l KeptLog) AppendAt(key string, value any, expect uint64) (uint64, error) {
	return l.append(map[string]any{"key": key, "value": value, "expect": expect})
}

// Read is the entries from the sequence from on, oldest first, at most limit of them (0 is the runtime's
// page, 100; it gives at most 1000). keys narrows it: a key, with `*` for one name and `>` for the rest;
// empty is every key. Read on from the page's Next until it is 0.
func (l KeptLog) Read(from uint64, keys string, limit int) (*LogPage, error) {
	if limit < 0 {
		return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", l.name, limit)
	}
	params := map[string]any{"from": from}
	if keys != "" {
		params["keys"] = keys
	}
	if limit > 0 {
		params["limit"] = limit
	}
	raw, err := l.ask("read", params)
	if err != nil {
		return nil, err
	}
	var p LogPage
	return &p, json.Unmarshal(raw, &p)
}

// Last is a key's last entry, or nil when the key has none.
func (l KeptLog) Last(key string) (*LogEntry, error) {
	raw, err := l.ask("last", map[string]any{"key": key})
	if err != nil || len(raw) == 0 || string(raw) == "null" {
		return nil, err
	}
	var e LogEntry
	return &e, json.Unmarshal(raw, &e)
}

// Info is how much the log holds and how much it may.
func (l KeptLog) Info() (*LogInfo, error) {
	raw, err := l.ask("info", map[string]any{})
	if err != nil {
		return nil, err
	}
	var i LogInfo
	return &i, json.Unmarshal(raw, &i)
}

// Newest is the newest entry of every key keys names (a key, `*` for one name, `**` for the rest; empty is every key),
// in the log's order: runtime verb `mesh/log.newest` {log, keys}, answered as a list of entries.
func (l KeptLog) Newest(keys string) ([]LogEntry, error) {
	params := map[string]any{}
	if keys != "" {
		params["keys"] = keys
	}
	raw, err := l.ask("newest", params)
	if err != nil {
		return nil, err
	}
	var out []LogEntry
	return out, json.Unmarshal(raw, &out)
}

// Since is one page of the entries from a time on, as Read pages them: runtime verb `mesh/log.read` with `since` (RFC
// 3339) in place of `from`.
func (l KeptLog) Since(at time.Time, keys string, limit int) (*LogPage, error) {
	if limit < 0 {
		return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", l.name, limit)
	}
	params := map[string]any{"since": at.UTC().Format(time.RFC3339Nano)}
	if keys != "" {
		params["keys"] = keys
	}
	if limit > 0 {
		params["limit"] = limit
	}
	raw, err := l.ask("read", params)
	if err != nil {
		return nil, err
	}
	var p LogPage
	return &p, json.Unmarshal(raw, &p)
}

// LogFrom is where a log's watch starts: LogFromNewest (the newest entry of each key, then every new one), LogFromAll
// (every entry the log keeps), LogFromSince (every entry from a time) or LogFromNow (only new entries).
type LogFrom struct {
	Kind  string
	Since time.Time
}

// The kinds of LogFrom.
const (
	LogFromNewestKind = "newest"
	LogFromAllKind    = "all"
	LogFromSinceKind  = "since"
	LogFromNowKind    = "now"
)

var (
	LogFromNewest = LogFrom{Kind: LogFromNewestKind}
	LogFromAll    = LogFrom{Kind: LogFromAllKind}
	LogFromNow    = LogFrom{Kind: LogFromNowKind}
)

// LogFromSince starts a watch at a time.
func LogFromSince(at time.Time) LogFrom { return LogFrom{Kind: LogFromSinceKind, Since: at} }

func (f LogFrom) params() (map[string]any, error) {
	p := map[string]any{"from": f.Kind}
	switch f.Kind {
	case LogFromNewestKind, LogFromAllKind, LogFromNowKind:
	case LogFromSinceKind:
		p["since"] = f.Since.UTC().Format(time.RFC3339Nano)
	default:
		return nil, fmt.Errorf("a log's watch starts from newest, all, since or now, not %q", f.Kind)
	}
	return p, nil
}

// Watch hands handle each entry of keys from where from says, in the log's order, and then every new one as it is
// appended, and returns once the entries there at its start have been handled: runtime verb `mesh/log.watch` {log,
// keys, from, since}, each entry then sent as `mesh/log` {watch, entry}. Event-driven: nothing reads the log again on
// a timer. The watch lasts as long as the process.
func (l KeptLog) Watch(keys string, from LogFrom, handle func(LogEntry) error) error {
	params, err := from.params()
	if err != nil {
		return err
	}
	c, err := live()
	if err != nil {
		return err
	}
	id := c.newID()
	c.mu.Lock()
	c.logWatches[id] = handle
	c.mu.Unlock()
	params["log"], params["keys"] = l.name, keys
	if _, err := c.askAs(id, "mesh/log.watch", params, logTimeout); err != nil {
		c.mu.Lock()
		delete(c.logWatches, id)
		c.mu.Unlock()
		return err
	}
	return nil
}