package showbridge

import (
        "context"
        _ "embed"
        "encoding/json"
        "fmt"
        "io"
        "net/http"
        "time"

        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/module"
        "github.com/jwetzell/showbridge-go/internal/route"
        "github.com/jwetzell/showbridge-go/internal/schema"
)

func (r *Router) startAPIServer(config config.ApiConfig) {
        if !config.Enabled {
                r.logger.Warn("API not enabled")
                return
        }
        r.logger.Debug("starting API server", "port", config.Port)
        mux := http.NewServeMux()
        mux.HandleFunc("/ws", r.handleWebsocket)
        mux.HandleFunc("/health", r.handleHealthHTTP)
        mux.HandleFunc("/api/v1/config", r.handleConfigHTTP)
        mux.HandleFunc("/schema/config.schema.json", handleConfigSchema)
        mux.HandleFunc("/schema/routes.schema.json", handleRoutesSchema)
        mux.HandleFunc("/schema/modules.schema.json", handleModulesSchema)
        mux.HandleFunc("/schema/processors.schema.json", handleProcessorsSchema)

        r.apiServerMu.Lock()
        defer r.apiServerMu.Unlock()
        r.apiServer = &http.Server{
                Addr:    fmt.Sprintf(":%d", config.Port),
                Handler: mux,
        }

        go func() {
                r.apiServer.ListenAndServe()
                r.apiServerShutdown()
        }()
}

func (r *Router) stopAPIServer() {
        if r.apiServer == nil {
                return
        }
        r.logger.Debug("stopping API server")
        r.apiServerMu.Lock()
        defer r.apiServerMu.Unlock()
        if r.apiServer != nil {
                apiShutdownCtx, apiShutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
                r.apiServerShutdown = apiShutdownCancel
                r.apiServer.Shutdown(apiShutdownCtx)
                <-apiShutdownCtx.Done()
                r.apiServer = nil
        }
}

func (r *Router) handleHealthHTTP(w http.ResponseWriter, req *http.Request) {
        switch req.Method {
        case http.MethodGet:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusOK)
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

func (r *Router) handleConfigHTTP(w http.ResponseWriter, req *http.Request) {

        switch req.Method {
        case http.MethodGet:
                configJSON, err := json.Marshal(r.runningConfig)
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }

                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Content-Type", "application/json")
                w.Write(configJSON)
        case http.MethodPut:
                if r.updatingConfig {
                        http.Error(w, "Config update in progress.", http.StatusConflict)
                        return
                }
                //TODO(jwetzell): again way too much marshaling
                cfgBytes, err := io.ReadAll(req.Body)
                if err != nil {
                        http.Error(w, "Bad request", http.StatusBadRequest)
                        return
                }
                cfgMap := make(map[string]any)
                err = json.Unmarshal(cfgBytes, &cfgMap)
                if err != nil {
                        http.Error(w, "Bad request", http.StatusBadRequest)
                        return
                }

                err = schema.ApplyDefaults(&cfgMap)

                if err != nil {
                        http.Error(w, err.Error(), http.StatusBadRequest)
                        return
                }

                err = schema.ValidateConfig(cfgMap)

                if err != nil {
                        http.Error(w, err.Error(), http.StatusBadRequest)
                        return
                }

                validCfgBytes, err := json.Marshal(cfgMap)
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }

                var newConfig config.Config
                err = json.Unmarshal(validCfgBytes, &newConfig)
                if err != nil {
                        http.Error(w, "Bad request", http.StatusBadRequest)
                        return
                }
                moduleErrors, routeErrors := r.UpdateConfig(newConfig)
                if len(moduleErrors) > 0 || len(routeErrors) > 0 {
                        errorResponse := struct {
                                ModuleErrors []module.ModuleError `json:"moduleErrors,omitempty"`
                                RouteErrors  []route.RouteError   `json:"routeErrors,omitempty"`
                        }{
                                ModuleErrors: moduleErrors,
                                RouteErrors:  routeErrors,
                        }
                        errorResponseJSON, err := json.Marshal(errorResponse)
                        if err != nil {
                                http.Error(w, "Internal server error", http.StatusInternalServerError)
                                return
                        }
                        w.Header().Set("Access-Control-Allow-Origin", "*")
                        w.Header().Set("Content-Type", "application/json")
                        w.WriteHeader(http.StatusBadRequest)
                        w.Write(errorResponseJSON)
                        return
                }
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusOK)
                r.ConfigChange <- newConfig
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

func handleConfigSchema(w http.ResponseWriter, req *http.Request) {
        switch req.Method {
        case http.MethodGet:
                schemaJSON, err := json.Marshal(schema.ConfigSchema)
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Content-Type", "application/json")
                w.Write(schemaJSON)
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

func handleRoutesSchema(w http.ResponseWriter, req *http.Request) {
        switch req.Method {
        case http.MethodGet:
                schemaJSON, err := json.Marshal(schema.RoutesConfigSchema)
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Content-Type", "application/json")
                w.Write(schemaJSON)
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

func handleModulesSchema(w http.ResponseWriter, req *http.Request) {
        switch req.Method {
        case http.MethodGet:
                schemaJSON, err := json.Marshal(schema.GetModulesSchema())
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Content-Type", "application/json")
                w.Write(schemaJSON)
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

func handleProcessorsSchema(w http.ResponseWriter, req *http.Request) {
        switch req.Method {
        case http.MethodGet:
                schemaJSON, err := json.Marshal(schema.GetProcessorsSchema())
                if err != nil {
                        http.Error(w, "Internal server error", http.StatusInternalServerError)
                        return
                }
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Content-Type", "application/json")
                w.Write(schemaJSON)
        case http.MethodOptions:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
                w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
                w.WriteHeader(http.StatusOK)
        default:
                w.Header().Set("Access-Control-Allow-Origin", "*")
                w.WriteHeader(http.StatusMethodNotAllowed)
        }
}

package showbridge

import (
        "encoding/json"

        "github.com/gorilla/websocket"
)

type Event struct {
        Type  string `json:"type"`
        Data  any    `json:"data,omitempty"`
        Error string `json:"error,omitempty"`
}

func (e Event) toJSON() ([]byte, error) {
        return json.Marshal(e)
}

func (r *Router) handleEvent(event Event, sender *websocket.Conn) {
        switch event.Type {
        case "ping":
                r.unicastEvent(Event{Type: "pong"}, sender)
        default:
                r.logger.Warn("unknown event type", "eventType", event.Type)
        }
}

func (r *Router) unicastEvent(event Event, conn *websocket.Conn) {
        eventJSON, err := event.toJSON()
        if err != nil {
                r.logger.Error("failed to marshal event to JSON", "error", err)
                return
        }
        err = conn.WriteMessage(websocket.TextMessage, eventJSON)
        if err != nil {
                r.logger.Error("failed to write message to websocket connection", "error", err)
        }
}

func (r *Router) broadcastEvent(event Event, excluded ...*websocket.Conn) {
        eventJSON, err := event.toJSON()
        if err != nil {
                r.logger.Error("failed to marshal event to JSON", "error", err)
                return
        }
        r.wsConnsMu.Lock()
        defer r.wsConnsMu.Unlock()
        for _, conn := range r.wsConns {
                exclude := false
                for _, excludedConn := range excluded {
                        if conn == excludedConn {
                                exclude = true
                                break
                        }
                }
                if exclude {
                        continue
                }
                err := conn.WriteMessage(websocket.TextMessage, eventJSON)
                if err != nil {
                        r.logger.Error("failed to write message to websocket connection", "error", err)
                }
        }
}

package common

import (
        "math"
        "reflect"
)

func GetAnyAs[T any](value any) (T, bool) {
        typed, ok := value.(T)
        return typed, ok
}

func GetAnyAsInt(value any) (int, bool) {

        intValue, ok := value.(int)
        if ok {
                return intValue, true
        }

        uintValue, ok := value.(uint)
        if ok {
                return int(uintValue), true
        }

        byteValue, ok := value.(byte)
        if ok {
                return int(byteValue), true
        }

        float32Value, ok := value.(float32)
        if ok {
                if float64(float32Value) != math.Floor(float64(float32Value)) {
                        return 0, false
                }
                return int(float32Value), true
        }

        float64Value, ok := value.(float64)
        if ok {
                if float64Value != math.Floor(float64Value) {
                        return 0, false
                }
                return int(float64Value), true
        }
        return 0, false
}

func GetAnyAsByte(value any) (byte, bool) {

        byteValue, ok := value.(byte)
        if ok {
                return byte(byteValue), true
        }

        intValue, ok := value.(int)
        if ok {
                return byte(intValue), true
        }

        uintValue, ok := value.(uint)
        if ok {
                return byte(uintValue), true
        }

        float32Value, ok := value.(float32)
        if ok {
                if float64(float32Value) != math.Floor(float64(float32Value)) {
                        return 0, false
                }
                return byte(float32Value), true
        }

        float64Value, ok := value.(float64)
        if ok {
                if float64Value != math.Floor(float64Value) {
                        return 0, false
                }
                return byte(float64Value), true
        }
        return 0, false
}

func GetAnyAsByteSlice(value any) ([]byte, bool) {
        v := reflect.ValueOf(value)
        if v.Kind() != reflect.Slice {
                return nil, false
        }

        result := make([]byte, v.Len())
        for i := 0; i < v.Len(); i++ {
                elem := v.Index(i).Interface()
                elemValue, ok := GetAnyAsByte(elem)
                if !ok {
                        return nil, false
                }
                result[i] = elemValue
        }
        return result, true
}

func GetAnyAsIntSlice(value any) ([]int, bool) {
        v := reflect.ValueOf(value)
        if v.Kind() != reflect.Slice {
                return nil, false
        }

        result := make([]int, v.Len())
        for i := 0; i < v.Len(); i++ {
                elem := v.Index(i).Interface()
                elemInt, ok := GetAnyAsInt(elem)
                if !ok {
                        return nil, false
                }
                result[i] = elemInt
        }
        return result, true
}

func GetAnyAsFloat32(value any) (float32, bool) {
        float32Value, ok := value.(float32)
        if ok {
                return float32Value, true
        }

        float64Value, ok := value.(float64)
        if ok {
                return float32(float64Value), true
        }

        intValue, ok := value.(int)
        if ok {
                return float32(intValue), true
        }

        uintValue, ok := value.(uint)
        if ok {
                return float32(uintValue), true
        }

        byteValue, ok := value.(byte)
        if ok {
                return float32(byteValue), true
        }

        return 0, false
}

func GetAnyAsFloat64(value any) (float64, bool) {
        float64Value, ok := value.(float64)
        if ok {
                return float64Value, true
        }

        float32Value, ok := value.(float32)
        if ok {
                return float64(float32Value), true
        }

        intValue, ok := value.(int)
        if ok {
                return float64(intValue), true
        }

        uintValue, ok := value.(uint)
        if ok {
                return float64(uintValue), true
        }

        byteValue, ok := value.(byte)
        if ok {
                return float64(byteValue), true
        }

        return 0, false
}

package common

import (
        "context"
)

type WrappedPayload struct {
        Payload any
        Modules map[string]Module
        Sender  any
        Source  string
        End     bool
}

func GetWrappedPayload(ctx context.Context, payload any) WrappedPayload {
        wrappedPayload := WrappedPayload{
                Payload: payload,
                End:     false,
        }
        modules := ctx.Value(ModulesContextKey)
        if modules != nil {
                moduleMap, ok := modules.(map[string]Module)
                if ok {
                        wrappedPayload.Modules = moduleMap
                } else {
                        wrappedPayload.Modules = make(map[string]Module)
                }
        }

        sender := ctx.Value(SenderContextKey)
        if sender != nil {
                wrappedPayload.Sender = sender
        }

        source := ctx.Value(SourceContextKey)
        if source != nil {
                wrappedPayload.Source = source.(string)
        }
        return wrappedPayload
}

package config

import (
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
)

type Params map[string]any

var (
        ErrParamNotFound       = errors.New("not found")
        ErrParamNotString      = errors.New("not a string")
        ErrParamNotNumber      = errors.New("not a number")
        ErrParamNotInteger     = errors.New("not an integer")
        ErrParamNotBool        = errors.New("not a boolean")
        ErrParamNotSlice       = errors.New("not a slice")
        ErrParamNotStringSlice = errors.New("not a string slice")
        ErrParamNotByteSlice   = errors.New("not a byte slice")
        ErrParamNotIntSlice    = errors.New("not an int slice")
)

func (p Params) GetString(key string) (string, error) {
        value, ok := p[key]
        if !ok {
                return "", ErrParamNotFound
        }

        stringValue, ok := value.(string)
        if !ok {
                return "", ErrParamNotString
        }
        return stringValue, nil
}

func (p Params) GetInt(key string) (int, error) {
        value, ok := p[key]
        if !ok {
                return 0, ErrParamNotFound
        }

        intValue, ok := common.GetAnyAsInt(value)
        if ok {
                return intValue, nil
        }

        return 0, ErrParamNotNumber
}

func (p Params) GetFloat32(key string) (float32, error) {
        value, ok := p[key]
        if !ok {
                return 0, ErrParamNotFound
        }

        floatValue, ok := common.GetAnyAsFloat32(value)
        if ok {
                return floatValue, nil
        }

        return 0, ErrParamNotNumber
}

func (p Params) GetFloat64(key string) (float64, error) {
        value, ok := p[key]
        if !ok {
                return 0, ErrParamNotFound
        }

        floatValue, ok := common.GetAnyAsFloat64(value)
        if ok {
                return floatValue, nil
        }

        return 0, ErrParamNotNumber
}

func (p Params) GetBool(key string) (bool, error) {
        value, ok := p[key]
        if !ok {
                return false, ErrParamNotFound
        }

        boolValue, ok := value.(bool)
        if !ok {
                return false, ErrParamNotBool
        }
        return boolValue, nil
}

func (p Params) GetStringSlice(key string) ([]string, error) {
        value, ok := p[key]
        if !ok {
                return nil, ErrParamNotFound
        }

        interfaceSlice, ok := value.([]any)
        if !ok {
                return nil, ErrParamNotSlice
        }

        stringSlice := make([]string, len(interfaceSlice))
        for i, v := range interfaceSlice {
                str, ok := v.(string)
                if !ok {
                        return nil, ErrParamNotStringSlice
                }
                stringSlice[i] = str
        }
        return stringSlice, nil
}

func (p Params) GetIntSlice(key string) ([]int, error) {
        value, ok := p[key]
        if !ok {
                return nil, ErrParamNotFound
        }

        intSlice, ok := common.GetAnyAsIntSlice(value)
        if !ok {
                return nil, ErrParamNotIntSlice
        }

        return intSlice, nil
}

func (p Params) GetByteSlice(key string) ([]byte, error) {
        value, ok := p[key]
        if !ok {
                return nil, ErrParamNotFound
        }

        byteSlice, ok := common.GetAnyAsByteSlice(value)

        if !ok {
                return nil, ErrParamNotByteSlice
        }

        return byteSlice, nil
}

package framer

type Framer interface {
        Decode([]byte) [][]byte
        Encode([]byte) []byte
        Clear()
        Buffer() []byte
}

func GetFramer(framingType string) Framer {
        switch framingType {
        case "CR":
                return NewByteSeparatorFramer([]byte{'\r'})
        case "LF":
                return NewByteSeparatorFramer([]byte{'\n'})
        case "CRLF":
                return NewByteSeparatorFramer([]byte{'\r', '\n'})
        case "SLIP":
                return NewSlipFramer()
        case "RAW":
                return NewRawFramer()
        default:
                return nil
        }
}

package framer

type RawFramer struct{}

func NewRawFramer() *RawFramer {
        return &RawFramer{}
}

func (rf *RawFramer) Decode(data []byte) [][]byte {
        return [][]byte{data}
}

func (rf *RawFramer) Encode(data []byte) []byte {
        return data
}

func (rf *RawFramer) Clear() {
        // NOTE(jwetzell): no internal state to clear
}

func (rf *RawFramer) Buffer() []byte {
        return []byte{}
}

package framer

import (
        "bytes"
)

type ByteSeparatorFramer struct {
        buffer    []byte
        separator []byte
}

func NewByteSeparatorFramer(separator []byte) *ByteSeparatorFramer {
        return &ByteSeparatorFramer{separator: separator, buffer: []byte{}}
}

func (bsf *ByteSeparatorFramer) Decode(data []byte) [][]byte {
        messages := [][]byte{}

        bsf.buffer = append(bsf.buffer, data...)

        parts := bytes.Split(bsf.buffer, bsf.separator)

        if len(parts) > 0 {
                bsf.buffer = parts[len(parts)-1]
                messages = parts[:len(parts)-1]
        }

        return messages
}

func (bsf *ByteSeparatorFramer) Encode(data []byte) []byte {
        return append(data, bsf.separator...)
}

func (bsf *ByteSeparatorFramer) Clear() {
        bsf.buffer = []byte{}
}

func (bsf *ByteSeparatorFramer) Buffer() []byte {
        return bsf.buffer
}

package framer

type SlipFramer struct {
        buffer []byte
}

func NewSlipFramer() *SlipFramer {
        return &SlipFramer{buffer: []byte{}}
}

func (sf *SlipFramer) Decode(data []byte) [][]byte {
        messages := [][]byte{}

        END := byte(0xc0)
        ESC := byte(0xdb)
        ESC_END := byte(0xdc)
        ESC_ESC := byte(0xdd)

        escapeNext := false
        for _, packetByte := range data {

                if packetByte == ESC {
                        escapeNext = true
                        continue
                }

                if escapeNext {
                        if packetByte == ESC_END {
                                sf.buffer = append(sf.buffer, END)
                        } else if packetByte == ESC_ESC {
                                sf.buffer = append(sf.buffer, ESC)
                        }
                        escapeNext = false
                } else if packetByte == END {
                        if len(sf.buffer) == 0 {
                                // opening END byte, can discard
                                continue
                        } else {
                                message := sf.buffer
                                messages = append(messages, message)
                        }
                        sf.buffer = []byte{}
                } else {
                        sf.buffer = append(sf.buffer, packetByte)
                }
        }

        return messages
}

func (sf *SlipFramer) Encode(data []byte) []byte {
        END := byte(0xc0)
        ESC := byte(0xdb)
        ESC_END := byte(0xdc)
        ESC_ESC := byte(0xdd)

        var encodedBytes = []byte{END}

        for _, byteToEncode := range data {
                switch byteToEncode {
                case END:
                        encodedBytes = append(encodedBytes, ESC, ESC_END)
                case ESC:
                        encodedBytes = append(encodedBytes, ESC, ESC_ESC)
                default:
                        encodedBytes = append(encodedBytes, byteToEncode)
                }
        }

        encodedBytes = append(encodedBytes, END)
        return encodedBytes
}

func (sf *SlipFramer) Clear() {
        sf.buffer = []byte{}
}

func (sf *SlipFramer) Buffer() []byte {
        return sf.buffer
}

package module

import (
        "context"
        "database/sql"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"

        _ "modernc.org/sqlite"
)

type DbSqlite struct {
        config config.ModuleConfig
        Dsn    string
        ctx    context.Context
        router common.RouteIO
        db     *sql.DB
        logger *slog.Logger
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "db.sqlite",
                Title: "SQLite Database",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "dsn": {
                                        Type:      "string",
                                        MinLength: jsonschema.Ptr(1),
                                },
                        },
                        Required:             []string{"dsn"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params

                        dsnString, err := params.GetString("dsn")
                        if err != nil {
                                return nil, fmt.Errorf("db.sqlite dsn error: %w", err)
                        }

                        return &DbSqlite{Dsn: dsnString, config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (t *DbSqlite) Id() string {
        return t.config.Id
}

func (t *DbSqlite) Type() string {
        return t.config.Type
}

func (t *DbSqlite) Start(ctx context.Context) error {
        t.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("db.sqlite unable to get router from context")
        }
        t.router = router
        t.ctx = ctx

        db, err := sql.Open("sqlite", t.Dsn)
        if err != nil {
                return fmt.Errorf("db.sqlite error opening database: %w", err)
        }
        t.db = db
        defer t.db.Close()
        <-ctx.Done()
        return nil
}

func (t *DbSqlite) Stop() {
        if t.db != nil {
                t.db.Close()
        }
}

func (t *DbSqlite) Database() *sql.DB {
        return t.db
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "net/http"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/processor"
)

type HTTPServer struct {
        config config.ModuleConfig
        Port   uint16
        ctx    context.Context
        router common.RouteIO
        logger *slog.Logger
        cancel context.CancelFunc
}

type ResponseIOError struct {
        Index        int     `json:"index"`
        OutputError  *string `json:"outputError"`
        ProcessError *string `json:"processError"`
        InputError   *string `json:"inputError"`
}

type IOResponseData struct {
        IOErrors []ResponseIOError `json:"ioErrors"`
        Message  string            `json:"message"`
        Status   string            `json:"status"`
}

type httpServerContextKey string

type HTTPServerResponseWriter struct {
        http.ResponseWriter
        done bool
}

func (hsrw *HTTPServerResponseWriter) WriteHeader(status int) {
        hsrw.done = true
        hsrw.ResponseWriter.WriteHeader(status)
}

func (hsrw *HTTPServerResponseWriter) Write(data []byte) (int, error) {
        hsrw.done = true
        return hsrw.ResponseWriter.Write(data)
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "http.server",
                Title: "HTTP Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                        },
                        Required:             []string{"port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("http.server port error: %w", err)
                        }
                        return &HTTPServer{Port: uint16(portNum), config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (hs *HTTPServer) Id() string {
        return hs.config.Id
}

func (hs *HTTPServer) Type() string {
        return hs.config.Type
}

func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
        responseWriter := HTTPServerResponseWriter{ResponseWriter: w}

        response := IOResponseData{
                Message: "routing successful",
                Status:  "ok",
        }
        if hs.router != nil {
                inputContext := context.WithValue(hs.ctx, httpServerContextKey("responseWriter"), &responseWriter)
                senderAddr, err := net.ResolveTCPAddr("tcp", r.RemoteAddr)
                if err == nil {
                        inputContext = context.WithValue(inputContext, common.SenderContextKey, senderAddr)
                }
                aRouteFound, routingErrors := hs.router.HandleInput(inputContext, hs.Id(), r)
                if !responseWriter.done {
                        if aRouteFound {
                                if routingErrors != nil {
                                        w.WriteHeader(http.StatusInternalServerError)
                                        response.Status = "error"
                                        response.Message = "routing failed"

                                        response.IOErrors = []ResponseIOError{}
                                        for _, responseIOError := range routingErrors {
                                                errorToAdd := ResponseIOError{
                                                        Index: responseIOError.Index,
                                                }

                                                if responseIOError.InputError != nil {
                                                        errorMsg := responseIOError.InputError.Error()
                                                        errorToAdd.InputError = &errorMsg
                                                }

                                                if responseIOError.ProcessError != nil {
                                                        errorMsg := responseIOError.ProcessError.Error()
                                                        errorToAdd.ProcessError = &errorMsg
                                                }

                                                if responseIOError.OutputError != nil {
                                                        errorMsg := responseIOError.OutputError.Error()
                                                        errorToAdd.OutputError = &errorMsg
                                                }

                                                response.IOErrors = append(response.IOErrors, errorToAdd)

                                        }
                                        json.NewEncoder(w).Encode(response)
                                        return
                                } else {
                                        w.WriteHeader(http.StatusOK)
                                        response.Message = "routing successful"
                                        json.NewEncoder(w).Encode(response)
                                        return
                                }
                        } else {
                                w.WriteHeader(http.StatusNotFound)
                                response.Status = "error"
                                response.Message = "no matching routes found"
                                json.NewEncoder(w).Encode(response)
                                return
                        }
                }
        } else {
                w.WriteHeader(http.StatusInternalServerError)
                response.Message = "no router registered"
                response.Status = "error"
                json.NewEncoder(w).Encode(response)
                return
        }
}

func (hs *HTTPServer) Start(ctx context.Context) error {
        hs.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("http.server unable to get router from context")
        }
        hs.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        hs.ctx = moduleContext
        hs.cancel = cancel

        httpServer := &http.Server{
                Addr:    fmt.Sprintf(":%d", hs.Port),
                Handler: hs,
        }

        go func() {
                <-hs.ctx.Done()
                httpServer.Close()
        }()

        err := httpServer.ListenAndServe()
        // TODO(jwetzell): handle server closed error differently
        if err != nil {
                if err.Error() != "http: Server closed" {
                        return err
                }
        }

        <-hs.ctx.Done()
        hs.logger.Debug("done")
        return nil
}

func (hs *HTTPServer) Output(ctx context.Context, payload any) error {
        responseWriter, ok := ctx.Value(httpServerContextKey("responseWriter")).(*HTTPServerResponseWriter)

        if !ok {
                return errors.New("http.server output must originate from an http.server input")
        }

        payloadResponse, ok := common.GetAnyAs[processor.HTTPResponse](payload)

        if !ok {
                return errors.New("http.server is only able to output HTTPResponse")
        }

        if responseWriter.done {
                return errors.New("http.server response writer has already been written to")
        }

        responseWriter.WriteHeader(payloadResponse.Status)
        responseWriter.Write(payloadResponse.Body)
        return nil
}

func (hs *HTTPServer) Stop() {
        hs.cancel()
}

//go:build cgo

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
        _ "gitlab.com/gomidi/midi/v2/drivers/rtmididrv"
)

type MIDIInput struct {
        config   config.ModuleConfig
        ctx      context.Context
        router   common.RouteIO
        Port     string
        SendFunc func(midi.Message) error
        logger   *slog.Logger
        cancel   context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "midi.input",
                Title: "MIDI Input",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "port": {
                                        Title: "Port",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        portString, err := params.GetString("port")
                        if err != nil {
                                return nil, fmt.Errorf("midi.input port error: %w", err)
                        }

                        return &MIDIInput{config: config, Port: portString, logger: CreateLogger(config)}, nil
                },
        })
}

func (mi *MIDIInput) Id() string {
        return mi.config.Id
}

func (mi *MIDIInput) Type() string {
        return mi.config.Type
}

func (mi *MIDIInput) Start(ctx context.Context) error {
        mi.logger.Debug("running")
        defer midi.CloseDriver()
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("midi.input unable to get router from context")
        }
        mi.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        mi.ctx = moduleContext
        mi.cancel = cancel

        in, err := midi.FindInPort(mi.Port)
        if err != nil {
                return fmt.Errorf("midi.input can't find input port: %s", mi.Port)
        }

        stop, err := midi.ListenTo(in, func(msg midi.Message, timestampms int32) {
                if mi.router != nil {
                        mi.router.HandleInput(mi.ctx, mi.Id(), msg)
                }
        }, midi.UseSysEx())

        if err != nil {
                return err
        }

        defer stop()

        <-mi.ctx.Done()
        mi.logger.Debug("done")
        return nil
}

func (mi *MIDIInput) Stop() {
        mi.cancel()
}

//go:build cgo

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
        _ "gitlab.com/gomidi/midi/v2/drivers/rtmididrv"
)

type MIDIOutput struct {
        config   config.ModuleConfig
        ctx      context.Context
        router   common.RouteIO
        Port     string
        SendFunc func(midi.Message) error
        logger   *slog.Logger
        cancel   context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "midi.output",
                Title: "MIDI Output",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "port": {
                                        Title: "Port",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params

                        portString, err := params.GetString("port")
                        if err != nil {
                                return nil, fmt.Errorf("midi.output port error: %w", err)
                        }

                        return &MIDIOutput{config: config, Port: portString, logger: CreateLogger(config)}, nil
                },
        })
}

