[MM-48523] Expose resumable uploads API to plugins (#21700)

* Expose resumable uploads API to plugins

* Update translations

api4/upload.go

@@ -65,6 +65,13 @@ func createUpload(c *Context, w http.ResponseWriter, r *http.Request) {
 	if c.AppContext.Session().UserId != "" {
 		us.UserId = c.AppContext.Session().UserId
 	}
+
+	if us.FileSize > *c.App.Config().FileSettings.MaxFileSize {
+		c.Err = model.NewAppError("createUpload", "api.upload.create.upload_too_large.app_error",
+			map[string]any{"channelId": us.ChannelId}, "", http.StatusRequestEntityTooLarge)
+		return
+	}
+
 	rus, err := c.App.CreateUploadSession(c.AppContext, &us)
 	if err != nil {
 		c.Err = err

api4/upload_test.go

@@ -45,6 +45,17 @@ func TestCreateUpload(t *testing.T) {
 		require.Equal(t, http.StatusForbidden, resp.StatusCode)
 	})
 
+	t.Run("FileSize over limit", func(t *testing.T) {
+		maxFileSize := *th.App.Config().FileSettings.MaxFileSize
+		th.App.UpdateConfig(func(cfg *model.Config) { *cfg.FileSettings.MaxFileSize = us.FileSize - 1 })
+		defer th.App.UpdateConfig(func(cfg *model.Config) { *cfg.FileSettings.MaxFileSize = maxFileSize })
+		us.ChannelId = th.BasicChannel.Id
+		u, resp, err := th.Client.CreateUpload(us)
+		require.Nil(t, u)
+		CheckErrorID(t, err, "api.upload.create.upload_too_large.app_error")
+		require.Equal(t, http.StatusRequestEntityTooLarge, resp.StatusCode)
+	})
+
 	t.Run("not allowed in cloud", func(t *testing.T) {
 		th.App.Srv().SetLicense(model.NewTestLicense("cloud"))
 		defer th.App.Srv().RemoveLicense()

app/plugin_api.go

@@ -1237,3 +1237,27 @@ func (api *PluginAPI) GetCloudLimits() (*model.ProductLimits, error) {
 func (api *PluginAPI) RegisterCollectionAndTopic(collectionType, topicType string) error {
 	return api.app.registerCollectionAndTopic(api.id, collectionType, topicType)
 }
+
+func (api *PluginAPI) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error) {
+	us, err := api.app.CreateUploadSession(api.ctx, us)
+	if err != nil {
+		return nil, err
+	}
+	return us, nil
+}
+
+func (api *PluginAPI) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error) {
+	fi, err := api.app.UploadData(api.ctx, us, rd)
+	if err != nil {
+		return nil, err
+	}
+	return fi, nil
+}
+
+func (api *PluginAPI) GetUploadSession(uploadID string) (*model.UploadSession, error) {
+	fi, err := api.app.GetUploadSession(uploadID)
+	if err != nil {
+		return nil, err
+	}
+	return fi, nil
+}

app/plugin_api_test.go

@@ -2100,3 +2100,98 @@ func TestRegisterCollectionAndTopic(t *testing.T) {
 	err = api.RegisterCollectionAndTopic("some other collection", "topicToBeRepeated")
 	assert.Error(t, err)
 }
+
+func TestPluginUploadsAPI(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	pluginCode := fmt.Sprintf(`
+    package main
+
+    import (
+		  "fmt"
+			"bytes"
+
+      "github.com/mattermost/mattermost-server/v6/model"
+      "github.com/mattermost/mattermost-server/v6/plugin"
+    )
+
+    type TestPlugin struct {
+      plugin.MattermostPlugin
+    }
+
+	  func (p *TestPlugin) OnActivate() error {
+		  data := []byte("some content to upload")
+			us, err := p.API.CreateUploadSession(&model.UploadSession{
+			  Id: "%s",
+				UserId: "%s",
+				ChannelId: "%s",
+				Type: model.UploadTypeAttachment,
+			  FileSize: int64(len(data)),
+				Filename: "upload.test",
+			})
+			if err != nil {
+			  return fmt.Errorf("failed to create upload session: %%w", err)
+			}
+
+			us2, err := p.API.GetUploadSession(us.Id)
+			if err != nil {
+			  return fmt.Errorf("failed to get upload session: %%w", err)
+			}
+
+			if us.Id != us2.Id {
+			  return fmt.Errorf("upload sessions should match")
+			}
+
+			fi, err := p.API.UploadData(us, bytes.NewBuffer(data))
+			if err != nil {
+			  return fmt.Errorf("failed to upload data: %%w", err)
+			}
+
+			if fi == nil || fi.Id == "" {
+			  return fmt.Errorf("fileinfo should be set")
+			}
+
+			fileData, appErr := p.API.GetFile(fi.Id)
+			if appErr != nil {
+			  return fmt.Errorf("failed to get file data: %%w", err)
+			}
+
+			if !bytes.Equal(data, fileData) {
+			  return fmt.Errorf("file data should match")
+			}
+
+		  return nil
+	  }
+
+    func main() {
+      plugin.ClientMain(&TestPlugin{})
+    }
+  `, model.NewId(), th.BasicUser.Id, th.BasicChannel.Id)
+
+	pluginDir, err := os.MkdirTemp("", "")
+	require.NoError(t, err)
+	webappPluginDir, err := os.MkdirTemp("", "")
+	require.NoError(t, err)
+	defer os.RemoveAll(pluginDir)
+	defer os.RemoveAll(webappPluginDir)
+
+	newPluginAPI := func(manifest *model.Manifest) plugin.API {
+		return th.App.NewPluginAPI(th.Context, manifest)
+	}
+	env, err := plugin.NewEnvironment(newPluginAPI, NewDriverImpl(th.App.Srv()), pluginDir, webappPluginDir, th.App.Log(), nil)
+	require.NoError(t, err)
+
+	th.App.ch.SetPluginsEnvironment(env)
+
+	pluginID := "testplugin"
+	pluginManifest := `{"id": "testplugin", "server": {"executable": "backend.exe"}}`
+	backend := filepath.Join(pluginDir, pluginID, "backend.exe")
+	utils.CompileGo(t, pluginCode, backend)
+
+	os.WriteFile(filepath.Join(pluginDir, pluginID, "plugin.json"), []byte(pluginManifest), 0600)
+	manifest, activated, reterr := env.Activate(pluginID)
+	require.NoError(t, reterr)
+	require.NotNil(t, manifest)
+	require.True(t, activated)
+}

app/upload.go

@@ -127,11 +127,6 @@ func (a *App) runPluginsHook(c *request.Context, info *model.FileInfo, file io.R
 }
 
 func (a *App) CreateUploadSession(c request.CTX, us *model.UploadSession) (*model.UploadSession, *model.AppError) {
-	if us.FileSize > *a.Config().FileSettings.MaxFileSize {
-		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.upload_too_large.app_error",
-			map[string]any{"channelId": us.ChannelId}, "", http.StatusRequestEntityTooLarge)
-	}
-
 	us.FileOffset = 0
 	now := time.Now()
 	us.CreateAt = model.GetMillisForTime(now)

app/upload_test.go

@@ -32,16 +32,6 @@ func TestCreateUploadSession(t *testing.T) {
 		FileSize:  8 * 1024 * 1024,
 	}
 
-	t.Run("FileSize over limit", func(t *testing.T) {
-		maxFileSize := *th.App.Config().FileSettings.MaxFileSize
-		th.App.UpdateConfig(func(cfg *model.Config) { *cfg.FileSettings.MaxFileSize = us.FileSize - 1 })
-		defer th.App.UpdateConfig(func(cfg *model.Config) { *cfg.FileSettings.MaxFileSize = maxFileSize })
-		u, err := th.App.CreateUploadSession(th.Context, us)
-		require.NotNil(t, err)
-		require.Equal(t, "app.upload.create.upload_too_large.app_error", err.Id)
-		require.Nil(t, u)
-	})
-
 	t.Run("invalid Id", func(t *testing.T) {
 		u, err := th.App.CreateUploadSession(th.Context, us)
 		require.NotNil(t, err)

i18n/en.json

@@ -3907,6 +3907,10 @@
     "id": "api.upgrade_to_enterprise_status.signature.app_error",
     "translation": "Mattermost was unable to upgrade to Enterprise Edition. The digital signature of the downloaded binary file could not be verified."
   },
+  {
+    "id": "api.upload.create.upload_too_large.app_error",
+    "translation": "Unable to upload file. File is too large."
+  },
   {
     "id": "api.upload.get_upload.forbidden.app_error",
     "translation": "Failed to get upload."
@@ -6539,10 +6543,6 @@
     "id": "app.upload.create.save.app_error",
     "translation": "Failed to save upload."
   },
-  {
-    "id": "app.upload.create.upload_too_large.app_error",
-    "translation": "Unable to upload file. File is too large."
-  },
   {
     "id": "app.upload.get.app_error",
     "translation": "Failed to get upload."

plugin/api.go

@@ -1168,6 +1168,24 @@ type API interface {
 	//
 	// Minimum server version: 7.6
 	RegisterCollectionAndTopic(collectionType, topicType string) error
+
+	// CreateUploadSession creates and returns a new (resumable) upload session.
+	//
+	// @tag Upload
+	// Minimum server version: 7.6
+	CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error)
+
+	// UploadData uploads the data for a given upload session.
+	//
+	// @tag Upload
+	// Minimum server version: 7.6
+	UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error)
+
+	// GetUploadSession returns the upload session for the provided id.
+	//
+	// @tag Upload
+	// Minimum server version: 7.6
+	GetUploadSession(uploadID string) (*model.UploadSession, error)
 }
 
 var handshake = plugin.HandshakeConfig{

plugin/api_timer_layer_generated.go

@@ -1246,3 +1246,24 @@ func (api *apiTimerLayer) RegisterCollectionAndTopic(collectionType, topicType s
 	api.recordTime(startTime, "RegisterCollectionAndTopic", _returnsA == nil)
 	return _returnsA
 }
+
+func (api *apiTimerLayer) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error) {
+	startTime := timePkg.Now()
+	_returnsA, _returnsB := api.apiImpl.CreateUploadSession(us)
+	api.recordTime(startTime, "CreateUploadSession", _returnsB == nil)
+	return _returnsA, _returnsB
+}
+
+func (api *apiTimerLayer) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error) {
+	startTime := timePkg.Now()
+	_returnsA, _returnsB := api.apiImpl.UploadData(us, rd)
+	api.recordTime(startTime, "UploadData", _returnsB == nil)
+	return _returnsA, _returnsB
+}
+
+func (api *apiTimerLayer) GetUploadSession(uploadID string) (*model.UploadSession, error) {
+	startTime := timePkg.Now()
+	_returnsA, _returnsB := api.apiImpl.GetUploadSession(uploadID)
+	api.recordTime(startTime, "GetUploadSession", _returnsB == nil)
+	return _returnsA, _returnsB
+}

plugin/client_rpc.go

@@ -866,3 +866,55 @@ func (s *apiRPCServer) InstallPlugin(args *Z_InstallPluginArgs, returns *Z_Insta
 	returns.A, returns.B = hook.InstallPlugin(pluginReader, args.B)
 	return nil
 }
+
+type Z_UploadDataArgs struct {
+	A              *model.UploadSession
+	PluginStreamID uint32
+}
+
+type Z_UploadDataReturns struct {
+	A *model.FileInfo
+	B error
+}
+
+func (g *apiRPCClient) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error) {
+	pluginStreamID := g.muxBroker.NextId()
+
+	go func() {
+		pluginConnection, err := g.muxBroker.Accept(pluginStreamID)
+		if err != nil {
+			log.Print("Failed to upload data. MuxBroker could not Accept connection", mlog.Err(err))
+			return
+		}
+		defer pluginConnection.Close()
+		serveIOReader(rd, pluginConnection)
+	}()
+
+	_args := &Z_UploadDataArgs{us, pluginStreamID}
+	_returns := &Z_UploadDataReturns{}
+	if err := g.client.Call("Plugin.UploadData", _args, _returns); err != nil {
+		log.Print("RPC call UploadData to plugin failed.", mlog.Err(err))
+	}
+
+	return _returns.A, _returns.B
+}
+
+func (s *apiRPCServer) UploadData(args *Z_UploadDataArgs, returns *Z_UploadDataReturns) error {
+	hook, ok := s.impl.(interface {
+		UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error)
+	})
+	if !ok {
+		return encodableError(fmt.Errorf("API UploadData called but not implemented"))
+	}
+
+	receivePluginConnection, err := s.muxBroker.Dial(args.PluginStreamID)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote plugin stream, error: %v", err.Error())
+		return err
+	}
+	pluginReader := connectIOReader(receivePluginConnection)
+	defer pluginReader.Close()
+
+	returns.A, returns.B = hook.UploadData(args.A, pluginReader)
+	return nil
+}

