[MM-38705] Move import validators into own package (#20914)
Этот коммит содержится в:
Tim Scheuermann
коммит произвёл
GitHub
GitHub
родитель
a5ea445bf9
Коммит
aef5165cdc
@@ -15,17 +15,20 @@ import (
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/app/imports"
|
||||
"github.com/mattermost/mattermost-server/v6/app/request"
|
||||
"github.com/mattermost/mattermost-server/v6/model"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
)
|
||||
|
||||
type ReactionImportData = imports.ReactionImportData // part of the app interface
|
||||
|
||||
const (
|
||||
importMultiplePostsThreshold = 1000
|
||||
maxScanTokenSize = 16 * 1024 * 1024 // Need to set a higher limit than default because some customers cross the limit. See MM-22314
|
||||
)
|
||||
|
||||
func stopOnError(c request.CTX, err LineImportWorkerError) bool {
|
||||
func stopOnError(c request.CTX, err imports.LineImportWorkerError) bool {
|
||||
switch err.Error.Id {
|
||||
case "api.file.upload_file.large_image.app_error":
|
||||
c.Logger().Warn("Large image import error", mlog.Err(err.Error))
|
||||
@@ -38,7 +41,7 @@ func stopOnError(c request.CTX, err LineImportWorkerError) bool {
|
||||
}
|
||||
}
|
||||
|
||||
func processAttachmentPaths(files *[]AttachmentImportData, basePath string, filesMap map[string]*zip.File) error {
|
||||
func processAttachmentPaths(files *[]imports.AttachmentImportData, basePath string, filesMap map[string]*zip.File) error {
|
||||
if files == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -58,11 +61,11 @@ func processAttachmentPaths(files *[]AttachmentImportData, basePath string, file
|
||||
return nil
|
||||
}
|
||||
|
||||
func processAttachments(line *LineImportData, basePath string, filesMap map[string]*zip.File) error {
|
||||
func processAttachments(line *imports.LineImportData, basePath string, filesMap map[string]*zip.File) error {
|
||||
var ok bool
|
||||
switch line.Type {
|
||||
case "post", "direct_post":
|
||||
var replies []ReplyImportData
|
||||
var replies []imports.ReplyImportData
|
||||
if line.Type == "direct_post" {
|
||||
if err := processAttachmentPaths(line.DirectPost.Attachments, basePath, filesMap); err != nil {
|
||||
return err
|
||||
@@ -108,48 +111,48 @@ func processAttachments(line *LineImportData, basePath string, filesMap map[stri
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *App) bulkImportWorker(c request.CTX, dryRun bool, wg *sync.WaitGroup, lines <-chan LineImportWorkerData, errors chan<- LineImportWorkerError) {
|
||||
postLines := []LineImportWorkerData{}
|
||||
directPostLines := []LineImportWorkerData{}
|
||||
func (a *App) bulkImportWorker(c request.CTX, dryRun bool, wg *sync.WaitGroup, lines <-chan imports.LineImportWorkerData, errors chan<- imports.LineImportWorkerError) {
|
||||
postLines := []imports.LineImportWorkerData{}
|
||||
directPostLines := []imports.LineImportWorkerData{}
|
||||
for line := range lines {
|
||||
switch {
|
||||
case line.LineImportData.Type == "post":
|
||||
postLines = append(postLines, line)
|
||||
if line.Post == nil {
|
||||
errors <- LineImportWorkerError{model.NewAppError("BulkImport", "app.import.import_line.null_post.error", nil, "", http.StatusBadRequest), line.LineNumber}
|
||||
errors <- imports.LineImportWorkerError{Error: model.NewAppError("BulkImport", "app.import.import_line.null_post.error", nil, "", http.StatusBadRequest), LineNumber: line.LineNumber}
|
||||
}
|
||||
if len(postLines) >= importMultiplePostsThreshold {
|
||||
if errLine, err := a.importMultiplePostLines(c, postLines, dryRun); err != nil {
|
||||
errors <- LineImportWorkerError{err, errLine}
|
||||
errors <- imports.LineImportWorkerError{Error: err, LineNumber: errLine}
|
||||
}
|
||||
postLines = []LineImportWorkerData{}
|
||||
postLines = []imports.LineImportWorkerData{}
|
||||
}
|
||||
case line.LineImportData.Type == "direct_post":
|
||||
directPostLines = append(directPostLines, line)
|
||||
if line.DirectPost == nil {
|
||||
errors <- LineImportWorkerError{model.NewAppError("BulkImport", "app.import.import_line.null_direct_post.error", nil, "", http.StatusBadRequest), line.LineNumber}
|
||||
errors <- imports.LineImportWorkerError{Error: model.NewAppError("BulkImport", "app.import.import_line.null_direct_post.error", nil, "", http.StatusBadRequest), LineNumber: line.LineNumber}
|
||||
}
|
||||
if len(directPostLines) >= importMultiplePostsThreshold {
|
||||
if errLine, err := a.importMultipleDirectPostLines(c, directPostLines, dryRun); err != nil {
|
||||
errors <- LineImportWorkerError{err, errLine}
|
||||
errors <- imports.LineImportWorkerError{Error: err, LineNumber: errLine}
|
||||
}
|
||||
directPostLines = []LineImportWorkerData{}
|
||||
directPostLines = []imports.LineImportWorkerData{}
|
||||
}
|
||||
default:
|
||||
if err := a.importLine(c, line.LineImportData, dryRun); err != nil {
|
||||
errors <- LineImportWorkerError{err, line.LineNumber}
|
||||
errors <- imports.LineImportWorkerError{Error: err, LineNumber: line.LineNumber}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(postLines) > 0 {
|
||||
if errLine, err := a.importMultiplePostLines(c, postLines, dryRun); err != nil {
|
||||
errors <- LineImportWorkerError{err, errLine}
|
||||
errors <- imports.LineImportWorkerError{Error: err, LineNumber: errLine}
|
||||
}
|
||||
}
|
||||
if len(directPostLines) > 0 {
|
||||
if errLine, err := a.importMultipleDirectPostLines(c, directPostLines, dryRun); err != nil {
|
||||
errors <- LineImportWorkerError{err, errLine}
|
||||
errors <- imports.LineImportWorkerError{Error: err, LineNumber: errLine}
|
||||
}
|
||||
}
|
||||
wg.Done()
|
||||
@@ -177,9 +180,9 @@ func (a *App) bulkImport(c request.CTX, jsonlReader io.Reader, attachmentsReader
|
||||
a.Srv().Store.LockToMaster()
|
||||
defer a.Srv().Store.UnlockFromMaster()
|
||||
|
||||
errorsChan := make(chan LineImportWorkerError, (2*workers)+1) // size chosen to ensure it never gets filled up completely.
|
||||
errorsChan := make(chan imports.LineImportWorkerError, (2*workers)+1) // size chosen to ensure it never gets filled up completely.
|
||||
var wg sync.WaitGroup
|
||||
var linesChan chan LineImportWorkerData
|
||||
var linesChan chan imports.LineImportWorkerData
|
||||
lastLineType := ""
|
||||
|
||||
var attachedFiles map[string]*zip.File
|
||||
@@ -194,7 +197,7 @@ func (a *App) bulkImport(c request.CTX, jsonlReader io.Reader, attachmentsReader
|
||||
decoder := json.NewDecoder(bytes.NewReader(scanner.Bytes()))
|
||||
lineNumber++
|
||||
|
||||
var line LineImportData
|
||||
var line imports.LineImportData
|
||||
if err := decoder.Decode(&line); err != nil {
|
||||
return model.NewAppError("BulkImport", "app.import.bulk_import.json_decode.error", nil, "", http.StatusBadRequest).Wrap(err), lineNumber
|
||||
}
|
||||
@@ -234,7 +237,7 @@ func (a *App) bulkImport(c request.CTX, jsonlReader io.Reader, attachmentsReader
|
||||
|
||||
// Set up the workers and channel for this type.
|
||||
lastLineType = line.Type
|
||||
linesChan = make(chan LineImportWorkerData, workers)
|
||||
linesChan = make(chan imports.LineImportWorkerData, workers)
|
||||
for i := 0; i < workers; i++ {
|
||||
wg.Add(1)
|
||||
go a.bulkImportWorker(c, dryRun, &wg, linesChan, errorsChan)
|
||||
@@ -242,7 +245,7 @@ func (a *App) bulkImport(c request.CTX, jsonlReader io.Reader, attachmentsReader
|
||||
}
|
||||
|
||||
select {
|
||||
case linesChan <- LineImportWorkerData{line, lineNumber}:
|
||||
case linesChan <- imports.LineImportWorkerData{LineImportData: line, LineNumber: lineNumber}:
|
||||
case err := <-errorsChan:
|
||||
if stopOnError(c, err) {
|
||||
close(linesChan)
|
||||
@@ -273,7 +276,7 @@ func (a *App) bulkImport(c request.CTX, jsonlReader io.Reader, attachmentsReader
|
||||
return nil, 0
|
||||
}
|
||||
|
||||
func processImportDataFileVersionLine(line LineImportData) (int, *model.AppError) {
|
||||
func processImportDataFileVersionLine(line imports.LineImportData) (int, *model.AppError) {
|
||||
if line.Type != "version" || line.Version == nil {
|
||||
return -1, model.NewAppError("BulkImport", "app.import.process_import_data_file_version_line.invalid_version.error", nil, "", http.StatusBadRequest)
|
||||
}
|
||||
@@ -281,7 +284,7 @@ func processImportDataFileVersionLine(line LineImportData) (int, *model.AppError
|
||||
return *line.Version, nil
|
||||
}
|
||||
|
||||
func (a *App) importLine(c request.CTX, line LineImportData, dryRun bool) *model.AppError {
|
||||
func (a *App) importLine(c request.CTX, line imports.LineImportData, dryRun bool) *model.AppError {
|
||||
switch {
|
||||
case line.Type == "scheme":
|
||||
if line.Scheme == nil {
|
||||
|
||||
Ссылка в новой задаче
Block a user