[MM-26812] Add support for resumable file uploads (#15252)

* Implement AppendFile for FileBackend

* Split test into subtests

* [MM-26812] Add support for resumable file uploads (#15252)

* Implement UploadSession

* Implement UploadSessionStore

* Add error strings

* Implement resumable file uploads

* Add UploadType

* Fix retry layer tests

* Regenerate store layers

* Fix store error handling

* Use base for filename

* Prevent concurrent uploads on the same upload session

* Fix erroneus error string

* Improve error handling

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

* Fix translations

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/app_iface.go

@@ -360,6 +360,7 @@ type AppIface interface {
 	AddUserToTeamByToken(userId string, tokenId string) (*model.Team, *model.AppError)
 	AdjustImage(file io.Reader) (*bytes.Buffer, *model.AppError)
 	AllowOAuthAppAccessToUser(userId string, authRequest *model.AuthorizeRequest) (string, *model.AppError)
+	AppendFile(fr io.Reader, path string) (int64, *model.AppError)
 	AsymmetricSigningKey() *ecdsa.PrivateKey
 	AttachDeviceId(sessionId string, deviceId string, expiresAt int64) *model.AppError
 	AttachSessionCookies(w http.ResponseWriter, r *http.Request)
@@ -427,6 +428,7 @@ type AppIface interface {
 	CreateTeam(team *model.Team) (*model.Team, *model.AppError)
 	CreateTeamWithUser(team *model.Team, userId string) (*model.Team, *model.AppError)
 	CreateTermsOfService(text, userId string) (*model.TermsOfService, *model.AppError)
+	CreateUploadSession(us *model.UploadSession) (*model.UploadSession, *model.AppError)
 	CreateUserAccessToken(token *model.UserAccessToken) (*model.UserAccessToken, *model.AppError)
 	CreateUserAsAdmin(user *model.User, redirect string) (*model.User, *model.AppError)
 	CreateUserFromSignup(user *model.User, redirect string) (*model.User, *model.AppError)
@@ -670,6 +672,8 @@ type AppIface interface {
 	GetTeamsForUser(userId string) ([]*model.Team, *model.AppError)
 	GetTeamsUnreadForUser(excludeTeamId string, userId string) ([]*model.TeamUnread, *model.AppError)
 	GetTermsOfService(id string) (*model.TermsOfService, *model.AppError)
+	GetUploadSession(uploadId string) (*model.UploadSession, *model.AppError)
+	GetUploadSessionsForUser(userId string) ([]*model.UploadSession, *model.AppError)
 	GetUser(userId string) (*model.User, *model.AppError)
 	GetUserAccessToken(tokenId string, sanitize bool) (*model.UserAccessToken, *model.AppError)
 	GetUserAccessTokens(page, perPage int) ([]*model.UserAccessToken, *model.AppError)
@@ -980,6 +984,7 @@ type AppIface interface {
 	UpdateUserAuth(userId string, userAuth *model.UserAuth) (*model.UserAuth, *model.AppError)
 	UpdateUserNotifyProps(userId string, props map[string]string) (*model.User, *model.AppError)
 	UpdateUserRoles(userId string, newRoles string, sendWebSocketEvent bool) (*model.User, *model.AppError)
+	UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, *model.AppError)
 	UploadEmojiImage(id string, imageData *multipart.FileHeader) *model.AppError
 	UploadMultipartFiles(teamId string, channelId string, userId string, fileHeaders []*multipart.FileHeader, clientIds []string, now time.Time) (*model.FileUploadResponse, *model.AppError)
 	UpsertGroupMember(groupID string, userID string) (*model.GroupMember, *model.AppError)

app/emoji.go

@@ -124,7 +124,7 @@ func (a *App) UploadEmojiImage(id string, imageData *multipart.FileHeader) *mode
 	if config.Width > MaxEmojiWidth || config.Height > MaxEmojiHeight {
 		data := buf.Bytes()
 		newbuf := bytes.NewBuffer(nil)
-		info, err := model.GetInfoForBytes(imageData.Filename, data)
+		info, err := model.GetInfoForBytes(imageData.Filename, bytes.NewReader(data), len(data))
 		if err != nil {
 			return err
 		}

app/file.go

@@ -118,6 +118,15 @@ func (a *App) WriteFile(fr io.Reader, path string) (int64, *model.AppError) {
 	return backend.WriteFile(fr, path)
 }
 
+func (a *App) AppendFile(fr io.Reader, path string) (int64, *model.AppError) {
+	backend, err := a.FileBackend()
+	if err != nil {
+		return 0, err
+	}
+
+	return backend.AppendFile(fr, path)
+}
+
 func (a *App) RemoveFile(path string) *model.AppError {
 	backend, err := a.FileBackend()
 	if err != nil {
@@ -157,7 +166,7 @@ func (a *App) getInfoForFilename(post *model.Post, teamId, channelId, userId, ol
 		return nil
 	}
 
-	info, err := model.GetInfoForBytes(name, data)
+	info, err := model.GetInfoForBytes(name, bytes.NewReader(data), len(data))
 	if err != nil {
 		mlog.Warn(
 			"Unable to fully decode file info when migrating post to use FileInfos",
@@ -902,7 +911,7 @@ func (a *App) DoUploadFileExpectModification(now time.Time, rawTeamId string, ra
 	channelId := filepath.Base(rawChannelId)
 	userId := filepath.Base(rawUserId)
 
-	info, err := model.GetInfoForBytes(filename, data)
+	info, err := model.GetInfoForBytes(filename, bytes.NewReader(data), len(data))
 	if err != nil {
 		err.StatusCode = http.StatusBadRequest
 		return nil, data, err

app/opentracing/opentracing_layer.go

@@ -589,6 +589,28 @@ func (a *OpenTracingAppLayer) AllowOAuthAppAccessToUser(userId string, authReque
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) AppendFile(fr io.Reader, path string) (int64, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.AppendFile")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.AppendFile(fr, path)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) AsymmetricSigningKey() *ecdsa.PrivateKey {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.AsymmetricSigningKey")
@@ -2135,6 +2157,28 @@ func (a *OpenTracingAppLayer) CreateTermsOfService(text string, userId string) (
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CreateUploadSession")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.CreateUploadSession(us)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) CreateUser(user *model.User) (*model.User, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CreateUser")
@@ -8270,6 +8314,50 @@ func (a *OpenTracingAppLayer) GetTotalUsersStats(viewRestrictions *model.ViewUse
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) GetUploadSession(uploadId string) (*model.UploadSession, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUploadSession")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.GetUploadSession(uploadId)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
+func (a *OpenTracingAppLayer) GetUploadSessionsForUser(userId string) ([]*model.UploadSession, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUploadSessionsForUser")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.GetUploadSessionsForUser(userId)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) GetUser(userId string) (*model.User, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUser")
@@ -15019,6 +15107,28 @@ func (a *OpenTracingAppLayer) UpdateWebConnUserActivity(session model.Session, a
 	a.app.UpdateWebConnUserActivity(session, activityAt)
 }
 
+func (a *OpenTracingAppLayer) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UploadData")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.UploadData(us, rd)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) UploadEmojiImage(id string, imageData *multipart.FileHeader) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UploadEmojiImage")

app/server.go

@@ -170,6 +170,12 @@ type Server struct {
 	CacheProvider cache.Provider
 
 	tracer *tracing.Tracer
+
+	// These are used to prevent concurrent upload requests
+	// for a given upload session which could cause inconsistencies
+	// and data corruption.
+	uploadLockMapMut sync.Mutex
+	uploadLockMap    map[string]bool
 }
 
 func NewServer(options ...Option) (*Server, error) {
@@ -182,6 +188,7 @@ func NewServer(options ...Option) (*Server, error) {
 		LocalRouter:         localRouter,
 		licenseListeners:    map[string]func(*model.License, *model.License){},
 		hashSeed:            maphash.MakeSeed(),
+		uploadLockMap:       map[string]bool{},
 	}
 
 	for _, option := range options {

app/upload.go

@@ -0,0 +1,248 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package app
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/mlog"
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/plugin"
+	"github.com/mattermost/mattermost-server/v5/store"
+)
+
+const minFirstPartSize = 5 * 1024 * 1024 // 5MB
+
+func (a *App) CreateUploadSession(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]interface{}{"channelId": us.ChannelId}, "", http.StatusRequestEntityTooLarge)
+	}
+
+	us.FileOffset = 0
+	now := time.Now()
+	us.CreateAt = model.GetMillisForTime(now)
+	us.Path = now.Format("20060102") + "/teams/noteam/channels/" + us.ChannelId + "/users/" + us.UserId + "/" + us.Id + "/" + filepath.Base(us.Filename)
+	if err := us.IsValid(); err != nil {
+		return nil, err
+	}
+
+	channel, err := a.GetChannel(us.ChannelId)
+	if err != nil {
+		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.incorrect_channel_id.app_error",
+			map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
+	}
+	if channel.DeleteAt != 0 {
+		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.cannot_upload_to_deleted_channel.app_error",
+			map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
+	}
+
+	us, storeErr := a.Srv().Store.UploadSession().Save(us)
+	if storeErr != nil {
+		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.save.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
+	}
+
+	return us, nil
+}
+
+func (a *App) GetUploadSession(uploadId string) (*model.UploadSession, *model.AppError) {
+	us, err := a.Srv().Store.UploadSession().Get(uploadId)
+	if err != nil {
+		var nfErr *store.ErrNotFound
+		switch {
+		case errors.As(err, &nfErr):
+			return nil, model.NewAppError("GetUpload", "app.upload.get.app_error",
+				nil, nfErr.Error(), http.StatusNotFound)
+		default:
+			return nil, model.NewAppError("GetUpload", "app.upload.get.app_error",
+				nil, err.Error(), http.StatusInternalServerError)
+		}
+	}
+	return us, nil
+}
+
+func (a *App) GetUploadSessionsForUser(userId string) ([]*model.UploadSession, *model.AppError) {
+	uss, err := a.Srv().Store.UploadSession().GetForUser(userId)
+	if err != nil {
+		return nil, model.NewAppError("GetUploadsForUser", "app.upload.get_for_user.app_error",
+			nil, err.Error(), http.StatusInternalServerError)
+	}
+	return uss, nil
+}
+
+func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, *model.AppError) {
+	// prevent more than one caller to upload data at the same time for a given upload session.
+	// This is to avoid possible inconsistencies.
+	a.Srv().uploadLockMapMut.Lock()
+	locked := a.Srv().uploadLockMap[us.Id]
+	if locked {
+		// session lock is already taken, return error.
+		a.Srv().uploadLockMapMut.Unlock()
+		return nil, model.NewAppError("UploadData", "app.upload.upload_data.concurrent.app_error",
+			nil, "", http.StatusBadRequest)
+	}
+	// grab the session lock.
+	a.Srv().uploadLockMap[us.Id] = true
+	a.Srv().uploadLockMapMut.Unlock()
+
+	// reset the session lock on exit.
+	defer func() {
+		a.Srv().uploadLockMapMut.Lock()
+		delete(a.Srv().uploadLockMap, us.Id)
+		a.Srv().uploadLockMapMut.Unlock()
+	}()
+
+	// make sure it's not possible to upload more data than what is expected.
+	lr := &io.LimitedReader{
+		R: rd,
+		N: us.FileSize - us.FileOffset,
+	}
+	var err *model.AppError
+	var written int64
+	if us.FileOffset == 0 {
+		// new upload
+		written, err = a.WriteFile(lr, us.Path)
+		if err != nil && written == 0 {
+			return nil, err
+		}
+		if written < minFirstPartSize && written != us.FileSize {
+			a.RemoveFile(us.Path)
+			var errStr string
+			if err != nil {
+				errStr = err.Error()
+			}
+			return nil, model.NewAppError("UploadData", "app.upload.upload_data.first_part_too_small.app_error",
+				map[string]interface{}{"Size": minFirstPartSize}, errStr, http.StatusBadRequest)
+		}
+	} else if us.FileOffset < us.FileSize {
+		// resume upload
+		written, err = a.AppendFile(lr, us.Path)
+	}
+	if written > 0 {
+		us.FileOffset += written
+		if storeErr := a.Srv().Store.UploadSession().Update(us); storeErr != nil {
+			return nil, model.NewAppError("UploadData", "app.upload.upload_data.update.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	// upload is incomplete
+	if us.FileOffset != us.FileSize {
+		return nil, nil
+	}
+
+	// upload is done, create FileInfo
+	file, err := a.FileReader(us.Path)
+	if err != nil {
+		return nil, model.NewAppError("UploadData", "app.upload.upload_data.read_file.app_error", nil, err.Error(), http.StatusInternalServerError)
+	}
+
+	info, err := model.GetInfoForBytes(us.Filename, file, int(us.FileSize))
+	file.Close()
+	if err != nil {
+		return nil, err
+	}
+
+	info.CreatorId = us.UserId
+	info.Path = us.Path
+
+	// call plugins upload hook
+	if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
+		// using a pipe to avoid loading the whole file content in memory.
+		r, w := io.Pipe()
+		errChan := make(chan *model.AppError, 1)
+		go func() {
+			defer w.Close()
+			defer close(errChan)
+			pluginContext := a.PluginContext()
+			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
+				newInfo, rejStr := hooks.FileWillBeUploaded(pluginContext, info, file, w)
+				if rejStr != "" {
+					errChan <- model.NewAppError("UploadData", "File rejected by plugin. "+rejStr, nil, "", http.StatusBadRequest)
+					return false
+				}
+				if newInfo != nil {
+					info = newInfo
+				}
+				return true
+			}, plugin.FileWillBeUploadedId)
+		}()
+
+		var written int64
+		tmpPath := us.Path + ".tmp"
+		written, err = a.WriteFile(r, tmpPath)
+		if err != nil {
+			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+				mlog.Error("Failed to remove file", mlog.Err(fileErr))
+			}
+			return nil, err
+		}
+
+		if err = <-errChan; err != nil {
+			if fileErr := a.RemoveFile(us.Path); fileErr != nil {
+				mlog.Error("Failed to remove file", mlog.Err(fileErr))
+			}
+			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+				mlog.Error("Failed to remove file", mlog.Err(fileErr))
+			}
+			return nil, err
+		}
+
+		if written > 0 {
+			info.Size = written
+			if fileErr := a.MoveFile(tmpPath, us.Path); fileErr != nil {
+				mlog.Error("Failed to move file", mlog.Err(fileErr))
+			}
+		} else {
+			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+				mlog.Error("Failed to remove file", mlog.Err(fileErr))
+			}
+		}
+	}
+
+	// image post-processing
+	if info.IsImage() {
+		// Check dimensions before loading the whole thing into memory later on
+		// This casting is done to prevent overflow on 32 bit systems (not needed
+		// in 64 bits systems because images can't have more than 32 bits height or
+		// width)
+		if int64(info.Width)*int64(info.Height) > MaxImageSize {
+			return nil, model.NewAppError("uploadData", "app.upload.upload_data.large_image.app_error",
+				map[string]interface{}{"Filename": us.Filename, "Width": info.Width, "Height": info.Height}, "", http.StatusBadRequest)
+		}
+		nameWithoutExtension := info.Name[:strings.LastIndex(info.Name, ".")]
+		info.PreviewPath = filepath.Dir(info.Path) + "/" + nameWithoutExtension + "_preview.jpg"
+		info.ThumbnailPath = filepath.Dir(info.Path) + "/" + nameWithoutExtension + "_thumb.jpg"
+		imgData, fileErr := a.ReadFile(us.Path)
+		if fileErr != nil {
+			return nil, fileErr
+		}
+		a.HandleImages([]string{info.PreviewPath}, []string{info.ThumbnailPath}, [][]byte{imgData})
+	}
+
+	var storeErr error
+	if info, storeErr = a.Srv().Store.FileInfo().Save(info); storeErr != nil {
+		var appErr *model.AppError
+		switch {
+		case errors.As(storeErr, &appErr):
+			return nil, appErr
+		default:
+			return nil, model.NewAppError("uploadData", "app.upload.upload_data.save.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
+		}
+	}
+
+	// delete upload session
+	if storeErr := a.Srv().Store.UploadSession().Delete(us.Id); storeErr != nil {
+		mlog.Error("Failed to delete UploadSession", mlog.Err(storeErr))
+	}
+
+	return info, nil
+}

app/upload_test.go

@@ -0,0 +1,308 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package app
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/utils/fileutils"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCreateUploadSession(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	us := &model.UploadSession{
+		Type:      model.UploadTypeAttachment,
+		UserId:    th.BasicUser.Id,
+		ChannelId: th.BasicChannel.Id,
+		Filename:  "upload",
+		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(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(us)
+		require.NotNil(t, err)
+		require.Equal(t, "model.upload_session.is_valid.id.app_error", err.Id)
+		require.Nil(t, u)
+	})
+
+	t.Run("invalid UserId", func(t *testing.T) {
+		us.Id = model.NewId()
+		us.UserId = ""
+		u, err := th.App.CreateUploadSession(us)
+		require.NotNil(t, err)
+		require.Equal(t, "model.upload_session.is_valid.user_id.app_error", err.Id)
+		require.Nil(t, u)
+	})
+
+	t.Run("invalid ChannelId", func(t *testing.T) {
+		us.UserId = th.BasicUser.Id
+		us.ChannelId = ""
+		u, err := th.App.CreateUploadSession(us)
+		require.NotNil(t, err)
+		require.Equal(t, "model.upload_session.is_valid.channel_id.app_error", err.Id)
+		require.Nil(t, u)
+	})
+
+	t.Run("non-existing channel", func(t *testing.T) {
+		us.ChannelId = model.NewId()
+		u, err := th.App.CreateUploadSession(us)
+		require.NotNil(t, err)
+		require.Equal(t, "app.upload.create.incorrect_channel_id.app_error", err.Id)
+		require.Nil(t, u)
+	})
+
+	t.Run("deleted channel", func(t *testing.T) {
+		ch := th.CreateChannel(th.BasicTeam)
+		th.App.DeleteChannel(ch, th.BasicUser.Id)
+		us.ChannelId = ch.Id
+		u, err := th.App.CreateUploadSession(us)
+		require.NotNil(t, err)
+		require.Equal(t, "app.upload.create.cannot_upload_to_deleted_channel.app_error", err.Id)
+		require.Nil(t, u)
+	})
+
+	t.Run("success", func(t *testing.T) {
+		us.ChannelId = th.BasicChannel.Id
+		u, err := th.App.CreateUploadSession(us)
+		require.Nil(t, err)
+		require.NotEmpty(t, u)
+	})
+}
+
+func TestUploadData(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	us := &model.UploadSession{
+		Id:        model.NewId(),
+		Type:      model.UploadTypeAttachment,
+		UserId:    th.BasicUser.Id,
+		ChannelId: th.BasicChannel.Id,
+		Filename:  "upload",
+		FileSize:  8 * 1024 * 1024,
+	}
+
+	var err error
+	us, err = th.App.CreateUploadSession(us)
+	require.Nil(t, err)
+	require.NotEmpty(t, us)
+
+	data := make([]byte, us.FileSize)
+	_, err2 := rand.Read(data)
+	require.NoError(t, err2)
+
+	t.Run("write error", func(t *testing.T) {
+		rd := &io.LimitedReader{
+			R: bytes.NewReader(data),
+			N: 1024 * 1024,
+		}
+
+		ok, err := th.App.FileExists(us.Path)
+		require.False(t, ok)
+		require.Nil(t, err)
+
+		u := *us
+		u.Path = ""
+		info, err := th.App.UploadData(&u, rd)
+		require.Nil(t, info)
+		require.NotNil(t, err)
+		require.NotEqual(t, "app.upload.upload_data.first_part_too_small.app_error", err.Id)
+	})
+
+	t.Run("first part too small", func(t *testing.T) {
+		rd := &io.LimitedReader{
+			R: bytes.NewReader(data),
+			N: 1024 * 1024,
+		}
+
+		ok, err := th.App.FileExists(us.Path)
+		require.False(t, ok)
+		require.Nil(t, err)
+
+		info, err := th.App.UploadData(us, rd)
+		require.Nil(t, info)
+		require.NotNil(t, err)
+		require.Equal(t, "app.upload.upload_data.first_part_too_small.app_error", err.Id)
+
+		ok, err = th.App.FileExists(us.Path)
+		require.False(t, ok)
+		require.Nil(t, err)
+	})
+
+	t.Run("resume success", func(t *testing.T) {
+		rd := &io.LimitedReader{
+			R: bytes.NewReader(data),
+			N: 5 * 1024 * 1024,
+		}
+		info, err := th.App.UploadData(us, rd)
+		require.Nil(t, info)
+		require.Nil(t, err)
+
+		rd = &io.LimitedReader{
+			R: bytes.NewReader(data[5*1024*1024:]),
+			N: 3 * 1024 * 1024,
+		}
+		info, err = th.App.UploadData(us, rd)
+		require.Nil(t, err)
+		require.NotEmpty(t, info)
+
+		d, err := th.App.ReadFile(us.Path)
+		require.Nil(t, err)
+		require.Equal(t, data, d)
+	})
+
+	t.Run("all at once success", func(t *testing.T) {
+		us.Id = model.NewId()
+		us, err = th.App.CreateUploadSession(us)
+		require.Nil(t, err)
+		require.NotEmpty(t, us)
+
+		info, err := th.App.UploadData(us, bytes.NewReader(data))
+		require.Nil(t, err)
+		require.NotEmpty(t, info)
+
+		d, err := th.App.ReadFile(us.Path)
+		require.Nil(t, err)
+		require.Equal(t, data, d)
+	})
+
+	t.Run("small file success", func(t *testing.T) {
+		us.Id = model.NewId()
+		us.FileSize = 1024 * 1024
+		us, err = th.App.CreateUploadSession(us)
+		require.Nil(t, err)
+		require.NotEmpty(t, us)
+
+		rd := &io.LimitedReader{
+			R: bytes.NewReader(data),
+			N: 1024 * 1024,
+		}
+		info, err := th.App.UploadData(us, rd)
+		require.Nil(t, err)
+		require.NotEmpty(t, info)
+
+		d, err := th.App.ReadFile(us.Path)
+		require.Nil(t, err)
+		require.Equal(t, data[:1024*1024], d)
+	})
+
+	t.Run("image processing", func(t *testing.T) {
+		testDir, _ := fileutils.FindDir("tests")
+		data, err := ioutil.ReadFile(filepath.Join(testDir, "test.png"))
+		require.Nil(t, err)
+		require.NotEmpty(t, data)
+
+		us.Id = model.NewId()
+		us.Filename = "test.png"
+		us.FileSize = int64(len(data))
+		us, err = th.App.CreateUploadSession(us)
+		require.Nil(t, err)
+		require.NotEmpty(t, us)
+
+		info, err := th.App.UploadData(us, bytes.NewReader(data))
+		require.Nil(t, err)
+		require.NotEmpty(t, info)
+		require.NotZero(t, info.Width)
+		require.NotZero(t, info.Height)
+		require.NotEmpty(t, info.ThumbnailPath)
+		require.NotEmpty(t, info.PreviewPath)
+	})
+}
+
+func TestUploadDataConcurrent(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	us := &model.UploadSession{
+		Id:        model.NewId(),
+		Type:      model.UploadTypeAttachment,
+		UserId:    th.BasicUser.Id,
+		ChannelId: th.BasicChannel.Id,
+		Filename:  "upload",
+		FileSize:  8 * 1024 * 1024,
+	}
+
+	var err error
+	us, err = th.App.CreateUploadSession(us)
+	require.Nil(t, err)
+	require.NotEmpty(t, us)
+
+	data := make([]byte, us.FileSize)
+	_, err2 := rand.Read(data)
+	require.NoError(t, err2)
+
+	var nErrs int32
+	var wg sync.WaitGroup
+	n := 8
+	wg.Add(n)
+
+	for i := 0; i < n; i++ {
+		go func() {
+			defer wg.Done()
+			rd := &io.LimitedReader{
+				R: bytes.NewReader(data),
+				N: 5 * 1024 * 1024,
+			}
+			u := *us
+			_, err := th.App.UploadData(&u, rd)
+			if err != nil && err.Id == "app.upload.upload_data.concurrent.app_error" {
+				atomic.AddInt32(&nErrs, 1)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// Verify that only 1 request was able to perform the upload.
+	require.Equal(t, int32(n-1), nErrs)
+
+	wg.Add(n)
+
+	for i := 0; i < n; i++ {
+		go func() {
+			defer wg.Done()
+			rd := &io.LimitedReader{
+				R: bytes.NewReader(data[5*1024*1024:]),
+				N: 3 * 1024 * 1024,
+			}
+			u := *us
+			u.FileOffset = 5 * 1024 * 1024
+			_, err := th.App.UploadData(&u, rd)
+			if err != nil && err.Id == "app.upload.upload_data.concurrent.app_error" {
+				atomic.AddInt32(&nErrs, 1)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// Verify that only 1 request was able to finish the upload.
+	require.Equal(t, int32(n*2-2), nErrs)
+
+	d, err := th.App.ReadFile(us.Path)
+	require.Nil(t, err)
+	require.Equal(t, data, d)
+}