rudder-module-go/data.go

package rudder

import (
	"encoding/json"
	"errors"
	"sort"
	"time"

	"hub.rudder.build/rudder/rudder-module-go/abi"
)

const (
	OwnerKindPlayer  = "player"
	OwnerKindProject = "project"

	catalogLimit = 1000
)

type Owner = abi.Owner

func Player(id string) Owner { return Owner{Kind: OwnerKindPlayer, ID: id} }

func (c *Context) Owner() Owner {
	if c.Player == nil {
		return Owner{}
	}
	return Owner{Kind: OwnerKindPlayer, ID: c.Player.ID}
}

type Op struct{ Key string }

func OpKey(op []Op) string {
	if len(op) == 0 {
		return ""
	}
	return op[0].Key
}

// Reward is what a designer types into a catalog row: some currencies, some items. It is granted
// as ONE mutation, all-or-nothing: a reward that fails halfway leaves nothing behind, and one
// idempotency key makes the whole grant idempotent.
type Reward struct {
	Currencies map[string]int64 `json:"currencies,omitempty"`
	Items      map[string]int64 `json:"items,omitempty"`
}

func (r Reward) Empty() bool {
	return len(r.Currencies) == 0 && len(r.Items) == 0
}

// Price is what a catalog field of type price carries. Without it a module hand-decodes a
// json.RawMessage every time it reads the number a designer typed into the price form.
type Price struct {
	Currency string `json:"currency"`
	Amount   int64  `json:"amount"`
}

// Version is a document's optimistic-concurrency token. Zero means the document does not exist,
// which is what Get reads back for a miss and what Put demands for "must not exist yet".
type Version int64

type Written struct {
	Version Version `json:"version"`
}

// Unchanged tells Patch not to write. Patch returns the document as it was, with no error.
var Unchanged = errors.New("rudder: document unchanged")

// IsCode reports whether err is a kernel business error carrying this code. Without it every
// module writes the same errors.As block to map one kernel code onto one of its own.
func IsCode(err error, code string) bool {
	var failure *Error
	return errors.As(err, &failure) && failure.Code == code
}

// JSON marshals a generated document type for a command. A type the generator emitted always
// marshals, so a failure here is a generator bug: it traps the call rather than letting the kernel
// store null.
func JSON(value any) json.RawMessage {
	body, err := json.Marshal(value)
	if err != nil {
		panic("rudder: generated document does not marshal: " + err.Error())
	}
	return body
}

// Stamp is the wire form of an optional timestamp. The ABI carries times as RFC3339 strings, and a
// nil time has to stay nil rather than become the zero year.
func Stamp(value *time.Time) *string {
	if value == nil {
		return nil
	}
	formatted := value.UTC().Format(time.RFC3339Nano)
	return &formatted
}

func One[Result any](command any) (Result, error) {
	var result Result
	if err := call(command, &result); err != nil {
		var zero Result
		return zero, err
	}
	return result, nil
}

type Batch struct {
	commands []json.RawMessage
	targets  []any
	ran      bool
}

func (b *Batch) add(command any, target any) {
	body, err := json.Marshal(command)
	if err != nil {
		panic(err)
	}
	b.commands = append(b.commands, body)
	b.targets = append(b.targets, target)
}

// Run sends every command in one data.exec call. The batch is all-or-nothing: the first error
// discards every command that already applied.
func (b *Batch) Run() error {
	if len(b.commands) == 0 {
		return nil
	}
	result, err := Data{}.Exec(abi.DataExecParameters{Commands: b.commands})
	if err != nil {
		return err
	}
	if len(result.Results) != len(b.targets) {
		return errors.New("rudder: data.exec returned a different number of results than commands")
	}
	for index, raw := range result.Results {
		if len(raw) == 0 {
			continue
		}
		if err := json.Unmarshal(raw, b.targets[index]); err != nil {
			return err
		}
	}
	b.ran = true
	return nil
}

type Future[Value any] struct {
	value Value
	batch *Batch
}

// NewFuture registers a command with the batch. Value panics until Run succeeds: reading a result
// that was never fetched is an author bug, and a trap rolls the call back instead of returning a
// zero score as if it were real.
func NewFuture[Value any](batch *Batch, command any) *Future[Value] {
	future := &Future[Value]{batch: batch}
	batch.add(command, &future.value)
	return future
}

func (f *Future[Value]) Value() Value {
	if !f.batch.ran {
		panic("rudder: batch result read before Run")
	}
	return f.value
}

