runtime_nakama_module.go

// Copyright 2017 The Nakama Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package server

import (
	"context"
	"io"
	"net/http"
	"time"

	"strings"

	"database/sql"

	"fmt"

	"encoding/json"

	"encoding/base64"

	"encoding/hex"
	"io/ioutil"

	"nakama/pkg/jsonpatch"

	"github.com/fatih/structs"
	"github.com/satori/go.uuid"
	"github.com/yuin/gopher-lua"
	"go.uber.org/zap"
)

const CALLBACKS = "runtime_callbacks"

type Callbacks struct {
	HTTP   map[string]*lua.LFunction
	RPC    map[string]*lua.LFunction
	Before map[string]*lua.LFunction
	After  map[string]*lua.LFunction
}

type NakamaModule struct {
	logger              *zap.Logger
	db                  *sql.DB
	notificationService *NotificationService
	client              *http.Client
}

func NewNakamaModule(logger *zap.Logger, db *sql.DB, l *lua.LState, notificationService *NotificationService) *NakamaModule {
	l.SetContext(context.WithValue(context.Background(), CALLBACKS, &Callbacks{
		RPC:    make(map[string]*lua.LFunction),
		Before: make(map[string]*lua.LFunction),
		After:  make(map[string]*lua.LFunction),
		HTTP:   make(map[string]*lua.LFunction),
	}))
	return &NakamaModule{
		logger:              logger,
		db:                  db,
		notificationService: notificationService,
		client: &http.Client{
			Timeout: 5 * time.Second,
		},
	}
}

func (n *NakamaModule) Loader(l *lua.LState) int {
	mod := l.SetFuncs(l.NewTable(), map[string]lua.LGFunction{
		"uuid_v4":                 n.uuidV4,
		"uuid_bytes_to_string":    n.uuidBytesToString,
		"uuid_string_to_bytes":    n.uuidStringToBytes,
		"http_request":            n.httpRequest,
		"json_encode":             n.jsonEncode,
		"json_decode":             n.jsonDecode,
		"base64_encode":           n.base64Encode,
		"base64_decode":           n.base64Decode,
		"base16_encode":           n.base16Encode,
		"base16_decode":           n.base16decode,
		"logger_info":             n.loggerInfo,
		"logger_warn":             n.loggerWarn,
		"logger_error":            n.loggerError,
		"register_rpc":            n.registerRPC,
		"register_before":         n.registerBefore,
		"register_after":          n.registerAfter,
		"register_http":           n.registerHTTP,
		"users_fetch_id":          n.usersFetchId,
		"users_fetch_handle":      n.usersFetchHandle,
		"users_update":            n.usersUpdate,
		"users_ban":               n.usersBan,
		"storage_list":            n.storageList,
		"storage_fetch":           n.storageFetch,
		"storage_write":           n.storageWrite,
		"storage_update":          n.storageUpdate,
		"storage_remove":          n.storageRemove,
		"leaderboard_create":      n.leaderboardCreate,
		"leaderboard_submit_incr": n.leaderboardSubmitIncr,
		"leaderboard_submit_decr": n.leaderboardSubmitDecr,
		"leaderboard_submit_set":  n.leaderboardSubmitSet,
		"leaderboard_submit_best": n.leaderboardSubmitBest,
		"groups_create":           n.groupsCreate,
		"groups_update":           n.groupsUpdate,
		"group_users_list":        n.groupUsersList,
		"groups_user_list":        n.groupsUserList,
		"notifications_send_id":   n.notificationsSendId,
	})

	l.Push(mod)
	return 1
}

func (n *NakamaModule) uuidV4(l *lua.LState) int {
	// TODO ensure there were no arguments to the function
	l.Push(lua.LString(uuid.NewV4().String()))
	return 1
}

func (n *NakamaModule) uuidBytesToString(l *lua.LState) int {
	uuidBytes := l.CheckString(1)
	if uuidBytes == "" {
		l.ArgError(1, "Expects a UUID byte string")
		return 0
	}
	u, err := uuid.FromBytes([]byte(uuidBytes))
	if err != nil {
		l.ArgError(1, "Not a valid UUID byte string")
		return 0
	}
	l.Push(lua.LString(u.String()))
	return 1
}

func (n *NakamaModule) uuidStringToBytes(l *lua.LState) int {
	uuidString := l.CheckString(1)
	if uuidString == "" {
		l.ArgError(1, "Expects a UUID string")
		return 0
	}
	u, err := uuid.FromString(uuidString)
	if err != nil {
		l.ArgError(1, "Not a valid UUID string")
		return 0
	}
	l.Push(lua.LString(u.Bytes()))
	return 1
}

func (n *NakamaModule) httpRequest(l *lua.LState) int {
	url := l.CheckString(1)
	method := l.CheckString(2)
	headers := l.CheckTable(3)
	body := l.OptString(4, "")
	if url == "" {
		l.ArgError(1, "Expects URL string")
		return 0
	}
	if method == "" {
		l.ArgError(2, "Expects method string")
		return 0
	}

	// Prepare request body, if any.
	var requestBody io.Reader
	if body != "" {
		requestBody = strings.NewReader(body)
	}
	// Prepare the request.
	req, err := http.NewRequest(method, url, requestBody)
	if err != nil {
		l.RaiseError("HTTP request error: %v", err.Error())
		return 0
	}
	// Apply any request headers.
	httpHeaders := ConvertLuaTable(headers)
	for k, v := range httpHeaders {
		if vs, ok := v.(string); !ok {
			l.RaiseError("HTTP header values must be strings")
			return 0
		} else {
			req.Header.Add(k, vs)
		}
	}
	// Execute the request.
	resp, err := n.client.Do(req)
	if err != nil {
		l.RaiseError("HTTP request error: %v", err.Error())
		return 0
	}
	// Read the response body.
	responseBody, err := ioutil.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		l.RaiseError("HTTP response body error: %v", err.Error())
		return 0
	}
	// Read the response headers.
	responseHeaders := make(map[string]interface{}, len(resp.Header))
	for k, vs := range resp.Header {
		// TODO accept multiple values per header
		for _, v := range vs {
			responseHeaders[k] = v
			break
		}
	}

	l.Push(lua.LNumber(resp.StatusCode))
	l.Push(ConvertMap(l, responseHeaders))
	l.Push(lua.LString(string(responseBody)))
	return 3
}

