Merge pull request #239 from jwetzell/feat/mysql-module

add support for connecting to mysql databases

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/eclipse/paho.mqtt.golang v1.5.1
 	github.com/expr-lang/expr v1.17.8
 	github.com/extism/go-sdk v1.7.1
+	github.com/go-sql-driver/mysql v1.10.1
 	github.com/google/jsonschema-go v0.4.3
 	github.com/gorilla/websocket v1.5.3
 	github.com/jackc/pgx/v5 v5.10.0
@@ -26,6 +27,7 @@ require (
 )
 
 require (
+	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect

go.sum

@@ -1,3 +1,5 @@
+filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
+filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op h1:p2zFsAzvhIpFya8AIOHIbWf7NGvO34QpLGclyf7nXj8=
 github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
@@ -19,6 +21,8 @@ github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM
 github.com/expr-lang/expr v1.17.8/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
 github.com/extism/go-sdk v1.7.1 h1:lWJos6uY+tRFdlIHR+SJjwFDApY7OypS/2nMhiVQ9Sw=
 github.com/extism/go-sdk v1.7.1/go.mod h1:IT+Xdg5AZM9hVtpFUA+uZCJMge/hbvshl8bwzLtFyKA=
+github.com/go-sql-driver/mysql v1.10.1 h1:arlSnNLq6a5yxGxV7qg9lF4j0C+KwD6NbQyKr9QL6ME=
+github.com/go-sql-driver/mysql v1.10.1/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk=
 github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
 github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=

internal/module/db-mysql.go

@@ -0,0 +1,110 @@
+package module
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/google/jsonschema-go/jsonschema"
+	"github.com/jwetzell/showbridge-go/internal/common"
+	"github.com/jwetzell/showbridge-go/internal/config"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+func init() {
+	RegisterModule(ModuleRegistration{
+		Type:  "db.mysql",
+		Title: "MySQL Database",
+		ParamsSchema: &jsonschema.Schema{
+			Type: "object",
+			Properties: map[string]*jsonschema.Schema{
+				"dsn": {
+					Title:       "Database DSN",
+					Description: "the connection DSN for the MySQL database",
+					Type:        "string",
+					MinLength:   new(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.mysql dsn error: %w", err)
+			}
+
+			return &DbMysql{Dsn: dsnString, config: config, logger: CreateLogger(config)}, nil
+		},
+	})
+}
+
+type DbMysql struct {
+	config       config.ModuleConfig
+	Dsn          string
+	ctx          context.Context
+	inputHandler common.InputHandler
+	db           *sql.DB
+	logger       *slog.Logger
+	dbMu         sync.Mutex
+	cancel       context.CancelFunc
+}
+
+func (dbs *DbMysql) Id() string {
+	return dbs.config.Id
+}
+
+func (dbs *DbMysql) Type() string {
+	return dbs.config.Type
+}
+
+func (dbs *DbMysql) Start(ctx context.Context, inputHandler common.InputHandler) error {
+	dbs.logger.Debug("running")
+	dbs.inputHandler = inputHandler
+	moduleContext, cancel := context.WithCancel(ctx)
+	dbs.ctx = moduleContext
+	dbs.cancel = cancel
+
+	db, err := sql.Open("mysql", dbs.Dsn)
+	if err != nil {
+		return fmt.Errorf("db.mysql error connecting to database: %w", err)
+	}
+
+	// TODO(jwetzell): make configurable
+	db.SetConnMaxLifetime(time.Minute * 3)
+	db.SetMaxOpenConns(10)
+	db.SetMaxIdleConns(10)
+
+	dbs.dbMu.Lock()
+	dbs.db = db
+	dbs.dbMu.Unlock()
+	<-dbs.ctx.Done()
+	dbs.logger.Debug("done")
+	return nil
+}
+
+func (dbs *DbMysql) Stop() {
+	if dbs.cancel != nil {
+		defer dbs.cancel()
+	}
+	dbs.dbMu.Lock()
+	defer dbs.dbMu.Unlock()
+	if dbs.db != nil {
+		dbs.db.Close()
+	}
+}
+
+func (dbs *DbMysql) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	dbs.dbMu.Lock()
+	defer dbs.dbMu.Unlock()
+	if dbs.db == nil {
+		return nil, fmt.Errorf("database not initialized")
+	}
+	return dbs.db.QueryContext(ctx, query, args...)
+}

internal/module/test/db-mysql_test.go

@@ -0,0 +1,126 @@
+package module_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/module"
+)
+
+func TestDbMySQLFromRegistry(t *testing.T) {
+	registration, ok := module.GetModuleRegistration("db.mysql")
+	if !ok {
+		t.Fatalf("db.mysql module not registered")
+	}
+
+	moduleInstance, err := registration.New(config.ModuleConfig{
+		Id:   "test",
+		Type: "db.mysql",
+		Params: map[string]any{
+			"dsn": "mysql:mysql@tcp(127.0.0.1:3306)/test",
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("failed to create db.mysql module: %s", err)
+	}
+
+	if moduleInstance.Id() != "test" {
+		t.Fatalf("db.mysql module has wrong id: %s", moduleInstance.Id())
+	}
+
+	if moduleInstance.Type() != "db.mysql" {
+		t.Fatalf("db.mysql module has wrong type: %s", moduleInstance.Type())
+	}
+}
+
+func TestGoodDbMySQL(t *testing.T) {
+
+	testCases := []struct {
+		name   string
+		params map[string]any
+	}{}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+
+			registration, ok := module.GetModuleRegistration("db.mysql")
+			if !ok {
+				t.Fatalf("db.mysql module not registered")
+			}
+
+			moduleInstance, err := registration.New(config.ModuleConfig{
+				Id:     "test",
+				Type:   "db.mysql",
+				Params: test.params,
+			})
+
+			if err != nil {
+				t.Fatalf("db.mysql failed to create module: %s", err)
+			}
+			// TODO(jwetzell) this is kind of hacky
+			go func() {
+				time.Sleep(1 * time.Second)
+				moduleInstance.Stop()
+			}()
+			err = moduleInstance.Start(t.Context(), nil)
+
+			if err != nil {
+				t.Fatalf("db.mysql failed to start: %s", err)
+			}
+		})
+	}
+}
+
+func TestBadDbMySQL(t *testing.T) {
+	tests := []struct {
+		name        string
+		params      map[string]any
+		errorString string
+	}{
+		{
+			name:        "no dsn param",
+			params:      map[string]any{},
+			errorString: "db.mysql dsn error: not found",
+		},
+		{
+			name:        "non-string dsn",
+			params:      map[string]any{"dsn": 123},
+			errorString: "db.mysql dsn error: not a string",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			registration, ok := module.GetModuleRegistration("db.mysql")
+			if !ok {
+				t.Fatalf("db.mysql module not registered")
+			}
+
+			moduleInstance, err := registration.New(config.ModuleConfig{
+				Id:     "test",
+				Type:   "db.mysql",
+				Params: test.params,
+			})
+
+			if err != nil {
+				if test.errorString != err.Error() {
+					t.Fatalf("db.mysql got error '%s', expected '%s'", err.Error(), test.errorString)
+				}
+				return
+			}
+
+			err = moduleInstance.Start(t.Context(), nil)
+
+			if err == nil {
+				t.Fatalf("db.mysql expected to fail")
+			}
+
+			if err.Error() != test.errorString {
+				t.Fatalf("db.mysql got error '%s', expected '%s'", err.Error(), test.errorString)
+			}
+		})
+	}
+}