func (mo *MIDIOutput) Id() string {
        return mo.config.Id
}

func (mo *MIDIOutput) Type() string {
        return mo.config.Type
}

func (mo *MIDIOutput) Start(ctx context.Context) error {
        mo.logger.Debug("running")
        defer midi.CloseDriver()
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("midi.output unable to get router from context")
        }
        mo.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        mo.ctx = moduleContext
        mo.cancel = cancel

        out, err := midi.FindOutPort(mo.Port)

        if err != nil {
                return fmt.Errorf("midi.output can't find output port: %s", mo.Port)
        }

        send, err := midi.SendTo(out)
        if err != nil {
                return err
        }

        mo.SendFunc = send

        <-mo.ctx.Done()
        mo.logger.Debug("done")
        return nil
}

func (mo *MIDIOutput) Output(ctx context.Context, payload any) error {
        if mo.SendFunc == nil {
                return errors.New("midi.output output is not setup")
        }

        payloadMessage, ok := common.GetAnyAs[midi.Message](payload)

        if !ok {
                return errors.New("midi.output can only output midi.Message")
        }

        return mo.SendFunc(payloadMessage)
}

func (mo *MIDIOutput) Stop() {
        mo.cancel()
}

package module

import (
        "fmt"
        "log/slog"
        "sync"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type ModuleError struct {
        Index  int                 `json:"index"`
        Config config.ModuleConfig `json:"config"`
        Error  string              `json:"error"`
}

type ModuleRegistration struct {
        Type         string             `json:"type"`
        Title        string             `json:"title,omitempty"`
        Description  string             `json:"description,omitempty"`
        ParamsSchema *jsonschema.Schema `json:"paramsSchema,omitempty"`
        New          func(config.ModuleConfig) (common.Module, error)
}

func RegisterModule(mod ModuleRegistration) {

        if mod.Type == "" {
                panic("module type is missing")
        }
        if mod.New == nil {
                panic("missing ModuleInfo.New")
        }

        moduleRegistryMu.Lock()
        defer moduleRegistryMu.Unlock()

        if _, ok := ModuleRegistry[string(mod.Type)]; ok {
                panic(fmt.Sprintf("module already registered: %s", mod.Type))
        }
        ModuleRegistry[string(mod.Type)] = mod
}

var (
        moduleRegistryMu sync.RWMutex
        ModuleRegistry   = make(map[string]ModuleRegistration)
)

func CreateLogger(config config.ModuleConfig) *slog.Logger {
        return slog.Default().With("component", "module", "id", config.Id, "type", config.Type)
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        mqtt "github.com/eclipse/paho.mqtt.golang"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type MQTTClient struct {
        config   config.ModuleConfig
        ctx      context.Context
        router   common.RouteIO
        Broker   string
        ClientID string
        Topic    string
        client   mqtt.Client
        logger   *slog.Logger
        cancel   context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "mqtt.client",
                Title: "MQTT Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "broker": {
                                        Title: "Broker URL",
                                        Type:  "string",
                                },
                                "topic": {
                                        Title: "Topic",
                                        Type:  "string",
                                },
                                "clientId": {
                                        Title: "Client ID",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"broker", "topic", "clientId"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        brokerString, err := params.GetString("broker")

                        if err != nil {
                                return nil, fmt.Errorf("mqtt.client broker error: %w", err)
                        }

                        topicString, err := params.GetString("topic")

                        if err != nil {
                                return nil, fmt.Errorf("mqtt.client topic error: %w", err)
                        }

                        clientIdString, err := params.GetString("clientId")

                        if err != nil {
                                return nil, fmt.Errorf("mqtt.client clientId error: %w", err)
                        }

                        return &MQTTClient{config: config, Broker: brokerString, Topic: topicString, ClientID: clientIdString, logger: CreateLogger(config)}, nil
                },
        })
}

func (mc *MQTTClient) Id() string {
        return mc.config.Id
}

func (mc *MQTTClient) Type() string {
        return mc.config.Type
}

func (mc *MQTTClient) Start(ctx context.Context) error {
        mc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("mqtt.client unable to get router from context")
        }
        mc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        mc.ctx = moduleContext
        mc.cancel = cancel

        opts := mqtt.NewClientOptions()
        opts.AddBroker(mc.Broker)
        opts.SetClientID(mc.ClientID)
        opts.SetAutoReconnect(true)
        opts.SetCleanSession(false)

        opts.OnConnect = func(c mqtt.Client) {
                token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) {
                        mc.router.HandleInput(mc.ctx, mc.Id(), m)
                })
                token.Wait()
        }

        mc.client = mqtt.NewClient(opts)
        defer mc.client.Disconnect(250)

        token := mc.client.Connect()

        token.Wait()
        err := token.Error()
        if err != nil {
                return err
        }

        <-mc.ctx.Done()
        mc.logger.Debug("done")
        return nil
}

func (mc *MQTTClient) Output(ctx context.Context, payload any) error {
        payloadMessage, ok := common.GetAnyAs[mqtt.Message](payload)

        if !ok {
                return errors.New("mqtt.client is only able to output a MQTTMessage")
        }

        if mc.client == nil {
                return errors.New("mqtt.client client is not setup")
        }

        if !mc.client.IsConnected() {
                return errors.New("mqtt.client is not connected")
        }

        token := mc.client.Publish(payloadMessage.Topic(), payloadMessage.Qos(), payloadMessage.Retained(), payloadMessage.Payload())

        token.Wait()

        return token.Error()
}

func (mc *MQTTClient) Stop() {
        mc.cancel()
}

package module

import (
        "context"
        "errors"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/processor"
        "github.com/nats-io/nats.go"
)

type NATSClient struct {
        config  config.ModuleConfig
        ctx     context.Context
        router  common.RouteIO
        URL     string
        Subject string
        client  *nats.Conn
        logger  *slog.Logger
        cancel  context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "nats.client",
                Title: "NATS Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "url": {
                                        Title: "NATS Server URL",
                                        Type:  "string",
                                },
                                "subject": {
                                        Title: "Subject",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"url", "subject"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        urlString, err := params.GetString("url")
                        if err != nil {
                                return nil, errors.New("nats.client url error: " + err.Error())
                        }

                        subjectString, err := params.GetString("subject")

                        if err != nil {
                                return nil, errors.New("nats.client subject error: " + err.Error())
                        }

                        return &NATSClient{config: config, URL: urlString, Subject: subjectString, logger: CreateLogger(config)}, nil
                },
        })
}

func (nc *NATSClient) Id() string {
        return nc.config.Id
}

func (nc *NATSClient) Type() string {
        return nc.config.Type
}

func (nc *NATSClient) Start(ctx context.Context) error {
        nc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("nats.client unable to get router from context")
        }

        nc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        nc.ctx = moduleContext
        nc.cancel = cancel

        client, err := nats.Connect(nc.URL, nats.RetryOnFailedConnect(true))

        if err != nil {
                return err
        }

        nc.client = client

        defer client.Drain()
        defer client.Close()

        sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) {
                if nc.router != nil {
                        nc.router.HandleInput(nc.ctx, nc.Id(), msg)
                }
        })

        if err != nil {
                return err
        }

        defer sub.Unsubscribe()

        <-nc.ctx.Done()
        nc.logger.Debug("done")
        return nil
}

func (nc *NATSClient) Output(ctx context.Context, payload any) error {

        payloadMessage, ok := common.GetAnyAs[processor.NATSMessage](payload)

        if !ok {
                return errors.New("nats.client is only able to output NATSMessage")
        }

        if nc.client == nil {
                return errors.New("nats.client client is not setup")
        }

        if !nc.client.IsConnected() {
                return errors.New("nats.client is not connected")
        }

        err := nc.client.Publish(payloadMessage.Subject, payloadMessage.Payload)

        return err
}

func (nc *NATSClient) Stop() {
        nc.cancel()
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/nats-io/nats-server/v2/server"
)

type NATSServer struct {
        config config.ModuleConfig
        ctx    context.Context
        Ip     string
        Port   int
        router common.RouteIO
        server *server.Server
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "nats.server",
                Title: "NATS Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title:   "IP",
                                        Type:    "string",
                                        Default: json.RawMessage(`"0.0.0.0"`),
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                        Default: json.RawMessage(`4222`),
                                },
                        },
                        Required:             []string{},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params
                        portNum, err := params.GetInt("port")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        portNum = 4222
                                } else {
                                        return nil, fmt.Errorf("nats.server port error: %w", err)
                                }
                        }

                        ipString, err := params.GetString("ip")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        ipString = "0.0.0.0"
                                } else {
                                        return nil, fmt.Errorf("nats.server ip error: %w", err)
                                }
                        }

                        _, err = net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", ipString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }
                        return &NATSServer{config: moduleConfig, logger: CreateLogger(moduleConfig), Ip: ipString, Port: portNum}, nil
                },
        })
}

func (ns *NATSServer) Id() string {
        return ns.config.Id
}

func (ns *NATSServer) Type() string {
        return ns.config.Type
}

func (ns *NATSServer) Start(ctx context.Context) error {
        ns.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("nats.server unable to get router from context")
        }

        ns.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        ns.ctx = moduleContext
        ns.cancel = cancel

        natsServer, err := server.NewServer(&server.Options{
                Host:  ns.Ip,
                Port:  ns.Port,
                NoLog: true,
        })

        if err != nil {
                return err
        }

        ns.server = natsServer
        natsServer.Start()
        defer natsServer.Shutdown()

        if !natsServer.ReadyForConnections(5 * time.Second) {
                return errors.New("nats.server failed to start")
        }
        ns.logger.Info("NATS server started", "client_url", natsServer.ClientURL())

        <-ns.ctx.Done()

        ns.logger.Debug("done")
        return nil
}

func (ns *NATSServer) Stop() {
        ns.cancel()
        if ns.server != nil {
                ns.server.Shutdown()
        }
}

package module

import (
        "context"
        "errors"
        "log/slog"
        "net"
        "time"

        "github.com/jwetzell/psn-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type PSNClient struct {
        config  config.ModuleConfig
        conn    *net.UDPConn
        ctx     context.Context
        router  common.RouteIO
        decoder *psn.Decoder
        logger  *slog.Logger
        cancel  context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "psn.client",
                Title: "PosiStageNet Client",
                New: func(config config.ModuleConfig) (common.Module, error) {
                        return &PSNClient{config: config, decoder: psn.NewDecoder(), logger: CreateLogger(config)}, nil
                },
        })
}

func (pc *PSNClient) Id() string {
        return pc.config.Id
}

func (pc *PSNClient) Type() string {
        return pc.config.Type
}

func (pc *PSNClient) Start(ctx context.Context) error {
        pc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("psn.client unable to get router from context")
        }
        pc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        pc.ctx = moduleContext
        pc.cancel = cancel

        addr, err := net.ResolveUDPAddr("udp", "236.10.10.10:56565")
        if err != nil {
                return err
        }

        client, err := net.ListenMulticastUDP("udp", nil, addr)
        if err != nil {
                return err
        }
        defer client.Close()

        pc.conn = client

        buffer := make([]byte, 2048)
        for {
                select {
                case <-pc.ctx.Done():
                        // TODO(jwetzell): cleanup?
                        pc.logger.Debug("done")
                        return nil
                default:
                        pc.conn.SetDeadline(time.Now().Add(time.Millisecond * 200))

                        numBytes, _, err := pc.conn.ReadFromUDP(buffer)
                        if err != nil {
                                //NOTE(jwetzell) we hit deadline
                                if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
                                        continue
                                }
                                return err
                        }

                        if numBytes > 0 {
                                message := buffer[:numBytes]
                                err := pc.decoder.Decode(message)
                                if err != nil {
                                        pc.logger.Error("problem decoding psn traffic", "error", err)
                                }

                                if pc.router != nil {
                                        // TODO(jwetzell): better input handling
                                        for _, tracker := range pc.decoder.Trackers {
                                                pc.router.HandleInput(pc.ctx, pc.Id(), tracker)
                                        }
                                } else {
                                        pc.logger.Error("has no router")
                                }
                        }
                }
        }
}

func (pc *PSNClient) Stop() {
        pc.cancel()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/redis/go-redis/v9"
)

type RedisClient struct {
        config config.ModuleConfig
        ctx    context.Context
        router common.RouteIO
        Host   string
        Port   uint16
        client *redis.Client
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "redis.client",
                Title: "Redis Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "host": {
                                        Type: "string",
                                },
                                "port": {
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                        },
                        Required:             []string{"host", "port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        hostString, err := params.GetString("host")
                        if err != nil {
                                return nil, errors.New("redis.client host error: " + err.Error())
                        }

                        portInt, err := params.GetInt("port")

                        if err != nil {
                                return nil, errors.New("redis.client port error: " + err.Error())
                        }

                        return &RedisClient{config: config, Host: hostString, Port: uint16(portInt), logger: CreateLogger(config)}, nil
                },
        })
}

func (rc *RedisClient) Id() string {
        return rc.config.Id
}

func (rc *RedisClient) Type() string {
        return rc.config.Type
}

func (rc *RedisClient) Printf(ctx context.Context, format string, v ...interface{}) {
        msg := fmt.Sprintf(format, v...)
        rc.logger.Debug(msg)
}

func (rc *RedisClient) Start(ctx context.Context) error {
        redis.SetLogger(rc)
        rc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("redis.client unable to get router from context")
        }

        rc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        rc.ctx = moduleContext
        rc.cancel = cancel

        client := redis.NewClient(&redis.Options{
                Addr:     fmt.Sprintf("%s:%d", rc.Host, rc.Port),
                Password: "",
                DB:       0,
        })

        rc.client = client

        defer client.Close()

        <-rc.ctx.Done()
        rc.logger.Debug("done")
        return nil
}

func (rc *RedisClient) Stop() {
        rc.cancel()
}

func (rc *RedisClient) Get(key string) (any, error) {
        if rc.client != nil {
                val, err := rc.client.Get(rc.ctx, key).Result()
                if err != nil {
                        return nil, err
                }
                return val, nil
        }
        return nil, errors.New("redis.client not setup")
}

func (rc *RedisClient) Set(key string, value any) error {
        if rc.client != nil {
                status := rc.client.Set(rc.ctx, key, value, 0)
                return status.Err()
        }
        return errors.New("redis.client not setup")
}

//go:build cgo

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/framer"
        "go.bug.st/serial"
)

type SerialClient struct {
        config config.ModuleConfig
        ctx    context.Context
        router common.RouteIO
        Port   string
        Framer framer.Framer
        Mode   *serial.Mode
        port   serial.Port
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "serial.client",
                Title: "Serial Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "port": {
                                        Title: "Port",
                                        Type:  "string",
                                },
                                "baudRate": {
                                        Title: "Baud Rate",
                                        Type:  "integer",
                                },
                        },
                        Required:             []string{"port", "baudRate"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        portString, err := params.GetString("port")
                        if err != nil {
                                return nil, fmt.Errorf("serial.client port error: %w", err)
                        }

                        framingMethodString, err := params.GetString("framing")
                        if err != nil {
                                return nil, fmt.Errorf("serial.client framing error: %w", err)
                        }

                        framer := framer.GetFramer(framingMethodString)

                        if framer == nil {
                                return nil, fmt.Errorf("serial.client unknown framing method: %s", framingMethodString)
                        }

                        baudRateInt, err := params.GetInt("baudRate")
                        if err != nil {
                                return nil, fmt.Errorf("serial.client baudRate error: %w", err)
                        }

                        mode := serial.Mode{
                                BaudRate: baudRateInt,
                        }

                        return &SerialClient{config: config, Port: portString, Framer: framer, Mode: &mode, logger: CreateLogger(config)}, nil
                },
        })
}

func (sc *SerialClient) Id() string {
        return sc.config.Id
}

func (sc *SerialClient) Type() string {
        return sc.config.Type
}

func (sc *SerialClient) SetupPort() error {

        port, err := serial.Open(sc.Port, sc.Mode)
        if err != nil {
                return fmt.Errorf("serial.client can't open input port: %s", sc.Port)
        }

        sc.port = port

        return nil
}

