mesh-sdk/go/parts/receive.go

package parts

// The receiving side of a file in parts (novox/hq ADR 0611, ADR 0627 §2, design 51 *Publishing*): a holder opens a
// transfer naming every part, takes the chunks of the parts it lacks into its own objects, checks each whole part's
// hash, drops a transfer left idle and says when its objects fill. Like the sender it names no seat, no verb and no
// seat's type: the caller's own shape of an opening rides along as M, its objects and its state are its own, and the
// checks before storing, the storing and the saying of a refusal are the caller's functions. Each holder binds it to
// the seat it holds.

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"time"
)

// The bounds design 51 gives the receiving side; a Receiver's own fields override them.
const (
	// MaxReceived is the most bytes one chunk may carry when it arrives.
	MaxReceived = 1 << 20
	// IdleAfter is how long a transfer with no chunk stays open.
	IdleAfter = 24 * time.Hour
	// FullPercent is the fill of the objects past which the receiver says they fill.
	FullPercent = 80
)

// Want is one part an opening names: its digest (sha256:<hex>) and its size.
type Want struct {
	Digest string `json:"digest"`
	Size   int64  `json:"size"`
}

// Transfer is one open transfer, kept in the caller's state; its chunks are objects of the caller's object store. Shape
// is the caller's own body of the opening, kept as it came.
type Transfer[M any] struct {
	ID     string `json:"id"`
	Opener string `json:"opener"`
	Parts  []Want `json:"parts"`
	// Held are the parts whose every chunk arrived and hashed right, or that the holder held at the open; Chunks the
	// chunks taken per part.
	Held   []string       `json:"held,omitempty"`
	Chunks map[string]int `json:"chunks,omitempty"`
	Opened time.Time      `json:"opened"`
	Last   time.Time      `json:"last"`
	Shape  M              `json:"shape"`
}

// Lacks is the digests of the parts the transfer does not hold yet, sorted.
func (t Transfer[M]) Lacks() []string {
	var out []string
	for _, p := range t.Parts {
		if !slices.Contains(t.Held, p.Digest) && !slices.Contains(out, p.Digest) {
			out = append(out, p.Digest)
		}
	}
	sort.Strings(out)
	return out
}

// Taken says whether any chunk of the part was taken in this transfer (a part held at the open was not).
func (t Transfer[M]) Taken(digest string) bool {
	_, ok := t.Chunks[digest]
	return ok
}

// Objects is the holder's own object store the chunks go into.
type Objects interface {
	Put(name string, data []byte) error
	Get(name string) ([]byte, error)
	Delete(name string) error
	// Used is the bytes the store holds.
	Used() (uint64, error)
}

// Transfers is the holder's own state of open transfers, keyed by Transfer.ID.
type Transfers[M any] interface {
	Get(id string) (Transfer[M], bool, error)
	Put(t Transfer[M]) error
	Delete(id string) error
	List() ([]Transfer[M], error)
}

// Received is the receiver's answer to an open or a chunk: the parts still lacked, and whether the transfer ended
// (stored or refused).
type Received struct {
	ID    string
	Lacks []string
	Ended bool
}

// AtCap opens the reason of a refusal for want of room: the objects would pass their cap. A holder tells it from
// the other refusals with RefusedAtCap, so the finding it raises for it can clear once it takes a file again.
const AtCap = "the objects are at their cap"

// RefusedAtCap says whether a refusal's reason is one for want of room (AtCap).
func RefusedAtCap(reason string) bool { return strings.HasPrefix(reason, AtCap) }

// Receiver takes transfers in parts into a holder's own objects.
type Receiver[M any] struct {
	Objects   Objects
	Transfers Transfers[M]
	// Prefix starts the name of every chunk object, so two receivers can share one object store.
	Prefix string
	// Cap is the bytes the objects may hold; an opening that would pass it is refused. Required.
	Cap uint64
	// MaxChunk, Idle and Full override MaxReceived, IdleAfter and FullPercent when set.
	MaxChunk int
	Idle     time.Duration
	Full     uint64
	// Holds says the holder already holds a part, which then counts as held at the open and is not sent. Optional.
	Holds func(t Transfer[M], digest string) bool
	// Check is the caller's check of a whole transfer before anything is stored; its error is the refusal's reason.
	// Optional.
	Check func(t Transfer[M]) error
	// Store stores a whole, checked transfer: the parts taken in it by digest (not those held at the open). It says
	// the outcome itself; its error is answered as it is.
	Store func(t Transfer[M], parts map[string][]byte) error
	// Refuse says a refusal by the receiver (the cap, the caller's check, a part it cannot read back). Its own error is
	// joined to the refusal's.
	Refuse func(t Transfer[M], reason string) error
	// Filled is told the objects' fill after every change: used, the cap and whether it is past Full percent. Optional.
	Filled func(used, cap uint64, full bool)
	Now    func() time.Time
}

