mesh-sdk/go/stateobjects/send.go

package stateobjects

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

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

// MessageLimit is the most a message of the bus carries when the sender does not know its connection's own
// (`nc.MaxPayload()`), and MessageHeaderRoom what the headers and the subject take of it.
const (
	MessageLimit      = 1 << 20
	MessageHeaderRoom = 8 << 10
)

// Sender sends the body of a verb's reply or an event: a body over the per-message limit is put in the sender's store of
// sent objects and its reference is sent instead (ADR 0666 §16). The store is the bus module's `<module>_sent-objects`,
// granted to the sender alone and read by the callers its seat definition names and the consumers that declare the event;
// a sent object lives as long as that store's age, which is transport and not kept data.
type Sender struct {
	// Objects is the sender's store of sent objects, and Store its name as a reader asks for it.
	Objects Objects
	Store   string
	// Limit is the connection's largest message; zero is MessageLimit.
	Limit int
	// Log is a foundation part's old `state` section `sent-log`, with Node the key of this node's sends in it and Now
	// the clock (zero is time.Now): the sender keeps the sequence of its own sends there and ages the store from it at
	// each send (design 47 G §410, §330). A sender with no Log ages nothing: the bus expires a template store's objects.
	Log  KV
	Node string
	Now  func() time.Time
}

// The sent store of a foundation part keeps an object SentMaxAge, and a send deletes at most SentAgeAtMost of the
// objects past it (design 47 G §330, §59).
const (
	SentLogState  = "sent-log"
	SentAgeAtMost = 20
)

type sentEntry struct {
	Object string    `json:"object"`
	At     time.Time `json:"at"`
}

// record writes this send into the sequence of the sender's sends by compare-and-set, and deletes the objects of the
// oldest sends that are past SentMaxAge, at most SentAgeAtMost; two sends at once lose none, the one that lost the
// compare reading the sequence again.
func (s Sender) record(object string) error {
	now := time.Now
	if s.Now != nil {
		now = s.Now
	}
	for {
		e, err := s.Log.Get(s.Node)
		if err != nil {
			return err
		}
		var log []sentEntry
		if e != nil {
			if err := json.Unmarshal(e.Value, &log); err != nil {
				return fmt.Errorf("the %s of %s: %w", SentLogState, s.Node, err)
			}
		}
		var aged []sentEntry
		for len(aged) < SentAgeAtMost && len(log) > 0 && now().Sub(log[0].At) > SentMaxAge {
			aged, log = append(aged, log[0]), log[1:]
		}
		log = append(log, sentEntry{object, now()})
		raw, _ := json.Marshal(log)
		if e == nil {
			err = s.Log.Create(s.Node, raw)
		} else {
			err = s.Log.Update(s.Node, raw, e.Revision)
		}
		if errors.Is(err, ErrChanged) {
			continue
		}
		if err != nil {
			return err
		}
		for _, a := range aged {
			if err := s.Objects.Delete(a.Object); err != nil {
				return fmt.Errorf("aging the sent object %s: %w", a.Object, err)
			}
		}
		return nil
	}
}

// Send is what to send for body: the body itself when it fits, else the reference of the object it was put in as
// `<label>.<sha256>`. label tells the object's use (the verb or event, the caller or the id).
func (s Sender) Send(label string, body []byte) ([]byte, error) {
	limit := s.Limit
	if limit <= 0 {
		limit = MessageLimit
	}
	if len(body) <= limit-MessageHeaderRoom {
		if _, isRef := ReadRef(body); isRef {
			return nil, fmt.Errorf("%s: %w", label, errLooksLikeRef)
		}
		return body, nil
	}
	if most := s.Objects.MostBytes(); most > 0 && len(body) > most {
		return nil, fmt.Errorf("%s: a bundle sends at most %d bytes of body, the runtime's channel carrying a message of 16 MiB; this one is %d", label, most, len(body))
	}
	ref := Ref{Store: s.Store, Object: objectName(label, digestOf(body)), Size: int64(len(body)), Digest: digestOf(body), Form: FormVersion}
	if err := s.Objects.Put(ref.Object, body); err != nil {
		return nil, fmt.Errorf("the object for %s: %w", label, err)
	}
	if s.Log != nil {
		if err := s.record(ref.Object); err != nil {
			return nil, fmt.Errorf("the %s for %s: %w", SentLogState, label, err)
		}
	}
	return Wrap(ref), nil
}

// Fetch reads the object a reference names.
type Fetch func(Ref) ([]byte, error)

// Receive is the body that was sent: the body itself, or the object its reference names, fetched and checked against the
// reference's size and digest.
func Receive(body []byte, fetch Fetch) ([]byte, error) {
	ref, ok := ReadRef(body)
	if !ok {
		return body, nil
	}
	data, err := fetch(ref)
	if err != nil {
		return nil, fmt.Errorf("the object %s of %s: %w", ref.Object, ref.Store, err)
	}
	if int64(len(data)) != ref.Size || digestOf(data) != ref.Digest {
		return nil, fmt.Errorf("the object %s of %s is not the one its reference names (size or digest differ)", ref.Object, ref.Store)
	}
	return data, nil
}

// BundleFetch reads a sent object through the runtime, which serves it under the standing grant the grants part derives
// from the manifests (buscalls.ObjectsSentGet).
func BundleFetch(ref Ref) ([]byte, error) {
	raw, err := stdio.BusState(buscalls.ObjectsSentGet, buscalls.ObjectsSentGetRequest{Store: ref.Store, Object: ref.Object, Size: ref.Size, Digest: ref.Digest})
	if err != nil {
		return nil, err
	}
	var d buscalls.ObjectData
	if err := json.Unmarshal(raw, &d); err != nil {
		return nil, fmt.Errorf("the runtime's answer to %s: %w", buscalls.ObjectsSentGet, err)
	}
	return d.Data, nil
}

// ProcessFetch reads a sent object through the reader's own object store.
func ProcessFetch(o Objects) Fetch {
	return func(ref Ref) ([]byte, error) { return o.Get(ref.Object) }
}