func (sc *SerialClient) Start(ctx context.Context) error {
        sc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("serial.client unable to get router from context")
        }

        sc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        sc.ctx = moduleContext
        sc.cancel = cancel

        // TODO(jwetzell): shutdown with router.Context properly
        go func() {
                <-sc.ctx.Done()
                sc.logger.Debug("done")
                if sc.port != nil {
                        sc.port.Close()
                }
        }()

        for {
                err := sc.SetupPort()
                if err != nil {
                        if sc.ctx.Err() != nil {
                                sc.logger.Debug("done")
                                return nil
                        }
                        sc.logger.Error("port setup error", "error", err.Error())
                        time.Sleep(time.Second * 2)
                        continue
                }

                buffer := make([]byte, 1024)
                select {
                case <-sc.ctx.Done():
                        sc.logger.Debug("done")
                        return nil
                default:
                READ:
                        for {
                                select {
                                case <-sc.ctx.Done():
                                        sc.logger.Debug("done")
                                        return nil
                                default:
                                        byteCount, err := sc.port.Read(buffer)

                                        if err != nil {
                                                sc.Framer.Clear()
                                                break READ
                                        }

                                        if sc.Framer != nil {
                                                if byteCount > 0 {
                                                        messages := sc.Framer.Decode(buffer[0:byteCount])
                                                        for _, message := range messages {
                                                                if sc.router != nil {
                                                                        sc.router.HandleInput(sc.ctx, sc.Id(), message)
                                                                } else {
                                                                        sc.logger.Error("input received but no router is configured")
                                                                }
                                                        }
                                                }
                                        }
                                }
                        }
                }
        }
}

func (sc *SerialClient) Output(ctx context.Context, payload any) error {

        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                return errors.New("serial.client can only output bytes")
        }

        _, err := sc.port.Write(sc.Framer.Encode(payloadBytes))
        return err
}

func (sc *SerialClient) Stop() {
        sc.cancel()
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "log/slog"
        "os"
        "sync"
        "time"

        "github.com/emiago/diago"
        "github.com/emiago/diago/media"
        "github.com/emiago/sipgo"
        "github.com/emiago/sipgo/sip"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/processor"
)

type SIPCallServer struct {
        config    config.ModuleConfig
        ctx       context.Context
        router    common.RouteIO
        IP        string
        Port      int
        Transport string
        UserAgent string
        dg        *diago.Diago
        logger    *slog.Logger
        cancel    context.CancelFunc
}

type SIPCallMessage struct {
        To string
}

type SIPCall struct {
        inDialog *diago.DialogServerSession
        lock     sync.Mutex
}

type sipCallContextKey string

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "sip.call.server",
                Title: "SIP Call Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title:   "IP",
                                        Type:    "string",
                                        Default: json.RawMessage(`"0.0.0.0"`),
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                        Default: json.RawMessage(`5060`),
                                },
                                "transport": {
                                        Title:   "Transport",
                                        Type:    "string",
                                        Enum:    []any{"udp", "tcp", "ws", "udp4", "tcp4"},
                                        Default: json.RawMessage(`"udp"`),
                                },
                                "userAgent": {
                                        Title:   "User Agent",
                                        Type:    "string",
                                        Default: json.RawMessage(`"showbridge"`),
                                },
                        },
                        Required:             []string{},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params
                        portNum, err := params.GetInt("port")
                        if err != nil {

                                if errors.Is(err, config.ErrParamNotFound) {
                                        portNum = 5060
                                } else {
                                        return nil, fmt.Errorf("sip.call.server port error: %w", err)
                                }
                        }

                        ipString, err := params.GetString("ip")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        ipString = "0.0.0.0"
                                } else {
                                        return nil, fmt.Errorf("sip.call.server ip error: %w", err)
                                }
                        }

                        transportString, err := params.GetString("transport")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        transportString = "udp"
                                } else {
                                        return nil, fmt.Errorf("sip.call.server transport error: %w", err)
                                }
                        }

                        userAgentString, err := params.GetString("userAgent")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        userAgentString = "showbridge"
                                } else {
                                        return nil, fmt.Errorf("sip.call.server userAgent error: %w", err)
                                }
                        }

                        return &SIPCallServer{config: moduleConfig, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, logger: CreateLogger(moduleConfig)}, nil
                },
        })
}

func (scs *SIPCallServer) Id() string {
        return scs.config.Id
}

func (scs *SIPCallServer) Type() string {
        return scs.config.Type
}

func (scs *SIPCallServer) Start(ctx context.Context) error {
        scs.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("sip.call.server unable to get router from context")
        }
        scs.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        scs.ctx = moduleContext
        scs.cancel = cancel

        diagoLogger := slog.New(slog.NewJSONHandler(io.Discard, nil))

        ua, _ := sipgo.NewUA(
                sipgo.WithUserAgent(scs.UserAgent),
                sipgo.WithUserAgentTransportLayerOptions(sip.WithTransportLayerLogger(diagoLogger)),
                sipgo.WithUserAgentTransactionLayerOptions(sip.WithTransactionLayerLogger(diagoLogger)),
        )
        defer ua.Close()

        sip.SetDefaultLogger(diagoLogger)
        media.SetDefaultLogger(diagoLogger)
        dg := diago.NewDiago(ua, diago.WithLogger(diagoLogger), diago.WithTransport(
                diago.Transport{
                        Transport: scs.Transport,
                        BindHost:  scs.IP,
                        BindPort:  scs.Port,
                },
        ))

        go func() {
                dg.Serve(scs.ctx, func(inDialog *diago.DialogServerSession) {
                        scs.HandleCall(inDialog)
                })
        }()

        scs.dg = dg

        <-scs.ctx.Done()
        scs.logger.Debug("done")
        return nil
}

func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) {
        inDialog.Trying()
        inDialog.Ringing()
        inDialog.Answer()

        dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{
                inDialog: inDialog,
        })
        scs.router.HandleInput(dialogContext, scs.Id(), SIPCallMessage{
                To: inDialog.ToUser(),
        })
}

func (scs *SIPCallServer) Output(ctx context.Context, payload any) error {

        call, ok := ctx.Value(sipCallContextKey("call")).(*SIPCall)

        if !ok {
                return errors.New("sip.call.server output must originate from sip.call.server input")
        }

        gotLock := call.lock.TryLock()

        if !gotLock {
                return errors.New("sip.call.server call is already locked")
        }

        if call.inDialog.LoadState() == sip.DialogStateEnded {
                return errors.New("sip.call.server inDialog already ended")
        }

        payloadDTMFResponse, ok := common.GetAnyAs[processor.SipDTMFResponse](payload)

        if ok {
                dtmfWriter := call.inDialog.AudioWriterDTMF()
                time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
                for i, dtmfRune := range payloadDTMFResponse.Digits {
                        err := dtmfWriter.WriteDTMF(dtmfRune)

                        if err != nil {
                                return fmt.Errorf("sip.dtmf.server error output dtmf digit at index %d", i)
                        }
                }
                time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PostWait))
                return nil
        }

        payloadAudioFileResponse, ok := common.GetAnyAs[processor.SipAudioFileResponse](payload)

        if ok {
                audioFile, err := os.Open(payloadAudioFileResponse.AudioFile)
                if err != nil {
                        return err
                }
                defer audioFile.Close()

                playback, err := call.inDialog.PlaybackCreate()

                if err != nil {
                        return err
                }

                time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PreWait))

                _, err = playback.Play(audioFile, "audio/wav")

                time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PostWait))

                if err != nil {
                        return err
                }
                return nil
        }
        return errors.New("sip.dtmf.server can only output SipDTMFResponse or SipAudioFileResponse")
}

func (scs *SIPCallServer) Stop() {
        scs.cancel()
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "log/slog"
        "os"
        "strings"
        "sync"
        "time"

        "github.com/emiago/diago"
        "github.com/emiago/diago/media"
        "github.com/emiago/sipgo"
        "github.com/emiago/sipgo/sip"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/processor"
)

type SIPDTMFServer struct {
        config    config.ModuleConfig
        ctx       context.Context
        router    common.RouteIO
        IP        string
        Port      int
        Transport string
        UserAgent string
        Separator string
        logger    *slog.Logger
        cancel    context.CancelFunc
}

type SIPDTMFMessage struct {
        To     string
        Digits string
}

type SIPDTMFCall struct {
        inDialog *diago.DialogServerSession
        lock     sync.Mutex
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "sip.dtmf.server",
                Title: "SIP DTMF Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title:   "IP",
                                        Type:    "string",
                                        Default: json.RawMessage(`"0.0.0.0"`),
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                        Default: json.RawMessage(`5060`),
                                },
                                "transport": {
                                        Title:   "Transport",
                                        Type:    "string",
                                        Enum:    []any{"udp", "tcp", "ws", "udp4", "tcp4"},
                                        Default: json.RawMessage(`"udp"`),
                                },
                                "userAgent": {
                                        Title:   "User Agent",
                                        Type:    "string",
                                        Default: json.RawMessage(`"showbridge"`),
                                },
                                "separator": {
                                        Title:     "DTMF Separator",
                                        Type:      "string",
                                        MinLength: jsonschema.Ptr(1),
                                        MaxLength: jsonschema.Ptr(1),
                                },
                        },
                        Required:             []string{"separator"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params

                        portNum, err := params.GetInt("port")
                        if err != nil {

                                if errors.Is(err, config.ErrParamNotFound) {
                                        portNum = 5060
                                } else {
                                        return nil, fmt.Errorf("sip.dtmf.server port error: %w", err)
                                }
                        }

                        ipString, err := params.GetString("ip")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        ipString = "0.0.0.0"
                                } else {
                                        return nil, fmt.Errorf("sip.dtmf.server ip error: %w", err)
                                }
                        }

                        transportString, err := params.GetString("transport")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        transportString = "udp"
                                } else {
                                        return nil, fmt.Errorf("sip.dtmf.server transport error: %w", err)
                                }
                        }

                        userAgentString, err := params.GetString("userAgent")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        userAgentString = "showbridge"
                                } else {
                                        return nil, fmt.Errorf("sip.dtmf.server userAgent error: %w", err)
                                }
                        }

                        separatorString, err := params.GetString("separator")
                        if err != nil {
                                return nil, fmt.Errorf("sip.dtmf.server separator error: %w", err)
                        }

                        if len(separatorString) != 1 {
                                return nil, errors.New("sip.dtmf.server separator must be a single character")
                        }

                        if !strings.ContainsRune("0123456789*#ABCD", rune(separatorString[0])) {
                                return nil, errors.New("sip.dtmf.server separator must be a valid DTMF character")
                        }
                        return &SIPDTMFServer{config: moduleConfig, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, Separator: separatorString, logger: CreateLogger(moduleConfig)}, nil
                },
        })
}

func (sds *SIPDTMFServer) Id() string {
        return sds.config.Id
}

func (sds *SIPDTMFServer) Type() string {
        return sds.config.Type
}

func (sds *SIPDTMFServer) Start(ctx context.Context) error {
        sds.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("sip.dtmf.server unable to get router from context")
        }
        sds.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        sds.ctx = moduleContext
        sds.cancel = cancel

        diagoLogger := slog.New(slog.NewJSONHandler(io.Discard, nil))

        ua, _ := sipgo.NewUA(
                sipgo.WithUserAgent(sds.UserAgent),
                sipgo.WithUserAgentTransportLayerOptions(sip.WithTransportLayerLogger(diagoLogger)),
                sipgo.WithUserAgentTransactionLayerOptions(sip.WithTransactionLayerLogger(diagoLogger)),
        )
        defer ua.Close()

        sip.SetDefaultLogger(diagoLogger)
        media.SetDefaultLogger(diagoLogger)
        dg := diago.NewDiago(ua, diago.WithLogger(diagoLogger), diago.WithTransport(
                diago.Transport{
                        Transport: sds.Transport,
                        BindHost:  sds.IP,
                        BindPort:  sds.Port,
                },
        ))

        err := dg.Serve(sds.ctx, func(inDialog *diago.DialogServerSession) {
                sds.HandleCall(inDialog)
        })

        if err != nil {
                return err
        }

        <-sds.ctx.Done()
        sds.logger.Debug("done")
        return nil
}

func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error {
        inDialog.Trying()
        inDialog.Ringing()
        inDialog.Answer()

        reader := inDialog.AudioReaderDTMF()
        userString := ""

        return reader.Listen(func(dtmf rune) error {
                if dtmf == rune(sds.Separator[0]) {
                        if sds.router != nil {
                                dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{
                                        inDialog: inDialog,
                                })
                                sds.router.HandleInput(dialogContext, sds.Id(), SIPDTMFMessage{
                                        To:     inDialog.ToUser(),
                                        Digits: userString,
                                })
                        }
                        userString = ""
                } else {
                        userString += string(dtmf)
                }
                return nil
        }, 5*time.Second)
}

func (sds *SIPDTMFServer) Output(ctx context.Context, payload any) error {
        call, ok := ctx.Value(sipCallContextKey("call")).(*SIPDTMFCall)

        if !ok {
                return errors.New("sip.dtmf.server output must originate from sip.dtmf.server input")
        }

        gotLock := call.lock.TryLock()

        if !gotLock {
                return errors.New("sip.dtmf.server call is already locked")
        }

        if call.inDialog.LoadState() == sip.DialogStateEnded {
                return errors.New("sip.dtmf.server inDialog already ended")
        }

        payloadDTMFResponse, ok := common.GetAnyAs[processor.SipDTMFResponse](payload)

        if ok {
                dtmfWriter := call.inDialog.AudioWriterDTMF()

                time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
                for i, dtmfRune := range payloadDTMFResponse.Digits {
                        err := dtmfWriter.WriteDTMF(dtmfRune)

                        if err != nil {
                                return fmt.Errorf("sip.dtmf.server error output dtmf digit at index %d", i)
                        }
                }
                time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PostWait))
                return nil
        }

        payloadAudioFileResponse, ok := common.GetAnyAs[processor.SipAudioFileResponse](payload)

        if ok {
                audioFile, err := os.Open(payloadAudioFileResponse.AudioFile)
                if err != nil {
                        return err
                }
                defer audioFile.Close()

                playback, err := call.inDialog.PlaybackCreate()

                if err != nil {
                        return err
                }

                time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PreWait))

                _, err = playback.Play(audioFile, "audio/wav")

                time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PostWait))

                if err != nil {
                        return err
                }
                return nil
        }

        return errors.New("sip.dtmf.server can only output SipDTMFResponse or SipAudioFileResponse")
}

func (sds *SIPDTMFServer) Stop() {
        sds.cancel()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/framer"
)

type TCPClient struct {
        config config.ModuleConfig
        framer framer.Framer
        conn   *net.TCPConn
        ctx    context.Context
        router common.RouteIO
        Addr   *net.TCPAddr
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "net.tcp.client",
                Title: "TCP Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "host": {
                                        Title: "Host",
                                        Type:  "string",
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                                "framing": {
                                        Title: "Framing Method",
                                        Type:  "string",
                                        Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
                                },
                        },
                        Required:             []string{"host", "port", "framing"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        hostString, err := params.GetString("host")
                        if err != nil {
                                return nil, fmt.Errorf("net.tcp.client host error: %w", err)
                        }

                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("net.tcp.client port error: %w", err)
                        }

                        addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", hostString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }

                        framingMethodString, err := params.GetString("framing")
                        if err != nil {
                                return nil, fmt.Errorf("net.tcp.client framing error: %w", err)
                        }

                        framer := framer.GetFramer(framingMethodString)

                        if framer == nil {
                                return nil, fmt.Errorf("net.tcp.client unknown framing method: %s", framingMethodString)
                        }
                        return &TCPClient{framer: framer, Addr: addr, config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (tc *TCPClient) Id() string {
        return tc.config.Id
}

func (tc *TCPClient) Type() string {
        return tc.config.Type
}

func (tc *TCPClient) Start(ctx context.Context) error {
        tc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.tcp.client unable to get router from context")
        }
        tc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        tc.ctx = moduleContext
        tc.cancel = cancel

        // TODO(jwetzell): shutdown with router.Context properly
        go func() {
                <-tc.ctx.Done()
                tc.logger.Debug("done")
                if tc.conn != nil {
                        tc.conn.Close()
                }
        }()

        for {
                err := tc.SetupConn()
                if err != nil {
                        if tc.ctx.Err() != nil {
                                tc.logger.Debug("done")
                                return nil
                        }
                        tc.logger.Error("connection error", "error", err.Error())
                        time.Sleep(time.Second * 2)
                        continue
                }

                buffer := make([]byte, 1024)
                select {
                case <-tc.ctx.Done():
                        tc.logger.Debug("done")
                        return nil
                default:
                READ:
                        for {
                                select {
                                case <-tc.ctx.Done():
                                        tc.logger.Debug("done")
                                        return nil
                                default:
                                        byteCount, err := tc.conn.Read(buffer)

                                        if err != nil {
                                                tc.framer.Clear()
                                                break READ
                                        }

                                        if tc.framer != nil {
                                                if byteCount > 0 {
                                                        messages := tc.framer.Decode(buffer[0:byteCount])
                                                        for _, message := range messages {
                                                                if tc.router != nil {
                                                                        tc.router.HandleInput(tc.ctx, tc.Id(), message)
                                                                } else {
                                                                        tc.logger.Error("input received but no router is configured")
                                                                }
                                                        }
                                                }
                                        }
                                }
                        }
                }
        }
}

func (tc *TCPClient) SetupConn() error {
        client, err := net.DialTCP("tcp", nil, tc.Addr)
        tc.conn = client
        return err
}

func (tc *TCPClient) Output(ctx context.Context, payload any) error {
        // NOTE(jwetzell): not sure how this would occur but
        if tc.conn == nil {
                err := tc.SetupConn()
                if err != nil {
                        return err
                }
        }
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)
        if !ok {
                return errors.New("net.tcp.client is only able to output bytes")
        }
        _, err := tc.conn.Write(tc.framer.Encode(payloadBytes))
        return err
}

func (tc *TCPClient) Stop() {
        tc.cancel()
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "slices"
        "sync"
        "syscall"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/framer"
)

type TCPServer struct {
        config        config.ModuleConfig
        Addr          *net.TCPAddr
        Framer        framer.Framer
        ctx           context.Context
        router        common.RouteIO
        quit          chan interface{}
        wg            sync.WaitGroup
        connections   []*net.TCPConn
        connectionsMu sync.RWMutex
        logger        *slog.Logger
        cancel        context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "net.tcp.server",
                Title: "TCP Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title:   "IP",
                                        Type:    "string",
                                        Default: json.RawMessage(`"0.0.0.0"`),
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                                "framing": {
                                        Title: "Framing Method",
                                        Type:  "string",
                                        Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
                                },
                        },
                        Required:             []string{"port", "framing"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params
                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("net.tcp.server port error: %w", err)
                        }

                        framingMethodString, err := params.GetString("framing")
                        if err != nil {
                                return nil, fmt.Errorf("net.tcp.server framing error: %w", err)
                        }

                        framer := framer.GetFramer(framingMethodString)

                        if framer == nil {
                                return nil, fmt.Errorf("net.tcp.server unknown framing method: %s", framingMethodString)
                        }

                        ipString, err := params.GetString("ip")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        ipString = "0.0.0.0"
                                } else {
                                        return nil, fmt.Errorf("net.tcp.server ip error: %w", err)
                                }
                        }

                        addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", ipString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }
                        return &TCPServer{Framer: framer, Addr: addr, config: moduleConfig, quit: make(chan interface{}), logger: CreateLogger(moduleConfig)}, nil
                },
        })
}

func (ts *TCPServer) Id() string {
        return ts.config.Id
}

func (ts *TCPServer) Type() string {
        return ts.config.Type
}

func (ts *TCPServer) handleClient(client *net.TCPConn) {
        ts.connectionsMu.Lock()
        ts.connections = append(ts.connections, client)
        ts.connectionsMu.Unlock()
        ts.logger.Debug("net.tcp.server connection accepted", "remoteAddr", client.RemoteAddr().String())
        defer client.Close()

        buffer := make([]byte, 1024)
ClientRead:
        for {
                select {
                case <-ts.quit:
                        client.Close()
                        ts.connectionsMu.Lock()
                        for i := 0; i < len(ts.connections); i++ {
                                if ts.connections[i] == client {
                                        ts.connections = slices.Delete(ts.connections, i, i+1)
                                        break
                                }
                        }
                        ts.connectionsMu.Unlock()
                        return
                default:
                        client.SetDeadline(time.Now().Add(time.Millisecond * 200))
                        byteCount, err := client.Read(buffer)

                        if err != nil {
                                if opErr, ok := err.(*net.OpError); ok {
                                        //NOTE(jwetzell) we hit deadline
                                        if opErr.Timeout() {
                                                continue ClientRead
                                        }
                                        if errors.Is(opErr, syscall.ECONNRESET) {
                                                ts.connectionsMu.Lock()
                                                for i := 0; i < len(ts.connections); i++ {
                                                        if ts.connections[i] == client {
                                                                ts.connections = slices.Delete(ts.connections, i, i+1)
                                                                break
                                                        }
                                                }
                                                ts.logger.Debug("connection reset", "remoteAddr", client.RemoteAddr().String())
                                                ts.connectionsMu.Unlock()
                                        }
                                }

                                if err.Error() == "EOF" {
                                        ts.connectionsMu.Lock()
                                        for i := 0; i < len(ts.connections); i++ {
                                                if ts.connections[i] == client {
                                                        ts.connections = slices.Delete(ts.connections, i, i+1)
                                                        break
                                                }
                                        }
                                        ts.logger.Debug("stream ended", "remoteAddr", client.RemoteAddr().String())
                                        ts.connectionsMu.Unlock()
                                }
                                return
                        }
                        if ts.Framer != nil {
                                if byteCount > 0 {
                                        messages := ts.Framer.Decode(buffer[0:byteCount])
                                        for _, message := range messages {
                                                if ts.router != nil {
                                                        senderAddr, ok := client.RemoteAddr().(*net.TCPAddr)
                                                        if ok {
                                                                senderCtx := context.WithValue(ts.ctx, common.SenderContextKey, senderAddr)
                                                                ts.router.HandleInput(senderCtx, ts.Id(), message)
                                                        } else {
                                                                ts.router.HandleInput(ts.ctx, ts.Id(), message)
                                                        }
                                                } else {
                                                        ts.logger.Error("input received but no router is configured")
                                                }
                                        }
                                }
                        }
                }
        }
}