plugin/client_rpc_generated.go

@@ -5968,3 +5968,63 @@ func (s *apiRPCServer) RegisterCollectionAndTopic(args *Z_RegisterCollectionAndT
 	}
 	return nil
 }
+
+type Z_CreateUploadSessionArgs struct {
+	A *model.UploadSession
+}
+
+type Z_CreateUploadSessionReturns struct {
+	A *model.UploadSession
+	B error
+}
+
+func (g *apiRPCClient) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error) {
+	_args := &Z_CreateUploadSessionArgs{us}
+	_returns := &Z_CreateUploadSessionReturns{}
+	if err := g.client.Call("Plugin.CreateUploadSession", _args, _returns); err != nil {
+		log.Printf("RPC call to CreateUploadSession API failed: %s", err.Error())
+	}
+	return _returns.A, _returns.B
+}
+
+func (s *apiRPCServer) CreateUploadSession(args *Z_CreateUploadSessionArgs, returns *Z_CreateUploadSessionReturns) error {
+	if hook, ok := s.impl.(interface {
+		CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error)
+	}); ok {
+		returns.A, returns.B = hook.CreateUploadSession(args.A)
+		returns.B = encodableError(returns.B)
+	} else {
+		return encodableError(fmt.Errorf("API CreateUploadSession called but not implemented."))
+	}
+	return nil
+}
+
+type Z_GetUploadSessionArgs struct {
+	A string
+}
+
+type Z_GetUploadSessionReturns struct {
+	A *model.UploadSession
+	B error
+}
+
+func (g *apiRPCClient) GetUploadSession(uploadID string) (*model.UploadSession, error) {
+	_args := &Z_GetUploadSessionArgs{uploadID}
+	_returns := &Z_GetUploadSessionReturns{}
+	if err := g.client.Call("Plugin.GetUploadSession", _args, _returns); err != nil {
+		log.Printf("RPC call to GetUploadSession API failed: %s", err.Error())
+	}
+	return _returns.A, _returns.B
+}
+
+func (s *apiRPCServer) GetUploadSession(args *Z_GetUploadSessionArgs, returns *Z_GetUploadSessionReturns) error {
+	if hook, ok := s.impl.(interface {
+		GetUploadSession(uploadID string) (*model.UploadSession, error)
+	}); ok {
+		returns.A, returns.B = hook.GetUploadSession(args.A)
+		returns.B = encodableError(returns.B)
+	} else {
+		return encodableError(fmt.Errorf("API GetUploadSession called but not implemented."))
+	}
+	return nil
+}