func (n *NakamaModule) jsonEncode(l *lua.LState) int {
	jsonTable := l.Get(1)
	if jsonTable == nil {
		l.ArgError(1, "Expects a non-nil value to encode")
		return 0
	}

	jsonData := convertLuaValue(jsonTable)
	jsonBytes, err := json.Marshal(jsonData)
	if err != nil {
		l.RaiseError("Error encoding to JSON: %v", err.Error())
		return 0
	}

	l.Push(lua.LString(string(jsonBytes)))
	return 1
}

func (n *NakamaModule) jsonDecode(l *lua.LState) int {
	jsonString := l.CheckString(1)
	if jsonString == "" {
		l.ArgError(1, "Expects JSON string")
		return 0
	}

	var jsonData interface{}
	if err := json.Unmarshal([]byte(jsonString), &jsonData); err != nil {
		l.RaiseError("Not a valid JSON string: %v", err.Error())
		return 0
	}

	l.Push(convertValue(l, jsonData))
	return 1
}

func (n *NakamaModule) base64Encode(l *lua.LState) int {
	input := l.CheckString(1)
	if input == "" {
		l.ArgError(1, "Expects string")
		return 0
	}

	output := base64.StdEncoding.EncodeToString([]byte(input))
	l.Push(lua.LString(output))
	return 1
}
func (n *NakamaModule) base64Decode(l *lua.LState) int {
	input := l.CheckString(1)
	if input == "" {
		l.ArgError(1, "Expects string")
		return 0
	}

	output, err := base64.StdEncoding.DecodeString(input)
	if err != nil {
		l.RaiseError("Not a valid base64 string: %v", err.Error())
		return 0
	}

	l.Push(lua.LString(output))
	return 1
}
func (n *NakamaModule) base16Encode(l *lua.LState) int {
	input := l.CheckString(1)
	if input == "" {
		l.ArgError(1, "Expects string")
		return 0
	}

	output := hex.EncodeToString([]byte(input))
	l.Push(lua.LString(output))
	return 1
}
func (n *NakamaModule) base16decode(l *lua.LState) int {
	input := l.CheckString(1)
	if input == "" {
		l.ArgError(1, "Expects string")
		return 0
	}

	output, err := hex.DecodeString(input)
	if err != nil {
		l.RaiseError("Not a valid base16 string: %v", err.Error())
		return 0
	}

	l.Push(lua.LString(output))
	return 1
}

func (n *NakamaModule) loggerInfo(l *lua.LState) int {
	message := l.CheckString(1)
	if message == "" {
		l.ArgError(1, "expects message string")
		return 0
	}
	n.logger.Info(message)
	l.Push(lua.LString(message))
	return 1
}

func (n *NakamaModule) loggerWarn(l *lua.LState) int {
	message := l.CheckString(1)
	if message == "" {
		l.ArgError(1, "expects message string")
		return 0
	}
	n.logger.Warn(message)
	l.Push(lua.LString(message))
	return 1
}

func (n *NakamaModule) loggerError(l *lua.LState) int {
	message := l.CheckString(1)
	if message == "" {
		l.ArgError(1, "expects message string")
		return 0
	}
	n.logger.Error(message)
	l.Push(lua.LString(message))
	return 1
}

func (n *NakamaModule) registerRPC(l *lua.LState) int {
	fn := l.CheckFunction(1)
	id := l.CheckString(2)

	if id == "" {
		l.ArgError(2, "expects rpc id")
		return 0
	}

	id = strings.ToLower(id)

	rc := l.Context().Value(CALLBACKS).(*Callbacks)
	rc.RPC[id] = fn
	n.logger.Info("Registered RPC function invocation", zap.String("id", id))
	return 0
}

func (n *NakamaModule) registerBefore(l *lua.LState) int {
	fn := l.CheckFunction(1)
	messageName := l.CheckString(2)

	if messageName == "" {
		l.ArgError(2, "expects message name")
		return 0
	}

	messageName = strings.ToLower(messageName)

	foundMessage := false
	for _, v := range RUNTIME_MESSAGES {
		if v == messageName {
			foundMessage = true
			break
		}
	}

	if !foundMessage {
		l.ArgError(2, "Invalid message name for register hook.")
		return 0
	}

	rc := l.Context().Value(CALLBACKS).(*Callbacks)
	rc.Before[messageName] = fn
	n.logger.Info("Registered Before function invocation", zap.String("message", messageName))
	return 0
}

func (n *NakamaModule) registerAfter(l *lua.LState) int {
	fn := l.CheckFunction(1)
	messageName := l.CheckString(2)

	if messageName == "" {
		l.ArgError(2, "expects message name")
		return 0
	}

	messageName = strings.ToLower(messageName)

	foundMessage := false
	for _, v := range RUNTIME_MESSAGES {
		if v == messageName {
			foundMessage = true
			break
		}
	}

	if !foundMessage {
		l.ArgError(2, "Invalid message name for register hook.")
		return 0
	}

	rc := l.Context().Value(CALLBACKS).(*Callbacks)
	rc.After[messageName] = fn
	n.logger.Info("Registered After function invocation", zap.String("message", messageName))
	return 0
}

func (n *NakamaModule) registerHTTP(l *lua.LState) int {
	fn := l.CheckFunction(1)
	path := l.CheckString(2)

	if path == "" {
		l.ArgError(2, "expects http path")
		return 0
	}

	if strings.HasPrefix(path, "/") {
		l.ArgError(2, "http path should not start with leading slash")
		return 0
	}

	path = strings.ToLower(path)

	rc := l.Context().Value(CALLBACKS).(*Callbacks)
	rc.HTTP[path] = fn
	n.logger.Info("Registered HTTP function invocation", zap.String("path", path))
	return 0
}

