package rudder import ( "encoding/json" "errors" "reflect" "slices" "sort" "time" "hub.rudder.build/rudder/rudder-module-go/abi" ) type Handler struct { run func(ctx *Context, args json.RawMessage) (any, error) arguments reflect.Type result reflect.Type } type Option func(function *function) type function struct { handler Handler errors []string } type Context struct { Trigger string Player *abi.Player Config json.RawMessage Now time.Time Module string Environment string Ledger Ledger Inventory Inventory Storage Storage ProjectStorage ProjectStorage Segments Segments Events Events Modules Modules Data Data } type Error struct { Code string Message string } func (e *Error) Error() string { return e.Code + ": " + e.Message } func Fail(code string, message string) error { return &Error{Code: code, Message: message} } var functions = map[string]*function{} func Register(name string, handler Handler, options ...Option) { registered := &function{handler: handler, errors: []string{}} for _, option := range options { option(registered) } functions[name] = registered } func Errors(codes ...string) Option { return func(function *function) { function.errors = append(function.errors, codes...) } } func Typed[Arguments any, Result any](handler func(ctx *Context, args Arguments) (Result, error)) Handler { return Handler{ run: func(ctx *Context, raw json.RawMessage) (any, error) { var args Arguments if len(raw) > 0 { if err := json.Unmarshal(raw, &args); err != nil { return nil, Fail(abi.ErrorCodeInvalidParameters, err.Error()) } } return handler(ctx, args) }, arguments: reflect.TypeFor[Arguments](), result: reflect.TypeFor[Result](), } } func runHandle() { output, err := invoke(readInput()) if err != nil { var failure *Error if !errors.As(err, &failure) { failure = &Error{Code: abi.ErrorCodeInternal, Message: err.Error()} } output = abi.Output{Error: &abi.ErrorBody{Code: failure.Code, Message: failure.Message}} } body, err := json.Marshal(output) if err != nil { panic(err) } writeOutput(body) } func invoke(input []byte) (abi.Output, error) { var request abi.Input if err := json.Unmarshal(input, &request); err != nil { return abi.Output{}, err } registered, ok := functions[request.Function] if !ok { return abi.Output{}, Fail(abi.ErrorCodeFunctionNotFound, "function "+request.Function+" is not registered") } now, err := time.Parse(time.RFC3339Nano, request.Context.Now) if err != nil { return abi.Output{}, err } ctx := &Context{ Trigger: request.Context.Trigger, Player: request.Context.Player, Config: request.Context.Config, Now: now, Module: request.Context.Module, Environment: request.Context.Environment, } result, err := registered.handler.run(ctx, request.Arguments) if err != nil { return abi.Output{}, err } body, err := json.Marshal(result) if err != nil { return abi.Output{}, err } return abi.Output{Result: body}, nil } func runDescribe() { names := make([]string, 0, len(functions)) for name := range functions { names = append(names, name) } sort.Strings(names) described := make([]abi.DescribedFunction, 0, len(names)) for _, name := range names { registered := functions[name] errorCodes := slices.Clone(registered.errors) slices.Sort(errorCodes) described = append(described, abi.DescribedFunction{ Name: name, Input: mustSchema(name+" arguments", registered.handler.arguments), Output: mustSchema(name+" result", registered.handler.result), Errors: slices.Compact(errorCodes), }) } body, err := json.Marshal(abi.Describe{Functions: described, DataHash: dataHash}) if err != nil { panic(err) } writeOutput(body) } func call(parameters any, result any) error { request, err := json.Marshal(parameters) if err != nil { return err } var response abi.CallResponse if err := json.Unmarshal(hostCall(request), &response); err != nil { return err } if response.Error != nil { return &Error{Code: response.Error.Code, Message: response.Error.Message} } if result == nil { return nil } return json.Unmarshal(response.Result, result) }