func (r *Receiver[M]) now() time.Time {
	if r.Now != nil {
		return r.Now().UTC()
	}
	return time.Now().UTC()
}

func (r *Receiver[M]) maxChunk() int {
	if r.MaxChunk > 0 {
		return r.MaxChunk
	}
	return MaxReceived
}

func (r *Receiver[M]) idle() time.Duration {
	if r.Idle > 0 {
		return r.Idle
	}
	return IdleAfter
}

func (r *Receiver[M]) fullPct() uint64 {
	if r.Full > 0 {
		return r.Full
	}
	return FullPercent
}

// Open opens a transfer for opener, or opens it again: a transfer already open answers what it lacks, so a cut transfer
// goes on from there, and only its opener may open it again. A new transfer that would pass the cap is refused; parts
// the holder holds count as held; when nothing is lacked the transfer ends at once.
func (r *Receiver[M]) Open(id, opener string, parts []Want, shape M) (Received, error) {
	switch {
	case strings.TrimSpace(id) == "" || strings.ContainsAny(id, "/. *>"):
		return Received{}, fmt.Errorf("transfer %q is no transfer id", id)
	case len(parts) == 0:
		return Received{}, errors.New("no parts")
	}
	for _, p := range parts {
		if !strings.HasPrefix(p.Digest, "sha256:") || len(p.Digest) != len("sha256:")+64 {
			return Received{}, fmt.Errorf("part digest %q is no sha256 digest", p.Digest)
		}
	}
	if t, found, err := r.Transfers.Get(id); err != nil {
		return Received{}, err
	} else if found {
		if t.Opener != opener {
			return Received{}, fmt.Errorf("refused for %s: transfer %s was opened by another caller", opener, id)
		}
		return Received{ID: id, Lacks: t.Lacks()}, nil
	}
	t := Transfer[M]{ID: id, Opener: opener, Parts: parts, Chunks: map[string]int{}, Opened: r.now(), Last: r.now(), Shape: shape}
	used, err := r.Objects.Used()
	if err != nil {
		return Received{}, fmt.Errorf("the objects' fill cannot be read: %w", err)
	}
	var size uint64
	for _, p := range parts {
		size += uint64(max(p.Size, 0))
	}
	if used+size > r.Cap {
		return Received{ID: id, Ended: true}, r.refuse(t, fmt.Sprintf(AtCap+" of %d bytes (%d held, %d asked)", r.Cap, used, size))
	}
	if r.Holds != nil {
		for _, p := range parts {
			if r.Holds(t, p.Digest) {
				t.Held = append(t.Held, p.Digest)
			}
		}
	}
	if err := r.Transfers.Put(t); err != nil {
		return Received{}, fmt.Errorf("transfer %s cannot be kept: %w", id, err)
	}
	r.fill()
	if len(t.Lacks()) == 0 {
		return Received{ID: id, Ended: true}, r.finish(t)
	}
	return Received{ID: id, Lacks: t.Lacks()}, nil
}

// Take takes one chunk of a part from the transfer's opener: chunks of a part arrive in sequence from 0, the last
// marked; the whole part must hash to its digest, else its chunks are dropped and the part is lacked again. Belongs,
// when given, refuses a transfer the caller's own shape says is not this call's (another seat's, say). When the last
// lacked part arrives the transfer ends: the caller's check, then its store.
func (r *Receiver[M]) Take(id, opener, digest string, sequence int, last bool, data []byte, belongs func(Transfer[M]) error) (Received, error) {
	t, found, err := r.Transfers.Get(id)
	switch {
	case err != nil:
		return Received{}, err
	case !found:
		return Received{}, fmt.Errorf("transfer %s is not open (dropped after %s with no chunk, or never opened)", id, r.idle())
	case t.Opener != opener:
		return Received{}, fmt.Errorf("refused for %s: transfer %s was opened by another caller", opener, id)
	}
	if belongs != nil {
		if err := belongs(t); err != nil {
			return Received{}, err
		}
	}
	switch {
	case len(data) > r.maxChunk():
		return Received{}, fmt.Errorf("a chunk of %d bytes is over the %d a chunk may carry", len(data), r.maxChunk())
	case !slices.ContainsFunc(t.Parts, func(p Want) bool { return p.Digest == digest }):
		return Received{}, fmt.Errorf("transfer %s names no part %s", id, digest)
	case slices.Contains(t.Held, digest):
		return Received{ID: id, Lacks: t.Lacks()}, nil
	case sequence != t.Chunks[digest]:
		return Received{}, fmt.Errorf("part %s of transfer %s waits for chunk %d, not %d", digest, id, t.Chunks[digest], sequence)
	}
	if err := r.Objects.Put(r.chunkName(id, digest, sequence), data); err != nil {
		return Received{}, fmt.Errorf("the objects did not take the chunk: %w", err)
	}
	// A transfer read back from a key-value state has no Chunks until one was taken: the empty map is kept omitted.
	if t.Chunks == nil {
		t.Chunks = map[string]int{}
	}
	t.Chunks[digest] = sequence + 1
	t.Last = r.now()
	if last {
		whole, err := r.part(t, digest)
		if err != nil {
			return Received{}, err
		}
		if got := "sha256:" + sum(whole); got != digest {
			r.dropChunks(t, digest)
			delete(t.Chunks, digest)
			_ = r.Transfers.Put(t)
			r.fill()
			return Received{}, fmt.Errorf("refused: the bytes of part %s hash to %s; send the part again", digest, got)
		}
		t.Held = append(t.Held, digest)
	}
	if err := r.Transfers.Put(t); err != nil {
		return Received{}, err
	}
	r.fill()
	if len(t.Lacks()) == 0 {
		return Received{ID: id, Ended: true}, r.finish(t)
	}
	return Received{ID: id, Lacks: t.Lacks()}, nil
}