func (n *NakamaModule) usersFetchId(l *lua.LState) int {
	lt := l.CheckTable(1)
	userIds, ok := convertLuaValue(lt).([]interface{})
	if !ok {
		l.ArgError(1, "invalid user id data")
		return 0
	}

	userIdBytes := make([][]byte, 0)
	for _, id := range userIds {
		if ids, ok := id.(string); !ok {
			l.ArgError(1, "each user id must be a string")
			return 0
		} else {
			if uid, err := uuid.FromString(ids); err != nil {
				l.ArgError(1, "invalid user id")
				return 0
			} else {
				userIdBytes = append(userIdBytes, uid.Bytes())
			}
		}
	}

	users, err := UsersFetchIds(n.logger, n.db, userIdBytes)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to retrieve users: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, u := range users {
		// Convert UUIDs to string representation.
		uid, _ := uuid.FromBytes(u.Id)
		u.Id = []byte(uid.String())
		um := structs.Map(u)

		metadataMap := make(map[string]interface{})
		err = json.Unmarshal(u.Metadata, &metadataMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert metadata to json: %s", err.Error()))
			return 0
		}

		ut := ConvertMap(l, um)
		ut.RawSetString("Metadata", ConvertMap(l, metadataMap))
		lv.RawSetInt(i+1, ut)
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) usersFetchHandle(l *lua.LState) int {
	lt := l.CheckTable(1)
	handles, ok := convertLuaValue(lt).([]interface{})
	if !ok {
		l.ArgError(1, "invalid user handle data")
		return 0
	}

	userHandles := make([]string, 0)
	for _, h := range handles {
		if hs, ok := h.(string); !ok {
			l.ArgError(1, "invalid user handle data, each handle must be a string")
			return 0
		} else {
			userHandles = append(userHandles, hs)
		}
	}

	users, err := UsersFetchHandle(n.logger, n.db, userHandles)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to retrieve users: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, u := range users {
		// Convert UUIDs to string representation.
		uid, _ := uuid.FromBytes(u.Id)
		u.Id = []byte(uid.String())
		um := structs.Map(u)

		metadataMap := make(map[string]interface{})
		err = json.Unmarshal(u.Metadata, &metadataMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert metadata to json: %s", err.Error()))
			return 0
		}

		ut := ConvertMap(l, um)
		ut.RawSetString("Metadata", ConvertMap(l, metadataMap))
		lv.RawSetInt(i+1, ut)
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) usersUpdate(l *lua.LState) int {
	updatesTable := l.CheckTable(1)
	if updatesTable == nil || updatesTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of user updates")
		return 0
	}

	conversionError := ""
	updates := make([]*SelfUpdateOp, 0)

	updatesTable.ForEach(func(i lua.LValue, u lua.LValue) {
		updateTable, ok := u.(*lua.LTable)
		if !ok {
			conversionError = "expects a valid set of user updates"
			return
		}

		update := &SelfUpdateOp{}
		updateTable.ForEach(func(k lua.LValue, v lua.LValue) {
			switch k.String() {
			case "UserId":
				if v.Type() != lua.LTString {
					conversionError = "expects valid user IDs in each update"
					return
				}
				if uid, err := uuid.FromString(v.String()); err != nil {
					conversionError = "expects valid user IDs in each update"
					return
				} else {
					update.UserId = uid.Bytes()
				}
			case "Handle":
				if v.Type() != lua.LTString {
					conversionError = "expects valid handles in each update"
					return
				}
				update.Handle = v.String()
			case "Fullname":
				if v.Type() != lua.LTString {
					conversionError = "expects valid fullnames in each update"
					return
				}
				update.Fullname = v.String()
			case "Timezone":
				if v.Type() != lua.LTString {
					conversionError = "expects valid timezones in each update"
					return
				}
				update.Timezone = v.String()
			case "Location":
				if v.Type() != lua.LTString {
					conversionError = "expects valid locations in each update"
					return
				}
				update.Location = v.String()
			case "Lang":
				if v.Type() != lua.LTString {
					conversionError = "expects valid langs in each update"
					return
				}
				update.Lang = v.String()
			case "Metadata":
				if v.Type() != lua.LTTable {
					conversionError = "expects Metadata to be a table"
					return
				}

				metadataMap := ConvertLuaTable(v.(*lua.LTable))
				metadataBytes, err := json.Marshal(metadataMap)
				if err != nil {
					conversionError = fmt.Sprintf("failed to convert metadata: %s", err.Error())
					return
				}

				update.Metadata = metadataBytes
			case "AvatarUrl":
				if v.Type() != lua.LTString {
					conversionError = "expects valid avatar urls in each update"
					return
				}
				update.AvatarUrl = v.String()
			default:
				conversionError = "unrecognised update key, expects a valid set of user updates"
				return
			}
		})

		// Check it's a valid update op.
		if len(update.UserId) == 0 {
			conversionError = "expects each update to contain a user ID"
			return
		}
		if update.Handle == "" && update.Fullname == "" && update.Timezone == "" && update.Location == "" && update.Lang == "" && len(update.Metadata) == 0 && update.AvatarUrl == "" {
			conversionError = "expects each update to contain at least one field to change"
			return
		}
		updates = append(updates, update)
	})

	if conversionError != "" {
		l.ArgError(1, conversionError)
		return 0
	}

	if _, err := SelfUpdate(n.logger, n.db, updates); err != nil {
		l.RaiseError(fmt.Sprintf("failed to update users: %s", err.Error()))
	}

	return 0
}

func (n *NakamaModule) usersBan(l *lua.LState) int {
	usersTable := l.CheckTable(1)
	if usersTable == nil || usersTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of users")
		return 0
	}
	usersRaw, ok := convertLuaValue(usersTable).([]interface{})
	if !ok {
		l.ArgError(1, "expects a valid set of users")
		return 0
	}

	ids := make([][]byte, 0)
	handles := make([]string, 0)
	for _, d := range usersRaw {
		if m, ok := d.(string); !ok {
			l.ArgError(1, "expects a valid set of user IDs or handles")
			return 0
		} else {
			uid, err := uuid.FromString(m)
			if err == nil {
				ids = append(ids, uid.Bytes())
			} else {
				handles = append(handles, m) // assume that they've passed in a handle
			}
		}
	}

	if err := UsersBan(n.logger, n.db, ids, handles); err != nil {
		l.RaiseError(fmt.Sprintf("failed to ban users: %s", err.Error()))
	}

	return 0
}

