mesh-sdk/go/process/chunks.go

package process

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"time"

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

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

// CallChunks calls a seat verb whose answer file is sent in chunks (a verb its seat definition marks `chunks`, design 47 G
// §681, A-139), the one way a process receives such an answer (A-291): it subscribes to the chunks of call (the 128 random
// bits the verb's input carries as its call) before it asks, so none is missed, asks key ("seat:<seat>.<verb>[@<node>]")
// with body as Call does, hands the verb's answer to accept (nil accepts every answer; an error ends the call before a chunk
// is taken), then receives the number of chunks the answer's member `chunks` names, in order, acknowledging each, into w.
// It answers the verb's answer. A chunk that does not come within buscalls.SeatAckWithin, or ctx ending, ends the receive.
func (c *Conn) CallChunks(ctx context.Context, key, call string, body any, accept func(answer json.RawMessage) error, w io.Writer) (json.RawMessage, error) {
	a, err := calladdress.Parse(key)
	if err != nil {
		return nil, err
	}
	if a.Seat == "" {
		return nil, fmt.Errorf("%q names no seat verb, so it sends no chunks", key)
	}
	if call == "" {
		return nil, fmt.Errorf("the call of %s names no call token for its chunks", key)
	}
	send := buscalls.SeatChunks{Seat: a.Seat, Verb: a.Verb, Call: call}
	chunks := make(chan *nats.Msg, 4)
	sub, err := c.nc.ChanSubscribe(send.Chunks(), chunks)
	if err != nil {
		return nil, fmt.Errorf("the chunks of %s cannot be heard: %w", key, err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	answer, err := c.Call(key, body)
	if err != nil {
		return nil, err
	}
	if accept != nil {
		if err := accept(answer); err != nil {
			return answer, err
		}
	}
	var count struct {
		Chunks int64 `json:"chunks"`
	}
	if err := json.Unmarshal(answer, &count); err != nil {
		return answer, fmt.Errorf("%s answered no count of its chunks: %w", key, err)
	}
	next := func() (string, []byte, error) {
		bound := time.NewTimer(buscalls.SeatAckWithin)
		defer bound.Stop()
		select {
		case m := <-chunks:
			return m.Subject, m.Data, nil
		case <-bound.C:
			return "", nil, buscalls.ErrSeatChunkMissed
		case <-ctx.Done():
			return "", nil, ctx.Err()
		}
	}
	ack := func(subject string) error {
		_, err := c.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin)
		return err
	}
	if _, err := send.Receive(count.Chunks, next, ack, w); err != nil {
		return answer, fmt.Errorf("the answer of %s: %w", key, err)
	}
	return answer, nil
}

// CallFile is CallChunks for an answer read whole, in the form a bundle's ordinary seat call answers a chunked verb (design 47
// G A-249, A-253): the verb's answer with the whole file added as its member `file` (base64), none when no chunk was sent. The
// call token is body's member `call`.
func (c *Conn) CallFile(key string, body any) (json.RawMessage, error) {
	call := callOf(body)
	var file bytes.Buffer
	answer, err := c.CallChunks(context.Background(), key, call, body, nil, &file)
	if err != nil {
		return nil, err
	}
	var whole map[string]any
	if err := json.Unmarshal(answer, &whole); err != nil {
		return nil, fmt.Errorf("%s answered what is no object: %w", key, err)
	}
	if file.Len() > 0 {
		whole["file"] = base64.StdEncoding.EncodeToString(file.Bytes())
	}
	return json.Marshal(whole)
}

// callOf is the member call of a verb's input.
func callOf(body any) string {
	raw, err := json.Marshal(body)
	if err != nil {
		return ""
	}
	var in struct {
		Call string `json:"call"`
	}
	_ = json.Unmarshal(raw, &in)
	return in.Call
}