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 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 }