func (n *NakamaModule) storageList(l *lua.LState) int {
	var userID []byte
	if us := l.OptString(1, ""); us != "" {
		if uid, err := uuid.FromString(us); err != nil {
			l.ArgError(1, "expects a valid user ID or nil")
			return 0
		} else {
			userID = uid.Bytes()
		}
	}
	bucket := l.OptString(2, "")
	collection := l.OptString(3, "")
	limit := l.CheckInt64(4)
	var cursor []byte
	if cs := l.OptString(5, ""); cs != "" {
		cb, err := base64.StdEncoding.DecodeString(cs)
		if err != nil {
			l.ArgError(5, "cursor is invalid")
			return 0
		}
		cursor = cb
	}

	values, newCursor, _, err := StorageList(n.logger, n.db, uuid.Nil, userID, bucket, collection, limit, cursor)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to list storage: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, v := range values {
		// Convert UUIDs to string representation if needed.
		if len(v.UserId) != 0 {
			uid, _ := uuid.FromBytes(v.UserId)
			v.UserId = []byte(uid.String())
		}
		vm := structs.Map(v)

		valueMap := make(map[string]interface{})
		err = json.Unmarshal(v.Value, &valueMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert value to json: %s", err.Error()))
			return 0
		}

		lt := ConvertMap(l, vm)
		lt.RawSetString("Value", ConvertMap(l, valueMap))
		lv.RawSetInt(i+1, lt)
	}
	l.Push(lv)

	// Convert and push the new cursor, if any.
	if len(newCursor) != 0 {
		newCursorString := base64.StdEncoding.EncodeToString(newCursor)
		l.Push(lua.LString(newCursorString))
	} else {
		l.Push(lua.LNil)
	}

	return 2
}