func (ts *TCPServer) Start(ctx context.Context) error {
        ts.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.tcp.server unable to get router from context")
        }
        ts.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        ts.ctx = moduleContext
        ts.cancel = cancel

        listener, err := net.ListenTCP("tcp", ts.Addr)
        if err != nil {
                return err
        }
        ts.wg.Add(1)

        go func() {
                <-ts.ctx.Done()
                close(ts.quit)
                listener.Close()
                ts.logger.Debug("done")
        }()

AcceptLoop:
        for {
                conn, err := listener.AcceptTCP()
                if err != nil {
                        select {
                        case <-ts.quit:
                                break AcceptLoop
                        default:
                                ts.logger.Debug("problem with listener", "error", err)
                        }
                } else {
                        ts.wg.Add(1)
                        go func() {
                                ts.handleClient(conn)
                                ts.wg.Done()
                        }()
                }
        }
        ts.wg.Done()
        ts.wg.Wait()
        return nil
}

func (ts *TCPServer) Output(ctx context.Context, payload any) error {
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                return errors.New("net.tcp.server is only able to output bytes")
        }
        ts.connectionsMu.Lock()
        errorString := ""

        for _, connection := range ts.connections {
                _, err := connection.Write(payloadBytes)
                if err != nil {
                        errorString += fmt.Sprintf("%s\n", err.Error())
                }
        }
        ts.connectionsMu.Unlock()

        if errorString == "" {
                return nil
        }
        return fmt.Errorf("net.tcp.server error during output: %s", errorString)
}

func (ts *TCPServer) Stop() {
        ts.cancel()
        ts.wg.Wait()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type TimeInterval struct {
        config   config.ModuleConfig
        Duration uint32
        ctx      context.Context
        router   common.RouteIO
        ticker   *time.Ticker
        logger   *slog.Logger
        cancel   context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "time.interval",
                Title: "Interval",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "duration": {
                                        Title:       "Duration",
                                        Type:        "integer",
                                        Description: "Interval duration in milliseconds",
                                },
                        },
                        Required:             []string{"duration"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params

                        durationInt, err := params.GetInt("duration")
                        if err != nil {
                                return nil, fmt.Errorf("time.interval duration error: %w", err)
                        }
                        return &TimeInterval{Duration: uint32(durationInt), config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (i *TimeInterval) Id() string {
        return i.config.Id
}

func (i *TimeInterval) Type() string {
        return i.config.Type
}

func (i *TimeInterval) Start(ctx context.Context) error {
        i.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("time.interval unable to get router from context")
        }
        i.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        i.ctx = moduleContext
        i.cancel = cancel

        ticker := time.NewTicker(time.Millisecond * time.Duration(i.Duration))
        i.ticker = ticker
        defer ticker.Stop()

        for {
                select {
                case <-i.ctx.Done():
                        i.logger.Debug("done")
                        return nil
                case <-ticker.C:
                        if i.router != nil {
                                i.router.HandleInput(i.ctx, i.Id(), time.Now())
                        }
                }
        }

}

func (i *TimeInterval) Stop() {
        i.cancel()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type TimeTimer struct {
        config   config.ModuleConfig
        Duration uint32
        ctx      context.Context
        router   common.RouteIO
        timer    *time.Timer
        logger   *slog.Logger
        cancel   context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "time.timer",
                Title: "Timer",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "duration": {
                                        Title:       "Duration",
                                        Type:        "integer",
                                        Description: "Interval duration in milliseconds",
                                },
                        },
                        Required:             []string{"duration"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params

                        durationNum, err := params.GetInt("duration")
                        if err != nil {
                                return nil, fmt.Errorf("time.timer duration error: %w", err)
                        }

                        return &TimeTimer{Duration: uint32(durationNum), config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (t *TimeTimer) Id() string {
        return t.config.Id
}

func (t *TimeTimer) Type() string {
        return t.config.Type
}

func (t *TimeTimer) Start(ctx context.Context) error {
        t.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.tcp.client unable to get router from context")
        }
        t.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        t.ctx = moduleContext
        t.cancel = cancel

        t.timer = time.NewTimer(time.Millisecond * time.Duration(t.Duration))
        defer t.timer.Stop()
        for {
                select {
                case <-t.ctx.Done():
                        t.timer.Stop()
                        t.logger.Debug("done")
                        return nil
                case time := <-t.timer.C:
                        if t.router != nil {
                                t.router.HandleInput(t.ctx, t.Id(), time)
                        }
                }
        }
}

func (t *TimeTimer) Stop() {
        t.cancel()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "net"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type UDPClient struct {
        config config.ModuleConfig
        Addr   *net.UDPAddr
        Port   uint16
        conn   *net.UDPConn
        ctx    context.Context
        router common.RouteIO
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "net.udp.client",
                Title: "UDP Client",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "host": {
                                        Title: "Host",
                                        Type:  "string",
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                        },
                        Required:             []string{"host", "port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ModuleConfig) (common.Module, error) {
                        params := config.Params
                        hostString, err := params.GetString("host")
                        if err != nil {
                                return nil, fmt.Errorf("net.udp.client host error: %w", err)
                        }

                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("net.udp.client port error: %w", err)
                        }

                        addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", hostString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }
                        return &UDPClient{Addr: addr, config: config, logger: CreateLogger(config)}, nil
                },
        })
}

func (uc *UDPClient) Id() string {
        return uc.config.Id
}

func (uc *UDPClient) Type() string {
        return uc.config.Type
}

func (uc *UDPClient) SetupConn() error {
        client, err := net.DialUDP("udp", nil, uc.Addr)
        uc.conn = client
        return err
}

func (uc *UDPClient) Start(ctx context.Context) error {
        uc.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.udp.client unable to get router from context")
        }
        uc.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        uc.ctx = moduleContext
        uc.cancel = cancel

        err := uc.SetupConn()
        if err != nil {
                return err
        }

        <-uc.ctx.Done()
        uc.logger.Debug("done")
        if uc.conn != nil {
                uc.conn.Close()
        }
        return nil
}

func (uc *UDPClient) Output(ctx context.Context, payload any) error {

        payloadBytes, ok := common.GetAnyAsByteSlice(payload)
        if !ok {
                return errors.New("net.udp.client is only able to output bytes")
        }
        if uc.conn != nil {
                _, err := uc.conn.Write(payloadBytes)

                if err != nil {
                        return err
                }
        } else {
                return errors.New("net.udp.client client is not setup")
        }
        return nil
}

func (uc *UDPClient) Stop() {
        uc.cancel()
}

package module

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type UDPMulticast struct {
        config config.ModuleConfig
        conn   *net.UDPConn
        ctx    context.Context
        router common.RouteIO
        Addr   *net.UDPAddr
        logger *slog.Logger
        cancel context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "net.udp.multicast",
                Title: "UDP Multicast",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title: "IP",
                                        Type:  "string",
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                        },
                        Required:             []string{"ip", "port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params
                        ipString, err := params.GetString("ip")
                        if err != nil {
                                return nil, fmt.Errorf("net.udp.multicast ip error: %w", err)
                        }

                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("net.udp.multicast port error: %w", err)
                        }

                        addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", ipString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }
                        return &UDPMulticast{config: moduleConfig, Addr: addr, logger: CreateLogger(moduleConfig)}, nil
                },
        })
}

func (um *UDPMulticast) Id() string {
        return um.config.Id
}

func (um *UDPMulticast) Type() string {
        return um.config.Type
}

func (um *UDPMulticast) Start(ctx context.Context) error {
        um.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.udp.multicast unable to get router from context")
        }
        um.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        um.ctx = moduleContext
        um.cancel = cancel

        client, err := net.ListenMulticastUDP("udp", nil, um.Addr)
        if err != nil {
                return err
        }
        defer client.Close()

        um.conn = client

        buffer := make([]byte, 2048)
        for {
                select {
                case <-um.ctx.Done():
                        // TODO(jwetzell): cleanup?
                        um.logger.Debug("done")
                        return nil
                default:
                        um.conn.SetDeadline(time.Now().Add(time.Millisecond * 200))

                        numBytes, _, err := um.conn.ReadFromUDP(buffer)
                        if err != nil {
                                //NOTE(jwetzell) we hit deadline
                                if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
                                        continue
                                }
                                return err
                        }

                        if numBytes > 0 {
                                message := buffer[:numBytes]

                                if um.router != nil {
                                        um.router.HandleInput(um.ctx, um.Id(), message)
                                } else {
                                        um.logger.Error("input received but no router is configured")
                                }
                        }
                }
        }
}

func (um *UDPMulticast) Output(ctx context.Context, payload any) error {

        payloadBytes, ok := common.GetAnyAsByteSlice(payload)
        if !ok {
                return errors.New("net.udp.multicast can only output bytes")
        }

        if um.conn == nil {
                return errors.New("net.udp.multicast connection is not setup")
        }

        _, err := um.conn.Write(payloadBytes)
        return err
}

func (um *UDPMulticast) Stop() {
        um.cancel()
}

package module

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "log/slog"
        "net"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type UDPServer struct {
        Addr       *net.UDPAddr
        BufferSize int
        config     config.ModuleConfig
        ctx        context.Context
        router     common.RouteIO
        logger     *slog.Logger
        cancel     context.CancelFunc
}

func init() {
        RegisterModule(ModuleRegistration{
                Type:  "net.udp.server",
                Title: "UDP Server",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "ip": {
                                        Title:   "IP",
                                        Type:    "string",
                                        Default: json.RawMessage(`"0.0.0.0"`),
                                },
                                "port": {
                                        Title:   "Port",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1024),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                },
                                "bufferSize": {
                                        Title:   "Buffer Size",
                                        Type:    "integer",
                                        Minimum: jsonschema.Ptr[float64](1),
                                        Maximum: jsonschema.Ptr[float64](65535),
                                        Default: json.RawMessage("2048"),
                                },
                        },
                        Required:             []string{"port"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
                        params := moduleConfig.Params
                        portNum, err := params.GetInt("port")
                        if err != nil {
                                return nil, fmt.Errorf("net.udp.server port error: %w", err)
                        }

                        ipString, err := params.GetString("ip")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        ipString = "0.0.0.0"
                                } else {
                                        return nil, fmt.Errorf("net.udp.server ip error: %w", err)
                                }
                        }

                        addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", ipString, uint16(portNum)))
                        if err != nil {
                                return nil, err
                        }

                        bufferSizeNum, err := params.GetInt("bufferSize")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        bufferSizeNum = 2048
                                } else {
                                        return nil, fmt.Errorf("net.udp.server bufferSize error: %w", err)
                                }
                        }
                        return &UDPServer{Addr: addr, BufferSize: bufferSizeNum, config: moduleConfig, logger: CreateLogger(moduleConfig)}, nil
                },
        })
}

func (us *UDPServer) Id() string {
        return us.config.Id
}

func (us *UDPServer) Type() string {
        return us.config.Type
}

func (us *UDPServer) Start(ctx context.Context) error {
        us.logger.Debug("running")
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)

        if !ok {
                return errors.New("net.udp.server unable to get router from context")
        }
        us.router = router
        moduleContext, cancel := context.WithCancel(ctx)
        us.ctx = moduleContext
        us.cancel = cancel

        listener, err := net.ListenUDP("udp", us.Addr)
        if err != nil {
                return err
        }

        defer listener.Close()

        buffer := make([]byte, us.BufferSize)
        for {
                select {
                case <-us.ctx.Done():
                        // TODO(jwetzell): cleanup?
                        us.logger.Debug("done")
                        return nil
                default:
                        listener.SetDeadline(time.Now().Add(time.Millisecond * 200))

                        numBytes, senderAddr, err := listener.ReadFromUDP(buffer)
                        if err != nil {
                                //NOTE(jwetzell) we hit deadline
                                if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
                                        continue
                                }
                                return err
                        }
                        message := buffer[:numBytes]
                        if us.router != nil {
                                senderCtx := context.WithValue(us.ctx, common.SenderContextKey, senderAddr)
                                us.router.HandleInput(senderCtx, us.Id(), message)
                        } else {
                                us.logger.Error("input received but no router is configured")
                        }
                }
        }

}

func (us *UDPServer) Output(ctx context.Context, payload any) error {
        return errors.New("net.udp.server output is not implemented")
}

func (us *UDPServer) Stop() {
        us.cancel()
}

package processor

import (
        "context"
        "fmt"

        "github.com/jwetzell/artnet-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type ArtNetPacketDecode struct {
        config config.ProcessorConfig
}

func (apd *ArtNetPacketDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("artnet.packet.decode processor only accepts a []byte")
        }

        payloadMessage, err := artnet.Decode(payloadBytes)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        wrappedPayload.Payload = payloadMessage

        return wrappedPayload, nil
}

func (apd *ArtNetPacketDecode) Type() string {
        return apd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "artnet.packet.decode",
                Title: "Decode ArtNet Packet",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &ArtNetPacketDecode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"

        "github.com/jwetzell/artnet-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type ArtNetPacketEncode struct {
        config config.ProcessorConfig
}

func (ape *ArtNetPacketEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadPacket, ok := common.GetAnyAs[artnet.ArtNetPacket](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("artnet.packet.encode processor only accepts an ArtNetPacket")
        }

        payloadBytes, err := payloadPacket.MarshalBinary()
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        wrappedPayload.Payload = payloadBytes

        return wrappedPayload, nil
}

func (ape *ArtNetPacketEncode) Type() string {
        return ape.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "artnet.packet.encode",
                Title: "Encode ArtNet Packet",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &ArtNetPacketEncode{config: config}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "errors"
        "fmt"
        "log/slog"
        "text/template"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type DbQuery struct {
        config   config.ProcessorConfig
        ModuleId string
        Query    *template.Template
        logger   *slog.Logger
}

func (dq *DbQuery) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        if wrappedPayload.Modules == nil {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("db.query wrapped payload has no modules")
        }

        module, ok := wrappedPayload.Modules[dq.ModuleId]
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("db.query unable to find module with id: %s", dq.ModuleId)
        }

        dbModule, ok := module.(common.DatabaseModule)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("db.query module with id %s is not a DatabaseModule", dq.ModuleId)
        }

        db := dbModule.Database()
        if db == nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("db.query module with id %s returned nil database", dq.ModuleId)
        }

        var queryBuffer bytes.Buffer
        err := dq.Query.Execute(&queryBuffer, wrappedPayload)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        // support proper parameterized queries
        rows, err := db.QueryContext(ctx, queryBuffer.String())
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("db.query error executing query: %w", err)
        }
        defer rows.Close()

        columns, err := rows.Columns()
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("db.query error getting columns: %w", err)
        }

        results := make([]map[string]any, 0)

        for rows.Next() {
                columnValues := make([]interface{}, len(columns))

                for i := range columnValues {
                        columnValues[i] = new(interface{})
                }

                if err := rows.Scan(columnValues...); err != nil {
                        wrappedPayload.End = true
                        return wrappedPayload, fmt.Errorf("db.query error scanning row: %w", err)
                }

                rowMap := make(map[string]any)
                for i, colName := range columns {
                        value := *columnValues[i].(*interface{})
                        rowMap[colName] = value
                }
                results = append(results, rowMap)
        }

        if len(results) == 0 {
                wrappedPayload.Payload = nil
                return wrappedPayload, nil
        } else if len(results) == 1 {
                wrappedPayload.Payload = results[0]
                return wrappedPayload, nil
        }
        wrappedPayload.Payload = results
        return wrappedPayload, nil
}

func (dq *DbQuery) Type() string {
        return dq.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "db.query",
                Title: "Query Database",
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        moduleIdString, err := params.GetString("module")
                        if err != nil {
                                return nil, fmt.Errorf("db.query module error: %w", err)
                        }

                        queryString, err := params.GetString("query")
                        if err != nil {
                                return nil, fmt.Errorf("db.query query error: %w", err)
                        }

                        queryTemplate, err := template.New("query").Parse(queryString)

                        if err != nil {
                                return nil, err
                        }
                        return &DbQuery{config: config, ModuleId: moduleIdString, Query: queryTemplate, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"
        "log/slog"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type DebugLog struct {
        config config.ProcessorConfig
        logger *slog.Logger
}

func (dl *DebugLog) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString := fmt.Sprintf("%+v", payload)
        payloadType := fmt.Sprintf("%T", payload)
        dl.logger.Debug("", "payload", payloadString, "payloadType", payloadType)
        return wrappedPayload, nil
}

func (dl *DebugLog) Type() string {
        return dl.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "debug.log",
                Title: "Debug Log",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &DebugLog{config: config, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

package processor

import (
        "context"
        "reflect"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FilterChange struct {
        config   config.ProcessorConfig
        previous any
}

func (fc *FilterChange) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload

        if reflect.DeepEqual(payload, fc.previous) {
                wrappedPayload.End = true
                return wrappedPayload, nil
        }
        fc.previous = payload

        return wrappedPayload, nil
}

func (fc *FilterChange) Type() string {
        return fc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "filter.change",
                Title: "Filter On Change",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &FilterChange{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"

        "github.com/expr-lang/expr"
        "github.com/expr-lang/expr/vm"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

// NOTE(jwetzell): see language definition https://expr-lang.org/docs/language-definition
type FilterExpr struct {
        config  config.ProcessorConfig
        Program *vm.Program
}

func (fe *FilterExpr) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        exprEnv := wrappedPayload

        output, err := expr.Run(fe.Program, exprEnv)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        outputBool, ok := output.(bool)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("filter.expr expression did not return a boolean")
        }
        if !outputBool {
                wrappedPayload.End = true
                return wrappedPayload, nil
        }

        wrappedPayload.Payload = exprEnv.Payload

        return wrappedPayload, nil
}

func (fe *FilterExpr) Type() string {
        return fe.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "filter.expr",
                Title: "Filter by Expr expression",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "expression": {
                                        Title: "Expression",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"expression"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        expressionString, err := params.GetString("expression")
                        if err != nil {
                                return nil, fmt.Errorf("filter.expr expression error: %w", err)
                        }

                        program, err := expr.Compile(expressionString)
                        if err != nil {
                                return nil, err
                        }

                        return &FilterExpr{config: config, Program: program}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "regexp"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FilterRegex struct {
        config  config.ProcessorConfig
        Pattern *regexp.Regexp
}

func (fr *FilterRegex) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString, ok := common.GetAnyAs[string](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("filter.regex processor only accepts a string")
        }

        if !fr.Pattern.MatchString(payloadString) {
                wrappedPayload.End = true
                return wrappedPayload, nil
        }

        wrappedPayload.Payload = payloadString
        return wrappedPayload, nil
}

func (fr *FilterRegex) Type() string {
        return fr.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "filter.regex",
                Title: "Filter by Regex",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "pattern": {
                                        Title: "Pattern",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"pattern"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        patternString, err := params.GetString("pattern")
                        if err != nil {
                                return nil, fmt.Errorf("filter.regex pattern error: %w", err)
                        }

                        patternRegexp, err := regexp.Compile(patternString)

                        if err != nil {
                                return nil, err
                        }

                        return &FilterRegex{config: config, Pattern: patternRegexp}, nil
                },
        })
}

package processor

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "strconv"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FloatParse struct {
        BitSize int
        config  config.ProcessorConfig
}

func (fp *FloatParse) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString, ok := common.GetAnyAs[string](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("float.parse processor only accepts a string")
        }

        payloadFloat, err := strconv.ParseFloat(payloadString, fp.BitSize)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }
        wrappedPayload.Payload = payloadFloat
        return wrappedPayload, nil
}

func (fp *FloatParse) Type() string {
        return fp.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "float.parse",
                Title: "Parse Float",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "bitSize": {
                                        Title:   "Bit Size",
                                        Type:    "integer",
                                        Enum:    []any{32, 64},
                                        Default: json.RawMessage("64"),
                                },
                        },
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ProcessorConfig) (Processor, error) {
                        params := moduleConfig.Params

                        bitSizeNum, err := params.GetInt("bitSize")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        bitSizeNum = 64
                                } else {
                                        return nil, fmt.Errorf("float.parse bitSize error: %w", err)
                                }
                        }
                        return &FloatParse{config: moduleConfig, BitSize: bitSizeNum}, nil
                },
        })
}

package processor

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "math/rand/v2"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FloatRandom struct {
        BitSize int
        Min     float64
        Max     float64
        config  config.ProcessorConfig
}

func (fr *FloatRandom) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        if fr.BitSize == 32 {
                payloadFloat := rand.Float32()*(float32(fr.Max)-float32(fr.Min)) + float32(fr.Min)
                wrappedPayload.Payload = payloadFloat
                return wrappedPayload, nil
        }
        if fr.BitSize == 64 {
                payloadFloat := rand.Float64()*(fr.Max-fr.Min) + fr.Min
                wrappedPayload.Payload = payloadFloat
                return wrappedPayload, nil
        }
        wrappedPayload.End = true
        return wrappedPayload, errors.New("float.random bitSize error: must be 32 or 64")
}

