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

package bus

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

	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"github.com/nats-io/nats.go/jetstream"
)

// A module's bus state in the new form (novox/hq ADR 0595, design 25 § 12.5; issue 1966 T1): an object store, a stream,
// a work queue, a ledger, each asked for in `requires` by a local name and bound in the module's membership by the bus
// names the bus module created. **A bundle reaches only its own module's bindings**: one account per machine carries
// every module on it, so what a module may reach is decided here, from its membership, and refused with the reason
// before anything is sent, as its state is (state.go).

// Binding is the module's binding of a local name, of the storage type storage, or the reason there is none.
func (c *Conn) Binding(module, local, storage string) (membership.BusBinding, error) {
	m := c.Membership(module)
	if m == nil {
		return membership.BusBinding{}, fmt.Errorf("%s has no membership issued on %s yet, so none of its bus state is "+
			"reachable until the mesh issues one (novox/hq ADR 0595)", module, c.node)
	}
	var names []string
	for _, b := range m.Bindings {
		if b.Local != local {
			names = append(names, b.Local+" ("+b.Storage+")")
			continue
		}
		if b.Storage != storage {
			return membership.BusBinding{}, fmt.Errorf("%s's %q is bound with the storage type %s, not %s (novox/hq ADR 0595)",
				module, local, b.Storage, storage)
		}
		return b, nil
	}
	sort.Strings(names)
	bound := "none"
	if len(names) > 0 {
		bound = strings.Join(names, ", ")
	}
	return membership.BusBinding{}, fmt.Errorf("%s was bound no %s called %q: a module reaches only the bus state its "+
		"manifest asks for in `requires`, by its local name, once the mesh created it; it was bound: %s (novox/hq ADR 0595)",
		module, storage, local, bound)
}

// Bindings is every binding a module's membership carries now.
func (c *Conn) Bindings(module string) []membership.BusBinding {
	if m := c.Membership(module); m != nil {
		return append([]membership.BusBinding(nil), m.Bindings...)
	}
	return nil
}

// jetStream is a JetStream handle on this connection.
func (c *Conn) jetStream() (jetstream.JetStream, error) { return jetstream.New(c.nc) }

func (c *Conn) objectStore(ctx context.Context, module, local string) (jetstream.ObjectStore, membership.BusBinding, error) {
	b, err := c.Binding(module, local, catalogue.StorageObjectStore)
	if err != nil {
		return nil, b, err
	}
	js, err := c.jetStream()
	if err != nil {
		return nil, b, err
	}
	obs, err := js.ObjectStore(ctx, b.ObjectStore)
	if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, b, fmt.Errorf("%s's objects %q are bound and their store %s is not on the bus", module, local, b.ObjectStore)
	}
	return obs, b, err
}

func objectInfoOf(i *jetstream.ObjectInfo) buscalls.ObjectInfo {
	return buscalls.ObjectInfo{Name: i.Name, Size: i.Size, Chunks: i.Chunks, Digest: i.Digest, Description: i.Description,
		Modified: i.ModTime, Deleted: i.Deleted}
}