func (n *NakamaModule) storageFetch(l *lua.LState) int {
	keysTable := l.CheckTable(1)
	if keysTable == nil || keysTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of keys")
		return 0
	}
	keysRaw, ok := convertLuaValue(keysTable).([]interface{})
	if !ok {
		l.ArgError(1, "expects a valid set of data")
		return 0
	}
	keyMap := make([]map[string]interface{}, 0)
	for _, d := range keysRaw {
		if m, ok := d.(map[string]interface{}); !ok {
			l.ArgError(1, "expects a valid set of data")
			return 0
		} else {
			keyMap = append(keyMap, m)
		}
	}

	keys := make([]*StorageKey, len(keyMap))
	idx := 0
	for _, k := range keyMap {
		var bucket string
		if b, ok := k["Bucket"]; !ok {
			l.ArgError(1, "expects a bucket in each key")
			return 0
		} else {
			if bs, ok := b.(string); !ok {
				l.ArgError(1, "bucket must be a string")
				return 0
			} else {
				bucket = bs
			}
		}
		var collection string
		if c, ok := k["Collection"]; !ok {
			l.ArgError(1, "expects a collection in each key")
			return 0
		} else {
			if cs, ok := c.(string); !ok {
				l.ArgError(1, "collection must be a string")
				return 0
			} else {
				collection = cs
			}
		}
		var record string
		if r, ok := k["Record"]; !ok {
			l.ArgError(1, "expects a record in each key")
			return 0
		} else {
			if rs, ok := r.(string); !ok {
				l.ArgError(1, "record must be a string")
				return 0
			} else {
				record = rs
			}
		}
		var userID []byte
		if u, ok := k["UserId"]; ok {
			if us, ok := u.(string); !ok {
				l.ArgError(1, "expects valid user IDs in each key, when provided")
				return 0
			} else {
				if uid, err := uuid.FromString(us); err != nil {
					l.ArgError(1, "expects valid user IDs in each key, when provided")
					return 0
				} else {
					userID = uid.Bytes()
				}
			}
		}

		keys[idx] = &StorageKey{
			Bucket:     bucket,
			Collection: collection,
			Record:     record,
			UserId:     userID,
		}
		idx++
	}

	values, _, err := StorageFetch(n.logger, n.db, uuid.Nil, keys)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to fetch storage: %s", err.Error()))
		return 0
	}

	lv := l.NewTable()
	for i, v := range values {
		// Convert UUIDs to string representation if needed.
		if len(v.UserId) != 0 {
			uid, _ := uuid.FromBytes(v.UserId)
			v.UserId = []byte(uid.String())
		}
		vm := structs.Map(v)

		valueMap := make(map[string]interface{})
		err = json.Unmarshal(v.Value, &valueMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert value to json: %s", err.Error()))
			return 0
		}

		lt := ConvertMap(l, vm)
		lt.RawSetString("Value", ConvertMap(l, valueMap))
		lv.RawSetInt(i+1, lt)
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) storageWrite(l *lua.LState) int {
	dataTable := l.CheckTable(1)
	if dataTable == nil || dataTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of data")
		return 0
	}
	dataRaw, ok := convertLuaValue(dataTable).([]interface{})
	if !ok {
		l.ArgError(1, "expects a valid set of data")
		return 0
	}
	dataMap := make([]map[string]interface{}, 0)
	for _, d := range dataRaw {
		if m, ok := d.(map[string]interface{}); !ok {
			l.ArgError(1, "expects a valid set of data")
			return 0
		} else {
			dataMap = append(dataMap, m)
		}
	}

	data := make([]*StorageData, len(dataMap))
	idx := 0
	for _, k := range dataMap {
		var bucket string
		if b, ok := k["Bucket"]; !ok {
			l.ArgError(1, "expects a bucket in each key")
			return 0
		} else {
			if bs, ok := b.(string); !ok {
				l.ArgError(1, "bucket must be a string")
				return 0
			} else {
				bucket = bs
			}
		}
		var collection string
		if c, ok := k["Collection"]; !ok {
			l.ArgError(1, "expects a collection in each key")
			return 0
		} else {
			if cs, ok := c.(string); !ok {
				l.ArgError(1, "collection must be a string")
				return 0
			} else {
				collection = cs
			}
		}
		var record string
		if r, ok := k["Record"]; !ok {
			l.ArgError(1, "expects a record in each key")
			return 0
		} else {
			if rs, ok := r.(string); !ok {
				l.ArgError(1, "record must be a string")
				return 0
			} else {
				record = rs
			}
		}
		var value []byte
		if v, ok := k["Value"]; !ok {
			l.ArgError(1, "expects a value in each key")
			return 0
		} else {
			if vs, ok := v.(map[string]interface{}); !ok {
				l.ArgError(1, "value must be a table")
				return 0
			} else {
				dataJson, err := json.Marshal(vs)
				if err != nil {
					l.RaiseError("could not convert value to JSON: %v", err.Error())
					return 0
				}
				value = dataJson
			}
		}
		var userID []byte
		if u, ok := k["UserId"]; ok {
			if us, ok := u.(string); !ok {
				l.ArgError(1, "expects valid user IDs in each value, when provided")
				return 0
			} else {
				uid, err := uuid.FromString(us)
				if err != nil {
					l.ArgError(1, "expects valid user IDs in each value, when provided")
					return 0
				}
				userID = uid.Bytes()
			}
		}
		var version []byte
		if v, ok := k["Version"]; ok {
			if vs, ok := v.(string); !ok {
				l.ArgError(1, "version must be a string")
				return 0
			} else {
				version = []byte(vs)
			}
		}
		readPermission := int64(1)
		if r, ok := k["PermissionRead"]; ok {
			if rf, ok := r.(float64); !ok {
				l.ArgError(1, "permission read must be a number")
				return 0
			} else {
				readPermission = int64(rf)
			}
		}
		writePermission := int64(1)
		if w, ok := k["PermissionWrite"]; ok {
			if wf, ok := w.(float64); !ok {
				l.ArgError(1, "permission read must be a number")
				return 0
			} else {
				writePermission = int64(wf)
			}
		}

		data[idx] = &StorageData{
			Bucket:          bucket,
			Collection:      collection,
			Record:          record,
			UserId:          userID,
			Value:           value,
			Version:         version,
			PermissionRead:  readPermission,
			PermissionWrite: writePermission,
		}
		idx++
	}

	keys, _, err := StorageWrite(n.logger, n.db, uuid.Nil, data)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to write storage: %s", err.Error()))
		return 0
	}

	lv := l.NewTable()
	for i, k := range keys {
		km := structs.Map(k)
		lv.RawSetInt(i+1, convertValue(l, km))
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) storageUpdate(l *lua.LState) int {
	updatesTable := l.CheckTable(1)
	if updatesTable == nil || updatesTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of user updates")
		return 0
	}

	conversionError := ""
	updates := make([]*StorageKeyUpdate, 0)

	updatesTable.ForEach(func(i lua.LValue, u lua.LValue) {
		updateTable, ok := u.(*lua.LTable)
		if !ok {
			conversionError = "expects a valid set of user updates"
			return
		}

		// Initialise fields where default values for their types are not the logical defaults needed.
		update := &StorageKeyUpdate{
			Key:             &StorageKey{},
			PermissionRead:  int64(1),
			PermissionWrite: int64(1),
		}

		updateTable.ForEach(func(k lua.LValue, v lua.LValue) {
			switch k.String() {
			case "Bucket":
				if v.Type() != lua.LTString {
					conversionError = "expects valid buckets in each update"
					return
				}
				update.Key.Bucket = v.String()
			case "Collection":
				if v.Type() != lua.LTString {
					conversionError = "expects valid collections in each update"
					return
				}
				update.Key.Collection = v.String()
			case "Record":
				if v.Type() != lua.LTString {
					conversionError = "expects valid records in each update"
					return
				}
				update.Key.Record = v.String()
			case "UserId":
				if v.Type() != lua.LTString {
					conversionError = "expects valid user IDs in each update"
					return
				}
				if uid, err := uuid.FromString(v.String()); err != nil {
					conversionError = "expects valid user IDs in each update"
					return
				} else {
					update.Key.UserId = uid.Bytes()
				}
			case "Version":
				if v.Type() != lua.LTString {
					conversionError = "expects valid versions in each update"
					return
				}
				update.Key.Version = []byte(v.String())
			case "PermissionRead":
				if v.Type() != lua.LTNumber {
					conversionError = "expects valid read permissions in each update"
					return
				}
				update.PermissionRead = int64(lua.LVAsNumber(v))
			case "PermissionWrite":
				if v.Type() != lua.LTNumber {
					conversionError = "expects valid write permissions in each update"
					return
				}
				update.PermissionWrite = int64(lua.LVAsNumber(v))
			case "Update":
				if v.Type() != lua.LTTable {
					conversionError = "expects valid patch op in each update"
					return
				}

				vi, ok := convertLuaValue(v).([]interface{})
				if !ok {
					conversionError = "expects valid patch op in each update"
					return
				}

				// Lowercase all key names in op declarations.
				// eg. the Lua patch op: {{Op = "incr", Path = "/foo", Value = 1}}
				// becomes the JSON:     [{"op": "incr", "path": "/foo", "value": 1}]
				for _, vim := range vi {
					vm, ok := vim.(map[string]interface{})
					if !ok {
						conversionError = "expects valid patch op in each update"
						return
					}
					for vmK, vmV := range vm {
						vmKlower := strings.ToLower(vmK)
						if vmKlower != vmK {
							delete(vm, vmK)
							vm[vmKlower] = vmV
						}
					}
				}

				ve, err := json.Marshal(vi)
				if err != nil {
					conversionError = "expects valid patch op in each update"
					return
				}
				patch, err := jsonpatch.DecodeExtendedPatch(ve)
				if err != nil {
					conversionError = "expects valid patch op in each update"
					return
				}
				update.Patch = patch
			default:
				conversionError = "unrecognised update key, expects a valid set of storage updates"
				return
			}
		})

		// If there was an inner error allow it to propagate.
		if conversionError != "" {
			return
		}

		// Check it's a valid update op.
		if update.Key.Bucket == "" || update.Key.Collection == "" || update.Key.Record == "" || update.Patch == nil {
			conversionError = "expects each update to contain at least bucket, collection, record, and update"
			return
		}
		updates = append(updates, update)
	})

	if conversionError != "" {
		l.ArgError(1, conversionError)
		return 0
	}

	keys, _, err := StorageUpdate(n.logger, n.db, uuid.Nil, updates)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to update storage: %s", err.Error()))
		return 0
	}

	lv := l.NewTable()
	for i, k := range keys {
		km := structs.Map(k)
		lv.RawSetInt(i+1, convertValue(l, km))
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) storageRemove(l *lua.LState) int {
	keysTable := l.CheckTable(1)
	if keysTable == nil || keysTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of keys")
		return 0
	}
	keysRaw, ok := convertLuaValue(keysTable).([]interface{})
	if !ok {
		l.ArgError(1, "expects a valid set of data")
		return 0
	}
	keyMap := make([]map[string]interface{}, 0)
	for _, d := range keysRaw {
		if m, ok := d.(map[string]interface{}); !ok {
			l.ArgError(1, "expects a valid set of data")
			return 0
		} else {
			keyMap = append(keyMap, m)
		}
	}

	keys := make([]*StorageKey, len(keyMap))
	idx := 0
	for _, k := range keyMap {
		var bucket string
		if b, ok := k["Bucket"]; !ok {
			l.ArgError(1, "expects a bucket in each key")
			return 0
		} else {
			if bs, ok := b.(string); !ok {
				l.ArgError(1, "bucket must be a string")
				return 0
			} else {
				bucket = bs
			}
		}
		var collection string
		if c, ok := k["Collection"]; !ok {
			l.ArgError(1, "expects a collection in each key")
			return 0
		} else {
			if cs, ok := c.(string); !ok {
				l.ArgError(1, "collection must be a string")
				return 0
			} else {
				collection = cs
			}
		}
		var record string
		if r, ok := k["Record"]; !ok {
			l.ArgError(1, "expects a record in each key")
			return 0
		} else {
			if rs, ok := r.(string); !ok {
				l.ArgError(1, "record must be a string")
				return 0
			} else {
				record = rs
			}
		}
		var userID []byte
		if u, ok := k["UserId"]; ok {
			if us, ok := u.(string); !ok {
				l.ArgError(1, "expects valid user IDs in each key, when provided")
				return 0
			} else {
				uid, err := uuid.FromString(us)
				if err != nil {
					l.ArgError(1, "expects valid user IDs in each key, when provided")
					return 0
				}
				userID = uid.Bytes()
			}
		}
		var version []byte
		if v, ok := k["Version"]; ok {
			if vs, ok := v.(string); !ok {
				l.ArgError(1, "version must be a string")
				return 0
			} else {
				version = []byte(vs)
			}
		}
		keys[idx] = &StorageKey{
			Bucket:     bucket,
			Collection: collection,
			Record:     record,
			UserId:     userID,
			Version:    version,
		}
		idx++
	}

	if _, err := StorageRemove(n.logger, n.db, uuid.Nil, keys); err != nil {
		l.RaiseError(fmt.Sprintf("failed to remove storage: %s", err.Error()))
	}
	return 0
}