func (fr *FloatRandom) Type() string {
        return fr.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "float.random",
                Title: "Random Float",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "bitSize": {
                                        Title:   "Bit Size",
                                        Type:    "integer",
                                        Enum:    []any{32, 64},
                                        Default: json.RawMessage("32"),
                                },
                                "min": {
                                        Title: "Minimum",
                                        Type:  "number",
                                },
                                "max": {
                                        Title: "Maximum",
                                        Type:  "number",
                                },
                        },
                        Required:             []string{"min", "max"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(processorConfig config.ProcessorConfig) (Processor, error) {
                        params := processorConfig.Params

                        bitSizeInt, err := params.GetInt("bitSize")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        bitSizeInt = 32
                                } else {
                                        return nil, fmt.Errorf("float.random bitSize error: %w", err)
                                }
                        }

                        if bitSizeInt != 32 && bitSizeInt != 64 {
                                return nil, errors.New("float.random bitSize error: must be 32 or 64")
                        }

                        minFloat, err := params.GetFloat64("min")
                        if err != nil {
                                return nil, fmt.Errorf("float.random min error: %w", err)
                        }

                        maxFloat, err := params.GetFloat64("max")
                        if err != nil {
                                return nil, fmt.Errorf("float.random max error: %w", err)
                        }

                        if maxFloat < minFloat {
                                return nil, errors.New("float.random max must be greater than min")
                        }

                        return &FloatRandom{config: processorConfig, Min: minFloat, Max: maxFloat, BitSize: bitSizeInt}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        freeD "github.com/jwetzell/free-d-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FreeDCreate struct {
        config config.ProcessorConfig
        Id     *template.Template
        Pan    *template.Template
        Tilt   *template.Template
        Roll   *template.Template
        PosX   *template.Template
        PosY   *template.Template
        PosZ   *template.Template
        Zoom   *template.Template
        Focus  *template.Template
}

func (fc *FreeDCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var idBuffer bytes.Buffer
        err := fc.Id.Execute(&idBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        idString := idBuffer.String()

        idNum, err := strconv.ParseUint(idString, 10, 8)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var panBuffer bytes.Buffer
        err = fc.Pan.Execute(&panBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        panString := panBuffer.String()

        panNum, err := strconv.ParseFloat(panString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var tiltBuffer bytes.Buffer
        err = fc.Tilt.Execute(&tiltBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        tiltString := tiltBuffer.String()

        tiltNum, err := strconv.ParseFloat(tiltString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var rollBuffer bytes.Buffer
        err = fc.Roll.Execute(&rollBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        rollString := rollBuffer.String()

        rollNum, err := strconv.ParseFloat(rollString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var posXBuffer bytes.Buffer
        err = fc.PosX.Execute(&posXBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        posXString := posXBuffer.String()

        posXNum, err := strconv.ParseFloat(posXString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var posYBuffer bytes.Buffer
        err = fc.PosY.Execute(&posYBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        posYString := posYBuffer.String()

        posYNum, err := strconv.ParseFloat(posYString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var posZBuffer bytes.Buffer
        err = fc.PosZ.Execute(&posZBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        posZString := posZBuffer.String()

        posZNum, err := strconv.ParseFloat(posZString, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var zoomBuffer bytes.Buffer
        err = fc.Zoom.Execute(&zoomBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        zoomString := zoomBuffer.String()

        zoomNum, err := strconv.ParseInt(zoomString, 10, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        var focusBuffer bytes.Buffer
        err = fc.Focus.Execute(&focusBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        focusString := focusBuffer.String()

        focusNum, err := strconv.ParseInt(focusString, 10, 32)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        payloadMessage := freeD.FreeDPosition{
                ID:    uint8(idNum),
                Pan:   float32(panNum),
                Tilt:  float32(tiltNum),
                Roll:  float32(rollNum),
                PosX:  float32(posXNum),
                PosY:  float32(posYNum),
                PosZ:  float32(posZNum),
                Zoom:  int32(zoomNum),
                Focus: int32(focusNum),
        }

        wrappedPayload.Payload = payloadMessage

        return wrappedPayload, nil
}

func (fc *FreeDCreate) Type() string {
        return fc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "freed.create",
                Title: "Create FreeD",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "id": {
                                        Title: "Camera ID",
                                        Type:  "string",
                                },
                                "pan": {
                                        Title: "Pan",
                                        Type:  "string",
                                },
                                "tilt": {
                                        Title: "Tilt",
                                        Type:  "string",
                                },
                                "roll": {
                                        Title: "Roll",
                                        Type:  "string",
                                },
                                "posX": {
                                        Title: "Position X",
                                        Type:  "string",
                                },
                                "posY": {
                                        Title: "Position Y",
                                        Type:  "string",
                                },
                                "posZ": {
                                        Title: "Position Z",
                                        Type:  "string",
                                },
                                "zoom": {
                                        Title: "Zoom",
                                        Type:  "string",
                                },
                                "focus": {
                                        Title: "Focus",
                                        Type:  "string",
                                },
                        },
                        Required: []string{
                                "id",
                                "pan",
                                "tilt",
                                "roll",
                                "posX",
                                "posY",
                                "posZ",
                                "zoom",
                                "focus",
                        },
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        // TODO(jwetzell): make some params optional
                        params := config.Params
                        idString, err := params.GetString("id")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create id error: %w", err)
                        }

                        idTemplate, err := template.New("id").Parse(idString)

                        if err != nil {
                                return nil, err
                        }

                        panString, err := params.GetString("pan")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create pan error: %w", err)
                        }

                        panTemplate, err := template.New("pan").Parse(panString)

                        if err != nil {
                                return nil, err
                        }

                        tiltString, err := params.GetString("tilt")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create tilt error: %w", err)
                        }

                        tiltTemplate, err := template.New("tilt").Parse(tiltString)

                        if err != nil {
                                return nil, err
                        }

                        rollString, err := params.GetString("roll")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create roll error: %w", err)
                        }

                        rollTemplate, err := template.New("roll").Parse(rollString)

                        if err != nil {
                                return nil, err
                        }

                        posXString, err := params.GetString("posX")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create posX error: %w", err)
                        }

                        posXTemplate, err := template.New("posX").Parse(posXString)

                        if err != nil {
                                return nil, err
                        }

                        posYString, err := params.GetString("posY")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create posY error: %w", err)
                        }

                        posYTemplate, err := template.New("posY").Parse(posYString)

                        if err != nil {
                                return nil, err
                        }

                        posZString, err := params.GetString("posZ")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create posZ error: %w", err)
                        }

                        posZTemplate, err := template.New("posZ").Parse(posZString)

                        if err != nil {
                                return nil, err
                        }

                        zoomString, err := params.GetString("zoom")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create zoom error: %w", err)
                        }

                        zoomTemplate, err := template.New("zoom").Parse(zoomString)

                        if err != nil {
                                return nil, err
                        }

                        focusString, err := params.GetString("focus")
                        if err != nil {
                                return nil, fmt.Errorf("freed.create focus error: %w", err)
                        }

                        focusTemplate, err := template.New("focus").Parse(focusString)
                        if err != nil {
                                return nil, err
                        }

                        return &FreeDCreate{
                                config: config,
                                Id:     idTemplate,
                                Pan:    panTemplate,
                                Tilt:   tiltTemplate,
                                Roll:   rollTemplate,
                                PosX:   posXTemplate,
                                PosY:   posYTemplate,
                                PosZ:   posZTemplate,
                                Zoom:   zoomTemplate,
                                Focus:  focusTemplate,
                        }, nil
                },
        })
}

package processor

import (
        "context"
        "errors"

        freeD "github.com/jwetzell/free-d-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FreeDDecode struct {
        config config.ProcessorConfig
}

func (fd *FreeDDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("freed.decode processor only accepts a []byte")
        }

        payloadMessage, err := freeD.Decode(payloadBytes)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }
        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (fd *FreeDDecode) Type() string {
        return fd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "freed.decode",
                Title: "Decode FreeD",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &FreeDDecode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"

        freeD "github.com/jwetzell/free-d-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type FreeDEncode struct {
        config config.ProcessorConfig
}

func (fe *FreeDEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadPosition, ok := common.GetAnyAs[freeD.FreeDPosition](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("freed.encode processor only accepts a FreeDPosition")
        }

        payloadBytes := freeD.Encode(payloadPosition)

        wrappedPayload.Payload = payloadBytes
        return wrappedPayload, nil
}

func (fe *FreeDEncode) Type() string {
        return fe.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "freed.encode",
                Title: "Encode FreeD",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &FreeDEncode{config: config}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "io"
        "net/http"
        "text/template"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type HTTPRequestDo struct {
        config config.ProcessorConfig
        client *http.Client
        Method string
        URL    *template.Template
}

func (hrd *HTTPRequestDo) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var urlBuffer bytes.Buffer
        err := hrd.URL.Execute(&urlBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        urlString := urlBuffer.String()

        //TODO(jwetzell): support body
        request, err := http.NewRequest(hrd.Method, urlString, bytes.NewBuffer([]byte{}))

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        response, err := hrd.client.Do(request)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        body, err := io.ReadAll(response.Body)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        //TODO(jwetzell): support headers, etc
        wrappedPayload.Payload = HTTPResponse{
                Status: response.StatusCode,
                Body:   body,
        }
        return wrappedPayload, nil
}

func (hrd *HTTPRequestDo) Type() string {
        return hrd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "http.request.do",
                Title: "Do HTTP Request",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "method": {
                                        Title: "HTTP Method",
                                        Type:  "string",
                                        Enum:  []any{"GET", "POST", "PUT", "PATCH", "DELETE"},
                                },
                                "url": {
                                        Title: "URL",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"method", "url"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        methodString, err := params.GetString("method")
                        if err != nil {
                                return nil, fmt.Errorf("http.request.do method error: %w", err)
                        }

                        urlString, err := params.GetString("url")
                        if err != nil {
                                return nil, fmt.Errorf("http.request.do url error: %w", err)
                        }

                        urlTemplate, err := template.New("url").Parse(urlString)

                        if err != nil {
                                return nil, err
                        }
                        client := &http.Client{
                                Timeout: 10 * time.Second,
                        }

                        return &HTTPRequestDo{config: config, URL: urlTemplate, Method: methodString, client: client}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type HTTPResponseCreate struct {
        Status   int
        BodyTmpl *template.Template
        config   config.ProcessorConfig
}

type HTTPResponse struct {
        Status int
        Body   []byte
}

func (hrc *HTTPResponseCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var bodyBuffer bytes.Buffer
        err := hrc.BodyTmpl.Execute(&bodyBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }
        wrappedPayload.Payload = HTTPResponse{
                Status: hrc.Status,
                Body:   bodyBuffer.Bytes(),
        }
        return wrappedPayload, nil
}

func (hrc *HTTPResponseCreate) Type() string {
        return hrc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "http.response.create",
                Title: "Create HTTP Response",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "status": {
                                        Title: "Status Code",
                                        Type:  "integer",
                                },
                                "body": {
                                        Title: "Body",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"status", "body"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        statusNum, err := params.GetInt("status")
                        if err != nil {
                                return nil, fmt.Errorf("http.response.create status error: %w", err)
                        }

                        bodyTemplateString, err := params.GetString("bodyTemplate")
                        if err != nil {
                                return nil, fmt.Errorf("http.response.create bodyTemplate error: %w", err)
                        }

                        bodyTemplate, err := template.New("body").Parse(bodyTemplateString)

                        if err != nil {
                                return nil, err
                        }

                        // TODO(jwetzell): support other body kind (direct bytes from input, from file?)
                        return &HTTPResponseCreate{config: config, Status: int(statusNum), BodyTmpl: bodyTemplate}, nil
                },
        })
}

package processor

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "strconv"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type IntParse struct {
        Base    int
        BitSize int
        config  config.ProcessorConfig
}

func (ip *IntParse) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString, ok := common.GetAnyAs[string](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("int.parse processor only accepts a string")
        }

        payloadInt, err := strconv.ParseInt(payloadString, ip.Base, ip.BitSize)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }
        wrappedPayload.Payload = payloadInt
        return wrappedPayload, nil
}

func (ip *IntParse) Type() string {
        return ip.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "int.parse",
                Title: "Parse Int",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "base": {
                                        Title:   "Base",
                                        Type:    "integer",
                                        Enum:    []any{0, 2, 8, 10, 16},
                                        Default: json.RawMessage("10"),
                                },
                                "bitSize": {
                                        Title:   "Bit Size",
                                        Type:    "integer",
                                        Enum:    []any{0, 8, 16, 32, 64},
                                        Default: json.RawMessage("64"),
                                },
                        },
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(moduleConfig config.ProcessorConfig) (Processor, error) {
                        params := moduleConfig.Params

                        baseNum, err := params.GetInt("base")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        baseNum = 10
                                } else {
                                        return nil, fmt.Errorf("int.parse base error: %w", err)
                                }
                        }

                        bitSizeNum, err := params.GetInt("bitSize")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        bitSizeNum = 64
                                } else {
                                        return nil, fmt.Errorf("int.parse bitSize error: %w", err)
                                }
                        }
                        return &IntParse{config: moduleConfig, Base: baseNum, BitSize: bitSizeNum}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "math/rand/v2"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type IntRandom struct {
        Min    int
        Max    int
        config config.ProcessorConfig
}

func (ir *IntRandom) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payloadInt := rand.IntN(ir.Max-ir.Min+1) + ir.Min
        wrappedPayload.Payload = payloadInt
        return wrappedPayload, nil
}

func (ir *IntRandom) Type() string {
        return ir.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "int.random",
                Title: "Random Int",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "min": {
                                        Title: "Minimum",
                                        Type:  "integer",
                                },
                                "max": {
                                        Title: "Maximum",
                                        Type:  "integer",
                                },
                        },
                        Required:             []string{"min", "max"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        minInt, err := params.GetInt("min")
                        if err != nil {
                                return nil, fmt.Errorf("int.random min error: %w", err)
                        }

                        maxInt, err := params.GetInt("max")
                        if err != nil {
                                return nil, fmt.Errorf("int.random max error: %w", err)
                        }

                        if maxInt < minInt {
                                return nil, errors.New("int.random max must be greater than min")
                        }

                        return &IntRandom{config: config, Min: int(minInt), Max: int(maxInt)}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type IntScale struct {
        OutMin int
        OutMax int
        InMin  int
        InMax  int
        config config.ProcessorConfig
}

func (ir *IntScale) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadInt, ok := common.GetAnyAs[int](payload)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("int.scale can only process an int")
        }

        payloadInt = (payloadInt-ir.InMin)*(ir.OutMax-ir.OutMin)/(ir.InMax-ir.InMin) + ir.OutMin
        wrappedPayload.Payload = payloadInt
        return wrappedPayload, nil
}

func (ir *IntScale) Type() string {
        return ir.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "int.scale",
                Title: "Scale Int",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "inMin": {
                                        Title: "Input Minimum",
                                        Type:  "integer",
                                },
                                "inMax": {
                                        Title: "Input Maximum",
                                        Type:  "integer",
                                },
                                "outMin": {
                                        Title: "Output Minimum",
                                        Type:  "integer",
                                },
                                "outMax": {
                                        Title: "Output Maximum",
                                        Type:  "integer",
                                },
                        },
                        Required:             []string{"inMin", "inMax", "outMin", "outMax"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        inMinInt, err := params.GetInt("inMin")
                        if err != nil {
                                return nil, fmt.Errorf("int.scale inMin error: %w", err)
                        }

                        inMaxInt, err := params.GetInt("inMax")
                        if err != nil {
                                return nil, fmt.Errorf("int.scale inMax error: %w", err)
                        }

                        if inMaxInt < inMinInt {
                                return nil, errors.New("int.scale inMax must be greater than inMin")
                        }

                        outMinInt, err := params.GetInt("outMin")
                        if err != nil {
                                return nil, fmt.Errorf("int.scale outMin error: %w", err)
                        }

                        outMaxInt, err := params.GetInt("outMax")
                        if err != nil {
                                return nil, fmt.Errorf("int.scale outMax error: %w", err)
                        }

                        if outMaxInt < outMinInt {
                                return nil, errors.New("int.scale outMax must be greater than outMin")
                        }

                        return &IntScale{config: config, InMin: inMinInt, InMax: inMaxInt, OutMin: outMinInt, OutMax: outMaxInt}, nil
                },
        })
}

package processor

import (
        "context"
        "encoding/json"
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type JsonDecode struct {
        config config.ProcessorConfig
}

func (jd *JsonDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload

        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                payloadString, ok := common.GetAnyAs[string](payload)
                if !ok {
                        wrappedPayload.End = true
                        return wrappedPayload, errors.New("json.decode can only process a string or []byte")
                }
                payloadBytes = []byte(payloadString)
        }

        payloadJson := make(map[string]any)

        err := json.Unmarshal(payloadBytes, &payloadJson)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        wrappedPayload.Payload = payloadJson
        return wrappedPayload, nil

}

func (jd *JsonDecode) Type() string {
        return jd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "json.decode",
                Title: "Decode JSON",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &JsonDecode{config: config}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "encoding/json"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type JsonEncode struct {
        config config.ProcessorConfig
}

func (je *JsonEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        var payloadBuffer bytes.Buffer

        err := json.NewEncoder(&payloadBuffer).Encode(payload)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        payloadBytes := payloadBuffer.Bytes()

        payloadBytes = payloadBytes[0 : len(payloadBytes)-1]

        wrappedPayload.Payload = payloadBytes
        return wrappedPayload, nil
}

func (je *JsonEncode) Type() string {
        return je.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "json.encode",
                Title: "Encode JSON",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &JsonEncode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type KVGet struct {
        config   config.ProcessorConfig
        ModuleId string
        Key      string
        logger   *slog.Logger
}

func (kvg *KVGet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        if wrappedPayload.Modules == nil {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("kv.get wrapped payload has no modules")
        }

        module, ok := wrappedPayload.Modules[kvg.ModuleId]
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.get unable to find module with id: %s", kvg.ModuleId)
        }

        kvModule, ok := module.(common.KeyValueModule)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.get module with id %s is not a KeyValueModule", kvg.ModuleId)
        }

        value, err := kvModule.Get(kvg.Key)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.get error getting key: %w", err)
        }

        wrappedPayload.Payload = value
        return wrappedPayload, nil
}

func (kvg *KVGet) Type() string {
        return kvg.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "kv.get",
                Title: "Get Key",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "module": {
                                        Title: "Module ID",
                                        Type:  "string",
                                },
                                "key": {
                                        Title: "Key",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"module", "key"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        moduleIdString, err := params.GetString("module")
                        if err != nil {
                                return nil, fmt.Errorf("kv.get module error: %w", err)
                        }

                        keyString, err := params.GetString("key")
                        if err != nil {
                                return nil, fmt.Errorf("kv.get key error: %w", err)
                        }
                        return &KVGet{config: config, ModuleId: moduleIdString, Key: keyString, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "errors"
        "fmt"
        "html/template"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type KVSet struct {
        config   config.ProcessorConfig
        ModuleId string
        Key      string
        Value    *template.Template
        logger   *slog.Logger
}

func (kvs *KVSet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        if wrappedPayload.Modules == nil {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("kv.set wrapped payload has no modules")
        }

        module, ok := wrappedPayload.Modules[kvs.ModuleId]
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.set unable to find module with id: %s", kvs.ModuleId)
        }

        kvModule, ok := module.(common.KeyValueModule)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.set module with id %s is not a KeyValueModule", kvs.ModuleId)
        }

        var valueBuffer bytes.Buffer
        err := kvs.Value.Execute(&valueBuffer, wrappedPayload)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        err = kvModule.Set(kvs.Key, valueBuffer.String())
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("kv.set error setting key: %w", err)
        }

        return wrappedPayload, nil
}

func (kvs *KVSet) Type() string {
        return kvs.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "kv.set",
                Title: "Set Key",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "module": {
                                        Title: "Module ID",
                                        Type:  "string",
                                },
                                "key": {
                                        Title: "Key",
                                        Type:  "string",
                                },
                                "value": {
                                        Title: "Value",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"module", "key", "value"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        moduleIdString, err := params.GetString("module")
                        if err != nil {
                                return nil, fmt.Errorf("kv.set module error: %w", err)
                        }

                        keyString, err := params.GetString("key")
                        if err != nil {
                                return nil, fmt.Errorf("kv.set key error: %w", err)
                        }

                        valueString, err := params.GetString("value")
                        if err != nil {
                                return nil, fmt.Errorf("kv.set value error: %w", err)
                        }
                        valueTemplate, err := template.New("template").Parse(valueString)

                        if err != nil {
                                return nil, err
                        }

                        return &KVSet{config: config, ModuleId: moduleIdString, Key: keyString, Value: valueTemplate, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

//go:build cgo

package processor

import (
        "bytes"
        "context"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDIControlChangeCreate struct {
        config  config.ProcessorConfig
        Channel *template.Template
        Control *template.Template
        Value   *template.Template
}

func (mccc *MIDIControlChangeCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var channelBuffer bytes.Buffer
        err := mccc.Channel.Execute(&channelBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)

        var controlBuffer bytes.Buffer
        err = mccc.Control.Execute(&controlBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        controlValue, err := strconv.ParseUint(controlBuffer.String(), 10, 8)

        var valueBuffer bytes.Buffer
        err = mccc.Value.Execute(&valueBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        valueValue, err := strconv.ParseUint(valueBuffer.String(), 10, 8)

        payloadMessage := midi.ControlChange(uint8(channelValue), uint8(controlValue), uint8(valueValue))
        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (mccc *MIDIControlChangeCreate) Type() string {
        return mccc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.control_change.create",
                Title: "Create MIDI Control Change Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "channel": {
                                        Title: "Channel",
                                        Type:  "string",
                                },
                                "control": {
                                        Title: "Control",
                                        Type:  "string",
                                },
                                "value": {
                                        Title: "Value",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"channel", "control", "value"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        channelString, err := params.GetString("channel")
                        if err != nil {
                                return nil, fmt.Errorf("midi.control_change.create channel error: %w", err)
                        }

                        channelTemplate, err := template.New("channel").Parse(channelString)

                        if err != nil {
                                return nil, err
                        }

                        controlString, err := params.GetString("control")
                        if err != nil {
                                return nil, fmt.Errorf("midi.control_change.create control error: %w", err)
                        }

                        controlTemplate, err := template.New("control").Parse(controlString)

                        if err != nil {
                                return nil, err
                        }

                        valueString, err := params.GetString("value")
                        if err != nil {
                                return nil, fmt.Errorf("midi.control_change.create value error: %w", err)
                        }

                        valueTemplate, err := template.New("value").Parse(valueString)

                        if err != nil {
                                return nil, err
                        }
                        return &MIDIControlChangeCreate{
                                config:  config,
                                Channel: channelTemplate,
                                Control: controlTemplate,
                                Value:   valueTemplate,
                        }, nil
                },
        })
}

//go:build cgo

package processor

import (
        "context"
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDIMessageDecode struct {
        config config.ProcessorConfig
}

func (mmd *MIDIMessageDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("midi.message.decode processor only accepts a []byte")
        }

        payloadMessage := midi.Message(payloadBytes)

        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (mmd *MIDIMessageDecode) Type() string {
        return mmd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.message.decode",
                Title: "Decode MIDI Message",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &MIDIMessageDecode{config: config}, nil
                },
        })
}

//go:build cgo

package processor

import (
        "context"
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDIMessageEncode struct {
        config config.ProcessorConfig
}

func (mme *MIDIMessageEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadMessage, ok := common.GetAnyAs[midi.Message](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("midi.message.encode processor only accepts a midi.Message")
        }

        wrappedPayload.Payload = payloadMessage.Bytes()
        return wrappedPayload, nil
}

func (mme *MIDIMessageEncode) Type() string {
        return mme.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.message.encode",
                Title: "Encode MIDI Message",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &MIDIMessageEncode{config: config}, nil
                },
        })
}