// ObjectsPut stores one object whole under its name in the module's object store.
func (c *Conn) ObjectsPut(module string, r buscalls.ObjectsPutRequest) (buscalls.ObjectInfo, error) {
	if r.Name == "" {
		return buscalls.ObjectInfo{}, errors.New("an object is put under a name")
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	obs, _, err := c.objectStore(ctx, module, r.Objects)
	if err != nil {
		return buscalls.ObjectInfo{}, err
	}
	info, err := obs.PutBytes(ctx, r.Name, r.Data)
	if err != nil {
		return buscalls.ObjectInfo{}, err
	}
	if r.Description != "" {
		if err := obs.UpdateMeta(ctx, r.Name, jetstream.ObjectMeta{Name: r.Name, Description: r.Description}); err != nil {
			return buscalls.ObjectInfo{}, err
		}
		if info, err = obs.GetInfo(ctx, r.Name); err != nil {
			return buscalls.ObjectInfo{}, err
		}
	}
	return objectInfoOf(info), nil
}

// ObjectsGet is one object of the module's object store, its info and its bytes; nil when there is none.
func (c *Conn) ObjectsGet(module string, r buscalls.ObjectsNameRequest) (*buscalls.ObjectData, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	obs, _, err := c.objectStore(ctx, module, r.Objects)
	if err != nil {
		return nil, err
	}
	res, err := obs.Get(ctx, r.Name)
	if errors.Is(err, jetstream.ErrObjectNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	defer res.Close()
	info, err := res.Info()
	if err != nil {
		return nil, err
	}
	data, err := obs.GetBytes(ctx, r.Name)
	if err != nil {
		return nil, err
	}
	return &buscalls.ObjectData{Info: objectInfoOf(info), Data: data}, nil
}

// ObjectsDelete deletes one object of the module's object store; one that is not there is not an error.
func (c *Conn) ObjectsDelete(module string, r buscalls.ObjectsNameRequest) error {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	obs, _, err := c.objectStore(ctx, module, r.Objects)
	if err != nil {
		return err
	}
	if err := obs.Delete(ctx, r.Name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) {
		return err
	}
	return nil
}

// ObjectsList is every object the module's object store holds.
func (c *Conn) ObjectsList(module string, r buscalls.ObjectsRequest) (buscalls.ObjectsListed, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	obs, _, err := c.objectStore(ctx, module, r.Objects)
	if err != nil {
		return buscalls.ObjectsListed{}, err
	}
	out := buscalls.ObjectsListed{Objects: []buscalls.ObjectInfo{}}
	infos, err := obs.List(ctx)
	if errors.Is(err, jetstream.ErrNoObjectsFound) {
		return out, nil
	}
	if err != nil {
		return out, err
	}
	for _, i := range infos {
		out.Objects = append(out.Objects, objectInfoOf(i))
	}
	return out, nil
}

// ObjectsInfo is how full the module's object store is: its objects, its bytes and its cap.
func (c *Conn) ObjectsInfo(module string, r buscalls.ObjectsRequest) (buscalls.Fill, error) {
	listed, err := c.ObjectsList(module, r)
	if err != nil {
		return buscalls.Fill{}, err
	}
	b, _ := c.Binding(module, r.Objects, catalogue.StorageObjectStore)
	fill, err := c.streamFill("OBJ_" + b.ObjectStore)
	if err != nil {
		return buscalls.Fill{}, err
	}
	fill.Count = uint64(len(listed.Objects))
	return fill.Fill, nil
}

// ObjectsWatch hands deliver every object of the module's object store as it is, then each stored or deleted.
func (c *Conn) ObjectsWatch(module string, r buscalls.ObjectsWatchRequest, deliver func(buscalls.ObjectsChange) error) (func(), error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	obs, _, err := c.objectStore(ctx, module, r.Objects)
	cancel()
	if err != nil {
		return nil, err
	}
	watchCtx, stop := context.WithCancel(context.Background())
	w, err := obs.Watch(watchCtx, jetstream.IncludeHistory())
	if err != nil {
		stop()
		return nil, err
	}
	current := true
	for current {
		select {
		case i, ok := <-w.Updates():
			if !ok {
				stop()
				return nil, errors.New("the object store's watch ended before its current objects were handed over")
			}
			if i == nil {
				current = false
				continue
			}
			if err := deliver(buscalls.ObjectsChange{Watch: r.Watch, Object: objectInfoOf(i), Current: true}); err != nil {
				c.Logf("[mesh-tools] %s did not take an object of %s: %v", module, r.Objects, err)
			}
		case <-time.After(StateTimeout):
			_ = w.Stop()
			stop()
			return nil, fmt.Errorf("the object store %s did not hand its current objects in %s", r.Objects, StateTimeout)
		}
	}
	go func() {
		for i := range w.Updates() {
			if i == nil {
				continue
			}
			if err := deliver(buscalls.ObjectsChange{Watch: r.Watch, Object: objectInfoOf(i)}); err != nil {
				c.Logf("[mesh-tools] %s did not take an object of %s: %v", module, r.Objects, err)
			}
		}
	}()
	return func() { _ = w.Stop(); stop() }, nil
}

// streamFill is a stream's fill and its first and last sequence, by its name on the bus.
func (c *Conn) streamFill(name string) (buscalls.StreamFilled, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	js, err := c.jetStream()
	if err != nil {
		return buscalls.StreamFilled{}, err
	}
	s, err := js.Stream(ctx, name)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return buscalls.StreamFilled{}, fmt.Errorf("the stream %s is bound and not on the bus", name)
	}
	if err != nil {
		return buscalls.StreamFilled{}, err
	}
	info, err := s.Info(ctx)
	if err != nil {
		return buscalls.StreamFilled{}, err
	}
	max := uint64(0)
	if info.Config.MaxBytes > 0 {
		max = uint64(info.Config.MaxBytes)
	}
	return buscalls.StreamFilled{Fill: buscalls.Fill{Count: info.State.Msgs, Bytes: info.State.Bytes, MaxBytes: max},
		First: info.State.FirstSeq, Last: info.State.LastSeq}, nil
}

// streamBinding is the module's binding of a stream by its local name, a work queue among them.
func (c *Conn) streamBinding(module, local string) (membership.BusBinding, error) {
	return c.Binding(module, local, catalogue.StorageStream)
}

// StreamFill is how full the module's stream is.
func (c *Conn) StreamFill(module string, r buscalls.StreamRequest) (buscalls.StreamFilled, error) {
	b, err := c.streamBinding(module, r.Stream)
	if err != nil {
		return buscalls.StreamFilled{}, err
	}
	return c.streamFill(b.Stream)
}

// StreamReadMost is the most messages one read answers.
const StreamReadMost = 1000

// StreamRead is the module's stream's messages from a sequence on, Subjects narrowing them, at most Limit of them.
func (c *Conn) StreamRead(module string, r buscalls.StreamReadRequest) (buscalls.StreamPage, error) {
	b, err := c.streamBinding(module, r.Stream)
	if err != nil {
		return buscalls.StreamPage{}, err
	}
	return c.readStream(b.Stream, r.Subjects, r.From, r.Limit)
}

// readStream reads a stream by its bus name from a sequence on, the messages on the subjects filter names (every
// subject when empty), at most limit of them.
func (c *Conn) readStream(name, filter string, from uint64, limit int) (buscalls.StreamPage, error) {
	if limit <= 0 || limit > StreamReadMost {
		limit = StreamReadMost
	}
	if filter == "" {
		filter = ">"
	}
	if from == 0 {
		from = 1
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	js, err := c.jetStream()
	if err != nil {
		return buscalls.StreamPage{}, err
	}
	s, err := js.Stream(ctx, name)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return buscalls.StreamPage{}, fmt.Errorf("the stream %s is bound and not on the bus", name)
	}
	if err != nil {
		return buscalls.StreamPage{}, err
	}
	info, err := s.Info(ctx)
	if err != nil {
		return buscalls.StreamPage{}, err
	}
	page := buscalls.StreamPage{Messages: []buscalls.StreamMessage{}, Last: info.State.LastSeq}
	for seq := from; seq <= info.State.LastSeq && len(page.Messages) < limit; {
		msg, err := s.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(filter))
		if errors.Is(err, jetstream.ErrMsgNotFound) {
			break
		}
		if err != nil {
			return page, err
		}
		body := json.RawMessage(msg.Data)
		if !json.Valid(msg.Data) {
			body, _ = json.Marshal(string(msg.Data))
		}
		page.Messages = append(page.Messages, buscalls.StreamMessage{Sequence: msg.Sequence, Subject: msg.Subject,
			Body: body, At: msg.Time})
		seq = msg.Sequence + 1
		if len(page.Messages) == limit && seq <= info.State.LastSeq {
			page.Next = seq
		}
	}
	return page, nil
}