func (n *NakamaModule) leaderboardCreate(l *lua.LState) int {
	id := l.CheckString(1)
	sort := l.CheckString(2)
	reset := l.OptString(3, "")
	metadata := l.OptTable(4, l.NewTable())
	authoritative := l.OptBool(5, false)

	if sort != "asc" && sort != "desc" {
		l.ArgError(2, "invalid sort - only acceptable values are 'asc' and 'desc'")
		return 0
	}

	metadataMap := ConvertLuaTable(metadata)
	metadataBytes, err := json.Marshal(metadataMap)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to convert metadata: %s", err.Error()))
		return 0
	}

	_, err = leaderboardCreate(n.logger, n.db, []byte(id), sort, reset, string(metadataBytes), authoritative)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to create leaderboard: %s", err.Error()))
		return 0
	}

	return 0
}

func (n *NakamaModule) leaderboardSubmitIncr(l *lua.LState) int {

	return n.leaderboardSubmit(l, "incr")
}
func (n *NakamaModule) leaderboardSubmitDecr(l *lua.LState) int {

	return n.leaderboardSubmit(l, "decr")
}
func (n *NakamaModule) leaderboardSubmitSet(l *lua.LState) int {
	return n.leaderboardSubmit(l, "set")
}
func (n *NakamaModule) leaderboardSubmitBest(l *lua.LState) int {
	return n.leaderboardSubmit(l, "best")
}

func (n *NakamaModule) leaderboardSubmit(l *lua.LState, op string) int {
	leaderboardID := l.CheckString(1)
	value := l.CheckInt64(2)
	oId := l.CheckString(3)
	handle := l.OptString(4, "")
	lang := l.OptString(5, "")
	location := l.OptString(6, "")
	timezone := l.OptString(7, "")
	metadata := l.OptTable(8, l.NewTable())

	ownerID, err := uuid.FromString(oId)
	if err != nil {
		l.ArgError(1, "invalid owner id")
		return 0
	}

	metadataMap := ConvertLuaTable(metadata)
	metadataBytes, err := json.Marshal(metadataMap)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to convert leaderboard record metadata: %s", err.Error()))
		return 0
	}

	record, err := leaderboardSubmit(n.logger, n.db, uuid.Nil, []byte(leaderboardID), ownerID, handle, lang, op, value, location, timezone, metadataBytes)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to submit leaderboard record: %s", err.Error()))
		return 0
	}

	oid, _ := uuid.FromBytes(record.OwnerId)
	record.OwnerId = []byte(oid.String())
	rm := structs.Map(record)

	outgoingMetadataMap := make(map[string]interface{})
	err = json.Unmarshal(record.Metadata, &outgoingMetadataMap)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to convert leaderboard record metadata to json: %s", err.Error()))
		return 0
	}

	lv := ConvertMap(l, rm)
	lv.RawSetString("Metadata", ConvertMap(l, outgoingMetadataMap))

	l.Push(lv)
	return 1
}