// Command registers a raw host command with a batch, so an op with no generated accessor runs in
// the same all-or-nothing batch as the generated ones. It is NOT an escape hatch for ops the
// generator covers.
func Command(batch *Batch, command any) *Future[struct{}] {
	return NewFuture[struct{}](batch, command)
}

var dataHash string

func SetDataHash(hash string) { dataHash = hash }

// docEntry is the wire shape of one document, the same one the generated <Name>Entry declares.
// DocPatch decodes into it because it is the only reader that does not know the generated type.
type docEntry[Value any] struct {
	Key       string    `json:"key"`
	Value     Value     `json:"value"`
	Version   Version   `json:"version"`
	UpdatedAt time.Time `json:"updatedAt"`
}

// DocPatch is get, apply, put, one retry. It lives here rather than in generated code because it is
// the only author operation with a loop in it, and generated code with a loop needs tests.
//
// Patch retries once on version_conflict, with the same idempotency key: a put that failed does
// not use up its key.
func DocPatch[Value any, Entry any](primitive string, owner Owner, key string, apply func(*Value) error, op []Op) (Entry, error) {
	var entry Entry
	for attempt := 0; attempt < 2; attempt++ {
		current, err := One[docEntry[Value]](abi.DocGetParameters{Primitive: primitive, Owner: owner, Key: key})
		if err != nil {
			return entry, err
		}
		value := current.Value
		if err := apply(&value); err != nil {
			if errors.Is(err, Unchanged) {
				return asEntry[Entry](key, current.Value, current.Version, current.UpdatedAt), nil
			}
			return entry, err
		}
		written, err := One[Written](abi.DocPutParameters{
			Primitive: primitive, Owner: owner, Key: key,
			Data: JSON(value), Version: int64(current.Version), OpKey: OpKey(op),
		})
		switch {
		case err == nil:
			return asEntry[Entry](key, value, written.Version, time.Time{}), nil
		case !IsCode(err, abi.ErrorCodeVersionConflict):
			return entry, err
		}
	}
	return entry, Fail(abi.ErrorCodeVersionConflict, "document "+primitive+"/"+key+" changed twice while patching")
}

// asEntry builds the generated entry type from the four fields every entry has. It goes through
// JSON because DocPatch is compiled once for a type the SDK cannot name; the generated struct is
// the same shape docEntry is, so a failure here is a generator bug and traps the call.
func asEntry[Entry any, Value any](key string, value Value, version Version, updatedAt time.Time) Entry {
	var entry Entry
	body, err := json.Marshal(docEntry[Value]{Key: key, Value: value, Version: version, UpdatedAt: updatedAt})
	if err == nil {
		err = json.Unmarshal(body, &entry)
	}
	if err != nil {
		panic("rudder: generated document entry does not fit the wire shape: " + err.Error())
	}
	return entry
}

// catalogPage fits every catalog result: list and active answer {rows, truncated}, get answers
// {row, found}. One shape is what lets every batched catalog read return *CatalogFuture[Row].
type catalogPage struct {
	Rows  []json.RawMessage `json:"rows"`
	Row   json.RawMessage   `json:"row"`
	Found bool              `json:"found"`
}

func (p catalogPage) raw() []json.RawMessage {
	if p.Found {
		return []json.RawMessage{p.Row}
	}
	return p.Rows
}

func CatalogAll[Row any](name string) ([]Row, error) {
	return catalogRows[Row](abi.CatalogListParameters{Primitive: name, Limit: catalogLimit})
}

func CatalogWhere[Row any](name string, field string, equals any) ([]Row, error) {
	return catalogRows[Row](abi.CatalogListParameters{
		Primitive: name, Field: field, Eq: catalogEq(equals), Limit: catalogLimit,
	})
}

func CatalogBy[Row any](name string, field string, equals any) (Row, bool, error) {
	return catalogOne[Row](abi.CatalogGetParameters{Primitive: name, Field: field, Eq: catalogEq(equals)})
}

// CatalogActive is the rows whose schedule is open now and whose segment this owner matches, in
// the declared order, both evaluated in the kernel. A zero Owner evaluates the schedule only.
func CatalogActive[Row any](name string, owner Owner, field string, equals any) ([]Row, error) {
	return catalogRows[Row](catalogActive(name, owner, field, equals, catalogLimit))
}

func CatalogActiveOne[Row any](name string, owner Owner, field string, equals any) (Row, bool, error) {
	return catalogOne[Row](catalogActive(name, owner, field, equals, 1))
}

