mostlymatter/server/enterprise/message_export/message_export.go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.enterprise for license information.

package message_export

import (
	"encoding/json"
	"errors"
	"strconv"
	"time"

	"github.com/mattermost/mattermost/server/public/model"
	"github.com/mattermost/mattermost/server/public/shared/mlog"
	"github.com/mattermost/mattermost/server/public/shared/request"
	"github.com/mattermost/mattermost/server/v8/channels/app"
	"github.com/mattermost/mattermost/server/v8/einterfaces"
	ejobs "github.com/mattermost/mattermost/server/v8/einterfaces/jobs"
	"github.com/mattermost/mattermost/server/v8/enterprise/message_export/actiance_export"
	"github.com/mattermost/mattermost/server/v8/enterprise/message_export/csv_export"
	"github.com/mattermost/mattermost/server/v8/enterprise/message_export/global_relay_export"
	"github.com/mattermost/mattermost/server/v8/enterprise/message_export/shared"
)

const GlobalRelayExportFilename = "global-relay.zip"

type MessageExportInterfaceImpl struct {
	Server *app.Server
}

type MessageExportJobInterfaceImpl struct {
	Server *app.Server
}

func init() {
	app.RegisterJobsMessageExportJobInterface(func(s *app.Server) ejobs.MessageExportJobInterface {
		return &MessageExportJobInterfaceImpl{s}
	})
	app.RegisterMessageExportInterface(func(app *app.App) einterfaces.MessageExportInterface {
		return &MessageExportInterfaceImpl{app.Srv()}
	})
}

func (m *MessageExportInterfaceImpl) StartSynchronizeJob(rctx request.CTX, exportFromTimestamp int64) (*model.Job, *model.AppError) {
	// if a valid export time was specified, put it in the job data
	jobData := make(map[string]string)
	if exportFromTimestamp >= 0 {
		jobData[shared.JobDataBatchStartTime] = strconv.FormatInt(exportFromTimestamp, 10)
	}

	// passing nil for job data will cause the worker to inherit start time from previously successful job
	job, err := m.Server.Jobs.CreateJob(rctx, model.JobTypeMessageExport, jobData)
	if err != nil {
		return nil, err
	}

	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for job.Status == model.JobStatusPending ||
		job.Status == model.JobStatusInProgress ||
		job.Status == model.JobStatusCancelRequested {
		select {
		case <-ticker.C:
			job, err = m.Server.Jobs.GetJob(rctx, job.Id)
			if err != nil {
				return nil, err
			}
		case <-rctx.Context().Done():
			return nil, model.NewAppError("StartSynchronizeJob", "ent.jobs.start_synchronize_job.timeout", nil, "", 0).Wrap(rctx.Context().Err())
		}
	}

	return job, nil
}

func RunBatch(rctx request.CTX, data shared.JobData, params shared.BackendParams) (shared.RunExportResults, shared.JobData, error) {
	start := time.Now()
	var err error
	var res shared.RunExportResults
	data, err = GetDataForBatch(rctx, data, params)
	if err != nil {
		return res, data, err
	}

	if data.Finished {
		return res, data, nil
	}

	// Now write the data to the export type.
	res, err = RunExportByType(rctx, DataToExportParams(data), params)
	if err != nil {
		return res, data, err
	}

	data.ProcessingPostsMs = append(data.ProcessingPostsMs, res.ProcessingPostsMs)
	data.ProcessingXmlMs = append(data.ProcessingXmlMs, res.ProcessingXmlMs)
	data.TransferringFilesMs = append(data.TransferringFilesMs, res.TransferringFilesMs)
	data.TransferringZipMs = append(data.TransferringZipMs, res.TransferringZipMs)
	data.TotalBatchMs = append(data.TotalBatchMs, time.Since(start).Milliseconds())
	data.WarningCount += res.NumWarnings
	data.BatchStartTime = data.BatchEndTime

	return res, data, err
}

// GetDataForBatch gets the posts for this batch and updates JobData with the current state.
func GetDataForBatch(rctx request.CTX, data shared.JobData, params shared.BackendParams) (shared.JobData, error) {
	start := time.Now()
	var err error
	// Using BatchSize+1 is a trick to test whether or not we've reached the final batch.
	data.PostsToExport, data.Cursor, err = params.Store.Compliance().MessageExport(rctx, data.Cursor, data.BatchSize+1)
	if err != nil {
		return data, err
	}
	data.MessageExportMs = append(data.MessageExportMs, time.Since(start).Milliseconds())

	if len(data.PostsToExport) == data.BatchSize+1 {
		// We still have posts after this current batch.
		// Remove the last post, we have to leave it for the next batch.
		lastPostIdx := len(data.PostsToExport) - 1
		data.PostsToExport = data.PostsToExport[:lastPostIdx]
		lastPostIdx = len(data.PostsToExport) - 1
		data.Cursor.LastPostUpdateAt = *data.PostsToExport[lastPostIdx].PostUpdateAt
		data.Cursor.LastPostId = *data.PostsToExport[lastPostIdx].PostId
		data.BatchEndTime = data.Cursor.LastPostUpdateAt
	} else {
		// We've reached the final batch; we need to include all join/leave events that occur after the lastpost.
		// This will let us also pick up the joins/leaves that occur after lastPostUpdateAt but before JobEndTime.
		data.BatchEndTime = data.JobEndTime
	}

	if len(data.PostsToExport) == 0 {
		data.Finished = true
		return data, nil
	}

	rctx.Logger().Debug("Found posts to export", mlog.Int("num_posts", len(data.PostsToExport)))
	data.MessagesExported += len(data.PostsToExport)
	data.BatchNumber++
	data.BatchPath = shared.GetBatchPath(data.ExportDir, data.BatchStartTime, data.BatchEndTime, data.BatchNumber)

	return data, nil
}