func (n *NakamaModule) groupsCreate(l *lua.LState) int {
	groupsTable := l.CheckTable(1)
	if groupsTable == nil || groupsTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of groups")
		return 0
	}

	conversionError := false
	groupParams := make([]*GroupCreateParam, 0)

	groupsTable.ForEach(func(i lua.LValue, g lua.LValue) {
		groupTable, ok := g.(*lua.LTable)
		if !ok {
			conversionError = true
			l.ArgError(1, "expects a valid group")
			return
		}

		p := &GroupCreateParam{}
		groupTable.ForEach(func(k lua.LValue, v lua.LValue) {
			switch k.String() {
			case "Name":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects Name to be string")
					return
				}
				p.Name = v.String()
			case "Description":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects Description to be string")
					return
				}
				p.Description = v.String()
			case "AvatarUrl":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects AvatarUrl to be string")
					return
				}
				p.AvatarURL = v.String()
			case "Lang":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects Lang to be string")
					return
				}
				p.Lang = v.String()
			case "Private":
				if v.Type() != lua.LTBool {
					conversionError = true
					l.ArgError(1, "expects Private to be boolean")
					return
				}
				p.Private = lua.LVAsBool(v)
			case "Metadata":
				if v.Type() != lua.LTTable {
					conversionError = true
					l.ArgError(1, "expects Metadata to be a table")
					return
				}

				metadataMap := ConvertLuaTable(v.(*lua.LTable))
				metadataBytes, err := json.Marshal(metadataMap)
				if err != nil {
					conversionError = true
					l.ArgError(1, fmt.Sprintf("failed to convert metadata: %s", err.Error()))
					return
				}

				p.Metadata = metadataBytes
			case "CreatorId":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects CreatorId to be string")
					return
				}

				u, err := uuid.FromString(v.String())
				if err != nil {
					conversionError = true
					l.ArgError(1, "invalid CreatorId")
					return
				}
				p.Creator = u
			}
		})

		// Check mandatory items.
		if p.Name == "" {
			conversionError = true
			l.ArgError(1, "missing group Name")
			return
		} else if len(p.Creator) == 0 {
			conversionError = true
			l.ArgError(1, "missing CreatorId")
			return
		}

		// Set defaults if the values are missing.
		if len(p.Metadata) == 0 {
			p.Metadata = []byte("{}")
		}

		groupParams = append(groupParams, p)
	})

	if conversionError {
		return 0
	}

	groups, err := GroupsCreate(n.logger, n.db, groupParams)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to create groups: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, g := range groups {
		// Convert UUIDs to string representation.
		gid, _ := uuid.FromBytes(g.Id)
		g.Id = []byte(gid.String())
		gm := structs.Map(g)

		metadataMap := make(map[string]interface{})
		err = json.Unmarshal(g.Metadata, &metadataMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert metadata to json: %s", err.Error()))
			return 0
		}

		gt := ConvertMap(l, gm)
		gt.RawSetString("Metadata", ConvertMap(l, metadataMap))
		lv.RawSetInt(i+1, gt)
	}

	l.Push(lv)
	return 1
}

func (n *NakamaModule) groupsUpdate(l *lua.LState) int {
	groupsTable := l.CheckTable(1)
	if groupsTable == nil || groupsTable.Len() == 0 {
		l.ArgError(1, "expects a valid set of groups")
		return 0
	}

	conversionError := ""
	groupUpdates := make([]*TGroupsUpdate_GroupUpdate, 0)

	groupsTable.ForEach(func(i lua.LValue, g lua.LValue) {
		groupTable, ok := g.(*lua.LTable)
		if !ok {
			conversionError = "expects a valid group"
			return
		}

		p := &TGroupsUpdate_GroupUpdate{}
		groupTable.ForEach(func(k lua.LValue, v lua.LValue) {
			switch k.String() {
			case "GroupId":
				if v.Type() != lua.LTString {
					conversionError = "expects GroupId to be a string"
					return
				}
				gid, err := uuid.FromString(v.String())
				if err != nil {
					conversionError = "expects GroupId to be a valid ID"
					return
				}
				p.GroupId = gid.Bytes()
			case "Name":
				if v.Type() != lua.LTString {
					conversionError = "expects Name to be a string"
					return
				}
				p.Name = v.String()
			case "Description":
				if v.Type() != lua.LTString {
					conversionError = "expects Description to be string"
					return
				}
				p.Description = v.String()
			case "AvatarUrl":
				if v.Type() != lua.LTString {
					conversionError = "expects AvatarUrl to be string"
					return
				}
				p.AvatarUrl = v.String()
			case "Lang":
				if v.Type() != lua.LTString {
					conversionError = "expects Lang to be string"
					return
				}
				p.Lang = v.String()
			case "Private":
				if v.Type() != lua.LTBool {
					conversionError = "expects Private to be boolean"
					return
				}
				p.Private = lua.LVAsBool(v)
			case "Metadata":
				if v.Type() != lua.LTTable {
					conversionError = "expects Metadata to be a table"
					return
				}

				metadataMap := ConvertLuaTable(v.(*lua.LTable))
				metadataBytes, err := json.Marshal(metadataMap)
				if err != nil {
					conversionError = "invalid Metadata"
					return
				}

				p.Metadata = metadataBytes
			default:
				conversionError = fmt.Sprintf("unknown group field: %v", k.String())
				return
			}
		})

		if conversionError != "" {
			return
		}

		// mandatory items
		if len(p.GroupId) == 0 {
			conversionError = "missing GroupId"
			return
		}

		groupUpdates = append(groupUpdates, p)
	})

	if conversionError != "" {
		l.ArgError(1, conversionError)
		return 0
	}

	_, err := GroupsUpdate(n.logger, n.db, uuid.Nil, groupUpdates)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to update groups: %s", err.Error()))
	}

	return 0
}

