diff --git a/app/plugin.go b/app/plugin.go index bba8b1eaa0..0937c707c3 100644 --- a/app/plugin.go +++ b/app/plugin.go @@ -199,7 +199,7 @@ func (s *Server) initPlugins(c *request.Context, pluginDir, webappPluginDir stri return New(ServerConnector(s)).NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newApiFunc, pluginDir, webappPluginDir, s.Log, s.Metrics) + env, err := plugin.NewEnvironment(newApiFunc, NewDriverImpl(s), pluginDir, webappPluginDir, s.Log, s.Metrics) if err != nil { mlog.Error("Failed to start up plugins", mlog.Err(err)) return diff --git a/app/plugin_api_test.go b/app/plugin_api_test.go index 3a77edd8e2..4a81a6c83a 100644 --- a/app/plugin_api_test.go +++ b/app/plugin_api_test.go @@ -69,7 +69,7 @@ func setDefaultPluginConfig(th *TestHelper, pluginID string) { }) } -func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests []string, pluginIDs []string, app *App, c *request.Context) string { +func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests []string, pluginIDs []string, asMain bool, app *App, c *request.Context) string { pluginDir, err := ioutil.TempDir("", "") require.NoError(t, err) t.Cleanup(func() { @@ -92,7 +92,7 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests return app.NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newPluginAPI, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(newPluginAPI, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) require.Equal(t, len(pluginCodes), len(pluginIDs)) @@ -100,7 +100,11 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests for i, pluginID := range pluginIDs { backend := filepath.Join(pluginDir, pluginID, "backend.exe") - utils.CompileGo(t, pluginCodes[i], backend) + if asMain { + utils.CompileGo(t, pluginCodes[i], backend) + } else { + utils.CompileGoTest(t, pluginCodes[i], backend) + } ioutil.WriteFile(filepath.Join(pluginDir, pluginID, "plugin.json"), []byte(pluginManifests[i]), 0600) manifest, activated, reterr := env.Activate(pluginID) @@ -121,7 +125,11 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests } func setupPluginApiTest(t *testing.T, pluginCode string, pluginManifest string, pluginID string, app *App, c *request.Context) string { - return setupMultiPluginApiTest(t, []string{pluginCode}, []string{pluginManifest}, []string{pluginID}, app, c) + + asMain := pluginID != "test_db_driver" + return setupMultiPluginApiTest(t, + []string{pluginCode}, []string{pluginManifest}, []string{pluginID}, + asMain, app, c) } func TestPublicFilesPathConfiguration(t *testing.T) { @@ -766,7 +774,7 @@ func TestPluginAPIGetPlugins(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), nil) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), nil) require.NoError(t, err) pluginIDs := []string{"pluginid1", "pluginid2", "pluginid3"} @@ -854,7 +862,7 @@ func TestInstallPlugin(t *testing.T) { return app.NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newPluginAPI, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(newPluginAPI, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) app.SetPluginsEnvironment(env) @@ -1053,6 +1061,7 @@ func pluginAPIHookTest(t *testing.T, th *TestHelper, fileName string, id string, if settingsSchema != "" { schema = settingsSchema } + th.App.srv.sqlStore = th.GetSqlStore() setupPluginApiTest(t, code, fmt.Sprintf(`{"id": "%v", "backend": {"executable": "backend.exe"}, "settings_schema": %v}`, id, schema), id, th.App, th.Context) @@ -1481,6 +1490,7 @@ func TestInterpluginPluginHTTP(t *testing.T) { "testplugininterserver", "testplugininterclient", }, + true, th.App, th.Context, ) @@ -1505,7 +1515,7 @@ func TestApiMetrics(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), metricsMock) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), metricsMock) require.NoError(t, err) th.App.SetPluginsEnvironment(env) diff --git a/app/plugin_api_tests/test_db_driver/main.go b/app/plugin_api_tests/test_db_driver/main.go new file mode 100644 index 0000000000..f7b5e771b1 --- /dev/null +++ b/app/plugin_api_tests/test_db_driver/main.go @@ -0,0 +1,57 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package main_test + +import ( + "database/sql" + "testing" + + "github.com/mattermost/mattermost-server/v5/app/plugin_api_tests" + "github.com/mattermost/mattermost-server/v5/model" + "github.com/mattermost/mattermost-server/v5/plugin" + "github.com/mattermost/mattermost-server/v5/shared/driver" + "github.com/mattermost/mattermost-server/v5/store/sqlstore" + "github.com/mattermost/mattermost-server/v5/store/storetest" +) + +type MyPlugin struct { + plugin.MattermostPlugin + config plugin_api_tests.BasicConfig + t *testing.T +} + +func (p *MyPlugin) OnConfigurationChange() error { + if err := p.API.LoadPluginConfiguration(&p.config); err != nil { + return err + } + return nil +} + +func (p *MyPlugin) MessageWillBePosted(_ *plugin.Context, _ *model.Post) (*model.Post, string) { + store := sqlstore.New(p.API.GetUnsanitizedConfig().SqlSettings, nil) + store.GetMaster().Db.Close() + + store.GetMaster().Db = sql.OpenDB(driver.NewConnector(p.Driver)) + defer store.GetMaster().Db.Close() + + // Testing with a handful of stores + storetest.TestPostStore(p.t, store, store) + storetest.TestUserStore(p.t, store, store) + storetest.TestTeamStore(p.t, store) + storetest.TestChannelStore(p.t, store, store) + storetest.TestBotStore(p.t, store, store) + + // Use the API to instantiate the driver + // And then run the full suite of tests. + return nil, "OK" +} + +// TestDBAPI is a test function which actually runs a plugin. The objective +// is to run the storetest suite from inside a plugin. +// +// The test runner compiles the test code to a binary, and runs it as a normal +// binary. But under the hood, a test runs. +func TestDBAPI(t *testing.T) { + plugin.ClientMain(&MyPlugin{t: t}) +} diff --git a/app/plugin_db_driver.go b/app/plugin_db_driver.go new file mode 100644 index 0000000000..65e352e926 --- /dev/null +++ b/app/plugin_db_driver.go @@ -0,0 +1,303 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "context" + "database/sql" + "database/sql/driver" + "sync" + + "github.com/mattermost/mattermost-server/v5/model" + "github.com/mattermost/mattermost-server/v5/plugin" +) + +// DriverImpl implements the plugin.Driver interface on the server-side. +// Each new request for a connection/statement/transaction etc, generates +// a new entry tracked centrally in a map. Further requests operate on the +// object ID. +type DriverImpl struct { + s *Server + connMut sync.RWMutex + connMap map[string]*sql.Conn + txMut sync.Mutex + txMap map[string]driver.Tx + stMut sync.RWMutex + stMap map[string]driver.Stmt + rowsMut sync.RWMutex + rowsMap map[string]driver.Rows +} + +func NewDriverImpl(s *Server) *DriverImpl { + return &DriverImpl{ + s: s, + connMap: make(map[string]*sql.Conn), + txMap: make(map[string]driver.Tx), + stMap: make(map[string]driver.Stmt), + rowsMap: make(map[string]driver.Rows), + } +} + +func (d *DriverImpl) Conn() (string, error) { + conn, err := d.s.sqlStore.GetMaster().Db.Conn(context.Background()) + if err != nil { + return "", err + } + connID := model.NewId() + d.connMut.Lock() + d.connMap[connID] = conn + d.connMut.Unlock() + return connID, nil +} + +// According to https://golang.org/pkg/database/sql/#Conn, a client can call +// Close on a connection, concurrently while running a query. +// +// Therefore, we have to handle the case where the connection is no longer +// present in the map because it has been closed. ErrBadConn is a good choice +// here which indicates the sql package to retry on a new connection. +// +// ConnPing, ConnQuery, ConnClose, Tx, and Stmt do this. + +func (d *DriverImpl) ConnPing(connID string) error { + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return driver.ErrBadConn + } + + return conn.Raw(func(innerConn interface{}) error { + return innerConn.(driver.Pinger).Ping(context.Background()) + }) +} + +func (d *DriverImpl) ConnQuery(connID, q string, args []driver.NamedValue) (_ string, err error) { + var rows driver.Rows + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + rows, err = innerConn.(driver.QueryerContext).QueryContext(context.Background(), q, args) + return err + }) + if err != nil { + return "", err + } + + rowsID := model.NewId() + d.rowsMut.Lock() + d.rowsMap[rowsID] = rows + d.rowsMut.Unlock() + + return rowsID, nil +} + +func (d *DriverImpl) ConnExec(connID, q string, args []driver.NamedValue) (_ plugin.ResultContainer, err error) { + var res driver.Result + var ret plugin.ResultContainer + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return ret, driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + res, err = innerConn.(driver.ExecerContext).ExecContext(context.Background(), q, args) + return err + }) + if err != nil { + return ret, err + } + + ret.LastID, ret.LastIDError = res.LastInsertId() + ret.RowsAffected, ret.RowsAffectedError = res.RowsAffected() + + return ret, nil +} + +func (d *DriverImpl) ConnClose(connID string) error { + d.connMut.Lock() + conn, ok := d.connMap[connID] + if !ok { + d.connMut.Unlock() + return driver.ErrBadConn + } + delete(d.connMap, connID) + d.connMut.Unlock() + + return conn.Close() +} + +func (d *DriverImpl) Tx(connID string, opts driver.TxOptions) (_ string, err error) { + var tx driver.Tx + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + tx, err = innerConn.(driver.ConnBeginTx).BeginTx(context.Background(), opts) + return err + }) + if err != nil { + return "", err + } + + txID := model.NewId() + d.txMut.Lock() + d.txMap[txID] = tx + d.txMut.Unlock() + return txID, nil +} + +func (d *DriverImpl) TxCommit(txID string) error { + d.txMut.Lock() + tx := d.txMap[txID] + delete(d.txMap, txID) + d.txMut.Unlock() + + return tx.Commit() +} + +func (d *DriverImpl) TxRollback(txID string) error { + d.txMut.Lock() + tx := d.txMap[txID] + delete(d.txMap, txID) + d.txMut.Unlock() + + return tx.Rollback() +} + +func (d *DriverImpl) Stmt(connID, q string) (_ string, err error) { + var stmt driver.Stmt + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + stmt, err = innerConn.(driver.Conn).Prepare(q) + return err + }) + if err != nil { + return "", err + } + + stID := model.NewId() + d.stMut.Lock() + d.stMap[stID] = stmt + d.stMut.Unlock() + return stID, nil +} + +func (d *DriverImpl) StmtClose(stID string) error { + d.stMut.Lock() + err := d.stMap[stID].Close() + delete(d.stMap, stID) + d.stMut.Unlock() + + return err +} + +func (d *DriverImpl) StmtNumInput(stID string) int { + d.stMut.RLock() + defer d.stMut.RUnlock() + return d.stMap[stID].NumInput() +} + +func (d *DriverImpl) StmtQuery(stID string, args []driver.NamedValue) (string, error) { + argVals := make([]driver.Value, len(args)) + for i, a := range args { + argVals[i] = a.Value + } + d.stMut.RLock() + st := d.stMap[stID] + d.stMut.RUnlock() + + rows, err := st.Query(argVals) //nolint:staticcheck + if err != nil { + return "", err + } + rowsID := model.NewId() + d.rowsMut.Lock() + d.rowsMap[rowsID] = rows + d.rowsMut.Unlock() + return rowsID, nil +} + +func (d *DriverImpl) StmtExec(stID string, args []driver.NamedValue) (plugin.ResultContainer, error) { + argVals := make([]driver.Value, len(args)) + for i, a := range args { + argVals[i] = a.Value + } + var ret plugin.ResultContainer + d.stMut.RLock() + st := d.stMap[stID] + d.stMut.RUnlock() + + res, err := st.Exec(argVals) //nolint:staticcheck + if err != nil { + return ret, err + } + + ret.LastID, ret.LastIDError = res.LastInsertId() + ret.RowsAffected, ret.RowsAffectedError = res.RowsAffected() + + return ret, nil +} + +func (d *DriverImpl) RowsColumns(rowsID string) []string { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].Columns() +} + +func (d *DriverImpl) RowsClose(rowsID string) error { + d.rowsMut.Lock() + defer d.rowsMut.Unlock() + err := d.rowsMap[rowsID].Close() + delete(d.rowsMap, rowsID) + return err +} + +func (d *DriverImpl) RowsNext(rowsID string, dest []driver.Value) error { + d.rowsMut.RLock() + rows := d.rowsMap[rowsID] + d.rowsMut.RUnlock() + return rows.Next(dest) +} + +func (d *DriverImpl) RowsHasNextResultSet(rowsID string) bool { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsNextResultSet).HasNextResultSet() +} + +func (d *DriverImpl) RowsNextResultSet(rowsID string) error { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsNextResultSet).NextResultSet() +} + +func (d *DriverImpl) RowsColumnTypeDatabaseTypeName(rowsID string, index int) string { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsColumnTypeDatabaseTypeName).ColumnTypeDatabaseTypeName(index) +} + +func (d *DriverImpl) RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsColumnTypePrecisionScale).ColumnTypePrecisionScale(index) +} diff --git a/app/plugin_hooks_test.go b/app/plugin_hooks_test.go index 02edfbfdb8..e429ea1756 100644 --- a/app/plugin_hooks_test.go +++ b/app/plugin_hooks_test.go @@ -34,7 +34,7 @@ func SetAppEnvironmentWithPlugins(t *testing.T, pluginCode []string, app *App, a webappPluginDir, err := ioutil.TempDir("", "") require.NoError(t, err) - env, err := plugin.NewEnvironment(apiFunc, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(apiFunc, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) app.SetPluginsEnvironment(env) @@ -1045,7 +1045,7 @@ func TestHookMetrics(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), metricsMock) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), metricsMock) require.NoError(t, err) th.App.SetPluginsEnvironment(env) diff --git a/plugin/client.go b/plugin/client.go index ed3d872d79..15ac0254c9 100644 --- a/plugin/client.go +++ b/plugin/client.go @@ -19,11 +19,13 @@ func ClientMain(pluginImplementation interface{}) { if impl, ok := pluginImplementation.(interface { SetAPI(api API) SetHelpers(helpers Helpers) + SetDriver(driver Driver) }); !ok { panic("Plugin implementation given must embed plugin.MattermostPlugin") } else { impl.SetAPI(nil) impl.SetHelpers(nil) + impl.SetDriver(nil) } pluginMap := map[string]plugin.Plugin{ @@ -40,6 +42,7 @@ type MattermostPlugin struct { // API exposes the plugin api, and becomes available just prior to the OnActive hook. API API Helpers Helpers + Driver Driver } // SetAPI persists the given API interface to the plugin. It is invoked just prior to the @@ -52,3 +55,7 @@ func (p *MattermostPlugin) SetAPI(api API) { func (p *MattermostPlugin) SetHelpers(helpers Helpers) { p.Helpers = helpers } + +func (p *MattermostPlugin) SetDriver(driver Driver) { + p.Driver = driver +} diff --git a/plugin/client_rpc.go b/plugin/client_rpc.go index 8382fe005e..58c0841a8c 100644 --- a/plugin/client_rpc.go +++ b/plugin/client_rpc.go @@ -7,6 +7,8 @@ package plugin import ( "bytes" + "database/sql" + "database/sql/driver" "encoding/gob" "encoding/json" "fmt" @@ -19,7 +21,9 @@ import ( "reflect" "github.com/dyatlov/go-opengraph/opengraph" + "github.com/go-sql-driver/mysql" "github.com/hashicorp/go-plugin" + "github.com/lib/pq" "github.com/mattermost/mattermost-server/v5/model" "github.com/mattermost/mattermost-server/v5/shared/mlog" @@ -32,6 +36,7 @@ type hooksRPCClient struct { log *mlog.Logger muxBroker *plugin.MuxBroker apiImpl API + driver Driver implemented [TotalHooksID]bool } @@ -43,9 +48,10 @@ type hooksRPCServer struct { // Implements hashicorp/go-plugin/plugin.Plugin interface to connect the hooks of a plugin type hooksPlugin struct { - hooks interface{} - apiImpl API - log *mlog.Logger + hooks interface{} + apiImpl API + driverImpl Driver + log *mlog.Logger } func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) { @@ -53,7 +59,12 @@ func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) { } func (p *hooksPlugin) Client(b *plugin.MuxBroker, client *rpc.Client) (interface{}, error) { - return &hooksRPCClient{client: client, log: p.log, muxBroker: b, apiImpl: p.apiImpl}, nil + return &hooksRPCClient{client: client, + log: p.log, + muxBroker: b, + apiImpl: p.apiImpl, + driver: p.driverImpl, + }, nil } type apiRPCClient struct { @@ -72,7 +83,8 @@ type apiRPCServer struct { // ErrorString merely preserves the string description of the error, while satisfying the error // interface itself to allow other registered types (such as model.AppError) to be sent unmodified. type ErrorString struct { - Err string + Code int // Code to map to various error variables + Err string } func (e ErrorString) Error() string { @@ -87,9 +99,58 @@ func encodableError(err error) error { return err } - return &ErrorString{ + if _, ok := err.(*pq.Error); ok { + return err + } + + if _, ok := err.(*mysql.MySQLError); ok { + return err + } + + ret := &ErrorString{ Err: err.Error(), } + + switch err { + case io.EOF: + ret.Code = 1 + case sql.ErrNoRows: + ret.Code = 2 + case sql.ErrConnDone: + ret.Code = 3 + case sql.ErrTxDone: + ret.Code = 4 + case driver.ErrSkip: + ret.Code = 5 + case driver.ErrBadConn: + ret.Code = 6 + case driver.ErrRemoveArgument: + ret.Code = 7 + } + + return ret +} + +func decodableError(err error) error { + if encErr, ok := err.(*ErrorString); ok { + switch encErr.Code { + case 1: + return io.EOF + case 2: + return sql.ErrNoRows + case 3: + return sql.ErrConnDone + case 4: + return sql.ErrTxDone + case 5: + return driver.ErrSkip + case 6: + return driver.ErrBadConn + case 7: + return driver.ErrRemoveArgument + } + } + return err } // Registering some types used by MM for encoding/gob used by rpc @@ -98,6 +159,8 @@ func init() { gob.Register([]interface{}{}) gob.Register(map[string]interface{}{}) gob.Register(&model.AppError{}) + gob.Register(&pq.Error{}) + gob.Register(&mysql.MySQLError{}) gob.Register(&ErrorString{}) gob.Register(&opengraph.OpenGraph{}) gob.Register(&model.AutocompleteDynamicListArg{}) @@ -167,7 +230,8 @@ func (s *hooksRPCServer) Implemented(args struct{}, reply *[]string) error { } type Z_OnActivateArgs struct { - APIMuxId uint32 + APIMuxId uint32 + DriverMuxId uint32 } type Z_OnActivateReturns struct { @@ -181,8 +245,14 @@ func (g *hooksRPCClient) OnActivate() error { muxBroker: g.muxBroker, }) + nextID := g.muxBroker.NextId() + go g.muxBroker.AcceptAndServe(nextID, &dbRPCServer{ + dbImpl: g.driver, + }) + _args := &Z_OnActivateArgs{ - APIMuxId: muxId, + APIMuxId: muxId, + DriverMuxId: nextID, } _returns := &Z_OnActivateReturns{} @@ -198,17 +268,28 @@ func (s *hooksRPCServer) OnActivate(args *Z_OnActivateArgs, returns *Z_OnActivat return err } + conn2, err := s.muxBroker.Dial(args.DriverMuxId) + if err != nil { + return err + } + s.apiRPCClient = &apiRPCClient{ client: rpc.NewClient(connection), muxBroker: s.muxBroker, } + dbClient := &dbRPCClient{ + client: rpc.NewClient(conn2), + } + if mmplugin, ok := s.impl.(interface { SetAPI(api API) SetHelpers(helpers Helpers) + SetDriver(driver Driver) }); ok { mmplugin.SetAPI(s.apiRPCClient) mmplugin.SetHelpers(&HelpersImpl{API: s.apiRPCClient}) + mmplugin.SetDriver(dbClient) } if mmplugin, ok := s.impl.(interface { diff --git a/plugin/db_rpc.go b/plugin/db_rpc.go new file mode 100644 index 0000000000..0d2701f24d --- /dev/null +++ b/plugin/db_rpc.go @@ -0,0 +1,461 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package plugin + +import ( + "database/sql/driver" + "log" + "net/rpc" +) + +// dbRPCClient contains the client-side logic to handle the RPC communication +// with the server. It's API is hand-written because we do not expect +// new methods to be added very frequently. +type dbRPCClient struct { + client *rpc.Client +} + +// dbRPCServer is the server-side component which is responsible for calling +// the driver methods and properly encoding the responses back to the RPC client. +type dbRPCServer struct { + dbImpl Driver +} + +var _ Driver = &dbRPCClient{} + +type Z_DbStrErrReturn struct { + A string + B error +} + +type Z_DbErrReturn struct { + A error +} + +type Z_DbInt64ErrReturn struct { + A int64 + B error +} + +type Z_DbBoolReturn struct { + A bool +} + +func (db *dbRPCClient) Conn() (string, error) { + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Conn", struct{}{}, ret) + if err != nil { + log.Printf("error during Plugin.Conn: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Conn(_ struct{}, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Conn() + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) ConnPing(connID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.ConnPing", connID, ret) + if err != nil { + log.Printf("error during Plugin.ConnPing: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) ConnPing(connID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.ConnPing(connID) + ret.A = encodableError(ret.A) + return nil +} + +func (db *dbRPCClient) ConnClose(connID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.ConnClose", connID, ret) + if err != nil { + log.Printf("error during Plugin.ConnClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) ConnClose(connID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.ConnClose(connID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbTxArgs struct { + A string + B driver.TxOptions +} + +func (db *dbRPCClient) Tx(connID string, opts driver.TxOptions) (string, error) { + args := &Z_DbTxArgs{ + A: connID, + B: opts, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Tx", args, ret) + if err != nil { + log.Printf("error during Plugin.Tx: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Tx(args *Z_DbTxArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Tx(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) TxCommit(txID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.TxCommit", txID, ret) + if err != nil { + log.Printf("error during Plugin.TxCommit: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) TxCommit(txID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.TxCommit(txID) + ret.A = encodableError(ret.A) + return nil +} + +func (db *dbRPCClient) TxRollback(txID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.TxRollback", txID, ret) + if err != nil { + log.Printf("error during Plugin.TxRollback: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) TxRollback(txID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.TxRollback(txID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbStmtArgs struct { + A string + B string +} + +func (db *dbRPCClient) Stmt(connID, q string) (string, error) { + args := &Z_DbStmtArgs{ + A: connID, + B: q, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Stmt", args, ret) + if err != nil { + log.Printf("error during Plugin.Stmt: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Stmt(args *Z_DbStmtArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Stmt(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) StmtClose(stID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.StmtClose", stID, ret) + if err != nil { + log.Printf("error during Plugin.StmtClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) StmtClose(stID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.StmtClose(stID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbIntReturn struct { + A int +} + +func (db *dbRPCClient) StmtNumInput(stID string) int { + ret := &Z_DbIntReturn{} + err := db.client.Call("Plugin.StmtNumInput", stID, ret) + if err != nil { + log.Printf("error during Plugin.StmtNumInput: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) StmtNumInput(stID string, ret *Z_DbIntReturn) error { + ret.A = db.dbImpl.StmtNumInput(stID) + return nil +} + +type Z_DbStmtQueryArgs struct { + A string + B []driver.NamedValue +} + +func (db *dbRPCClient) StmtQuery(stID string, argVals []driver.NamedValue) (string, error) { + args := &Z_DbStmtQueryArgs{ + A: stID, + B: argVals, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.StmtQuery", args, ret) + if err != nil { + log.Printf("error during Plugin.StmtQuery: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) StmtQuery(args *Z_DbStmtQueryArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.StmtQuery(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) StmtExec(stID string, argVals []driver.NamedValue) (ResultContainer, error) { + args := &Z_DbStmtQueryArgs{ + A: stID, + B: argVals, + } + ret := &Z_DbResultContErrReturn{} + err := db.client.Call("Plugin.StmtExec", args, ret) + if err != nil { + log.Printf("error during Plugin.StmtExec: %v", err) + } + ret.A.LastIDError = decodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError) + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) StmtExec(args *Z_DbStmtQueryArgs, ret *Z_DbResultContErrReturn) error { + ret.A, ret.B = db.dbImpl.StmtExec(args.A, args.B) + ret.A.LastIDError = encodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbConnArgs struct { + A string + B string + C []driver.NamedValue +} + +func (db *dbRPCClient) ConnQuery(connID, q string, argVals []driver.NamedValue) (string, error) { + args := &Z_DbConnArgs{ + A: connID, + B: q, + C: argVals, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.ConnQuery", args, ret) + if err != nil { + log.Printf("error during Plugin.ConnQuery: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) ConnQuery(args *Z_DbConnArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.ConnQuery(args.A, args.B, args.C) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbResultContErrReturn struct { + A ResultContainer + B error +} + +func (db *dbRPCClient) ConnExec(connID, q string, argVals []driver.NamedValue) (ResultContainer, error) { + args := &Z_DbConnArgs{ + A: connID, + B: q, + C: argVals, + } + ret := &Z_DbResultContErrReturn{} + err := db.client.Call("Plugin.ConnExec", args, ret) + if err != nil { + log.Printf("error during Plugin.ConnExec: %v", err) + } + ret.A.LastIDError = decodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError) + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) ConnExec(args *Z_DbConnArgs, ret *Z_DbResultContErrReturn) error { + ret.A, ret.B = db.dbImpl.ConnExec(args.A, args.B, args.C) + ret.A.LastIDError = encodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbStrSliceReturn struct { + A []string +} + +func (db *dbRPCClient) RowsColumns(rowsID string) []string { + ret := &Z_DbStrSliceReturn{} + err := db.client.Call("Plugin.RowsColumns", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsColumns: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) RowsColumns(rowsID string, ret *Z_DbStrSliceReturn) error { + ret.A = db.dbImpl.RowsColumns(rowsID) + return nil +} + +func (db *dbRPCClient) RowsClose(resID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.RowsClose", resID, ret) + if err != nil { + log.Printf("error during Plugin.RowsClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) RowsClose(resID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.RowsClose(resID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbRowScanReturn struct { + A error + B []driver.Value +} + +type Z_DbRowScanArg struct { + A string + B []driver.Value +} + +func (db *dbRPCClient) RowsNext(rowsID string, dest []driver.Value) error { + args := &Z_DbRowScanArg{ + A: rowsID, + B: dest, + } + ret := &Z_DbRowScanReturn{} + err := db.client.Call("Plugin.RowsNext", args, ret) + if err != nil { + log.Printf("error during Plugin.RowsNext: %v", err) + } + ret.A = decodableError(ret.A) + for i, v := range ret.B { + dest[i] = v + } + return ret.A +} + +func (db *dbRPCServer) RowsNext(args *Z_DbRowScanArg, ret *Z_DbRowScanReturn) error { + ret.A = db.dbImpl.RowsNext(args.A, args.B) + ret.A = encodableError(ret.A) + // Trick to populate the dest slice. RPC doesn't have a semantic to populate + // pointer type args. So the only way to pass values is via args, and only way + // to return values is via the return struct. + ret.B = args.B + return nil +} + +func (db *dbRPCClient) RowsHasNextResultSet(rowsID string) bool { + ret := &Z_DbBoolReturn{} + err := db.client.Call("Plugin.RowsHasNextResultSet", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsHasNextResultSet: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) RowsHasNextResultSet(rowsID string, ret *Z_DbBoolReturn) error { + ret.A = db.dbImpl.RowsHasNextResultSet(rowsID) + return nil +} + +func (db *dbRPCClient) RowsNextResultSet(rowsID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.RowsNextResultSet", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsNextResultSet: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) RowsNextResultSet(rowsID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.RowsNextResultSet(rowsID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbRowsColumnArg struct { + A string + B int +} + +func (db *dbRPCClient) RowsColumnTypeDatabaseTypeName(rowsID string, index int) string { + args := &Z_DbRowsColumnArg{ + A: rowsID, + B: index, + } + var ret string + err := db.client.Call("Plugin.RowsColumnTypeDatabaseTypeName", args, &ret) + if err != nil { + log.Printf("error during Plugin.RowsColumnTypeDatabaseTypeName: %v", err) + } + return ret +} + +func (db *dbRPCServer) RowsColumnTypeDatabaseTypeName(args *Z_DbRowsColumnArg, ret *string) error { + *ret = db.dbImpl.RowsColumnTypeDatabaseTypeName(args.A, args.B) + return nil +} + +type Z_DbRowsColumnTypePrecisionScaleReturn struct { + A int64 + B int64 + C bool +} + +func (db *dbRPCClient) RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) { + args := &Z_DbRowsColumnArg{ + A: rowsID, + B: index, + } + ret := &Z_DbRowsColumnTypePrecisionScaleReturn{} + err := db.client.Call("Plugin.RowsColumnTypePrecisionScale", args, ret) + if err != nil { + log.Printf("error during Plugin.RowsColumnTypePrecisionScale: %v", err) + } + return ret.A, ret.B, ret.C +} + +func (db *dbRPCServer) RowsColumnTypePrecisionScale(args *Z_DbRowsColumnArg, ret *Z_DbRowsColumnTypePrecisionScaleReturn) error { + ret.A, ret.B, ret.C = db.dbImpl.RowsColumnTypePrecisionScale(args.A, args.B) + return nil +} diff --git a/plugin/environment.go b/plugin/environment.go index 6d44047dde..50c2b037cd 100644 --- a/plugin/environment.go +++ b/plugin/environment.go @@ -54,17 +54,23 @@ type Environment struct { logger *mlog.Logger metrics einterfaces.MetricsInterface newAPIImpl apiImplCreatorFunc + dbDriver Driver pluginDir string webappPluginDir string prepackagedPlugins []*PrepackagedPlugin prepackagedPluginsLock sync.RWMutex } -func NewEnvironment(newAPIImpl apiImplCreatorFunc, pluginDir string, webappPluginDir string, logger *mlog.Logger, metrics einterfaces.MetricsInterface) (*Environment, error) { +func NewEnvironment(newAPIImpl apiImplCreatorFunc, + dbDriver Driver, + pluginDir string, webappPluginDir string, + logger *mlog.Logger, + metrics einterfaces.MetricsInterface) (*Environment, error) { return &Environment{ logger: logger, metrics: metrics, newAPIImpl: newAPIImpl, + dbDriver: dbDriver, pluginDir: pluginDir, webappPluginDir: webappPluginDir, }, nil @@ -263,7 +269,7 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated } if pluginInfo.Manifest.HasServer() { - sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.logger, env.metrics) + sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.dbDriver, env.logger, env.metrics) if err != nil { return nil, false, errors.Wrapf(err, "unable to start plugin: %v", id) } diff --git a/plugin/health_check_test.go b/plugin/health_check_test.go index 7aa559576c..8a7d929a26 100644 --- a/plugin/health_check_test.go +++ b/plugin/health_check_test.go @@ -59,7 +59,7 @@ func testPluginHealthCheckSuccess(t *testing.T) { EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.NoError(t, err) require.NotNil(t, supervisor) defer supervisor.Shutdown() @@ -106,7 +106,7 @@ func testPluginHealthCheckPanic(t *testing.T) { EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.NoError(t, err) require.NotNil(t, supervisor) defer supervisor.Shutdown() diff --git a/plugin/helpers.go b/plugin/helpers.go index 914fc2e1f2..68885dd78f 100644 --- a/plugin/helpers.go +++ b/plugin/helpers.go @@ -4,6 +4,8 @@ package plugin import ( + "database/sql/driver" + "github.com/mattermost/mattermost-server/v5/model" ) @@ -95,3 +97,55 @@ type Helpers interface { type HelpersImpl struct { API API } + +// ResultContainer contains the output from the LastInsertID +// and RowsAffected methods for a given set of rows. +// It is used to embed another round-trip to the server, +// and helping to avoid tracking results on the server. +type ResultContainer struct { + LastID int64 + LastIDError error + RowsAffected int64 + RowsAffectedError error +} + +type Driver interface { + // Connection + Conn() (string, error) + ConnPing(connID string) error + ConnClose(connID string) error + ConnQuery(connID, q string, args []driver.NamedValue) (string, error) // rows + ConnExec(connID, q string, args []driver.NamedValue) (ResultContainer, error) // result + + // Transaction + Tx(connID string, opts driver.TxOptions) (string, error) + TxCommit(txID string) error + TxRollback(txID string) error + + // Statement + Stmt(connID, q string) (string, error) + StmtClose(stID string) error + StmtNumInput(stID string) int + StmtQuery(stID string, args []driver.NamedValue) (string, error) // rows + StmtExec(stID string, args []driver.NamedValue) (ResultContainer, error) // result + + // Rows + RowsColumns(rowsID string) []string + RowsClose(rowsID string) error + RowsNext(rowsID string, dest []driver.Value) error + RowsHasNextResultSet(rowsID string) bool + RowsNextResultSet(rowsID string) error + RowsColumnTypeDatabaseTypeName(rowsID string, index int) string + RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) + + // TODO: add this + // RowsColumnScanType(rowsID string, index int) reflect.Type + + // Note: the following cannot be implemented because either MySQL or PG + // does not support it. So this implementation has to be a common subset + // of both DB implementations. + // RowsColumnTypeLength(rowsID string, index int) (int64, bool) + // RowsColumnTypeNullable(rowsID string, index int) (bool, bool) + // ResetSession(ctx context.Context) error + // IsValid() bool +} diff --git a/plugin/supervisor.go b/plugin/supervisor.go index ce067aff39..de1fbf3715 100644 --- a/plugin/supervisor.go +++ b/plugin/supervisor.go @@ -27,7 +27,7 @@ type supervisor struct { pid int } -func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) { +func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, driver Driver, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) { sup := supervisor{} defer func() { if retErr != nil { @@ -44,8 +44,9 @@ func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog pluginMap := map[string]plugin.Plugin{ "hooks": &hooksPlugin{ - log: wrappedLogger, - apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics}, + log: wrappedLogger, + driverImpl: driver, + apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics}, }, } diff --git a/plugin/supervisor_test.go b/plugin/supervisor_test.go index 0f2bdef9fa..921841b704 100644 --- a/plugin/supervisor_test.go +++ b/plugin/supervisor_test.go @@ -41,7 +41,7 @@ func testSupervisorInvalidExecutablePath(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) assert.Nil(t, supervisor) assert.Error(t, err) } @@ -60,7 +60,7 @@ func testSupervisorNonExistentExecutablePath(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.Error(t, err) require.Nil(t, supervisor) } @@ -90,7 +90,7 @@ func testSupervisorStartTimeout(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.Error(t, err) require.Nil(t, supervisor) } diff --git a/services/telemetry/telemetry_test.go b/services/telemetry/telemetry_test.go index 207ec8d5e1..9a57ee8320 100644 --- a/services/telemetry/telemetry_test.go +++ b/services/telemetry/telemetry_test.go @@ -52,7 +52,12 @@ func initializeMocks(cfg *model.Config) (*mocks.ServerIface, *storeMocks.Store, os.RemoveAll(webappPluginDir) } pluginsAPIMock := &plugintest.API{} - pluginEnv, _ := plugin.NewEnvironment(func(m *model.Manifest) plugin.API { return pluginsAPIMock }, pluginDir, webappPluginDir, mlog.NewLogger(&mlog.LoggerConfiguration{}), nil) + pluginEnv, _ := plugin.NewEnvironment( + func(m *model.Manifest) plugin.API { return pluginsAPIMock }, + nil, + pluginDir, webappPluginDir, + mlog.NewLogger(&mlog.LoggerConfiguration{}), + nil) serverIfaceMock.On("GetPluginsEnvironment").Return(pluginEnv, nil) serverIfaceMock.On("License").Return(model.NewTestLicense(), nil) diff --git a/shared/driver/conn.go b/shared/driver/conn.go index 196a1ad892..c3cfa0e548 100644 --- a/shared/driver/conn.go +++ b/shared/driver/conn.go @@ -5,15 +5,16 @@ package driver import ( "context" - "database/sql" "database/sql/driver" + + "github.com/mattermost/mattermost-server/v5/plugin" ) // Conn is a DB driver conn implementation -// which will just pass-through all queries to its -// underlying connection. +// which executes queries using the Plugin DB API. type Conn struct { - conn *sql.Conn + id string + api plugin.Driver } // driverConn is a super-interface combining the basic @@ -33,66 +34,84 @@ var ( ) func (c *Conn) Begin() (tx driver.Tx, err error) { - err = c.conn.Raw(func(innerConn interface{}) error { - tx, err = innerConn.(driver.Conn).Begin() //nolint:staticcheck - return err - }) - return tx, err + txID, err := c.api.Tx(c.id, driver.TxOptions{}) + if err != nil { + return nil, err + } + + t := &wrapperTx{ + id: txID, + api: c.api, + } + return t, nil } -func (c *Conn) BeginTx(ctx context.Context, opts driver.TxOptions) (_ driver.Tx, err error) { - t := &wrapperTx{} - err = c.conn.Raw(func(innerConn interface{}) error { - t.Tx, err = innerConn.(driver.ConnBeginTx).BeginTx(ctx, opts) - return err - }) - return t, err +func (c *Conn) BeginTx(_ context.Context, opts driver.TxOptions) (driver.Tx, error) { + txID, err := c.api.Tx(c.id, opts) + if err != nil { + return nil, err + } + + t := &wrapperTx{ + id: txID, + api: c.api, + } + return t, nil } -func (c *Conn) Prepare(q string) (_ driver.Stmt, err error) { - st := &wrapperStmt{} - err = c.conn.Raw(func(innerConn interface{}) error { - st.Stmt, err = innerConn.(driver.Conn).Prepare(q) - return err - }) - return st, err +func (c *Conn) Prepare(q string) (driver.Stmt, error) { + stID, err := c.api.Stmt(c.id, q) + if err != nil { + return nil, err + } + + st := &wrapperStmt{ + id: stID, + api: c.api, + } + return st, nil } -func (c *Conn) PrepareContext(ctx context.Context, q string) (_ driver.Stmt, err error) { - st := &wrapperStmt{} - err = c.conn.Raw(func(innerConn interface{}) error { - st.Stmt, err = innerConn.(driver.ConnPrepareContext).PrepareContext(ctx, q) - return err - }) - return st, err +func (c *Conn) PrepareContext(_ context.Context, q string) (driver.Stmt, error) { + stID, err := c.api.Stmt(c.id, q) + if err != nil { + return nil, err + } + st := &wrapperStmt{ + id: stID, + api: c.api, + } + return st, nil } -func (c *Conn) ExecContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Result, err error) { - res := &wrapperResult{} - err = c.conn.Raw(func(innerConn interface{}) error { - res.Result, err = innerConn.(driver.ExecerContext).ExecContext(ctx, q, args) - return err - }) - return res, err +func (c *Conn) ExecContext(_ context.Context, q string, args []driver.NamedValue) (driver.Result, error) { + resultContainer, err := c.api.ConnExec(c.id, q, args) + if err != nil { + return nil, err + } + res := &wrapperResult{ + res: resultContainer, + } + return res, nil } -func (c *Conn) QueryContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Rows, err error) { - rows := &wrapperRows{} - err = c.conn.Raw(func(innerConn interface{}) error { - rows.Rows, err = innerConn.(driver.QueryerContext).QueryContext(ctx, q, args) - return err - }) - return rows, err +func (c *Conn) QueryContext(_ context.Context, q string, args []driver.NamedValue) (driver.Rows, error) { + rowsID, err := c.api.ConnQuery(c.id, q, args) + if err != nil { + return nil, err + } + + rows := &wrapperRows{ + id: rowsID, + api: c.api, + } + return rows, nil } -func (c *Conn) Ping(ctx context.Context) error { - return c.conn.Raw(func(innerConn interface{}) error { - return innerConn.(driver.Pinger).Ping(ctx) - }) +func (c *Conn) Ping(_ context.Context) error { + return c.api.ConnPing(c.id) } func (c *Conn) Close() error { - return c.conn.Raw(func(innerConn interface{}) error { - return innerConn.(driver.Conn).Close() - }) + return c.api.ConnClose(c.id) } diff --git a/shared/driver/driver.go b/shared/driver/driver.go index a38854a4b9..a917c87404 100644 --- a/shared/driver/driver.go +++ b/shared/driver/driver.go @@ -1,12 +1,18 @@ // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. // See LICENSE.txt for license information. +// package driver implements a DB driver that can be used by plugins +// to make SQL queries using RPC. This helps to avoid opening new connections +// for every plugin, and lets everyone use the central connection +// pool in the server. +// The tests for this package are at app/plugin_api_tests/test_db_driver/main.go. package driver import ( "context" - "database/sql" "database/sql/driver" + + "github.com/mattermost/mattermost-server/v5/plugin" ) var ( @@ -15,32 +21,22 @@ var ( ) // Connector is the DB connector which is used to -// initialize the underlying DB. +// communicate with the DB API. type Connector struct { - driverName string - dsn string - db *sql.DB + api plugin.Driver } -func NewConnector(driverName, dsn string) (*Connector, error) { - db, err := sql.Open(driverName, dsn) - if err != nil { - return nil, err - } - return &Connector{ - driverName: driverName, - dsn: dsn, - db: db, - }, nil +func NewConnector(api plugin.Driver) *Connector { + return &Connector{api: api} } -func (c *Connector) Connect(ctx context.Context) (driver.Conn, error) { - conn, err := c.db.Conn(ctx) +func (c *Connector) Connect(_ context.Context) (driver.Conn, error) { + connID, err := c.api.Conn() if err != nil { return nil, err } - return &Conn{conn: conn}, nil + return &Conn{id: connID, api: c.api}, nil } func (c *Connector) Driver() driver.Driver { diff --git a/shared/driver/objects.go b/shared/driver/objects.go index 008c189ccf..f6c5ef19e5 100644 --- a/shared/driver/objects.go +++ b/shared/driver/objects.go @@ -6,93 +6,109 @@ package driver import ( "context" "database/sql/driver" - "reflect" + + "github.com/mattermost/mattermost-server/v5/plugin" ) type wrapperTx struct { driver.Tx + id string + api plugin.Driver } func (t *wrapperTx) Commit() error { - return t.Tx.Commit() + return t.api.TxCommit(t.id) } func (t *wrapperTx) Rollback() error { - return t.Tx.Rollback() + return t.api.TxRollback(t.id) } type wrapperStmt struct { driver.Stmt + id string + api plugin.Driver } func (s *wrapperStmt) Close() error { - return s.Stmt.Close() + return s.api.StmtClose(s.id) } func (s *wrapperStmt) NumInput() int { - return s.Stmt.NumInput() + return s.api.StmtNumInput(s.id) } -func (s *wrapperStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) { - return s.Stmt.(driver.StmtExecContext).ExecContext(ctx, args) +func (s *wrapperStmt) ExecContext(_ context.Context, args []driver.NamedValue) (driver.Result, error) { + resultContainer, err := s.api.StmtExec(s.id, args) + if err != nil { + return nil, err + } + res := &wrapperResult{ + res: resultContainer, + } + return res, nil } -func (s *wrapperStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) { - return s.Stmt.(driver.StmtQueryContext).QueryContext(ctx, args) +func (s *wrapperStmt) QueryContext(_ context.Context, args []driver.NamedValue) (driver.Rows, error) { + rowsID, err := s.api.StmtQuery(s.id, args) + if err != nil { + return nil, err + } + rows := &wrapperRows{ + id: rowsID, + api: s.api, + } + return rows, nil } +// wrapperResult implements the driver.Result interface. +// This differs from other objects because it already contains the +// information for its methods. This does two things: +// +// 1. Simplifies server-side code by avoiding to track result ids +// in a map. +// 2. Avoids round-trip to compute result methods. type wrapperResult struct { - driver.Result + res plugin.ResultContainer } func (r *wrapperResult) LastInsertId() (int64, error) { - return r.Result.LastInsertId() + return r.res.LastID, r.res.LastIDError } func (r *wrapperResult) RowsAffected() (int64, error) { - return r.Result.RowsAffected() + return r.res.RowsAffected, r.res.RowsAffectedError } type wrapperRows struct { - driver.Rows + id string + api plugin.Driver } func (r *wrapperRows) Columns() []string { - return r.Rows.Columns() + return r.api.RowsColumns(r.id) } func (r *wrapperRows) Close() error { - return r.Rows.Close() + return r.api.RowsClose(r.id) } func (r *wrapperRows) Next(dest []driver.Value) error { - return r.Rows.Next(dest) + return r.api.RowsNext(r.id, dest) } func (r *wrapperRows) HasNextResultSet() bool { - return r.Rows.(driver.RowsNextResultSet).HasNextResultSet() + return r.api.RowsHasNextResultSet(r.id) } func (r *wrapperRows) NextResultSet() error { - return r.Rows.(driver.RowsNextResultSet).NextResultSet() -} - -func (r *wrapperRows) ColumnTypeScanType(index int) reflect.Type { - return r.Rows.(driver.RowsColumnTypeScanType).ColumnTypeScanType(index) + return r.api.RowsNextResultSet(r.id) } func (r *wrapperRows) ColumnTypeDatabaseTypeName(index int) string { - return r.Rows.(driver.RowsColumnTypeDatabaseTypeName).ColumnTypeDatabaseTypeName(index) -} - -func (r *wrapperRows) ColumnTypeLength(index int) (length int64, ok bool) { - return r.Rows.(driver.RowsColumnTypeLength).ColumnTypeLength(index) -} - -func (r *wrapperRows) ColumnTypeNullable(index int) (nullable, ok bool) { - return r.Rows.(driver.RowsColumnTypeNullable).ColumnTypeNullable(index) + return r.api.RowsColumnTypeDatabaseTypeName(r.id, index) } func (r *wrapperRows) ColumnTypePrecisionScale(index int) (precision, scale int64, ok bool) { - return r.Rows.(driver.RowsColumnTypePrecisionScale).ColumnTypePrecisionScale(index) + return r.api.RowsColumnTypePrecisionScale(r.id, index) } diff --git a/store/sqlstore/store_test.go b/store/sqlstore/store_test.go index 3bec50df4b..55081f0663 100644 --- a/store/sqlstore/store_test.go +++ b/store/sqlstore/store_test.go @@ -4,7 +4,6 @@ package sqlstore import ( - "database/sql" "fmt" "os" "regexp" @@ -21,7 +20,6 @@ import ( "github.com/mattermost/mattermost-server/v5/einterfaces/mocks" "github.com/mattermost/mattermost-server/v5/model" - "github.com/mattermost/mattermost-server/v5/shared/driver" "github.com/mattermost/mattermost-server/v5/store" "github.com/mattermost/mattermost-server/v5/store/searchtest" "github.com/mattermost/mattermost-server/v5/store/storetest" @@ -97,47 +95,6 @@ func StoreTestWithSqlStore(t *testing.T, f func(*testing.T, store.Store, storete } } -func TestDBConnector(t *testing.T) { - testDrivers := []string{ - model.DATABASE_DRIVER_POSTGRES, - model.DATABASE_DRIVER_MYSQL, - } - - for _, dr := range testDrivers { - settings := makeSqlSettings(dr) - - store := &SqlStore{ - settings: settings, - } - connector, err := driver.NewConnector(*settings.DriverName, *settings.DataSource) - require.NoError(t, err) - db := sql.OpenDB(connector) - - store.master = getDBMap(settings, db) - store.stores.post = newSqlPostStore(store, nil) - store.stores.channel = newSqlChannelStore(store, nil) - store.stores.team = newSqlTeamStore(store) - store.stores.thread = newSqlThreadStore(store) - store.stores.user = newSqlUserStore(store, nil) - store.stores.preference = newSqlPreferenceStore(store) - store.stores.bot = newSqlBotStore(store, nil) - store.stores.scheme = newSqlSchemeStore(store) - err = store.GetMaster().CreateTablesIfNotExists() - require.NoError(t, err) - - store.stores.post.(*SqlPostStore).createIndexesIfNotExists() - store.stores.channel.(*SqlChannelStore).createIndexesIfNotExists() - - t.Run(dr, func(t *testing.T) { - // Just testing post store for now. - // This will eventually go away when it is replaced with RPC. - storetest.TestPostStore(t, store, store) - }) - - store.Close() - } -} - func initStores() { if testing.Short() { return diff --git a/utils/test_files_compiler.go b/utils/test_files_compiler.go index ff6bc14c29..77ad84ce02 100644 --- a/utils/test_files_compiler.go +++ b/utils/test_files_compiler.go @@ -43,3 +43,32 @@ func CompileGo(t *testing.T, sourceCode, outputPath string) { } require.NoError(t, err, "failed to compile go") } + +func CompileGoTest(t *testing.T, sourceCode, outputPath string) { + dir, err := ioutil.TempDir(".", "") + require.NoError(t, err) + defer os.RemoveAll(dir) + + dir, err = filepath.Abs(dir) + require.NoError(t, err) + + // Write out main.go given the source code. + main := filepath.Join(dir, "main_test.go") + err = ioutil.WriteFile(main, []byte(sourceCode), 0600) + require.NoError(t, err) + + _, sourceFile, _, ok := runtime.Caller(0) + require.True(t, ok) + serverPath := filepath.Dir(filepath.Dir(sourceFile)) + + out := &bytes.Buffer{} + cmd := exec.Command("go", "test", "-c", "-o", outputPath, main) + cmd.Dir = serverPath + cmd.Stdout = out + cmd.Stderr = out + err = cmd.Run() + if err != nil { + t.Log("Go compile errors:\n", out.String()) + } + require.NoError(t, err, "failed to compile go") +} diff --git a/web/web_test.go b/web/web_test.go index dfc13ced1c..d4862503a3 100644 --- a/web/web_test.go +++ b/web/web_test.go @@ -270,7 +270,7 @@ func TestPublicFilesRequest(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), nil) + env, err := plugin.NewEnvironment(th.NewPluginAPI, app.NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), nil) require.NoError(t, err) pluginID := "com.mattermost.sample"