// WorkQueueSize is how many messages wait in the module's work queue, against its cap.
func (c *Conn) WorkQueueSize(module string, r buscalls.WorkQueueRequest) (buscalls.WorkQueueSized, error) {
	b, err := c.streamBinding(module, r.Queue)
	if err != nil {
		return buscalls.WorkQueueSized{}, err
	}
	fill, err := c.streamFill(b.Stream)
	if err != nil {
		return buscalls.WorkQueueSized{}, err
	}
	return buscalls.WorkQueueSized{Fill: fill.Fill}, nil
}

// WorkQueueDead is the module's work queue's dead letters: in a work queue a message is gone once settled, so one still
// kept below a consumer's acknowledgement floor was delivered as often as the consumer allows and never settled.
func (c *Conn) WorkQueueDead(module string, r buscalls.WorkQueueDeadRequest) (buscalls.WorkQueueDeadLetters, error) {
	b, err := c.streamBinding(module, r.Queue)
	if err != nil {
		return buscalls.WorkQueueDeadLetters{}, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	js, err := c.jetStream()
	if err != nil {
		return buscalls.WorkQueueDeadLetters{}, err
	}
	s, err := js.Stream(ctx, b.Stream)
	if err != nil {
		return buscalls.WorkQueueDeadLetters{}, fmt.Errorf("the work queue %s: %w", b.Stream, err)
	}
	out := buscalls.WorkQueueDeadLetters{Dead: []buscalls.DeadLetter{}}
	consumers := s.ListConsumers(ctx)
	for ci := range consumers.Info() {
		floor := ci.AckFloor.Stream
		if ci.Config.MaxDeliver <= 0 || floor == 0 {
			continue
		}
		page, err := c.readStream(b.Stream, ci.Config.FilterSubject, 1, r.Limit)
		if err != nil {
			return out, err
		}
		for _, m := range page.Messages {
			if m.Sequence > floor {
				break
			}
			out.Dead = append(out.Dead, buscalls.DeadLetter{Sequence: m.Sequence, Subject: m.Subject,
				Deliveries: ci.Config.MaxDeliver, Body: m.Body, At: m.At})
		}
	}
	if err := consumers.Err(); err != nil {
		return out, err
	}
	if r.Limit > 0 && len(out.Dead) > r.Limit {
		out.Dead = out.Dead[:r.Limit]
	}
	return out, nil
}