func (n *NakamaModule) groupUsersList(l *lua.LState) int {
	group := l.CheckString(1)
	if group == "" {
		l.ArgError(1, "expects a valid group ID")
		return 0
	}
	groupID, err := uuid.FromString(group)
	if err != nil {
		l.ArgError(1, "expects a valid group ID")
		return 0
	}

	users, _, err := GroupUsersList(n.logger, n.db, uuid.Nil, groupID)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to list group users: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, u := range users {
		// Convert UUIDs to string representation.
		uid, _ := uuid.FromBytes(u.User.Id)
		u.User.Id = []byte(uid.String())
		um := structs.Map(u)

		metadataMap := make(map[string]interface{})
		err = json.Unmarshal(u.User.Metadata, &metadataMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert metadata to json: %s", err.Error()))
			return 0
		}

		ut := ConvertMap(l, um)
		ut.RawGetString("User").(*lua.LTable).RawSetString("Metadata", ConvertMap(l, metadataMap))
		lv.RawSetInt(i+1, ut)
	}

	l.Push(lv)

	return 1
}

func (n *NakamaModule) groupsUserList(l *lua.LState) int {
	user := l.CheckString(1)
	if user == "" {
		l.ArgError(1, "expects a valid user ID")
		return 0
	}
	userID, err := uuid.FromString(user)
	if err != nil {
		l.ArgError(1, "expects a valid user ID")
		return 0
	}

	groups, _, err := GroupsSelfList(n.logger, n.db, uuid.Nil, userID)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to list user groups: %s", err.Error()))
		return 0
	}

	// Convert and push the values.
	lv := l.NewTable()
	for i, g := range groups {
		// Convert UUIDs to string representation.
		gid, _ := uuid.FromBytes(g.Group.Id)
		g.Group.Id = []byte(gid.String())
		gm := structs.Map(g)

		metadataMap := make(map[string]interface{})
		err = json.Unmarshal(g.Group.Metadata, &metadataMap)
		if err != nil {
			l.RaiseError(fmt.Sprintf("failed to convert metadata to json: %s", err.Error()))
			return 0
		}

		gt := ConvertMap(l, gm)
		gt.RawGetString("Group").(*lua.LTable).RawSetString("Metadata", ConvertMap(l, metadataMap))
		lv.RawSetInt(i+1, gt)
	}

	l.Push(lv)

	return 1
}

func (n *NakamaModule) notificationsSendId(l *lua.LState) int {
	notificationsTable := l.CheckTable(1)
	if notificationsTable == nil {
		l.ArgError(1, "expects a valid set of notifications")
		return 0
	}

	conversionError := false
	notifications := make([]*NNotification, 0)
	notificationsTable.ForEach(func(i lua.LValue, g lua.LValue) {
		notificationTable, ok := g.(*lua.LTable)
		if !ok {
			conversionError = true
			l.ArgError(1, "expects a valid set of notifications")
			return
		}

		notification := &NNotification{}
		notificationTable.ForEach(func(k lua.LValue, v lua.LValue) {
			switch k.String() {
			case "Persistent":
				if v.Type() != lua.LTBool {
					conversionError = true
					l.ArgError(1, "expects Persistent to be boolean")
					return
				}
				notification.Persistent = lua.LVAsBool(v)
			case "Subject":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects Subject to be string")
					return
				}
				notification.Subject = v.String()
			case "Content":
				if v.Type() != lua.LTTable {
					conversionError = true
					l.ArgError(1, "expects Content to be a table")
					return
				}

				contentMap := ConvertLuaTable(v.(*lua.LTable))
				contentBytes, err := json.Marshal(contentMap)
				if err != nil {
					conversionError = true
					l.ArgError(1, fmt.Sprintf("failed to convert content: %s", err.Error()))
					return
				}

				notification.Content = contentBytes
			case "Code":
				if v.Type() != lua.LTNumber {
					conversionError = true
					l.ArgError(1, "expects Code to be number")
					return
				}
				number := int64(lua.LVAsNumber(v))
				if number <= 100 {
					l.ArgError(1, "expects Code to number above 100")
					return
				}
				notification.Code = int64(number)
			case "UserId":
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects UserId to be string")
					return
				}
				u, err := uuid.FromString(v.String())
				if err != nil {
					l.ArgError(1, "expects UserId to be a valid UUID")
					return
				}
				notification.UserID = u.Bytes()
			case "SenderId":
				if v.Type() == lua.LTNil {
					return
				}
				if v.Type() != lua.LTString {
					conversionError = true
					l.ArgError(1, "expects SenderId to be string")
					return
				}
				u, err := uuid.FromString(v.String())
				if err != nil {
					l.ArgError(1, "expects SenderId to be a valid UUID")
					return
				}
				notification.SenderID = u.Bytes()
			}
		})

		if notification.Subject == "" {
			l.ArgError(1, "expects Subject to be non-empty")
			return
		} else if len(notification.Content) == 0 {
			l.ArgError(1, "expects Content to be a valid JSON")
			return
		} else if len(notification.UserID) == 0 {
			l.ArgError(1, "expects UserId to be a valid UUID")
			return
		}

		notifications = append(notifications, notification)
	})

	if conversionError {
		return 0
	}

	err := n.notificationService.NotificationSend(notifications)
	if err != nil {
		l.RaiseError(fmt.Sprintf("failed to send notifications: %s", err.Error()))
	}

	return 0
}