//go:build cgo

package processor

import (
        "context"
        "errors"
        "fmt"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDIMessageUnpack struct {
        config config.ProcessorConfig
}

type MIDINoteOn struct {
        Channel  uint8
        Note     uint8
        Velocity uint8
}

type MIDINoteOff struct {
        Channel  uint8
        Note     uint8
        Velocity uint8
}

type MIDIControlChange struct {
        Channel uint8
        Control uint8
        Value   uint8
}

type MIDIProgramChange struct {
        Channel uint8
        Program uint8
}

type MIDIPitchBend struct {
        Channel  uint8
        Relative int16
        Absolute uint16
}

func (mmu *MIDIMessageUnpack) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadMidi, ok := common.GetAnyAs[midi.Message](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("midi.message.unpack processor only accepts a midi.Message")
        }

        switch payloadMidi.Type() {
        case midi.NoteOnMsg:
                noteOnMsg := MIDINoteOn{}
                payloadMidi.GetNoteOn(&noteOnMsg.Channel, &noteOnMsg.Note, &noteOnMsg.Velocity)
                wrappedPayload.Payload = noteOnMsg
                return wrappedPayload, nil
        case midi.NoteOffMsg:
                noteOffMsg := MIDINoteOff{}
                payloadMidi.GetNoteOff(&noteOffMsg.Channel, &noteOffMsg.Note, &noteOffMsg.Velocity)
                wrappedPayload.Payload = noteOffMsg
                return wrappedPayload, nil
        case midi.ControlChangeMsg:
                controlChangeMsg := MIDIControlChange{}
                payloadMidi.GetControlChange(&controlChangeMsg.Channel, &controlChangeMsg.Control, &controlChangeMsg.Value)
                wrappedPayload.Payload = controlChangeMsg
                return wrappedPayload, nil
        case midi.ProgramChangeMsg:
                programChangeMsg := MIDIProgramChange{}
                payloadMidi.GetProgramChange(&programChangeMsg.Channel, &programChangeMsg.Program)
                wrappedPayload.Payload = programChangeMsg
                return wrappedPayload, nil
        case midi.PitchBendMsg:
                pitchBendMsg := MIDIPitchBend{}
                payloadMidi.GetPitchBend(&pitchBendMsg.Channel, &pitchBendMsg.Relative, &pitchBendMsg.Absolute)
                wrappedPayload.Payload = pitchBendMsg
                return wrappedPayload, nil
        default:
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("midi.message.unpack message type not supported %v", payloadMidi.Type())
        }
}

func (mmu *MIDIMessageUnpack) Type() string {
        return mmu.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.message.unpack",
                Title: "Unpack MIDI Message",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &MIDIMessageUnpack{config: config}, nil
                },
        })
}

//go:build cgo

package processor

import (
        "bytes"
        "context"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDINoteOffCreate struct {
        config   config.ProcessorConfig
        Channel  *template.Template
        Note     *template.Template
        Velocity *template.Template
}

func (mnoc *MIDINoteOffCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var channelBuffer bytes.Buffer
        err := mnoc.Channel.Execute(&channelBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)

        var noteBuffer bytes.Buffer
        err = mnoc.Note.Execute(&noteBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        noteValue, err := strconv.ParseUint(noteBuffer.String(), 10, 8)

        var velocityBuffer bytes.Buffer
        err = mnoc.Velocity.Execute(&velocityBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        velocityValue, err := strconv.ParseUint(velocityBuffer.String(), 10, 8)
        payloadMessage := midi.NoteOffVelocity(uint8(channelValue), uint8(noteValue), uint8(velocityValue))
        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (mnoc *MIDINoteOffCreate) Type() string {
        return mnoc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.note_off.create",
                Title: "Create MIDI Note Off Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "channel": {
                                        Title: "Channel",
                                        Type:  "string",
                                },
                                "note": {
                                        Title: "Note",
                                        Type:  "string",
                                },
                                "velocity": {
                                        Title: "Velocity",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"channel", "note", "velocity"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        channelString, err := params.GetString("channel")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_off.create channel error: %w", err)
                        }

                        channelTemplate, err := template.New("channel").Parse(channelString)

                        if err != nil {
                                return nil, err
                        }

                        noteString, err := params.GetString("note")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_off.create note error: %w", err)
                        }

                        noteTemplate, err := template.New("note").Parse(noteString)

                        if err != nil {
                                return nil, err
                        }

                        velocityString, err := params.GetString("velocity")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_off.create velocity error: %w", err)
                        }

                        velocityTemplate, err := template.New("velocity").Parse(velocityString)

                        if err != nil {
                                return nil, err
                        }
                        return &MIDINoteOffCreate{
                                config:   config,
                                Channel:  channelTemplate,
                                Note:     noteTemplate,
                                Velocity: velocityTemplate,
                        }, nil
                },
        })
}

//go:build cgo

package processor

import (
        "bytes"
        "context"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDINoteOnCreate struct {
        config   config.ProcessorConfig
        Channel  *template.Template
        Note     *template.Template
        Velocity *template.Template
}

func (mnoc *MIDINoteOnCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var channelBuffer bytes.Buffer
        err := mnoc.Channel.Execute(&channelBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)

        var noteBuffer bytes.Buffer
        err = mnoc.Note.Execute(&noteBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        noteValue, err := strconv.ParseUint(noteBuffer.String(), 10, 8)

        var velocityBuffer bytes.Buffer
        err = mnoc.Velocity.Execute(&velocityBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        velocityValue, err := strconv.ParseUint(velocityBuffer.String(), 10, 8)
        payloadMessage := midi.NoteOn(uint8(channelValue), uint8(noteValue), uint8(velocityValue))
        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (mnoc *MIDINoteOnCreate) Type() string {
        return mnoc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.note_on.create",
                Title: "Create MIDI Note On Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "channel": {
                                        Title: "Channel",
                                        Type:  "string",
                                },
                                "note": {
                                        Title: "Note",
                                        Type:  "string",
                                },
                                "velocity": {
                                        Title: "Velocity",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"channel", "note", "velocity"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        channelString, err := params.GetString("channel")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_on.create channel error: %w", err)
                        }

                        channelTemplate, err := template.New("channel").Parse(channelString)

                        if err != nil {
                                return nil, err
                        }

                        noteString, err := params.GetString("note")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_on.create note error: %w", err)
                        }

                        noteTemplate, err := template.New("note").Parse(noteString)

                        if err != nil {
                                return nil, err
                        }

                        velocityString, err := params.GetString("velocity")
                        if err != nil {
                                return nil, fmt.Errorf("midi.note_on.create velocity error: %w", err)
                        }

                        velocityTemplate, err := template.New("velocity").Parse(velocityString)

                        if err != nil {
                                return nil, err
                        }
                        return &MIDINoteOnCreate{
                                config:   config,
                                Channel:  channelTemplate,
                                Note:     noteTemplate,
                                Velocity: velocityTemplate,
                        }, nil
                },
        })
}

//go:build cgo

package processor

import (
        "bytes"
        "context"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "gitlab.com/gomidi/midi/v2"
)

type MIDIProgramChangeCreate struct {
        config  config.ProcessorConfig
        Channel *template.Template
        Program *template.Template
}

func (mpcc *MIDIProgramChangeCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var channelBuffer bytes.Buffer
        err := mpcc.Channel.Execute(&channelBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)

        var programBuffer bytes.Buffer
        err = mpcc.Program.Execute(&programBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        programValue, err := strconv.ParseUint(programBuffer.String(), 10, 8)

        payloadMessage := midi.ProgramChange(uint8(channelValue), uint8(programValue))
        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (mpcc *MIDIProgramChangeCreate) Type() string {
        return mpcc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "midi.program_change.create",
                Title: "Create MIDI Prgoram Change Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "channel": {
                                        Title: "Channel",
                                        Type:  "string",
                                },
                                "program": {
                                        Title: "Program",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"type", "channel", "program"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        channelString, err := params.GetString("channel")
                        if err != nil {
                                return nil, fmt.Errorf("midi.program_change.create channel error: %w", err)
                        }

                        channelTemplate, err := template.New("channel").Parse(channelString)

                        if err != nil {
                                return nil, err
                        }

                        programString, err := params.GetString("program")
                        if err != nil {
                                return nil, fmt.Errorf("midi.program_change.create program error: %w", err)
                        }

                        programTemplate, err := template.New("program").Parse(programString)

                        if err != nil {
                                return nil, err
                        }
                        return &MIDIProgramChangeCreate{
                                config:  config,
                                Channel: channelTemplate,
                                Program: programTemplate,
                        }, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type MQTTMessage struct {
        topic    string
        qos      byte
        payload  []byte
        retained bool
}

type MQTTMessageCreate struct {
        config   config.ProcessorConfig
        Topic    string
        QoS      byte
        Retained bool
        Payload  []byte
}

func NewMQTTMessage(topic string, qos byte, retained bool, payload []byte) MQTTMessage {
        return MQTTMessage{
                topic:    topic,
                qos:      qos,
                retained: retained,
                payload:  payload,
        }
}

func (mm MQTTMessage) Duplicate() bool {
        // TODO(jwetzell): implement?
        return false
}

func (mm MQTTMessage) Qos() byte {
        return mm.qos
}

func (mm MQTTMessage) Retained() bool {
        return mm.retained
}

func (mm MQTTMessage) Topic() string {
        return mm.topic
}

func (mm MQTTMessage) MessageID() uint16 {
        // TODO(jwetzell): implement?
        return 0
}

func (mm MQTTMessage) Payload() []byte {
        return mm.payload
}

func (mm MQTTMessage) Ack() {}

func (mmc *MQTTMessageCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        // TODO(jwetzell): support templating

        wrappedPayload.Payload = MQTTMessage{
                topic:    mmc.Topic,
                qos:      mmc.QoS,
                retained: mmc.Retained,
                payload:  mmc.Payload,
        }

        return wrappedPayload, nil
}

func (mmc *MQTTMessageCreate) Type() string {
        return mmc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "mqtt.message.create",
                Title: "Create MQTT Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "topic": {
                                        Title: "Topic",
                                        Type:  "string",
                                },
                                "qos": {
                                        Title: "QoS",
                                        Type:  "number",
                                },
                                "retained": {
                                        Title: "Retained",
                                        Type:  "boolean",
                                },
                                "payload": {
                                        Title: "Payload",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"topic", "qos", "retained", "payload"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(processorConfig config.ProcessorConfig) (Processor, error) {
                        params := processorConfig.Params
                        topicString, err := params.GetString("topic")
                        if err != nil {
                                return nil, fmt.Errorf("mqtt.message.create topic error: %w", err)
                        }

                        qosByte, err := params.GetInt("qos")
                        if err != nil {
                                return nil, fmt.Errorf("mqtt.message.create qos error: %w", err)
                        }

                        retainedBool, err := params.GetBool("retained")
                        if err != nil {
                                return nil, fmt.Errorf("mqtt.message.create retained error: %w", err)
                        }

                        //TODO(jwetzell): convert payload into []byte or string for sending
                        payloadString, err := params.GetString("payload")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotString) {
                                        payloadBytes, err := params.GetByteSlice("payload")
                                        if err != nil {
                                                return nil, fmt.Errorf("mqtt.message.create payload error: %w", err)
                                        }
                                        return &MQTTMessageCreate{config: processorConfig, Topic: topicString, QoS: byte(qosByte), Retained: retainedBool, Payload: payloadBytes}, nil
                                } else {
                                        return nil, fmt.Errorf("mqtt.message.create payload error: %w", err)
                                }
                        }

                        payloadBytes := []byte(payloadString)

                        return &MQTTMessageCreate{config: processorConfig, Topic: topicString, QoS: byte(qosByte), Retained: retainedBool, Payload: payloadBytes}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type NATSMessage struct {
        Subject string
        Payload []byte
}

type NATSMessageCreate struct {
        config  config.ProcessorConfig
        Subject *template.Template
        Payload *template.Template
}

func (nmc *NATSMessageCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var payloadBuffer bytes.Buffer
        err := nmc.Payload.Execute(&payloadBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        payloadString := payloadBuffer.String()

        var subjectBuffer bytes.Buffer
        err = nmc.Subject.Execute(&subjectBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        subjectString := subjectBuffer.String()

        wrappedPayload.Payload = NATSMessage{
                Subject: subjectString,
                Payload: []byte(payloadString),
        }

        return wrappedPayload, nil
}

func (nmc *NATSMessageCreate) Type() string {
        return nmc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "nats.message.create",
                Title: "Create NATS Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "subject": {
                                        Title: "Subject",
                                        Type:  "string",
                                },
                                "payload": {
                                        Title: "Payload",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"subject", "payload"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params
                        subjectString, err := params.GetString("subject")
                        if err != nil {
                                return nil, fmt.Errorf("nats.message.create subject error: %w", err)
                        }

                        subjectTemplate, err := template.New("subject").Parse(subjectString)

                        if err != nil {
                                return nil, err
                        }

                        payloadString, err := params.GetString("payload")
                        if err != nil {
                                return nil, fmt.Errorf("nats.message.create payload error: %w", err)
                        }

                        payloadTemplate, err := template.New("payload").Parse(payloadString)

                        if err != nil {
                                return nil, err
                        }

                        return &NATSMessageCreate{config: config, Subject: subjectTemplate, Payload: payloadTemplate}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "encoding/hex"
        "errors"
        "fmt"
        "strconv"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/osc-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type OSCMessageCreate struct {
        config  config.ProcessorConfig
        Address *template.Template
        Args    []*template.Template
        Types   string
}

func (omc *OSCMessageCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var addressBuffer bytes.Buffer
        err := omc.Address.Execute(&addressBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        addressString := addressBuffer.String()

        if len(addressString) == 0 {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.create address must not be empty")
        }

        if addressString[0] != '/' {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.create address must start with '/'")
        }

        payloadMessage := &osc.OSCMessage{
                Address: addressString,
        }

        args := []osc.OSCArg{}

        for argIndex, argTemplate := range omc.Args {
                var argBuffer bytes.Buffer
                err := argTemplate.Execute(&argBuffer, templateData)

                if err != nil {
                        wrappedPayload.End = true
                        return wrappedPayload, err
                }

                argString := argBuffer.String()

                typedArg, err := argToTypedArg(argString, omc.Types[argIndex])

                if err != nil {
                        wrappedPayload.End = true
                        return wrappedPayload, err
                }

                args = append(args, typedArg)
        }

        if len(args) > 0 {
                payloadMessage.Args = args
        }

        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (omc *OSCMessageCreate) Type() string {
        return omc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "osc.message.create",
                Title: "Create OSC Message",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "address": {
                                        Title: "Address",
                                        Type:  "string",
                                },
                                "args": {
                                        Title: "Arguments",
                                        Type:  "array",
                                        Items: &jsonschema.Schema{
                                                Type: "string",
                                        },
                                },
                                "types": {
                                        Title: "Argument Types",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"address"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(processorConfig config.ProcessorConfig) (Processor, error) {
                        params := processorConfig.Params
                        addressString, err := params.GetString("address")
                        if err != nil {
                                return nil, fmt.Errorf("osc.message.create address error: %w", err)
                        }

                        addressTemplate, err := template.New("address").Parse(addressString)

                        if err != nil {
                                return nil, err
                        }

                        argStrings, err := params.GetStringSlice("args")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        return &OSCMessageCreate{config: processorConfig, Address: addressTemplate}, nil
                                } else {
                                        return nil, fmt.Errorf("osc.message.create args error: %w", err)
                                }
                        }

                        typesString, err := params.GetString("types")
                        if err != nil {
                                return nil, fmt.Errorf("osc.message.create types error: %w", err)
                        }

                        if len(argStrings) != len(typesString) {
                                return nil, errors.New("osc.message.create args and types must be the same length")
                        }

                        argTemplates := []*template.Template{}

                        for _, argString := range argStrings {

                                argTemplate, err := template.New("arg").Parse(argString)

                                if err != nil {
                                        return nil, err
                                }
                                argTemplates = append(argTemplates, argTemplate)
                        }
                        return &OSCMessageCreate{config: processorConfig, Address: addressTemplate, Args: argTemplates, Types: typesString}, nil
                },
        })
}

func argToTypedArg(rawArg string, oscType byte) (osc.OSCArg, error) {

        switch oscType {
        case 's':
                return osc.OSCArg{
                        Value: rawArg,
                        Type:  "s",
                }, nil
        case 'i':
                number, err := strconv.ParseInt(rawArg, 10, 32)
                if err != nil {
                        return osc.OSCArg{}, err
                }
                return osc.OSCArg{
                        Value: int32(number),
                        Type:  "i",
                }, nil
        case 'f':
                number, err := strconv.ParseFloat(rawArg, 32)
                if err != nil {
                        return osc.OSCArg{}, err
                }
                return osc.OSCArg{
                        Value: float32(number),
                        Type:  "f",
                }, nil
        case 'b':
                data, err := hex.DecodeString(rawArg)
                if err != nil {
                        return osc.OSCArg{}, err
                }
                return osc.OSCArg{
                        Value: data,
                        Type:  "b",
                }, nil
        case 'h':
                number, err := strconv.ParseInt(rawArg, 10, 64)
                if err != nil {
                        return osc.OSCArg{}, err
                }
                return osc.OSCArg{
                        Value: int64(number),
                        Type:  "h",
                }, nil
        case 'd':
                number, err := strconv.ParseFloat(rawArg, 64)
                if err != nil {
                        return osc.OSCArg{}, err
                }
                return osc.OSCArg{
                        Value: float64(number),
                        Type:  "d",
                }, nil
        case 'T':
                return osc.OSCArg{
                        Value: true,
                        Type:  "T",
                }, nil
        case 'F':
                return osc.OSCArg{
                        Value: false,
                        Type:  "F",
                }, nil
        case 'N':
                return osc.OSCArg{
                        Value: nil,
                        Type:  "N",
                }, nil
        default:
                return osc.OSCArg{}, fmt.Errorf("osc.message.create unhandled osc type: %c", oscType)
        }
}

package processor

import (
        "context"
        "errors"
        "fmt"

        osc "github.com/jwetzell/osc-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type OSCMessageDecode struct {
        config config.ProcessorConfig
}

func (omd *OSCMessageDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.decode processor only accepts a []byte payload")
        }

        if len(payloadBytes) == 0 {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.decode processor can't work on empty []byte")
        }

        if payloadBytes[0] != '/' {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.decode processor needs an OSC looking []byte")
        }

        message, err := osc.MessageFromBytes(payloadBytes)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("osc.message.decode processor failed to decode OSC message: %w", err)
        }
        wrappedPayload.Payload = message
        return wrappedPayload, nil
}

func (omd *OSCMessageDecode) Type() string {
        return omd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "osc.message.decode",
                Title: "Decode OSC Message",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &OSCMessageDecode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"

        osc "github.com/jwetzell/osc-go"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type OSCMessageEncode struct {
        config config.ProcessorConfig
}

func (ome *OSCMessageEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadMessage, ok := common.GetAnyAs[*osc.OSCMessage](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("osc.message.encode processor only accepts an *OSCMessage")
        }

        bytes, err := payloadMessage.ToBytes()
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("osc.message.encode processor failed to encode OSCMessage: %w", err)
        }
        wrappedPayload.Payload = bytes
        return wrappedPayload, nil
}

func (ome *OSCMessageEncode) Type() string {
        return ome.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "osc.message.encode",
                Title: "Encode OSC Message",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &OSCMessageEncode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"
        "sync"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type Processor interface {
        Type() string
        Process(context.Context, common.WrappedPayload) (common.WrappedPayload, error)
}

type ProcessorRegistration struct {
        Type         string             `json:"type"`
        Title        string             `json:"title,omitempty"`
        Description  string             `json:"description,omitempty"`
        ParamsSchema *jsonschema.Schema `json:"paramsSchema,omitempty"`
        New          func(config.ProcessorConfig) (Processor, error)
}

func RegisterProcessor(processor ProcessorRegistration) {

        if processor.Type == "" {
                panic("processor type is missing")
        }
        if processor.New == nil {
                panic("missing ProcessorRegistration.New")
        }

        processorRegistryMu.Lock()
        defer processorRegistryMu.Unlock()

        if _, ok := ProcessorRegistry[string(processor.Type)]; ok {
                panic(fmt.Sprintf("processor already registered: %s", processor.Type))
        }
        ProcessorRegistry[string(processor.Type)] = processor
}

var (
        processorRegistryMu sync.RWMutex
        ProcessorRegistry   = make(map[string]ProcessorRegistration)
)

package processor

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type RouterInput struct {
        config   config.ProcessorConfig
        SourceId string
        logger   *slog.Logger
}

func (ro *RouterInput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        payload := wrappedPayload.Payload
        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
        if !ok {

                wrappedPayload.End = true
                return wrappedPayload, errors.New("router.input no router found")
        }

        _, err := router.HandleInput(ctx, ro.SourceId, payload)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("router.input failed to send input")
        }

        wrappedPayload.Payload = payload

        return wrappedPayload, nil
}

func (ro *RouterInput) Type() string {
        return ro.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "router.input",
                Title: "Router Input",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "source": {
                                        Title:       "Source",
                                        Type:        "string",
                                        Description: "source to report as to the router",
                                },
                        },
                        Required:             []string{"source"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        sourceId, err := params.GetString("source")

                        if err != nil {
                                return nil, fmt.Errorf("router.input source error: %w", err)
                        }

                        return &RouterInput{config: config, SourceId: sourceId, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "log/slog"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type RouterOutput struct {
        config   config.ProcessorConfig
        ModuleId string
        logger   *slog.Logger
}

func (ro *RouterOutput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("router.output no router found")
        }

        err := router.HandleOutput(ctx, ro.ModuleId, wrappedPayload.Payload)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("router.output failed to send output: %w", err)
        }

        return wrappedPayload, nil
}

func (ro *RouterOutput) Type() string {
        return ro.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "router.output",
                Title: "Router Output",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "module": {
                                        Title:       "Module ID",
                                        Type:        "string",
                                        Description: "ID of module to send output to",
                                },
                        },
                        Required:             []string{"module"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {

                        params := config.Params

                        moduleId, err := params.GetString("module")

                        if err != nil {
                                return nil, fmt.Errorf("router.output module error: %w", err)
                        }

                        return &RouterOutput{config: config, ModuleId: moduleId, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"

        "github.com/expr-lang/expr"
        "github.com/expr-lang/expr/vm"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

// NOTE(jwetzell): see language definition https://expr-lang.org/docs/language-definition
type ScriptExpr struct {
        config  config.ProcessorConfig
        Program *vm.Program
}

func (se *ScriptExpr) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        exprEnv := wrappedPayload

        output, err := expr.Run(se.Program, exprEnv)
        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        wrappedPayload.Payload = output
        return wrappedPayload, nil
}

func (se *ScriptExpr) Type() string {
        return se.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "script.expr",
                Title: "Evaluate Expr Expression",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "expression": {
                                        Title: "Expression",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"expression"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        expressionString, err := params.GetString("expression")
                        if err != nil {
                                return nil, fmt.Errorf("script.expr expression error: %w", err)
                        }

                        program, err := expr.Compile(expressionString)
                        if err != nil {
                                return nil, err
                        }

                        return &ScriptExpr{config: config, Program: program}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "modernc.org/quickjs"
)

type ScriptJS struct {
        config      config.ProcessorConfig
        vm          *quickjs.VM
        payloadAtom quickjs.Atom
        senderAtom  quickjs.Atom
        Program     string
}

func (sj *ScriptJS) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        //NOTE(jwetzell): some weird conversion going on with these types
        _, isUint8Slice := common.GetAnyAs[[]uint8](wrappedPayload.Payload)
        _, isByteSlice := common.GetAnyAs[[]byte](wrappedPayload.Payload)

        if isUint8Slice || isByteSlice {
                intSlice, ok := common.GetAnyAsIntSlice(wrappedPayload.Payload)

                if ok {
                        wrappedPayload.Payload = intSlice
                }
        }

        sj.vm.SetProperty(sj.vm.GlobalObject(), sj.payloadAtom, wrappedPayload.Payload)

        sj.vm.SetProperty(sj.vm.GlobalObject(), sj.senderAtom, wrappedPayload.Sender)

        _, err := sj.vm.Eval(sj.Program, quickjs.EvalGlobal)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        output, err := sj.vm.GetProperty(sj.vm.GlobalObject(), sj.payloadAtom)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        // NOTE(jwetzell): turn undefined into nil
        _, ok := output.(quickjs.Undefined)

        if ok {
                wrappedPayload.End = true
                wrappedPayload.Payload = nil
                return wrappedPayload, nil
        }

        // NOTE(jwetzell): turn object into map[string]interface{}
        outputObject, ok := output.(*quickjs.Object)

        if ok {
                var outputSlice []interface{}

                err = outputObject.Into(&outputSlice)

                if err != nil {
                        var outputMap map[string]interface{}
                        err = outputObject.Into(&outputMap)
                        if err != nil {
                                wrappedPayload.End = true
                                return wrappedPayload, err
                        } else {
                                wrappedPayload.Payload = outputMap
                                return wrappedPayload, nil
                        }

                } else {
                        wrappedPayload.Payload = outputSlice
                        return wrappedPayload, nil
                }
        }

        wrappedPayload.Payload = output
        return wrappedPayload, nil
}

func (sj *ScriptJS) Type() string {
        return sj.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "script.js",
                Title: "Run JavaScript",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "program": {
                                        Title: "Program",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"program"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        programString, err := params.GetString("program")
                        if err != nil {
                                return nil, fmt.Errorf("script.js program error: %w", err)
                        }

                        vm, err := quickjs.NewVM()

                        if err != nil {
                                return nil, err
                        }

                        payloadAtom, err := vm.NewAtom("payload")
                        if err != nil {
                                return nil, err
                        }

                        senderAtom, err := vm.NewAtom("sender")
                        if err != nil {
                                return nil, err
                        }

                        return &ScriptJS{config: config, Program: programString, vm: vm, payloadAtom: payloadAtom, senderAtom: senderAtom}, nil
                },
        })
}

package processor

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"

        extism "github.com/extism/go-sdk"
        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type ScriptWASM struct {
        config   config.ProcessorConfig
        Program  *extism.CompiledPlugin
        Function string
}

func (sw *ScriptWASM) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("script.wasm can only process a byte array")
        }

        program, err := sw.Program.Instance(ctx, extism.PluginInstanceConfig{})

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        _, output, err := program.Call(sw.Function, payloadBytes)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }
        wrappedPayload.Payload = output

        return wrappedPayload, nil
}

