mesh-delivery/modules/mesh-delivery-view/internal/view/owner.go

package view

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

// OwnerRead reads an owner's state through the owner's ordinary read tool (design 25 § 12.2, ADR 0615 §1, DLV-M4): the
// view's tools answer the indexer's and the assigner's states by asking it, never by reading their buckets. ask is the
// bundle's ask of another module's tool (mesh-sdk stdio Ask), tool the read tool as the manifest's `invokes` names it
// (`<owner>.<short>_read_<state>`). With no argument the tool answers the keys; with key, the key's value and revision;
// with values, the values whole by key in pages; with key and history, the values it keeps for the key with their times.
func OwnerRead(ask func(key string, body any) (json.RawMessage, error), tool string) StateRead {
	entry := func(key string) (json.RawMessage, uint64, error) {
		raw, err := ask(tool, map[string]any{"key": key})
		if err != nil {
			return nil, 0, fmt.Errorf("%s: %w", tool, err)
		}
		var a struct {
			Found    bool            `json:"found"`
			Value    json.RawMessage `json:"value"`
			Revision uint64          `json:"revision"`
		}
		if err := json.Unmarshal(raw, &a); err != nil {
			return nil, 0, fmt.Errorf("%s answered what is not a key's value: %w", tool, err)
		}
		if !a.Found || len(a.Value) == 0 {
			return nil, 0, nil
		}
		return a.Value, a.Revision, nil
	}
	return StateRead{
		Keys: func() ([]string, error) {
			raw, err := ask(tool, map[string]any{})
			if err != nil {
				return nil, fmt.Errorf("%s: %w", tool, err)
			}
			var a struct {
				Keys []string `json:"keys"`
			}
			if err := json.Unmarshal(raw, &a); err != nil {
				return nil, fmt.Errorf("%s answered what is not a state's keys: %w", tool, err)
			}
			return a.Keys, nil
		},
		GetEntry: entry,
		Get: func(key string) (json.RawMessage, error) {
			v, _, err := entry(key)
			return v, err
		},
		All: func() (map[string]json.RawMessage, error) {
			out := map[string]json.RawMessage{}
			for after, pages := "", 0; ; pages++ {
				if pages == maxOwnerPages {
					return nil, fmt.Errorf("%s answered more than %d pages", tool, maxOwnerPages)
				}
				body := map[string]any{"values": true}
				if after != "" {
					body["after"] = after
				}
				raw, err := ask(tool, body)
				if err != nil {
					return nil, fmt.Errorf("%s: %w", tool, err)
				}
				var a struct {
					Values map[string]json.RawMessage `json:"values"`
					Next   string                     `json:"next"`
				}
				if err := json.Unmarshal(raw, &a); err != nil {
					return nil, fmt.Errorf("%s answered what is not a page of values: %w", tool, err)
				}
				for k, v := range a.Values {
					out[k] = v
				}
				if a.Next == "" || a.Next == after {
					return out, nil
				}
				after = a.Next
			}
		},
		History: func(key string) ([]Revision, error) {
			raw, err := ask(tool, map[string]any{"key": key, "history": true})
			if err != nil {
				return nil, fmt.Errorf("%s: %w", tool, err)
			}
			var a struct {
				History *[]struct {
					Revision uint64          `json:"revision"`
					At       time.Time       `json:"at"`
					Op       string          `json:"op"`
					Value    json.RawMessage `json:"value"`
				} `json:"history"`
			}
			if err := json.Unmarshal(raw, &a); err != nil {
				return nil, fmt.Errorf("%s answered what is not a key's history: %w", tool, err)
			}
			if a.History == nil {
				return nil, fmt.Errorf("%s answers no history of a key: its owner's read tool takes no history argument yet", tool)
			}
			out := make([]Revision, 0, len(*a.History))
			for _, h := range *a.History {
				out = append(out, Revision{Revision: h.Revision, At: h.At, Op: h.Op, Value: h.Value})
			}
			return out, nil
		},
	}
}

// maxOwnerPages bounds one read of a state's values page by page (a page is at most 768 KiB, mesh-sdk views PageBytes).
const maxOwnerPages = 256