// finish checks and stores a whole transfer; its chunks and its key go either way.
func (r *Receiver[M]) finish(t Transfer[M]) error {
	defer func() {
		for _, p := range t.Parts {
			r.dropChunks(t, p.Digest)
		}
		_ = r.Transfers.Delete(t.ID)
		r.fill()
	}()
	if r.Check != nil {
		if err := r.Check(t); err != nil {
			return r.refuse(t, err.Error())
		}
	}
	parts := map[string][]byte{}
	for _, p := range t.Parts {
		if !t.Taken(p.Digest) {
			continue // held by the holder already
		}
		b, err := r.part(t, p.Digest)
		if err != nil {
			return r.refuse(t, "part "+p.Digest+" cannot be read from the objects: "+err.Error())
		}
		parts[p.Digest] = b
	}
	return r.Store(t, parts)
}

func (r *Receiver[M]) refuse(t Transfer[M], reason string) error {
	err := errors.New("refused: " + reason)
	if r.Refuse != nil {
		if serr := r.Refuse(t, reason); serr != nil {
			return fmt.Errorf("%w; and the refusal could not be said: %w", err, serr)
		}
	}
	return err
}

func (r *Receiver[M]) part(t Transfer[M], digest string) ([]byte, error) {
	var buf bytes.Buffer
	for seq := 0; seq < t.Chunks[digest]; seq++ {
		b, err := r.Objects.Get(r.chunkName(t.ID, digest, seq))
		if err != nil {
			return nil, err
		}
		buf.Write(b)
	}
	return buf.Bytes(), nil
}

func (r *Receiver[M]) dropChunks(t Transfer[M], digest string) {
	for seq := 0; seq < t.Chunks[digest]; seq++ {
		_ = r.Objects.Delete(r.chunkName(t.ID, digest, seq))
	}
}

func (r *Receiver[M]) chunkName(id, digest string, seq int) string {
	return fmt.Sprintf("%s%s/%s/%d", r.Prefix, id, strings.TrimPrefix(digest, "sha256:"), seq)
}

// fill tells Filled the objects' fill.
func (r *Receiver[M]) fill() {
	if r.Filled == nil || r.Cap == 0 {
		return
	}
	used, err := r.Objects.Used()
	if err != nil {
		return
	}
	r.Filled(used, r.Cap, used*100 >= r.Cap*r.fullPct())
}

// Drop drops a transfer, by hand or because it was idle: its chunks and its key.
func (r *Receiver[M]) Drop(id string) (Transfer[M], error) {
	t, found, err := r.Transfers.Get(id)
	if err != nil {
		return Transfer[M]{}, err
	}
	if !found {
		return Transfer[M]{}, fmt.Errorf("transfer %s is not open", id)
	}
	for _, p := range t.Parts {
		r.dropChunks(t, p.Digest)
	}
	if err := r.Transfers.Delete(id); err != nil {
		return Transfer[M]{}, err
	}
	r.fill()
	return t, nil
}

// DropIdle drops every transfer with no chunk for Idle and answers their ids. It runs when its caller calls it (on an
// open, on a listing), never on a timer.
func (r *Receiver[M]) DropIdle() ([]string, error) {
	all, err := r.Transfers.List()
	if err != nil {
		return nil, err
	}
	var out []string
	for _, t := range all {
		if r.now().Sub(t.Last) >= r.idle() {
			if _, err := r.Drop(t.ID); err == nil {
				out = append(out, t.ID)
			}
		}
	}
	sort.Strings(out)
	return out, nil
}

func sum(b []byte) string {
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}