mesh-host/internal/link/files.go

package link

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.novox.be/novox/mesh-sdk/go/declaration"
	"github.com/nats-io/nats.go"
)

// FilesBus is a link that can read an object of the bucket the controller keeps a declaration's large files in
// (novox/hq ADR 0486 §1, §2). Separate from Bus, as HaveBus is: a link that cannot is still a link, and a declaration
// whose files are given by digest is then refused whole, naming that.
type FilesBus interface {
	// FileObject is the object of that name, no larger than limit bytes, read within ctx. **Reads by name only**: a
	// node-engine's account may neither list the bucket nor write it (ADR 0486 §2).
	FileObject(ctx context.Context, name string, limit int64) ([]byte, error)
}

type filesKey struct{}

// WithFiles is ctx carrying the link a declaration's file references are fetched on.
func WithFiles(ctx context.Context, bus any) context.Context {
	if f, ok := bus.(FilesBus); ok {
		return context.WithValue(ctx, filesKey{}, f)
	}
	return ctx
}

// FilesFrom is the fetch the apply's context carries, or nil when the node has no such link.
func FilesFrom(ctx context.Context) func(context.Context, string, int64) ([]byte, error) {
	if f, ok := ctx.Value(filesKey{}).(FilesBus); ok {
		return f.FileObject
	}
	return nil
}

// objectInfo is the part of an object store's metadata record this read needs (nats.go's ObjectInfo, as the store writes it).
type objectInfo struct {
	Name    string `json:"name"`
	NUID    string `json:"nuid"`
	Size    uint64 `json:"size"`
	Digest  string `json:"digest"`
	Deleted bool   `json:"deleted"`
}

// FileObject reads one object by name from the declaration-files bucket, without listing the bucket or looking at
// anything but the object it names (novox/hq ADR 0486 §2: a node-engine's account may read by name and may neither
// list nor write).
//
// **The bus subjects this uses, and no others** (the grant of issue 1697 is exactly these; reviewer's note 5 there), for
// bucket B = declaration-files, stream S = OBJ_B, and E the base64 (URL alphabet) of the object's name:
//
//	request  $JS.API.DIRECT.GET.S.$O.B.M.E          the object's metadata. Grant only as ...DIRECT.GET.S.$O.B.M.*, never
//	                                                 the bare ...DIRECT.GET.S, which takes a sequence number
//	request  $JS.API.CONSUMER.CREATE.S.<hash>.$O.B.C.<nuid>
//	                                                 an ordered consumer on the object's chunks. Grant only as
//	                                                 ...CREATE.S.*.$O.B.C.*, never with ">" (a consumer filtered on
//	                                                 $O.B.M.> would read every object's name). Repeated on a reset
//	subscribe  a new inbox under the connection's own inbox prefix, where the chunks are delivered
//	publish  $JS.FC.S.>                              the flow-control replies of the ordered consumer; without them a
//	                                                 large object stalls at the flow-control window
//	request  $JS.API.CONSUMER.DELETE.S.<hash>       on unsubscribe; best effort, the consumer's inactivity threshold
//	                                                 removes it if refused
//
// It calls no STREAM.INFO and no STREAM.MSG.GET: js.ObjectStore asks for the stream's info (which lists every object's name)
// and store.Get reads the metadata with a message get a grant cannot keep from reading by sequence.
func (b OverNATS) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
	js := b.JS
	wait := 10 * time.Second
	if at, ok := ctx.Deadline(); ok {
		if wait = time.Until(at); wait <= 0 {
			return nil, context.DeadlineExceeded
		}
	}
	if b.Conn != nil {
		bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
		if err != nil {
			return nil, err
		}
		js = bounded
	}
	stream := "OBJ_" + declaration.FileBucket
	meta, err := js.GetLastMsg(stream, "$O."+declaration.FileBucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)),
		nats.DirectGet(), nats.Context(ctx))
	if err != nil {
		if errors.Is(err, nats.ErrMsgNotFound) {
			return nil, fmt.Errorf("no object %s in the bucket %s", name, declaration.FileBucket)
		}
		return nil, fmt.Errorf("the metadata of object %s: %w", name, err)
	}
	var info objectInfo
	if err := json.Unmarshal(meta.Data, &info); err != nil {
		return nil, fmt.Errorf("the metadata of object %s is not readable: %w", name, err)
	}
	if info.Deleted {
		return nil, fmt.Errorf("the object %s was deleted from the bucket %s", name, declaration.FileBucket)
	}
	if info.Size > uint64(limit) {
		return nil, fmt.Errorf("the object is larger than the %d bytes the declaration names", limit)
	}
	if info.Size == 0 {
		return []byte{}, nil
	}
	sub, err := js.SubscribeSync("$O."+declaration.FileBucket+".C."+info.NUID,
		nats.OrderedConsumer(), nats.BindStream(stream))
	if err != nil {
		return nil, fmt.Errorf("reading the chunks of object %s: %w", name, err)
	}
	defer sub.Unsubscribe()
	body := make([]byte, 0, info.Size)
	for uint64(len(body)) < info.Size {
		msg, err := sub.NextMsgWithContext(ctx)
		if err != nil {
			return nil, fmt.Errorf("reading the chunks of object %s (%d of %d bytes): %w", name, len(body), info.Size, err)
		}
		if uint64(len(body))+uint64(len(msg.Data)) > info.Size {
			return nil, fmt.Errorf("the chunks of object %s are more than the %d bytes its metadata says", name, info.Size)
		}
		body = append(body, msg.Data...)
	}
	sum := sha256.Sum256(body)
	if want := "SHA-256=" + base64.URLEncoding.EncodeToString(sum[:]); info.Digest != "" && info.Digest != want {
		return nil, fmt.Errorf("the chunks of object %s do not hash to the digest of its metadata", name)
	}
	return body, nil
}

func (l *natsLink) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
	return OverNATS{Conn: l.conn, JS: l.js}.FileObject(ctx, name, limit)
}

var (
	_ FilesBus = OverNATS{}
	_ FilesBus = (*natsLink)(nil)
)