func (sw *ScriptWASM) Type() string {
        return sw.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "script.wasm",
                Title: "Run WASM Plugin",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "path": {
                                        Title: "Path",
                                        Type:  "string",
                                },
                                "function": {
                                        Title:   "Function",
                                        Type:    "string",
                                        Default: json.RawMessage(`"process"`),
                                },
                                "enableWasi": {
                                        Title:   "Enable WASI",
                                        Type:    "boolean",
                                        Default: json.RawMessage("false"),
                                },
                        },
                        Required:             []string{"path"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(processorConfig config.ProcessorConfig) (Processor, error) {
                        params := processorConfig.Params

                        pathString, err := params.GetString("path")
                        if err != nil {
                                return nil, fmt.Errorf("script.wasm path error: %w", err)
                        }

                        functionString, err := params.GetString("function")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        functionString = "process"
                                } else {
                                        return nil, fmt.Errorf("script.wasm function error: %w", err)
                                }
                        }

                        enableWasiBool, err := params.GetBool("enableWasi")
                        if err != nil {
                                if errors.Is(err, config.ErrParamNotFound) {
                                        enableWasiBool = false
                                } else {
                                        return nil, fmt.Errorf("script.wasm enableWasi error: %w", err)
                                }
                        }

                        manifest := extism.Manifest{
                                Wasm: []extism.Wasm{
                                        extism.WasmFile{
                                                Path: pathString,
                                        },
                                },
                        }

                        program, err := extism.NewCompiledPlugin(context.Background(), manifest, extism.PluginConfig{
                                EnableWasi: enableWasiBool,
                        }, []extism.HostFunction{})

                        if err != nil {
                                return nil, err
                        }

                        return &ScriptWASM{config: processorConfig, Program: program, Function: functionString}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type SipResponseAudioCreate struct {
        config    config.ProcessorConfig
        PreWait   int
        PostWait  int
        AudioFile *template.Template
}

type SipAudioFileResponse struct {
        PreWait   int
        PostWait  int
        AudioFile string
}

func (srac *SipResponseAudioCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var audioFileBuffer bytes.Buffer
        err := srac.AudioFile.Execute(&audioFileBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        audioFileString := audioFileBuffer.String()

        wrappedPayload.Payload = SipAudioFileResponse{
                PreWait:   srac.PreWait,
                PostWait:  srac.PostWait,
                AudioFile: audioFileString,
        }
        return wrappedPayload, nil
}

func (srac *SipResponseAudioCreate) Type() string {
        return srac.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "sip.response.audio.create",
                Title: "Create SIP Audio Response",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "preWait": {
                                        Title: "Pre Wait (ms)",
                                        Type:  "integer",
                                },
                                "postWait": {
                                        Title: "Post Wait (ms)",
                                        Type:  "integer",
                                },
                                "audioFile": {
                                        Type: "string",
                                },
                        },
                        Required:             []string{"preWait", "postWait", "audioFile"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        preWaitNum, err := params.GetInt("preWait")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.audio.create preWait error: %w", err)
                        }

                        postWaitNum, err := params.GetInt("postWait")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.audio.create postWait error: %w", err)
                        }

                        audioFileString, err := params.GetString("audioFile")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.audio.create audioFile error: %w", err)
                        }

                        audioFileTemplate, err := template.New("audioFile").Parse(audioFileString)

                        if err != nil {
                                return nil, err
                        }
                        return &SipResponseAudioCreate{config: config, AudioFile: audioFileTemplate, PreWait: int(preWaitNum), PostWait: int(postWaitNum)}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "errors"
        "fmt"
        "regexp"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type SipResponseDTMFCreate struct {
        config    config.ProcessorConfig
        PreWait   int
        PostWait  int
        Digits    *template.Template
        validDTMF *regexp.Regexp
}

type SipDTMFResponse struct {
        PreWait  int
        PostWait int
        Digits   string
}

func (srdc *SipResponseDTMFCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {

        templateData := wrappedPayload

        var digitsBuffer bytes.Buffer
        err := srdc.Digits.Execute(&digitsBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        digitsString := digitsBuffer.String()

        if !srdc.validDTMF.MatchString(digitsString) {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("sip.response.dtmf.create result of digits template contains invalid characters")
        }

        wrappedPayload.Payload = SipDTMFResponse{
                PreWait:  srdc.PreWait,
                PostWait: srdc.PostWait,
                Digits:   digitsString,
        }
        return wrappedPayload, nil
}

func (srdc *SipResponseDTMFCreate) Type() string {
        return srdc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "sip.response.dtmf.create",
                Title: "Create SIP DTMF Response",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "preWait": {
                                        Title: "Pre Wait (ms)",
                                        Type:  "integer",
                                },
                                "postWait": {
                                        Title: "Post Wait (ms)",
                                        Type:  "integer",
                                },
                                "digits": {
                                        Type: "string",
                                },
                        },
                        Required:             []string{"preWait", "postWait", "digits"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        preWaitNum, err := params.GetInt("preWait")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.dtmf.create preWait error: %w", err)
                        }

                        postWaitNum, err := params.GetInt("postWait")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.dtmf.create postWait error: %w", err)
                        }

                        digitsString, err := params.GetString("digits")
                        if err != nil {
                                return nil, fmt.Errorf("sip.response.dtmf.create digits error: %w", err)
                        }

                        digitsTemplate, err := template.New("digits").Parse(digitsString)

                        if err != nil {
                                return nil, err
                        }
                        return &SipResponseDTMFCreate{config: config, Digits: digitsTemplate, PreWait: int(preWaitNum), PostWait: int(postWaitNum), validDTMF: regexp.MustCompile(`^[0-9*#A-Da-d]+$`)}, nil
                },
        })
}

package processor

import (
        "bytes"
        "context"
        "fmt"
        "text/template"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StringCreate struct {
        config   config.ProcessorConfig
        Template *template.Template
}

func (sc *StringCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        templateData := wrappedPayload

        var templateBuffer bytes.Buffer
        err := sc.Template.Execute(&templateBuffer, templateData)

        if err != nil {
                wrappedPayload.End = true
                return wrappedPayload, err
        }

        wrappedPayload.Payload = templateBuffer.String()

        return wrappedPayload, nil
}

func (sc *StringCreate) Type() string {
        return sc.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "string.create",
                Title: "Create String",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "template": {
                                        Title: "Template",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"template"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params
                        templateString, err := params.GetString("template")
                        if err != nil {
                                return nil, fmt.Errorf("string.create template error: %w", err)
                        }

                        templateTemplate, err := template.New("template").Parse(templateString)

                        if err != nil {
                                return nil, err
                        }
                        return &StringCreate{config: config, Template: templateTemplate}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StringDecode struct {
        config config.ProcessorConfig
}

func (sd *StringDecode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadBytes, ok := common.GetAnyAsByteSlice(payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("string.decode processor only accepts a []byte")
        }

        payloadMessage := string(payloadBytes)

        wrappedPayload.Payload = payloadMessage
        return wrappedPayload, nil
}

func (sd *StringDecode) Type() string {
        return sd.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "string.decode",
                Title: "Decode String",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &StringDecode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StringEncode struct {
        config config.ProcessorConfig
}

func (se *StringEncode) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString, ok := common.GetAnyAs[string](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("string.encode processor only accepts a string")
        }

        wrappedPayload.Payload = []byte(payloadString)

        return wrappedPayload, nil
}

func (se *StringEncode) Type() string {
        return se.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "string.encode",
                Title: "Encode String",
                New: func(config config.ProcessorConfig) (Processor, error) {
                        return &StringEncode{config: config}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "strings"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StringSplit struct {
        config    config.ProcessorConfig
        Separator string
}

func (ss *StringSplit) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        payloadString, ok := common.GetAnyAs[string](payload)

        if !ok {
                wrappedPayload.End = true
                return wrappedPayload, errors.New("string.split only accepts a string")
        }

        wrappedPayload.Payload = strings.Split(payloadString, ss.Separator)

        return wrappedPayload, nil
}

func (ss *StringSplit) Type() string {
        return ss.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "string.split",
                Title: "Split String",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "separator": {
                                        Title: "Separator",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"separator"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        separatorString, err := params.GetString("separator")
                        if err != nil {
                                return nil, fmt.Errorf("string.split separator error: %w", err)
                        }

                        return &StringSplit{config: config, Separator: separatorString}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "reflect"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StructFieldGet struct {
        config config.ProcessorConfig
        Name   string
}

func (sf *StructFieldGet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        s := reflect.ValueOf(payload)

        if s.Kind() != reflect.Struct {
                if s.Kind() == reflect.Pointer && s.Elem().Kind() == reflect.Struct {
                        s = s.Elem()
                } else {
                        wrappedPayload.End = true
                        return wrappedPayload, errors.New("struct.field.get processor only accepts a struct payload")
                }
        }

        field := s.FieldByName(sf.Name)
        if !field.IsValid() {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("struct.field.get field '%s' does not exist", sf.Name)
        }

        wrappedPayload.Payload = field.Interface()
        return wrappedPayload, nil
}

func (sf *StructFieldGet) Type() string {
        return sf.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "struct.field.get",
                Title: "Get Struct Field",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "name": {
                                        Title: "Field Name",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"name"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params
                        nameString, err := params.GetString("name")
                        if err != nil {
                                return nil, fmt.Errorf("struct.field.get name error: %w", err)
                        }

                        return &StructFieldGet{config: config, Name: nameString}, nil
                },
        })
}

package processor

import (
        "context"
        "errors"
        "fmt"
        "reflect"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type StructMethodGet struct {
        config config.ProcessorConfig
        Name   string
}

func (sm *StructMethodGet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        payload := wrappedPayload.Payload
        s := reflect.ValueOf(payload)

        if s.Kind() != reflect.Struct {
                if s.Kind() == reflect.Pointer && s.Elem().Kind() == reflect.Struct {
                        s = s.Elem()
                } else {
                        wrappedPayload.End = true
                        return wrappedPayload, errors.New("struct.method.get processor only accepts a struct payload")
                }
        }

        method := s.MethodByName(sm.Name)
        if !method.IsValid() {
                wrappedPayload.End = true
                return wrappedPayload, fmt.Errorf("struct.method.get method '%s' does not exist", sm.Name)
        }

        value := method.Call(nil)

        if len(value) == 0 {
                wrappedPayload.End = true
                wrappedPayload.Payload = nil
                return wrappedPayload, nil
        }

        if len(value) == 1 {
                wrappedPayload.Payload = value[0].Interface()
                return wrappedPayload, nil
        }

        results := make([]any, len(value))

        for i, v := range value {
                results[i] = v.Interface()
        }

        wrappedPayload.Payload = results
        return wrappedPayload, nil
}

func (sm *StructMethodGet) Type() string {
        return sm.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "struct.method.get",
                Title: "Get Struct Method",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "name": {
                                        Title: "Method Name",
                                        Type:  "string",
                                },
                        },
                        Required:             []string{"name"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params
                        nameString, err := params.GetString("name")
                        if err != nil {
                                return nil, fmt.Errorf("struct.method.get name error: %w", err)
                        }

                        return &StructMethodGet{config: config, Name: nameString}, nil
                },
        })
}

package processor

import (
        "context"
        "fmt"
        "time"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
)

type TimeSleep struct {
        config   config.ProcessorConfig
        Duration time.Duration
}

func (ts *TimeSleep) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        time.Sleep(ts.Duration)
        return wrappedPayload, nil
}

func (ts *TimeSleep) Type() string {
        return ts.config.Type
}

func init() {
        RegisterProcessor(ProcessorRegistration{
                Type:  "time.sleep",
                Title: "Sleep",
                ParamsSchema: &jsonschema.Schema{
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "duration": {
                                        Title:       "Duration",
                                        Type:        "integer",
                                        Description: "Duration to sleep in milliseconds",
                                },
                        },
                        Required:             []string{"duration"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                },
                New: func(config config.ProcessorConfig) (Processor, error) {
                        params := config.Params

                        durationNum, err := params.GetInt("duration")
                        if err != nil {
                                return nil, fmt.Errorf("time.sleep duration error: %w", err)
                        }

                        return &TimeSleep{config: config, Duration: time.Millisecond * time.Duration(durationNum)}, nil
                },
        })
}

package route

import (
        "context"
        "fmt"

        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/processor"
        "go.opentelemetry.io/otel"
        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/codes"
        "go.opentelemetry.io/otel/trace"
)

type RouteError struct {
        Index  int                `json:"index"`
        Config config.RouteConfig `json:"config"`
        Error  string             `json:"error"`
}
type Route struct {
        input      string
        processors []processor.Processor
}

func NewRoute(config config.RouteConfig) (*Route, error) {
        processors := []processor.Processor{}

        if len(config.Processors) > 0 {
                for _, processorDecl := range config.Processors {
                        processorInfo, ok := processor.ProcessorRegistry[processorDecl.Type]
                        if !ok {
                                return nil, fmt.Errorf("problem loading processor registration for processor type: %s", processorDecl.Type)
                        }

                        processor, err := processorInfo.New(processorDecl)
                        if err != nil {
                                return nil, err
                        }
                        processors = append(processors, processor)
                }
        }

        return &Route{input: config.Input, processors: processors}, nil
}

func (r *Route) Input() string {
        return r.input
}

func (r *Route) ProcessPayload(ctx context.Context, payload any) (any, error) {
        wrappedPayload := common.GetWrappedPayload(ctx, payload)
        tracer := otel.Tracer("route")
        processCtx, processSpan := tracer.Start(ctx, "ProcessPayload", trace.WithAttributes(attribute.String("payload.type", fmt.Sprintf("%T", payload))))
        defer processSpan.End()
        for processorIndex, processor := range r.processors {
                processorCtx, processorSpan := otel.Tracer("processor").Start(processCtx, "process", trace.WithAttributes(attribute.Int("processor.index", processorIndex), attribute.String("processor.type", processor.Type())))
                processedPayload, err := processor.Process(processorCtx, wrappedPayload)
                if err != nil {
                        processorSpan.SetStatus(codes.Error, "route processor error")
                        processorSpan.RecordError(err)
                        processorSpan.End()
                        processSpan.SetStatus(codes.Error, "route processing error")
                        processSpan.RecordError(err)
                        return nil, err
                }
                processorSpan.SetStatus(codes.Ok, "processor successful")
                //NOTE(jwetzell) payload has been marked as an end
                if processedPayload.End {
                        processSpan.SetStatus(codes.Ok, "route processing terminated early due to processor signal")
                        processorSpan.End()
                        return processedPayload.Payload, nil
                }
                wrappedPayload = processedPayload
                processorSpan.End()
        }
        processSpan.SetStatus(codes.Ok, "route processing successful")

        return wrappedPayload.Payload, nil
}

package schema

import (
        "github.com/google/jsonschema-go/jsonschema"
)

var ConfigSchema = jsonschema.Schema{
        Schema:      "https://json-schema.org/draft/2020-12/schema",
        ID:          "https://showbridge.io/config.schema.json",
        Title:       "Config",
        Description: "showbridge configuration",
        Type:        "object",
        Properties: map[string]*jsonschema.Schema{
                "api": &ApiConfigSchema,
                "modules": {
                        Ref: "https://showbridge.io/modules.schema.json",
                },
                "routes": {
                        Ref: "https://showbridge.io/routes.schema.json",
                },
        },
        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
}

func ApplyDefaults(cfg *map[string]any) error {
        resolvedSchema, err := GetResolvedConfigSchema()
        if err != nil {
                return err
        }

        return resolvedSchema.ApplyDefaults(cfg)
}

func ValidateConfig(cfg map[string]any) error {
        resolvedSchema, err := GetResolvedConfigSchema()
        if err != nil {
                return err
        }
        return resolvedSchema.Validate(cfg)
}

package schema

import (
        "encoding/json"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/module"
)

func GetModulesSchema() *jsonschema.Schema {

        schema := &jsonschema.Schema{
                Schema:      "https://json-schema.org/draft/2020-12/schema",
                ID:          "https://showbridge.io/modules.schema.json",
                Title:       "Modules",
                Description: "module configurations",
                Type:        "array",
                Default:     json.RawMessage(`[]`),
        }

        moduleDefinitionSchemas := []*jsonschema.Schema{}
        for _, mod := range module.ModuleRegistry {
                moduleSchema := &jsonschema.Schema{
                        ID:   mod.Type,
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "id": {
                                        Type:      "string",
                                        MinLength: jsonschema.Ptr(1),
                                },
                                "type": {
                                        Const: jsonschema.Ptr[any](mod.Type),
                                },
                        },
                        Required:             []string{"id", "type"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                }
                if mod.Title != "" {
                        moduleSchema.Title = mod.Title
                }
                if mod.Description != "" {
                        moduleSchema.Description = mod.Description
                }
                if mod.ParamsSchema != nil {
                        moduleSchema.Properties["params"] = mod.ParamsSchema
                        moduleSchema.Required = append(moduleSchema.Required, "params")
                }
                moduleDefinitionSchemas = append(moduleDefinitionSchemas, moduleSchema)
        }
        schema.Items = &jsonschema.Schema{
                OneOf: moduleDefinitionSchemas,
        }
        return schema
}

package schema

import (
        "encoding/json"

        "github.com/google/jsonschema-go/jsonschema"
        "github.com/jwetzell/showbridge-go/internal/processor"
)

func GetProcessorsSchema() *jsonschema.Schema {

        schema := &jsonschema.Schema{
                Schema:      "https://json-schema.org/draft/2020-12/schema",
                ID:          "https://showbridge.io/processors.schema.json",
                Title:       "Processors",
                Description: "processor configurations",
                Type:        "array",
                Default:     json.RawMessage(`[]`),
        }

        processorDefinitionSchemas := []*jsonschema.Schema{}
        for _, proc := range processor.ProcessorRegistry {
                processorSchema := &jsonschema.Schema{
                        ID:   proc.Type,
                        Type: "object",
                        Properties: map[string]*jsonschema.Schema{
                                "type": {
                                        Const: jsonschema.Ptr[any](proc.Type),
                                },
                        },
                        Required:             []string{"type"},
                        AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
                }
                if proc.Title != "" {
                        processorSchema.Title = proc.Title
                }
                if proc.Description != "" {
                        processorSchema.Description = proc.Description
                }
                if proc.ParamsSchema != nil {
                        processorSchema.Properties["params"] = proc.ParamsSchema
                        processorSchema.Required = append(processorSchema.Required, "params")
                }
                processorDefinitionSchemas = append(processorDefinitionSchemas, processorSchema)
        }
        schema.Items = &jsonschema.Schema{
                OneOf: processorDefinitionSchemas,
        }
        return schema
}

package schema

import (
        "fmt"
        "net/url"

        "github.com/google/jsonschema-go/jsonschema"
)

func GetResolvedConfigSchema() (*jsonschema.Resolved, error) {
        return ConfigSchema.Resolve(&jsonschema.ResolveOptions{
                Loader: func(uri *url.URL) (*jsonschema.Schema, error) {
                        switch uri.String() {
                        case "https://showbridge.io/modules.schema.json":
                                return GetModulesSchema(), nil
                        case "https://showbridge.io/processors.schema.json":
                                return GetProcessorsSchema(), nil
                        case "https://showbridge.io/routes.schema.json":
                                return &RoutesConfigSchema, nil
                        default:
                                return nil, fmt.Errorf("unknown schema reference: %s", uri.String())
                        }
                },
                ValidateDefaults: true,
        })
}

package test

import (
        "context"

        "github.com/jwetzell/showbridge-go/internal/common"
)

func GetContextWithModules(ctx context.Context, modules map[string]common.Module) context.Context {
        ctx = context.WithValue(ctx, common.ModulesContextKey, modules)
        return ctx
}

func GetContextWithRouter(ctx context.Context) context.Context {
        ctx = context.WithValue(ctx, common.RouterContextKey, GetNewTestRouter())
        return ctx
}

func GetContextWithSender(ctx context.Context, sender any) context.Context {
        ctx = context.WithValue(ctx, common.SenderContextKey, sender)
        return ctx
}

func GetContextWithSource(ctx context.Context, source string) context.Context {
        ctx = context.WithValue(ctx, common.SourceContextKey, source)
        return ctx
}

package test

import (
        "context"
        "database/sql"

        _ "modernc.org/sqlite"
)

type TestModule struct {
}

func (m *TestModule) Start(ctx context.Context) error {
        <-ctx.Done()
        return nil
}

func (m *TestModule) Stop() {}

func (m *TestModule) Type() string {
        return "test.plain"
}

func (m *TestModule) Id() string {
        return "test"
}

func NewTestKVModule(id string) *TestKVModule {
        return &TestKVModule{
                id: id,
        }
}

type TestKVModule struct {
        id     string
        kvData map[string]any
}

func (m *TestKVModule) Start(ctx context.Context) error {
        <-ctx.Done()
        return nil
}

func (m *TestKVModule) Stop() {}

func (m *TestKVModule) Type() string {
        return "test.kv"
}

func (m *TestKVModule) Id() string {
        return m.id
}

func (m *TestKVModule) Get(key string) (any, error) {
        return key, nil
}

func (m *TestKVModule) Set(key string, value any) error {
        if m.kvData == nil {
                m.kvData = make(map[string]any)
        }
        m.kvData[key] = value
        return nil
}
func NewTestDBModule(id string) *TestDBModule {
        return &TestDBModule{
                id: id,
        }
}

type TestDBModule struct {
        id string
        db *sql.DB
}

func (m *TestDBModule) Start(ctx context.Context) error {
        <-ctx.Done()
        return nil
}

func (m *TestDBModule) Database() *sql.DB {
        if m.db == nil {
                db, _ := sql.Open("sqlite", ":memory:")

                db.Exec(`
                CREATE TABLE test (
                        id INTEGER PRIMARY KEY,
                        value TEXT
                );
                INSERT INTO test (id, value) VALUES (1, 'test-1'), (2, 'test-2');
                
                `)
                m.db = db
        }
        return m.db
}

func (m *TestDBModule) Stop() {}

func (m *TestDBModule) Type() string {
        return "test.db"
}

func (m *TestDBModule) Id() string {
        return m.id
}

package test

import (
        "context"

        "github.com/jwetzell/showbridge-go/internal/common"
)

type TestProcessor struct {
}

func (p *TestProcessor) Type() string {
        return "test"
}
func (p *TestProcessor) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
        return wrappedPayload, nil
}