func catalogActive(name string, owner Owner, field string, equals any, limit int64) abi.CatalogActiveParameters {
	return abi.CatalogActiveParameters{
		Primitive: name, Owner: owner, Field: field, Eq: catalogEq(equals), Limit: limit,
	}
}

func catalogRows[Row any](command any) ([]Row, error) {
	page, err := One[catalogPage](command)
	if err != nil {
		return nil, err
	}
	return decodeRows[Row](page.raw())
}

func catalogOne[Row any](command any) (Row, bool, error) {
	rows, err := catalogRows[Row](command)
	var row Row
	if err != nil || len(rows) == 0 {
		return row, false, err
	}
	return rows[0], true, nil
}

func decodeRows[Row any](items []json.RawMessage) ([]Row, error) {
	rows := make([]Row, 0, len(items))
	for _, item := range items {
		var row Row
		if err := json.Unmarshal(item, &row); err != nil {
			return nil, err
		}
		rows = append(rows, row)
	}
	return rows, nil
}

// CatalogFuture is the batched form of a catalog read. The rows decode when the value is read
// rather than in Run, because only the generated package knows the row type; a row that does not
// fit it is a generated-code bug, so it traps the call the way Value does before Run.
type CatalogFuture[Row any] struct {
	page *Future[catalogPage]
}

func newCatalogFuture[Row any](batch *Batch, command any) *CatalogFuture[Row] {
	return &CatalogFuture[Row]{page: NewFuture[catalogPage](batch, command)}
}

func (f *CatalogFuture[Row]) Value() []Row {
	rows, err := decodeRows[Row](f.page.Value().raw())
	if err != nil {
		panic("rudder: catalog row does not fit the generated type: " + err.Error())
	}
	return rows
}

func (f *CatalogFuture[Row]) One() (Row, bool) {
	rows := f.Value()
	if len(rows) == 0 {
		var row Row
		return row, false
	}
	return rows[0], true
}

func CatalogAllFuture[Row any](batch *Batch, name string) *CatalogFuture[Row] {
	return newCatalogFuture[Row](batch, abi.CatalogListParameters{Primitive: name, Limit: catalogLimit})
}

func CatalogWhereFuture[Row any](batch *Batch, name string, field string, equals any) *CatalogFuture[Row] {
	return newCatalogFuture[Row](batch, abi.CatalogListParameters{
		Primitive: name, Field: field, Eq: catalogEq(equals), Limit: catalogLimit,
	})
}

func CatalogByFuture[Row any](batch *Batch, name string, field string, equals any) *CatalogFuture[Row] {
	return newCatalogFuture[Row](batch, abi.CatalogGetParameters{Primitive: name, Field: field, Eq: catalogEq(equals)})
}

func CatalogActiveFuture[Row any](batch *Batch, name string, owner Owner, field string, equals any) *CatalogFuture[Row] {
	return newCatalogFuture[Row](batch, catalogActive(name, owner, field, equals, catalogLimit))
}

func CatalogActiveOneFuture[Row any](batch *Batch, name string, owner Owner, field string, equals any) *CatalogFuture[Row] {
	return newCatalogFuture[Row](batch, catalogActive(name, owner, field, equals, 1))
}

func catalogEq(equals any) json.RawMessage {
	if equals == nil {
		return nil
	}
	body, err := json.Marshal(equals)
	if err != nil {
		panic(err)
	}
	return body
}

// Grant pays a reward to an owner. Concurrent grants to the same owner are safe.
func (c *Context) Grant(owner Owner, reward Reward, op ...Op) error {
	return call(abi.RewardGrantParameters{
		Owner:      owner,
		Currencies: amounts(reward.Currencies),
		Items:      stacks(reward.Items),
		OpKey:      OpKey(op),
	}, nil)
}

func amounts(values map[string]int64) []abi.Amount {
	out := make([]abi.Amount, 0, len(values))
	for _, currency := range sortedKeys(values) {
		out = append(out, abi.Amount{Currency: currency, Amount: values[currency]})
	}
	return out
}

func stacks(values map[string]int64) []abi.Stack {
	out := make([]abi.Stack, 0, len(values))
	for _, item := range sortedKeys(values) {
		out = append(out, abi.Stack{Item: item, Quantity: values[item]})
	}
	return out
}

func sortedKeys(values map[string]int64) []string {
	keys := make([]string, 0, len(values))
	for key := range values {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}