type ExportParams struct {
	ExportType             string
	ChannelMetadata        map[string]*shared.MetadataChannel
	ChannelMemberHistories map[string][]*model.ChannelMemberHistoryResult
	PostsToExport          []*model.MessageExport
	JobStartTime           int64
	BatchPath              string
	BatchStartTime         int64
	BatchEndTime           int64
}

func DataToExportParams(data shared.JobData) ExportParams {
	return ExportParams{
		ExportType:             data.ExportType,
		ChannelMetadata:        data.ChannelMetadata,
		ChannelMemberHistories: data.ChannelMemberHistories,
		PostsToExport:          data.PostsToExport,
		JobStartTime:           data.JobStartTime,
		BatchPath:              data.BatchPath,
		BatchStartTime:         data.BatchStartTime,
		BatchEndTime:           data.BatchEndTime,
	}
}

func RunExportByType(rctx request.CTX, p ExportParams, b shared.BackendParams) (results shared.RunExportResults, err error) {
	preparePosts(rctx, p.PostsToExport)

	exportParams := shared.ExportParams{
		ExportType:             p.ExportType,
		ChannelMetadata:        p.ChannelMetadata,
		Posts:                  p.PostsToExport,
		ChannelMemberHistories: p.ChannelMemberHistories,
		JobStartTime:           p.JobStartTime,
		BatchPath:              p.BatchPath,
		BatchStartTime:         p.BatchStartTime,
		BatchEndTime:           p.BatchEndTime,
		Config:                 b.Config,
		Db:                     b.Store,
		FileAttachmentBackend:  b.FileAttachmentBackend,
		ExportBackend:          b.ExportBackend,
		Templates:              b.HtmlTemplates,
	}

	switch p.ExportType {
	case model.ComplianceExportTypeCsv:
		rctx.Logger().Debug("Exporting CSV")
		return csv_export.CsvExport(rctx, exportParams)

	case model.ComplianceExportTypeActiance:
		rctx.Logger().Debug("Exporting Actiance")
		return actiance_export.ActianceExport(rctx, exportParams)

	case model.ComplianceExportTypeGlobalrelay, model.ComplianceExportTypeGlobalrelayZip:
		rctx.Logger().Debug("Exporting GlobalRelay")
		return global_relay_export.GlobalRelayExport(rctx, exportParams)

	default:
		return results, errors.New("Unknown output format: " + p.ExportType)
	}
}

func preparePosts(rctx request.CTX, postsToExport []*model.MessageExport) {
	// go through all the posts and if the post's props contain 'from_bot' - override the IsBot field, since it's possible that the sender is not a user, but was a Bot and vise-versa
	for _, post := range postsToExport {
		if post.PostProps != nil {
			props := map[string]any{}

			if json.Unmarshal([]byte(*post.PostProps), &props) == nil {
				if val, ok := props["from_bot"]; ok {
					post.IsBot = val == "true"
				}
			}
		}

		// Team info can be null for DM/GM channels.
		if post.TeamId == nil {
			post.TeamId = new(string)
		}
		if post.TeamName == nil {
			post.TeamName = new(string)
		}
		if post.TeamDisplayName == nil {
			post.TeamDisplayName = new(string)
		}

		// make sure user information is present. Set defaults and log an error otherwise.
		if post.ChannelId == nil {
			rctx.Logger().Warn("ChannelId is missing for post", mlog.String("post_id", *post.PostId))
			post.ChannelId = new(string)
		}
		if post.ChannelName == nil {
			rctx.Logger().Warn("ChannelName is missing for post", mlog.String("post_id", *post.PostId))
			post.ChannelName = new(string)
		}
		if post.ChannelDisplayName == nil {
			rctx.Logger().Warn("ChannelDisplayName is missing for post", mlog.String("post_id", *post.PostId))
			post.ChannelDisplayName = new(string)
		}
		if post.ChannelType == nil {
			rctx.Logger().Warn("ChannelType is missing for post", mlog.String("post_id", *post.PostId))
			post.ChannelType = new(model.ChannelType)
		}

		if post.UserId == nil {
			rctx.Logger().Warn("UserId is missing for post", mlog.String("post_id", *post.PostId))
			post.UserId = new(string)
		}
		if post.UserEmail == nil {
			rctx.Logger().Warn("UserEmail is missing for post", mlog.String("post_id", *post.PostId))
			post.UserEmail = new(string)
		}
		if post.Username == nil {
			rctx.Logger().Warn("Username is missing for post", mlog.String("post_id", *post.PostId))
			post.Username = new(string)
		}

		if post.PostType == nil {
			rctx.Logger().Warn("Type is missing for post", mlog.String("post_id", *post.PostId))
			post.PostType = new(string)
		}
		if post.PostMessage == nil {
			rctx.Logger().Warn("Message is missing for post", mlog.String("post_id", *post.PostId))
			post.PostMessage = new(string)
		}
		if post.PostCreateAt == nil {
			rctx.Logger().Warn("CreateAt is missing for post", mlog.String("post_id", *post.PostId))
			post.PostCreateAt = new(int64)
		}
	}
}