package test

import (
        "context"

        "github.com/jwetzell/showbridge-go/internal/common"
)

type TestRouter struct {
}

func (r *TestRouter) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []common.RouteIOError) {
        return false, nil
}

func (r *TestRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error {
        return nil
}

func GetNewTestRouter() *TestRouter {
        return &TestRouter{}
}

package test

type TestStruct struct {
        String   string
        Int      int
        Float    float64
        Bool     bool
        Data     any
        IntSlice []int
}

func (t TestStruct) GetString() string {
        return t.String
}

func (t TestStruct) GetInt() int {
        return t.Int
}

func (t TestStruct) GetFloat() float64 {
        return t.Float
}

func (t TestStruct) GetBool() bool {
        return t.Bool
}

func (t TestStruct) GetData() any {
        return t.Data
}

func (t TestStruct) GetIntSlice() []int {
        return t.IntSlice
}

func (t TestStruct) Void() {}

func (t TestStruct) MultipleReturnValues() (string, int) {
        return t.String, t.Int
}

package showbridge

import (
        "context"
        "errors"
        "log/slog"
        "net/http"
        "reflect"
        "sync"

        "github.com/gorilla/websocket"
        "github.com/jwetzell/showbridge-go/internal/common"
        "github.com/jwetzell/showbridge-go/internal/config"
        "github.com/jwetzell/showbridge-go/internal/module"
        "github.com/jwetzell/showbridge-go/internal/route"

        "go.opentelemetry.io/otel"
        "go.opentelemetry.io/otel/attribute"
        "go.opentelemetry.io/otel/codes"
        "go.opentelemetry.io/otel/trace"
)

// TODO(jwetzell): can/should this be split into different components?
type Router struct {
        contextCancel context.CancelFunc
        Context       context.Context
        // TODO(jwetzell): do these need to be guarded against concurrency?
        ModuleInstances map[string]common.Module
        // TODO(jwetzell): change to something easier to lookup
        RouteInstances    []*route.Route
        ConfigChange      chan config.Config
        moduleWait        sync.WaitGroup
        logger            *slog.Logger
        runningConfig     config.Config
        runningConfigMu   sync.RWMutex
        wsConns           []*websocket.Conn
        wsConnsMu         sync.Mutex
        apiServer         *http.Server
        apiServerMu       sync.Mutex
        apiServerShutdown context.CancelFunc
        updatingConfig    bool
}

func (r *Router) addModule(moduleDecl config.ModuleConfig) error {
        if moduleDecl.Id == "" {
                return errors.New("module id cannot be empty")
        }
        moduleInfo, ok := module.ModuleRegistry[moduleDecl.Type]
        if !ok {
                return errors.New("module type not defined")
        }

        _, ok = r.ModuleInstances[moduleDecl.Id]
        if ok {
                return errors.New("module id already exists")
        }

        moduleInstance, err := moduleInfo.New(moduleDecl)
        if err != nil {
                return err
        }

        r.ModuleInstances[moduleDecl.Id] = moduleInstance
        return nil
}

func (r *Router) removeModule(moduleId string) error {
        err := r.stopModule(moduleId)
        if err != nil {
                return err
        }
        delete(r.ModuleInstances, moduleId)
        return nil
}

func (r *Router) startModule(ctx context.Context, moduleId string) error {
        moduleInstance := r.getModule(moduleId)
        if moduleInstance == nil {
                return errors.New("module id not found")
        }
        r.moduleWait.Go(func() {
                err := moduleInstance.Start(ctx)
                if err != nil {
                        // TODO(jwetzell): propagate module run errors better
                        r.logger.Error("error encountered running module", "moduleId", moduleId, "error", err)
                }
        })
        return nil
}

func (r *Router) stopModule(moduleId string) error {
        moduleInstance := r.getModule(moduleId)
        if moduleInstance == nil {
                return errors.New("module id not found")
        }
        moduleInstance.Stop()
        return nil
}

// TODO(jwetzell): support removing route
func (r *Router) addRoute(routeDecl config.RouteConfig) error {
        routeInstance, err := route.NewRoute(routeDecl)
        if err != nil {
                return err
        }
        r.RouteInstances = append(r.RouteInstances, routeInstance)
        return nil
}

func (r *Router) getModule(moduleId string) common.Module {
        moduleInstance, ok := r.ModuleInstances[moduleId]
        if !ok {
                return nil
        }
        return moduleInstance
}

func NewRouter(routerConfig config.Config) (*Router, []module.ModuleError, []route.RouteError) {

        router := Router{
                ModuleInstances: make(map[string]common.Module),
                RouteInstances:  []*route.Route{},
                ConfigChange:    make(chan config.Config, 1),
                logger:          slog.Default().With("component", "router"),
                runningConfig:   routerConfig,
                updatingConfig:  false,
        }
        router.logger.Debug("creating")

        var moduleErrors []module.ModuleError

        for moduleIndex, moduleDecl := range routerConfig.Modules {

                err := router.addModule(moduleDecl)
                if err != nil {
                        if moduleErrors == nil {
                                moduleErrors = []module.ModuleError{}
                        }
                        moduleErrors = append(moduleErrors, module.ModuleError{
                                Index:  moduleIndex,
                                Config: moduleDecl,
                                Error:  err.Error(),
                        })
                        continue
                }

        }

        var routeErrors []route.RouteError
        for routeIndex, routeDecl := range routerConfig.Routes {
                err := router.addRoute(routeDecl)
                if err != nil {
                        if routeErrors == nil {
                                routeErrors = []route.RouteError{}
                        }
                        routeErrors = append(routeErrors, route.RouteError{
                                Index:  routeIndex,
                                Config: routeDecl,
                                Error:  err.Error(),
                        })
                        continue
                }
        }

        return &router, moduleErrors, routeErrors
}

func (r *Router) Start(ctx context.Context) {
        r.logger.Info("running")
        routerContext, cancel := context.WithCancel(ctx)
        r.Context = routerContext
        r.contextCancel = cancel
        r.startModules()
        r.startAPIServer(r.runningConfig.Api)
        <-r.Context.Done()
        r.logger.Debug("shutting down api server")
        r.stopAPIServer()
        r.logger.Debug("waiting for modules to exit")
        r.moduleWait.Wait()
        r.logger.Info("done")
}

func (r *Router) Stop() {
        r.logger.Info("stopping")
        r.contextCancel()
}

func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []common.RouteIOError) {
        r.runningConfigMu.RLock()
        defer r.runningConfigMu.RUnlock()

        spanCtx, span := otel.Tracer("router").Start(ctx, "input", trace.WithAttributes(attribute.String("source.id", sourceId)))
        defer span.End()
        var routeIOErrors []common.RouteIOError
        routeFound := false

        r.broadcastEvent(Event{
                Type: "input",
                Data: map[string]any{
                        "source": sourceId,
                },
        })

        var routeWaitGroup sync.WaitGroup

        for routeIndex, routeInstance := range r.RouteInstances {
                if routeInstance == nil {
                        r.logger.Error("nil route instance found", "routeIndex", routeIndex)
                        continue
                }
                if routeInstance.Input() == sourceId {
                        routeWaitGroup.Go(func() {

                                routeFound = true
                                routeContext := context.WithValue(spanCtx, common.SourceContextKey, sourceId)
                                routeContext = context.WithValue(routeContext, common.RouterContextKey, r)
                                routeContext = context.WithValue(routeContext, common.ModulesContextKey, r.ModuleInstances)

                                routeCtx, routeSpan := otel.Tracer("router").Start(routeContext, "route", trace.WithAttributes(attribute.Int("route.index", routeIndex), attribute.String("route.input", routeInstance.Input())))
                                _, err := routeInstance.ProcessPayload(routeCtx, payload)
                                if err != nil {
                                        if routeIOErrors == nil {
                                                routeIOErrors = []common.RouteIOError{}
                                        }
                                        r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
                                        routeIOErrors = append(routeIOErrors, common.RouteIOError{
                                                Index:        routeIndex,
                                                ProcessError: err,
                                        })
                                        r.broadcastEvent(Event{
                                                Type: "route",
                                                Data: map[string]any{
                                                        "index": routeIndex,
                                                },
                                                Error: err.Error(),
                                        })
                                        return
                                }
                                r.broadcastEvent(Event{
                                        Type: "route",
                                        Data: map[string]any{
                                                "index": routeIndex,
                                        },
                                })
                                routeSpan.End()
                        })
                }
        }
        routeWaitGroup.Wait()
        return routeFound, routeIOErrors
}

func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error {
        spanCtx, span := otel.Tracer("router").Start(ctx, "output", trace.WithAttributes(attribute.String("destination.id", destinationId)))
        defer span.End()
        outputEvent := Event{
                Type: "output",
                Data: map[string]any{
                        "destination": destinationId,
                },
        }
        destinationModule := r.getModule(destinationId)

        if destinationModule == nil {
                err := errors.New("no module found for destination id")
                span.SetStatus(codes.Error, err.Error())
                span.RecordError(err)
                r.logger.Error("no module found for destination id", "destinationId", destinationId)
                outputEvent.Error = err.Error()
                r.broadcastEvent(outputEvent)
                return err
        }

        outputModule, ok := destinationModule.(common.OutputModule)
        if !ok {
                err := errors.New("module does not support output")
                span.SetStatus(codes.Error, err.Error())
                span.RecordError(err)
                r.logger.Error("module does not support output", "destinationId", destinationId)
                outputEvent.Error = err.Error()
                r.broadcastEvent(outputEvent)
                return err
        }

        moduleOutputCtx, moduleOutputSpan := otel.Tracer("module").Start(spanCtx, "output", trace.WithAttributes(attribute.String("module.id", destinationModule.Id()), attribute.String("module.type", destinationModule.Type())))
        defer moduleOutputSpan.End()
        err := outputModule.Output(moduleOutputCtx, payload)
        if err != nil {
                moduleOutputSpan.SetStatus(codes.Error, err.Error())
                moduleOutputSpan.RecordError(err)
                r.logger.ErrorContext(moduleOutputCtx, "module output encountered error", "module", destinationModule.Id(), "error", err)
                outputEvent.Error = err.Error()
                r.broadcastEvent(outputEvent)
                return err
        } else {
                moduleOutputSpan.SetStatus(codes.Ok, "module output successful")
        }
        r.broadcastEvent(outputEvent)
        return nil
}

func (r *Router) startModules() {
        contextWithRouter := context.WithValue(r.Context, common.RouterContextKey, r)

        for moduleId := range r.ModuleInstances {
                // TODO(jwetzell): handle module run errors
                err := r.startModule(contextWithRouter, moduleId)
                if err != nil {
                        r.logger.Error("error starting module", "moduleId", moduleId, "error", err)
                }
        }
}

func (r *Router) RunningConfig() config.Config {
        r.runningConfigMu.RLock()
        defer r.runningConfigMu.RLock()
        return r.runningConfig
}

func (r *Router) UpdateConfig(newConfig config.Config) ([]module.ModuleError, []route.RouteError) {
        r.runningConfigMu.Lock()
        defer r.runningConfigMu.Unlock()
        r.updatingConfig = true
        defer func() {
                r.updatingConfig = false
        }()
        oldConfig := r.runningConfig
        r.logger.Debug("received config update", "oldConfig", oldConfig, "newConfig", newConfig)

        if !reflect.DeepEqual(oldConfig.Api, newConfig.Api) {
                r.logger.Info("applying new API config")
                r.stopAPIServer()
                r.startAPIServer(newConfig.Api)
                r.runningConfig.Api = newConfig.Api
        }

        // TODO(jwetzell): handle config update errors better
        for _, moduleInstance := range r.ModuleInstances {
                moduleInstance.Stop()
        }
        r.logger.Debug("waiting for modules to exit")
        r.moduleWait.Wait()

        r.ModuleInstances = make(map[string]common.Module)
        r.RouteInstances = []*route.Route{}

        var moduleErrors []module.ModuleError

        for moduleIndex, moduleDecl := range newConfig.Modules {

                err := r.addModule(moduleDecl)
                if err != nil {
                        if moduleErrors == nil {
                                moduleErrors = []module.ModuleError{}
                        }
                        moduleErrors = append(moduleErrors, module.ModuleError{
                                Index:  moduleIndex,
                                Config: moduleDecl,
                                Error:  err.Error(),
                        })
                        continue
                }

        }

        var routeErrors []route.RouteError
        for routeIndex, routeDecl := range newConfig.Routes {
                err := r.addRoute(routeDecl)
                if err != nil {
                        if routeErrors == nil {
                                routeErrors = []route.RouteError{}
                        }
                        routeErrors = append(routeErrors, route.RouteError{
                                Index:  routeIndex,
                                Config: routeDecl,
                                Error:  err.Error(),
                        })
                        continue
                }
        }
        r.runningConfig = newConfig
        r.startModules()

        return moduleErrors, routeErrors
}

package showbridge

import (
        "encoding/json"
        "net/http"

        "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
        CheckOrigin: func(r *http.Request) bool {
                return true
        },
}

func (r *Router) handleWebsocket(w http.ResponseWriter, req *http.Request) {
        conn, err := upgrader.Upgrade(w, req, nil)
        if err != nil {
                r.logger.Error("websocket upgrade error", "error", err)
                return
        }
        defer conn.Close()

        r.wsConnsMu.Lock()
        r.wsConns = append(r.wsConns, conn)
        r.wsConnsMu.Unlock()
READ_LOOP:
        for {
                messageType, message, err := conn.ReadMessage()
                if err != nil {
                        _, ok := err.(*websocket.CloseError)
                        if ok {
                                break READ_LOOP
                        }
                }

                switch messageType {
                case websocket.TextMessage, websocket.BinaryMessage:
                        event := Event{}
                        err = json.Unmarshal(message, &event)
                        if err != nil {
                                r.logger.Error("websocket message unmarshal error", "error", err)
                                continue
                        }
                        r.handleEvent(event, conn)
                case websocket.CloseMessage:
                        break READ_LOOP
                case websocket.PingMessage:
                        err = conn.WriteMessage(websocket.PongMessage, nil)
                        if err != nil {
                                r.logger.Error("websocket pong error", "error", err)
                        }
                default:
                        r.logger.Warn("unsupported websocket message type", "type", messageType)
                        continue
                }

        }
        //NOTE(jwetzell): remove ws connection
        r.wsConnsMu.Lock()
        for i, c := range r.wsConns {
                if c == conn {
                        r.wsConns = append(r.wsConns[:i], r.wsConns[i+1:]...)
                        break
                }
        }
        r.wsConnsMu.Unlock()
}