plugin/interface_generator/main.go

@@ -35,6 +35,7 @@ var excludedPluginHooks = []string{
 	"OnActivate",
 	"PluginHTTP",
 	"ServeHTTP",
+	"UploadData",
 }
 
 var excludedProductHooks = []string{

plugin/plugintest/api.go

@@ -391,6 +391,29 @@ func (_m *API) CreateTeamMembersGracefully(teamID string, userIds []string, requ
 	return r0, r1
 }
 
+// CreateUploadSession provides a mock function with given fields: us
+func (_m *API) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, error) {
+	ret := _m.Called(us)
+
+	var r0 *model.UploadSession
+	if rf, ok := ret.Get(0).(func(*model.UploadSession) *model.UploadSession); ok {
+		r0 = rf(us)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*model.UploadSession)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*model.UploadSession) error); ok {
+		r1 = rf(us)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // CreateUser provides a mock function with given fields: user
 func (_m *API) CreateUser(user *model.User) (*model.User, *model.AppError) {
 	ret := _m.Called(user)
@@ -2181,6 +2204,29 @@ func (_m *API) GetUnsanitizedConfig() *model.Config {
 	return r0
 }
 
+// GetUploadSession provides a mock function with given fields: uploadID
+func (_m *API) GetUploadSession(uploadID string) (*model.UploadSession, error) {
+	ret := _m.Called(uploadID)
+
+	var r0 *model.UploadSession
+	if rf, ok := ret.Get(0).(func(string) *model.UploadSession); ok {
+		r0 = rf(uploadID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*model.UploadSession)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(uploadID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetUser provides a mock function with given fields: userID
 func (_m *API) GetUser(userID string) (*model.User, *model.AppError) {
 	ret := _m.Called(userID)
@@ -3717,6 +3763,29 @@ func (_m *API) UpdateUserStatus(userID string, status string) (*model.Status, *m
 	return r0, r1
 }
 
+// UploadData provides a mock function with given fields: us, rd
+func (_m *API) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error) {
+	ret := _m.Called(us, rd)
+
+	var r0 *model.FileInfo
+	if rf, ok := ret.Get(0).(func(*model.UploadSession, io.Reader) *model.FileInfo); ok {
+		r0 = rf(us, rd)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*model.FileInfo)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*model.UploadSession, io.Reader) error); ok {
+		r1 = rf(us, rd)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // UploadFile provides a mock function with given fields: data, channelId, filename
 func (_m *API) UploadFile(data []byte, channelId string, filename string) (*model.FileInfo, *model.AppError) {
 	ret := _m.Called(data, channelId, filename)