mesh-sdk/go/publish/publish.go

// Package publish is the runtime's helper for sending a file to a registry seat in parts (novox/hq ADR 0611, design
// 51 *Publishing*): `publish` opens a transfer naming every part's digest, the holder answers the parts it lacks, and
// only those travel, by `publish-part`, in chunks a bus message carries. A cut transfer goes on from what the holder
// answers when the same transfer is opened again. The seat's messages are mesh-contracts' (parts/registry); the call is
// the caller's own (a bundle's stdio.Ask, a process's call of a seat verb), so this package holds no transport.
package publish

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

	"git.novox.be/novox/mesh-contracts/go/parts/registry"
)

// MaxChunk is the most bytes of a part one publish-part carries: base64 makes it about 1 000 KiB, under 1 MiB, the
// server's largest message, with the message's own fields (design 51: each part travels in chunks of at most 1 MiB).
const MaxChunk = 760 << 10

// Part is one part of a transfer: its digest (sha256:<hex>), its size and how to read its bytes.
type Part struct {
	Digest string
	Size   int64
	Open   func() (io.ReadCloser, error)
}

// Call asks one verb of the seat, `seat:<seat>.<verb>`, with a body, and answers its JSON.
type Call func(key string, body any) (json.RawMessage, error)

// Sent is what a Publish did: the parts the holder lacked and were sent, and the chunks they took.
type Sent struct {
	Transfer string
	Held     bool
	Parts    []string
	Chunks   int
}

// Publish opens (or opens again) the transfer in on the seat and sends the parts the holder answers it lacks, each in
// chunks of at most MaxChunk, in sequence, the last marked. A part whose bytes do not hash to its digest is refused
// before anything of it is sent; a part the holder does not name is never read. Its outcome, published or refused, is
// the holder's event, which the caller waits for.
func Publish(ctx context.Context, call Call, seat string, in registry.PublishInput, parts []Part) (Sent, error) {
	in.Parts = in.Parts[:0]
	byDigest := map[string]Part{}
	for _, p := range parts {
		in.Parts = append(in.Parts, registry.Part{Digest: p.Digest, Size: p.Size})
		byDigest[p.Digest] = p
	}
	raw, err := call("seat:"+seat+"."+registry.VerbPublish, in)
	if err != nil {
		return Sent{}, fmt.Errorf("%s's publish of %s: %w", seat, in.Name, err)
	}
	var opened registry.PublishAnswer
	if err := json.Unmarshal(raw, &opened); err != nil {
		return Sent{}, fmt.Errorf("%s's answer to publish: %w", seat, err)
	}
	sent := Sent{Transfer: in.Transfer, Held: opened.Held}
	for _, d := range opened.Lacks {
		p, ok := byDigest[d]
		if !ok {
			return sent, fmt.Errorf("%s lacks the part %s, which the transfer %s does not name", seat, d, in.Transfer)
		}
		if err := ctx.Err(); err != nil {
			return sent, err
		}
		if err := check(p); err != nil {
			return sent, err
		}
		n, err := sendPart(ctx, call, seat, in.Transfer, p)
		sent.Chunks += n
		if err != nil {
			return sent, err
		}
		sent.Parts = append(sent.Parts, d)
	}
	return sent, nil
}

// check reads the part once and refuses it when its bytes are not its digest or its size.
func check(p Part) error {
	r, err := p.Open()
	if err != nil {
		return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err)
	}
	defer r.Close()
	h := sha256.New()
	n, err := io.Copy(h, r)
	if err != nil {
		return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err)
	}
	if got := "sha256:" + hex.EncodeToString(h.Sum(nil)); got != p.Digest || n != p.Size {
		return fmt.Errorf("the part %s holds %d bytes of digest %s, not its own %d: nothing of it is sent", p.Digest, n, got, p.Size)
	}
	return nil
}

func sendPart(ctx context.Context, call Call, seat, transfer string, p Part) (int, error) {
	r, err := p.Open()
	if err != nil {
		return 0, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err)
	}
	defer r.Close()
	buf := make([]byte, MaxChunk)
	// A chunk is sent once the next read says whether it is the last, so the last is always marked: an empty part is
	// one empty chunk, marked last.
	var prev []byte
	sent := 0
	send := func(data []byte, last bool) error {
		if err := sendChunk(call, seat, transfer, p.Digest, sent, last, data); err != nil {
			return err
		}
		sent++
		return nil
	}
	for {
		if err := ctx.Err(); err != nil {
			return sent, err
		}
		n, err := io.ReadFull(r, buf)
		end := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)
		if err != nil && !end {
			return sent, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err)
		}
		if end && n == 0 {
			if prev == nil {
				prev = []byte{}
			}
			return sent, send(prev, true)
		}
		if prev != nil {
			if err := send(prev, false); err != nil {
				return sent, err
			}
		}
		prev = append([]byte(nil), buf[:n]...)
		if end {
			return sent, send(prev, true)
		}
	}
}

func sendChunk(call Call, seat, transfer, part string, sequence int, last bool, data []byte) error {
	raw, err := call("seat:"+seat+"."+registry.VerbPublishPart, registry.PublishPartInput{Transfer: transfer, Part: part,
		Sequence: int64(sequence), Last: last, Data: base64.StdEncoding.EncodeToString(data)})
	if err != nil {
		return fmt.Errorf("%s's publish-part %d of %s: %w", seat, sequence, part, err)
	}
	var a registry.PublishPartAnswer
	if err := json.Unmarshal(raw, &a); err != nil {
		return fmt.Errorf("%s's answer to publish-part: %w", seat, err)
	}
	return nil
}