status-go/protocol/messenger_communities_import_discord.go
Samuel Hawksby-Robinson 3d68013332 chore(no-torrent)_: Renamed Torrent to Archive
I've renamed TorrentManager to ArchiveManager, ArchiveManager to ArchiveFileManager, TorrentContract to ArchiveService, ArchiveContract to ArchiveFileService. I've also renamed all init functions and struct fields to the appropriate archive-centric naming.
2024-06-07 13:44:12 +01:00

1869 lines
65 KiB
Go

package protocol
import (
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"sync"
"time"
"github.com/golang/protobuf/proto"
"github.com/meirf/gopart"
"go.uber.org/zap"
"github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/images"
"github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/discord"
"github.com/status-im/status-go/protocol/protobuf"
"github.com/status-im/status-go/protocol/requests"
"github.com/status-im/status-go/protocol/transport"
v1protocol "github.com/status-im/status-go/protocol/v1"
)
func (m *Messenger) ExtractDiscordDataFromImportFiles(filesToImport []string) (*discord.ExtractedData, map[string]*discord.ImportError) {
extractedData := &discord.ExtractedData{
Categories: map[string]*discord.Category{},
ExportedData: make([]*discord.ExportedData, 0),
OldestMessageTimestamp: 0,
MessageCount: 0,
}
errors := map[string]*discord.ImportError{}
for _, fileToImport := range filesToImport {
filePath := strings.Replace(fileToImport, "file://", "", -1)
fileInfo, err := os.Stat(filePath)
if err != nil {
errors[fileToImport] = discord.Error(err.Error())
continue
}
fileSize := fileInfo.Size()
if fileSize > discord.MaxImportFileSizeBytes {
errors[fileToImport] = discord.Error(discord.ErrImportFileTooBig.Error())
continue
}
bytes, err := os.ReadFile(filePath)
if err != nil {
errors[fileToImport] = discord.Error(err.Error())
continue
}
var discordExportedData discord.ExportedData
err = json.Unmarshal(bytes, &discordExportedData)
if err != nil {
errors[fileToImport] = discord.Error(err.Error())
continue
}
if len(discordExportedData.Messages) == 0 {
errors[fileToImport] = discord.Error(discord.ErrNoMessageData.Error())
continue
}
discordExportedData.Channel.FilePath = filePath
categoryID := discordExportedData.Channel.CategoryID
discordCategory := discord.Category{
ID: categoryID,
Name: discordExportedData.Channel.CategoryName,
}
_, ok := extractedData.Categories[categoryID]
if !ok {
extractedData.Categories[categoryID] = &discordCategory
}
extractedData.MessageCount = extractedData.MessageCount + discordExportedData.MessageCount
extractedData.ExportedData = append(extractedData.ExportedData, &discordExportedData)
if len(discordExportedData.Messages) > 0 {
msgTime, err := time.Parse(discordTimestampLayout, discordExportedData.Messages[0].Timestamp)
if err != nil {
m.logger.Error("failed to parse discord message timestamp", zap.Error(err))
continue
}
if extractedData.OldestMessageTimestamp == 0 || int(msgTime.Unix()) <= extractedData.OldestMessageTimestamp {
// Exported discord channel data already comes with `messages` being
// sorted, starting with the oldest, so we can safely rely on the first
// message
extractedData.OldestMessageTimestamp = int(msgTime.Unix())
}
}
}
return extractedData, errors
}
func (m *Messenger) ExtractDiscordChannelsAndCategories(filesToImport []string) (*MessengerResponse, map[string]*discord.ImportError) {
response := &MessengerResponse{}
extractedData, errs := m.ExtractDiscordDataFromImportFiles(filesToImport)
for _, category := range extractedData.Categories {
response.AddDiscordCategory(category)
}
for _, export := range extractedData.ExportedData {
response.AddDiscordChannel(&export.Channel)
}
if extractedData.OldestMessageTimestamp != 0 {
response.DiscordOldestMessageTimestamp = extractedData.OldestMessageTimestamp
}
return response, errs
}
func (m *Messenger) RequestExtractDiscordChannelsAndCategories(filesToImport []string) {
go func() {
response, errors := m.ExtractDiscordChannelsAndCategories(filesToImport)
m.config.messengerSignalsHandler.DiscordCategoriesAndChannelsExtracted(
response.DiscordCategories,
response.DiscordChannels,
int64(response.DiscordOldestMessageTimestamp),
errors)
}()
}
func (m *Messenger) saveDiscordAuthorIfNotExists(discordAuthor *protobuf.DiscordMessageAuthor) *discord.ImportError {
exists, err := m.persistence.HasDiscordMessageAuthor(discordAuthor.GetId())
if err != nil {
m.logger.Error("failed to check if message author exists in database", zap.Error(err))
return discord.Error(err.Error())
}
if !exists {
err := m.persistence.SaveDiscordMessageAuthor(discordAuthor)
if err != nil {
return discord.Error(err.Error())
}
}
return nil
}
func (m *Messenger) convertDiscordMessageTimeStamp(discordMessage *protobuf.DiscordMessage, timestamp time.Time) *discord.ImportError {
discordMessage.Timestamp = fmt.Sprintf("%d", timestamp.Unix())
if discordMessage.TimestampEdited != "" {
timestampEdited, err := time.Parse(discordTimestampLayout, discordMessage.TimestampEdited)
if err != nil {
m.logger.Error("failed to parse discord message timestamp", zap.Error(err))
return discord.Warning(err.Error())
}
// Convert timestamp to unix timestamp
discordMessage.TimestampEdited = fmt.Sprintf("%d", timestampEdited.Unix())
}
return nil
}
func (m *Messenger) createPinMessageFromDiscordMessage(message *common.Message, pinnedMessageID string, channelID string, community *communities.Community) (*common.PinMessage, *discord.ImportError) {
pinMessage := protobuf.PinMessage{
Clock: message.WhisperTimestamp,
MessageId: pinnedMessageID,
ChatId: message.LocalChatID,
MessageType: protobuf.MessageType_COMMUNITY_CHAT,
Pinned: true,
}
encodedPayload, err := proto.Marshal(&pinMessage)
if err != nil {
m.logger.Error("failed to parse marshal pin message", zap.Error(err))
return nil, discord.Warning(err.Error())
}
wrappedPayload, err := v1protocol.WrapMessageV1(encodedPayload, protobuf.ApplicationMetadataMessage_PIN_MESSAGE, community.PrivateKey())
if err != nil {
m.logger.Error("failed to wrap pin message", zap.Error(err))
return nil, discord.Warning(err.Error())
}
pinMessageToSave := &common.PinMessage{
ID: types.EncodeHex(v1protocol.MessageID(&community.PrivateKey().PublicKey, wrappedPayload)),
PinMessage: &pinMessage,
LocalChatID: channelID,
From: message.From,
SigPubKey: message.SigPubKey,
WhisperTimestamp: message.WhisperTimestamp,
}
return pinMessageToSave, nil
}
func (m *Messenger) processDiscordMessages(discordChannel *discord.ExportedData,
channel *Chat,
importProgress *discord.ImportProgress,
progressUpdates chan *discord.ImportProgress,
fromDate int64,
community *communities.Community) (
map[string]*common.Message,
[]*common.PinMessage,
map[string]*protobuf.DiscordMessageAuthor,
[]*protobuf.DiscordMessageAttachment) {
messagesToSave := make(map[string]*common.Message, 0)
pinMessagesToSave := make([]*common.PinMessage, 0)
authorProfilesToSave := make(map[string]*protobuf.DiscordMessageAuthor, 0)
messageAttachmentsToDownload := make([]*protobuf.DiscordMessageAttachment, 0)
for _, discordMessage := range discordChannel.Messages {
timestamp, err := time.Parse(discordTimestampLayout, discordMessage.Timestamp)
if err != nil {
m.logger.Error("failed to parse discord message timestamp", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
if timestamp.Unix() < fromDate {
progressUpdates <- importProgress
continue
}
importErr := m.saveDiscordAuthorIfNotExists(discordMessage.Author)
if importErr != nil {
importProgress.AddTaskError(discord.ImportMessagesTask, importErr)
progressUpdates <- importProgress
continue
}
hasPayload, err := m.persistence.HasDiscordMessageAuthorImagePayload(discordMessage.Author.GetId())
if err != nil {
m.logger.Error("failed to check if message avatar payload exists in database", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
if !hasPayload {
authorProfilesToSave[discordMessage.Author.Id] = discordMessage.Author
}
// Convert timestamp to unix timestamp
importErr = m.convertDiscordMessageTimeStamp(discordMessage, timestamp)
if importErr != nil {
importProgress.AddTaskError(discord.ImportMessagesTask, importErr)
progressUpdates <- importProgress
continue
}
for i := range discordMessage.Attachments {
discordMessage.Attachments[i].MessageId = discordMessage.Id
}
messageAttachmentsToDownload = append(messageAttachmentsToDownload, discordMessage.Attachments...)
clockAndTimestamp := uint64(timestamp.Unix()) * 1000
communityPubKey := community.PrivateKey().PublicKey
chatMessage := protobuf.ChatMessage{
Timestamp: clockAndTimestamp,
MessageType: protobuf.MessageType_COMMUNITY_CHAT,
ContentType: protobuf.ChatMessage_DISCORD_MESSAGE,
Clock: clockAndTimestamp,
ChatId: channel.ID,
Payload: &protobuf.ChatMessage_DiscordMessage{
DiscordMessage: discordMessage,
},
}
// Handle message replies
if discordMessage.Type == string(discord.MessageTypeReply) && discordMessage.Reference != nil {
repliedMessageID := community.IDString() + discordMessage.Reference.MessageId
if _, exists := messagesToSave[repliedMessageID]; exists {
chatMessage.ResponseTo = repliedMessageID
}
}
messageToSave := &common.Message{
ID: community.IDString() + discordMessage.Id,
WhisperTimestamp: clockAndTimestamp,
From: types.EncodeHex(crypto.FromECDSAPub(&communityPubKey)),
Seen: true,
LocalChatID: channel.ID,
SigPubKey: &communityPubKey,
CommunityID: community.IDString(),
ChatMessage: &chatMessage,
}
err = messageToSave.PrepareContent(common.PubkeyToHex(&m.identity.PublicKey))
if err != nil {
m.logger.Error("failed to prepare message content", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
// Handle pin messages
if discordMessage.Type == string(discord.MessageTypeChannelPinned) && discordMessage.Reference != nil {
pinnedMessageID := community.IDString() + discordMessage.Reference.MessageId
_, exists := messagesToSave[pinnedMessageID]
if exists {
pinMessageToSave, importErr := m.createPinMessageFromDiscordMessage(messageToSave, pinnedMessageID, channel.ID, community)
if importErr != nil {
importProgress.AddTaskError(discord.ImportMessagesTask, importErr)
progressUpdates <- importProgress
continue
}
pinMessagesToSave = append(pinMessagesToSave, pinMessageToSave)
// Generate SystemMessagePinnedMessage
systemMessage, importErr := m.generateSystemPinnedMessage(pinMessageToSave, channel, clockAndTimestamp, pinnedMessageID)
if importErr != nil {
importProgress.AddTaskError(discord.ImportMessagesTask, importErr)
progressUpdates <- importProgress
continue
}
messagesToSave[systemMessage.ID] = systemMessage
}
} else {
messagesToSave[messageToSave.ID] = messageToSave
}
}
return messagesToSave, pinMessagesToSave, authorProfilesToSave, messageAttachmentsToDownload
}
func calculateProgress(i int, t int, currentProgress float32) float32 {
current := float32(1) / float32(t) * currentProgress
if i > 1 {
return float32(i-1)/float32(t) + current
}
return current
}
func (m *Messenger) MarkDiscordCommunityImportAsCancelled(communityID string) {
m.importingCommunities[communityID] = true
}
func (m *Messenger) MarkDiscordChannelImportAsCancelled(channelID string) {
m.importingChannels[channelID] = true
}
func (m *Messenger) DiscordImportMarkedAsCancelled(communityID string) bool {
cancelled, exists := m.importingCommunities[communityID]
return exists && cancelled
}
func (m *Messenger) DiscordImportChannelMarkedAsCancelled(channelID string) bool {
cancelled, exists := m.importingChannels[channelID]
return exists && cancelled
}
func (m *Messenger) cleanUpImports() {
for id := range m.importingCommunities {
m.cleanUpImport(id)
}
}
func (m *Messenger) cleanUpImport(communityID string) {
community, err := m.communitiesManager.GetByIDString(communityID)
if err != nil {
m.logger.Error("clean up failed, couldn't delete community", zap.Error(err))
return
}
deleteErr := m.communitiesManager.DeleteCommunity(community.ID())
if deleteErr != nil {
m.logger.Error("clean up failed, couldn't delete community", zap.Error(deleteErr))
}
deleteErr = m.persistence.DeleteMessagesByCommunityID(community.IDString())
if deleteErr != nil {
m.logger.Error("clean up failed, couldn't delete community messages", zap.Error(deleteErr))
}
m.config.messengerSignalsHandler.DiscordCommunityImportCleanedUp(communityID)
}
func (m *Messenger) cleanUpImportChannel(communityID string, channelID string) {
_, err := m.DeleteCommunityChat(types.HexBytes(communityID), channelID)
if err != nil {
m.logger.Error("clean up failed, couldn't delete community chat", zap.Error(err))
return
}
err = m.persistence.DeleteMessagesByChatID(channelID)
if err != nil {
m.logger.Error("clean up failed, couldn't delete community chat messages", zap.Error(err))
return
}
}
func (m *Messenger) publishImportProgress(progress *discord.ImportProgress) {
m.config.messengerSignalsHandler.DiscordCommunityImportProgress(progress)
}
func (m *Messenger) publishChannelImportProgress(progress *discord.ImportProgress) {
m.config.messengerSignalsHandler.DiscordChannelImportProgress(progress)
}
func (m *Messenger) startPublishImportProgressInterval(c chan *discord.ImportProgress, cancel chan string, done chan struct{}) {
var currentProgress *discord.ImportProgress
go func() {
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if currentProgress != nil {
m.publishImportProgress(currentProgress)
if currentProgress.Stopped {
return
}
}
case progressUpdate := <-c:
currentProgress = progressUpdate
case <-done:
if currentProgress != nil {
m.publishImportProgress(currentProgress)
}
return
case communityID := <-cancel:
if currentProgress != nil {
m.publishImportProgress(currentProgress)
}
m.cleanUpImport(communityID)
m.config.messengerSignalsHandler.DiscordCommunityImportCancelled(communityID)
return
case <-m.quit:
m.cleanUpImports()
return
}
}
}()
}
func (m *Messenger) startPublishImportChannelProgressInterval(c chan *discord.ImportProgress, cancel chan []string, done chan struct{}) {
var currentProgress *discord.ImportProgress
go func() {
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if currentProgress != nil {
m.publishChannelImportProgress(currentProgress)
if currentProgress.Stopped {
return
}
}
case progressUpdate := <-c:
currentProgress = progressUpdate
case <-done:
if currentProgress != nil {
m.publishChannelImportProgress(currentProgress)
}
return
case ids := <-cancel:
if currentProgress != nil {
m.publishImportProgress(currentProgress)
}
if len(ids) > 0 {
communityID := ids[0]
channelID := ids[1]
discordChannelID := ids[2]
m.cleanUpImportChannel(communityID, channelID)
m.config.messengerSignalsHandler.DiscordChannelImportCancelled(discordChannelID)
}
return
case <-m.quit:
m.cleanUpImports()
return
}
}
}()
}
func createCommunityChannelForImport(request *requests.ImportDiscordChannel) *protobuf.CommunityChat {
return &protobuf.CommunityChat{
Permissions: &protobuf.CommunityPermissions{
Access: protobuf.CommunityPermissions_AUTO_ACCEPT,
},
Identity: &protobuf.ChatIdentity{
DisplayName: request.Name,
Emoji: request.Emoji,
Description: request.Description,
Color: request.Color,
},
CategoryId: "",
HideIfPermissionsNotMet: false,
}
}
func (m *Messenger) RequestImportDiscordChannel(request *requests.ImportDiscordChannel) {
go func() {
totalImportChunkCount := len(request.FilesToImport)
progressUpdates := make(chan *discord.ImportProgress)
done := make(chan struct{})
cancel := make(chan []string)
var newChat *Chat
m.startPublishImportChannelProgressInterval(progressUpdates, cancel, done)
importProgress := &discord.ImportProgress{}
importProgress.Init(totalImportChunkCount, []discord.ImportTask{
discord.ChannelsCreationTask,
discord.ImportMessagesTask,
discord.DownloadAssetsTask,
discord.InitCommunityTask,
})
importProgress.ChannelID = request.DiscordChannelID
importProgress.ChannelName = request.Name
// initial progress immediately
if err := request.Validate(); err != nil {
errmsg := fmt.Sprintf("Request validation failed: '%s'", err.Error())
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error(errmsg))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
// Here's 3 steps: Find the corrent channel in files, get the community and create the channel
progressValue := float32(0.3)
m.publishChannelImportProgress(importProgress)
community, err := m.GetCommunityByID(request.CommunityID)
if err != nil {
errmsg := fmt.Sprintf("Couldn't get the community '%s': '%s'", request.CommunityID, err.Error())
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error(errmsg))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, progressValue)
progressUpdates <- importProgress
for i, importFile := range request.FilesToImport {
m.importingChannels[request.DiscordChannelID] = false
exportData, errs := m.ExtractDiscordDataFromImportFiles([]string{importFile})
if len(errs) > 0 {
for _, err := range errs {
importProgress.AddTaskError(discord.ChannelsCreationTask, err)
}
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
var channel *discord.ExportedData
for _, ch := range exportData.ExportedData {
if ch.Channel.ID == request.DiscordChannelID {
channel = ch
}
}
if channel == nil {
if i < len(request.FilesToImport)-1 {
// skip this file
continue
} else if i == len(request.FilesToImport)-1 {
errmsg := fmt.Sprintf("Couldn't find the target channel id in files: '%s'", request.DiscordChannelID)
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error(errmsg))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
}
progressValue := float32(0.6)
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, progressValue)
progressUpdates <- importProgress
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
if len(channel.Channel.ID) == 0 {
// skip this file and try to find in the next file
continue
}
exists := false
for _, chatID := range community.ChatIDs() {
if strings.HasSuffix(chatID, request.DiscordChannelID) {
exists = true
break
}
}
if !exists {
communityChat := createCommunityChannelForImport(request)
changes, err := m.communitiesManager.CreateChat(request.CommunityID, communityChat, false, channel.Channel.ID)
if err != nil {
errmsg := err.Error()
if errors.Is(err, communities.ErrInvalidCommunityDescriptionDuplicatedName) {
errmsg = fmt.Sprintf("Couldn't create channel '%s': %s", communityChat.Identity.DisplayName, err.Error())
fmt.Println(errmsg)
}
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error(errmsg))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
community = changes.Community
for chatID, chat := range changes.ChatsAdded {
newChat = CreateCommunityChat(request.CommunityID.String(), chatID, chat, m.getTimesource())
}
progressValue = float32(1.0)
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, progressValue)
progressUpdates <- importProgress
} else {
// When channel with current discord id already exist we should skip import
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error("Channel already imported to this community"))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), "", request.DiscordChannelID}
return
}
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
messagesToSave, pinMessagesToSave, authorProfilesToSave, messageAttachmentsToDownload :=
m.processDiscordMessages(channel, newChat, importProgress, progressUpdates, request.From, community)
// If either there were no messages in the channel or something happened and all the messages errored, we
// we still up the percent to 100%
if len(messagesToSave) == 0 {
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, 1.0)
progressUpdates <- importProgress
}
var discordMessages []*protobuf.DiscordMessage
for _, msg := range messagesToSave {
if msg.ChatMessage.ContentType == protobuf.ChatMessage_DISCORD_MESSAGE {
discordMessages = append(discordMessages, msg.GetDiscordMessage())
}
}
// We save these messages in chunks, so we don't block the database
// for a longer period of time
discordMessageChunks := chunkSlice(discordMessages, maxChunkSizeMessages)
chunksCount := len(discordMessageChunks)
for ii, msgs := range discordMessageChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d discord messages", ii+1, chunksCount, len(msgs)))
err := m.persistence.SaveDiscordMessages(msgs)
if err != nil {
m.cleanUpImportChannel(request.CommunityID.String(), newChat.ID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
// We're multiplying `chunksCount` by `0.25` so we leave 25% for additional save operations
// 0.5 are the previous 50% of progress
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.5+(float32(currentCount)/float32(chunksCount))*0.25)
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of message chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
// Get slice of all values in `messagesToSave` map
var messages = make([]*common.Message, 0, len(messagesToSave))
for _, msg := range messagesToSave {
messages = append(messages, msg)
}
// Same as above, we save these messages in chunks so we don't block
// the database for a longer period of time
messageChunks := chunkSlice(messages, maxChunkSizeMessages)
chunksCount = len(messageChunks)
for ii, msgs := range messageChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d app messages", ii+1, chunksCount, len(msgs)))
err := m.persistence.SaveMessages(msgs)
if err != nil {
m.cleanUpImportChannel(request.CommunityID.String(), request.DiscordChannelID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
// 0.75 are the previous 75% of progress, hence we multiply our chunk progress
// by 0.25
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.75+(float32(currentCount)/float32(chunksCount))*0.25)
// progressValue := 0.75 + ((float32(currentCount) / float32(chunksCount)) * 0.25)
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of message chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
pinMessageChunks := chunkSlice(pinMessagesToSave, maxChunkSizeMessages)
for _, pinMsgs := range pinMessageChunks {
err := m.persistence.SavePinMessages(pinMsgs)
if err != nil {
m.cleanUpImportChannel(request.CommunityID.String(), request.DiscordChannelID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
}
totalAssetsCount := len(messageAttachmentsToDownload) + len(authorProfilesToSave)
var assetCounter discord.AssetCounter
var wg sync.WaitGroup
for id, author := range authorProfilesToSave {
wg.Add(1)
go func(id string, author *protobuf.DiscordMessageAuthor) {
defer wg.Done()
m.logger.Debug(fmt.Sprintf("downloading asset %d/%d", assetCounter.Value()+1, totalAssetsCount))
imagePayload, err := discord.DownloadAvatarAsset(author.AvatarUrl)
if err != nil {
errmsg := fmt.Sprintf("Couldn't download profile avatar '%s': %s", author.AvatarUrl, err.Error())
importProgress.AddTaskError(
discord.DownloadAssetsTask,
discord.Warning(errmsg),
)
progressUpdates <- importProgress
return
}
err = m.persistence.UpdateDiscordMessageAuthorImage(author.Id, imagePayload)
if err != nil {
importProgress.AddTaskError(discord.DownloadAssetsTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
return
}
author.AvatarImagePayload = imagePayload
authorProfilesToSave[id] = author
if m.DiscordImportMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
assetCounter.Increase()
progressValue := calculateProgress(i+1, totalImportChunkCount, (float32(assetCounter.Value())/float32(totalAssetsCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
}(id, author)
}
wg.Wait()
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
for idxRange := range gopart.Partition(len(messageAttachmentsToDownload), 100) {
attachments := messageAttachmentsToDownload[idxRange.Low:idxRange.High]
wg.Add(1)
go func(attachments []*protobuf.DiscordMessageAttachment) {
defer wg.Done()
for ii, attachment := range attachments {
m.logger.Debug(fmt.Sprintf("downloading asset %d/%d", assetCounter.Value()+1, totalAssetsCount))
assetPayload, contentType, err := discord.DownloadAsset(attachment.Url)
if err != nil {
errmsg := fmt.Sprintf("Couldn't download message attachment '%s': %s", attachment.Url, err.Error())
importProgress.AddTaskError(
discord.DownloadAssetsTask,
discord.Warning(errmsg),
)
progressUpdates <- importProgress
continue
}
attachment.Payload = assetPayload
attachment.ContentType = contentType
messageAttachmentsToDownload[ii] = attachment
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
assetCounter.Increase()
progressValue := calculateProgress(i+1, totalImportChunkCount, (float32(assetCounter.Value())/float32(totalAssetsCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
}
}(attachments)
}
wg.Wait()
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
attachmentChunks := chunkAttachmentsByByteSize(messageAttachmentsToDownload, maxChunkSizeBytes)
chunksCount = len(attachmentChunks)
for ii, attachments := range attachmentChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d discord message attachments", ii+1, chunksCount, len(attachments)))
err := m.persistence.SaveDiscordMessageAttachments(attachments)
if err != nil {
importProgress.AddTaskError(discord.DownloadAssetsTask, discord.Warning(err.Error()))
importProgress.Stop()
progressUpdates <- importProgress
continue
}
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
// 0.5 are the previous 50% of progress, hence we multiply our chunk progress
// by 0.5
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.5+(float32(currentCount)/float32(chunksCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of attachment chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
if len(attachmentChunks) == 0 {
progressValue := calculateProgress(i+1, totalImportChunkCount, 1.0)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
}
_, err := m.transport.JoinPublic(newChat.ID)
if err != nil {
m.logger.Error("failed to load filter for chat", zap.Error(err))
continue
}
wakuChatMessages, err := m.chatMessagesToWakuMessages(messages, community)
if err != nil {
m.logger.Error("failed to convert chat messages into waku messages", zap.Error(err))
continue
}
wakuPinMessages, err := m.pinMessagesToWakuMessages(pinMessagesToSave, community)
if err != nil {
m.logger.Error("failed to convert pin messages into waku messages", zap.Error(err))
continue
}
wakuMessages := append(wakuChatMessages, wakuPinMessages...)
topics, err := m.archiveManager.GetCommunityChatsTopics(request.CommunityID)
if err != nil {
m.logger.Error("failed to get community chat topics", zap.Error(err))
continue
}
startDate := time.Unix(int64(exportData.OldestMessageTimestamp), 0)
endDate := time.Now()
_, err = m.archiveManager.CreateHistoryArchiveTorrentFromMessages(
request.CommunityID,
wakuMessages,
topics,
startDate,
endDate,
messageArchiveInterval,
community.Encrypted(),
)
if err != nil {
m.logger.Error("failed to create history archive torrent", zap.Error(err))
continue
}
communitySettings, err := m.communitiesManager.GetCommunitySettingsByID(request.CommunityID)
if err != nil {
m.logger.Error("Failed to get community settings", zap.Error(err))
continue
}
if m.archiveManager.IsReady() && communitySettings.HistoryArchiveSupportEnabled {
err = m.archiveManager.SeedHistoryArchiveTorrent(request.CommunityID)
if err != nil {
m.logger.Error("failed to seed history archive", zap.Error(err))
}
go m.archiveManager.StartHistoryArchiveTasksInterval(community, messageArchiveInterval)
}
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, float32(0.0))
if m.DiscordImportChannelMarkedAsCancelled(request.DiscordChannelID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), newChat.ID, request.DiscordChannelID}
return
}
// Chats need to be saved after the community has been published,
// hence we make this part of the `InitCommunityTask`
err = m.saveChat(newChat)
if err != nil {
m.cleanUpImportChannel(request.CommunityID.String(), request.DiscordChannelID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.Stop()
progressUpdates <- importProgress
cancel <- []string{request.CommunityID.String(), request.DiscordChannelID}
return
}
// Make sure all progress tasks are at 100%, in case one of the steps had errors
// The front-end doesn't understand that the import is done until all tasks are at 100%
importProgress.UpdateTaskProgress(discord.CommunityCreationTask, float32(1.0))
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, float32(1.0))
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, float32(1.0))
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, float32(1.0))
importProgress.UpdateTaskProgress(discord.InitCommunityTask, float32(1.0))
m.config.messengerSignalsHandler.DiscordChannelImportFinished(request.CommunityID.String(), newChat.ID)
close(done)
}()
}
func (m *Messenger) RequestImportDiscordCommunity(request *requests.ImportDiscordCommunity) {
go func() {
totalImportChunkCount := len(request.FilesToImport)
progressUpdates := make(chan *discord.ImportProgress)
done := make(chan struct{})
cancel := make(chan string)
m.startPublishImportProgressInterval(progressUpdates, cancel, done)
importProgress := &discord.ImportProgress{}
importProgress.Init(totalImportChunkCount, []discord.ImportTask{
discord.CommunityCreationTask,
discord.ChannelsCreationTask,
discord.ImportMessagesTask,
discord.DownloadAssetsTask,
discord.InitCommunityTask,
})
importProgress.CommunityName = request.Name
// initial progress immediately
m.publishImportProgress(importProgress)
createCommunityRequest := request.ToCreateCommunityRequest()
// We're calling `CreateCommunity` on `communitiesManager` directly, instead of
// using the `Messenger` API, so we get more control over when we set up filters,
// the community is published and data is being synced (we don't want the community
// to show up in clients while the import is in progress)
discordCommunity, err := m.communitiesManager.CreateCommunity(createCommunityRequest, false)
if err != nil {
importProgress.AddTaskError(discord.CommunityCreationTask, discord.Error(err.Error()))
importProgress.StopTask(discord.CommunityCreationTask)
progressUpdates <- importProgress
return
}
communitySettings := communities.CommunitySettings{
CommunityID: discordCommunity.IDString(),
HistoryArchiveSupportEnabled: true,
}
err = m.communitiesManager.SaveCommunitySettings(communitySettings)
if err != nil {
m.cleanUpImport(discordCommunity.IDString())
importProgress.AddTaskError(discord.CommunityCreationTask, discord.Error(err.Error()))
importProgress.StopTask(discord.CommunityCreationTask)
progressUpdates <- importProgress
return
}
communityID := discordCommunity.IDString()
// marking import as not cancelled
m.importingCommunities[communityID] = false
importProgress.CommunityID = communityID
importProgress.CommunityImages = make(map[string]images.IdentityImage)
imgs := discordCommunity.Images()
for t, i := range imgs {
importProgress.CommunityImages[t] = images.IdentityImage{Name: t, Payload: i.Payload}
}
importProgress.UpdateTaskProgress(discord.CommunityCreationTask, 1)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.CommunityCreationTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
var chatsToSave []*Chat
createdChats := make(map[string]*Chat, 0)
processedChannelIds := make(map[string]string, 0)
processedCategoriesIds := make(map[string]string, 0)
// The map with counts of duplicated channel names
uniqueChatNames := make(map[string]int, 0)
for i, importFile := range request.FilesToImport {
exportData, errs := m.ExtractDiscordDataFromImportFiles([]string{importFile})
if len(errs) > 0 {
for _, err := range errs {
importProgress.AddTaskError(discord.CommunityCreationTask, err)
}
progressUpdates <- importProgress
return
}
totalChannelsCount := len(exportData.ExportedData)
totalMessageCount := exportData.MessageCount
if totalChannelsCount == 0 || totalMessageCount == 0 {
importError := discord.Error(fmt.Errorf("No channel to import messages from in file: %s", importFile).Error())
if totalMessageCount == 0 {
importError.Message = fmt.Errorf("No messages to import in file: %s", importFile).Error()
}
importProgress.AddTaskError(discord.ChannelsCreationTask, importError)
progressUpdates <- importProgress
continue
}
importProgress.CurrentChunk = i + 1
// We actually only ever receive a single category
// from `exportData` but since it's a map, we still have to
// iterate over it to access its values
for _, category := range exportData.Categories {
categories := discordCommunity.Categories()
exists := false
for catID := range categories {
if strings.HasSuffix(catID, category.ID) {
exists = true
break
}
}
if !exists {
createCommunityCategoryRequest := &requests.CreateCommunityCategory{
CommunityID: discordCommunity.ID(),
CategoryName: category.Name,
ThirdPartyID: category.ID,
ChatIDs: make([]string, 0),
}
// We call `CreateCategory` on `communitiesManager` directly so we can control
// whether or not the community update should be published (it should not until the
// import has finished)
communityWithCategories, changes, err := m.communitiesManager.CreateCategory(createCommunityCategoryRequest, false)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.CommunityCreationTask, discord.Error(err.Error()))
importProgress.StopTask(discord.CommunityCreationTask)
progressUpdates <- importProgress
return
}
discordCommunity = communityWithCategories
// This looks like we keep overriding the same field but there's
// only one `CategoriesAdded` change at this point.
for _, addedCategory := range changes.CategoriesAdded {
processedCategoriesIds[category.ID] = addedCategory.CategoryId
}
}
}
progressValue := calculateProgress(i+1, totalImportChunkCount, (float32(1) / 2))
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, progressValue)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.CommunityCreationTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
messagesToSave := make(map[string]*common.Message, 0)
pinMessagesToSave := make([]*common.PinMessage, 0)
authorProfilesToSave := make(map[string]*protobuf.DiscordMessageAuthor, 0)
messageAttachmentsToDownload := make([]*protobuf.DiscordMessageAttachment, 0)
// Save to access the first item here as we process
// exported data by files which only holds a single channel
channel := exportData.ExportedData[0]
chatIDs := discordCommunity.ChatIDs()
exists := false
for _, chatID := range chatIDs {
if strings.HasSuffix(chatID, channel.Channel.ID) {
exists = true
break
}
}
if !exists {
channelUniqueName := channel.Channel.Name
if count, ok := uniqueChatNames[channelUniqueName]; ok {
uniqueChatNames[channelUniqueName] = count + 1
channelUniqueName = fmt.Sprintf("%s_%d", channelUniqueName, uniqueChatNames[channelUniqueName])
} else {
uniqueChatNames[channelUniqueName] = 1
}
communityChat := &protobuf.CommunityChat{
Permissions: &protobuf.CommunityPermissions{
Access: protobuf.CommunityPermissions_AUTO_ACCEPT,
},
Identity: &protobuf.ChatIdentity{
DisplayName: channelUniqueName,
Emoji: "",
Description: channel.Channel.Description,
Color: discordCommunity.Color(),
},
CategoryId: processedCategoriesIds[channel.Channel.CategoryID],
HideIfPermissionsNotMet: false,
}
// We call `CreateChat` on `communitiesManager` directly to get more control
// over whether we want to publish the updated community description.
changes, err := m.communitiesManager.CreateChat(discordCommunity.ID(), communityChat, false, channel.Channel.ID)
if err != nil {
m.cleanUpImport(communityID)
errmsg := err.Error()
if errors.Is(err, communities.ErrInvalidCommunityDescriptionDuplicatedName) {
errmsg = fmt.Sprintf("Couldn't create channel '%s': %s", communityChat.Identity.DisplayName, err.Error())
}
importProgress.AddTaskError(discord.ChannelsCreationTask, discord.Error(errmsg))
importProgress.StopTask(discord.ChannelsCreationTask)
progressUpdates <- importProgress
return
}
discordCommunity = changes.Community
// This looks like we keep overriding the chat id value
// as we iterate over `ChatsAdded`, however at this point we
// know there was only a single such change (and it's a map)
for chatID, chat := range changes.ChatsAdded {
c := CreateCommunityChat(communityID, chatID, chat, m.getTimesource())
createdChats[c.ID] = c
chatsToSave = append(chatsToSave, c)
processedChannelIds[channel.Channel.ID] = c.ID
}
}
progressValue = calculateProgress(i+1, totalImportChunkCount, 1)
importProgress.UpdateTaskProgress(discord.ChannelsCreationTask, progressValue)
progressUpdates <- importProgress
for ii, discordMessage := range channel.Messages {
timestamp, err := time.Parse(discordTimestampLayout, discordMessage.Timestamp)
if err != nil {
m.logger.Error("failed to parse discord message timestamp", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
if timestamp.Unix() < request.From {
progressUpdates <- importProgress
continue
}
exists, err := m.persistence.HasDiscordMessageAuthor(discordMessage.Author.GetId())
if err != nil {
m.logger.Error("failed to check if message author exists in database", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
if !exists {
err := m.persistence.SaveDiscordMessageAuthor(discordMessage.Author)
if err != nil {
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
}
hasPayload, err := m.persistence.HasDiscordMessageAuthorImagePayload(discordMessage.Author.GetId())
if err != nil {
m.logger.Error("failed to check if message avatar payload exists in database", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
if !hasPayload {
authorProfilesToSave[discordMessage.Author.Id] = discordMessage.Author
}
// Convert timestamp to unix timestamp
discordMessage.Timestamp = fmt.Sprintf("%d", timestamp.Unix())
if discordMessage.TimestampEdited != "" {
timestampEdited, err := time.Parse(discordTimestampLayout, discordMessage.TimestampEdited)
if err != nil {
m.logger.Error("failed to parse discord message timestamp", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
// Convert timestamp to unix timestamp
discordMessage.TimestampEdited = fmt.Sprintf("%d", timestampEdited.Unix())
}
for i := range discordMessage.Attachments {
discordMessage.Attachments[i].MessageId = discordMessage.Id
}
messageAttachmentsToDownload = append(messageAttachmentsToDownload, discordMessage.Attachments...)
clockAndTimestamp := uint64(timestamp.Unix()) * 1000
communityPubKey := discordCommunity.PrivateKey().PublicKey
chatMessage := protobuf.ChatMessage{
Timestamp: clockAndTimestamp,
MessageType: protobuf.MessageType_COMMUNITY_CHAT,
ContentType: protobuf.ChatMessage_DISCORD_MESSAGE,
Clock: clockAndTimestamp,
ChatId: processedChannelIds[channel.Channel.ID],
Payload: &protobuf.ChatMessage_DiscordMessage{
DiscordMessage: discordMessage,
},
}
// Handle message replies
if discordMessage.Type == string(discord.MessageTypeReply) && discordMessage.Reference != nil {
repliedMessageID := communityID + discordMessage.Reference.MessageId
if _, exists := messagesToSave[repliedMessageID]; exists {
chatMessage.ResponseTo = repliedMessageID
}
}
messageToSave := &common.Message{
ID: communityID + discordMessage.Id,
WhisperTimestamp: clockAndTimestamp,
From: types.EncodeHex(crypto.FromECDSAPub(&communityPubKey)),
Seen: true,
LocalChatID: processedChannelIds[channel.Channel.ID],
SigPubKey: &communityPubKey,
CommunityID: communityID,
ChatMessage: &chatMessage,
}
err = messageToSave.PrepareContent(common.PubkeyToHex(&m.identity.PublicKey))
if err != nil {
m.logger.Error("failed to prepare message content", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
progressUpdates <- importProgress
continue
}
// Handle pin messages
if discordMessage.Type == string(discord.MessageTypeChannelPinned) && discordMessage.Reference != nil {
pinnedMessageID := communityID + discordMessage.Reference.MessageId
_, exists := messagesToSave[pinnedMessageID]
if exists {
pinMessage := protobuf.PinMessage{
Clock: messageToSave.WhisperTimestamp,
MessageId: pinnedMessageID,
ChatId: messageToSave.LocalChatID,
MessageType: protobuf.MessageType_COMMUNITY_CHAT,
Pinned: true,
}
encodedPayload, err := proto.Marshal(&pinMessage)
if err != nil {
m.logger.Error("failed to parse marshal pin message", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
wrappedPayload, err := v1protocol.WrapMessageV1(encodedPayload, protobuf.ApplicationMetadataMessage_PIN_MESSAGE, discordCommunity.PrivateKey())
if err != nil {
m.logger.Error("failed to wrap pin message", zap.Error(err))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
pinMessageToSave := common.PinMessage{
ID: types.EncodeHex(v1protocol.MessageID(&communityPubKey, wrappedPayload)),
PinMessage: &pinMessage,
LocalChatID: processedChannelIds[channel.Channel.ID],
From: messageToSave.From,
SigPubKey: messageToSave.SigPubKey,
WhisperTimestamp: messageToSave.WhisperTimestamp,
}
pinMessagesToSave = append(pinMessagesToSave, &pinMessageToSave)
// Generate SystemMessagePinnedMessage
chat, ok := createdChats[pinMessageToSave.LocalChatID]
if !ok {
err := errors.New("failed to get chat for pin message")
m.logger.Warn(err.Error(),
zap.String("PinMessageId", pinMessageToSave.ID),
zap.String("ChatID", pinMessageToSave.LocalChatID))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
id, err := generatePinMessageNotificationID(&m.identity.PublicKey, &pinMessageToSave, chat)
if err != nil {
m.logger.Warn("failed to generate pin message notification ID",
zap.String("PinMessageId", pinMessageToSave.ID))
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
continue
}
systemMessage := &common.Message{
ChatMessage: &protobuf.ChatMessage{
Clock: pinMessageToSave.Clock,
Timestamp: clockAndTimestamp,
ChatId: chat.ID,
MessageType: pinMessageToSave.MessageType,
ResponseTo: pinMessage.MessageId,
ContentType: protobuf.ChatMessage_SYSTEM_MESSAGE_PINNED_MESSAGE,
},
WhisperTimestamp: clockAndTimestamp,
ID: id,
LocalChatID: chat.ID,
From: messageToSave.From,
Seen: true,
}
messagesToSave[systemMessage.ID] = systemMessage
}
} else {
messagesToSave[messageToSave.ID] = messageToSave
}
progressValue := calculateProgress(i+1, totalImportChunkCount, float32(ii+1)/float32(len(channel.Messages))*0.5)
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, progressValue)
progressUpdates <- importProgress
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
var discordMessages []*protobuf.DiscordMessage
for _, msg := range messagesToSave {
if msg.ChatMessage.ContentType == protobuf.ChatMessage_DISCORD_MESSAGE {
discordMessages = append(discordMessages, msg.GetDiscordMessage())
}
}
// We save these messages in chunks, so we don't block the database
// for a longer period of time
discordMessageChunks := chunkSlice(discordMessages, maxChunkSizeMessages)
chunksCount := len(discordMessageChunks)
for ii, msgs := range discordMessageChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d discord messages", ii+1, chunksCount, len(msgs)))
err = m.persistence.SaveDiscordMessages(msgs)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
return
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
// We're multiplying `chunksCount` by `0.25` so we leave 25% for additional save operations
// 0.5 are the previous 50% of progress
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.5+(float32(currentCount)/float32(chunksCount))*0.25)
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of message chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
// Get slice of all values in `messagesToSave` map
var messages = make([]*common.Message, 0, len(messagesToSave))
for _, msg := range messagesToSave {
messages = append(messages, msg)
}
// Same as above, we save these messages in chunks so we don't block
// the database for a longer period of time
messageChunks := chunkSlice(messages, maxChunkSizeMessages)
chunksCount = len(messageChunks)
for ii, msgs := range messageChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d app messages", ii+1, chunksCount, len(msgs)))
err = m.persistence.SaveMessages(msgs)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
return
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
// 0.75 are the previous 75% of progress, hence we multiply our chunk progress
// by 0.25
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.75+(float32(currentCount)/float32(chunksCount))*0.25)
// progressValue := 0.75 + ((float32(currentCount) / float32(chunksCount)) * 0.25)
importProgress.UpdateTaskProgress(discord.ImportMessagesTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of message chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
pinMessageChunks := chunkSlice(pinMessagesToSave, maxChunkSizeMessages)
for _, pinMsgs := range pinMessageChunks {
err = m.persistence.SavePinMessages(pinMsgs)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.ImportMessagesTask, discord.Error(err.Error()))
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
return
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.ImportMessagesTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
}
totalAssetsCount := len(messageAttachmentsToDownload) + len(authorProfilesToSave)
var assetCounter discord.AssetCounter
var wg sync.WaitGroup
for id, author := range authorProfilesToSave {
wg.Add(1)
go func(id string, author *protobuf.DiscordMessageAuthor) {
defer wg.Done()
m.logger.Debug(fmt.Sprintf("downloading asset %d/%d", assetCounter.Value()+1, totalAssetsCount))
imagePayload, err := discord.DownloadAvatarAsset(author.AvatarUrl)
if err != nil {
errmsg := fmt.Sprintf("Couldn't download profile avatar '%s': %s", author.AvatarUrl, err.Error())
importProgress.AddTaskError(
discord.DownloadAssetsTask,
discord.Warning(errmsg),
)
progressUpdates <- importProgress
return
}
err = m.persistence.UpdateDiscordMessageAuthorImage(author.Id, imagePayload)
if err != nil {
importProgress.AddTaskError(discord.DownloadAssetsTask, discord.Warning(err.Error()))
progressUpdates <- importProgress
return
}
author.AvatarImagePayload = imagePayload
authorProfilesToSave[id] = author
if m.DiscordImportMarkedAsCancelled(discordCommunity.IDString()) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- discordCommunity.IDString()
return
}
assetCounter.Increase()
progressValue := calculateProgress(i+1, totalImportChunkCount, (float32(assetCounter.Value())/float32(totalAssetsCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
}(id, author)
}
wg.Wait()
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
for idxRange := range gopart.Partition(len(messageAttachmentsToDownload), 100) {
attachments := messageAttachmentsToDownload[idxRange.Low:idxRange.High]
wg.Add(1)
go func(attachments []*protobuf.DiscordMessageAttachment) {
defer wg.Done()
for ii, attachment := range attachments {
m.logger.Debug(fmt.Sprintf("downloading asset %d/%d", assetCounter.Value()+1, totalAssetsCount))
assetPayload, contentType, err := discord.DownloadAsset(attachment.Url)
if err != nil {
errmsg := fmt.Sprintf("Couldn't download message attachment '%s': %s", attachment.Url, err.Error())
importProgress.AddTaskError(
discord.DownloadAssetsTask,
discord.Warning(errmsg),
)
progressUpdates <- importProgress
continue
}
attachment.Payload = assetPayload
attachment.ContentType = contentType
messageAttachmentsToDownload[ii] = attachment
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
assetCounter.Increase()
progressValue := calculateProgress(i+1, totalImportChunkCount, (float32(assetCounter.Value())/float32(totalAssetsCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
}
}(attachments)
}
wg.Wait()
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
attachmentChunks := chunkAttachmentsByByteSize(messageAttachmentsToDownload, maxChunkSizeBytes)
chunksCount = len(attachmentChunks)
for ii, attachments := range attachmentChunks {
m.logger.Debug(fmt.Sprintf("saving %d/%d chunk with %d discord message attachments", ii+1, chunksCount, len(attachments)))
err = m.persistence.SaveDiscordMessageAttachments(attachments)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.DownloadAssetsTask, discord.Error(err.Error()))
importProgress.Stop()
progressUpdates <- importProgress
return
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.DownloadAssetsTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
// 0.5 are the previous 50% of progress, hence we multiply our chunk progress
// by 0.5
currentCount := ii + 1
progressValue := calculateProgress(i+1, totalImportChunkCount, 0.5+(float32(currentCount)/float32(chunksCount))*0.5)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
progressUpdates <- importProgress
// We slow down the saving of attachment chunks to keep the database responsive
if currentCount < chunksCount {
time.Sleep(2 * time.Second)
}
}
if len(attachmentChunks) == 0 {
progressValue := calculateProgress(i+1, totalImportChunkCount, 1.0)
importProgress.UpdateTaskProgress(discord.DownloadAssetsTask, progressValue)
}
_, err := m.transport.JoinPublic(processedChannelIds[channel.Channel.ID])
if err != nil {
m.logger.Error("failed to load filter for chat", zap.Error(err))
continue
}
wakuChatMessages, err := m.chatMessagesToWakuMessages(messages, discordCommunity)
if err != nil {
m.logger.Error("failed to convert chat messages into waku messages", zap.Error(err))
continue
}
wakuPinMessages, err := m.pinMessagesToWakuMessages(pinMessagesToSave, discordCommunity)
if err != nil {
m.logger.Error("failed to convert pin messages into waku messages", zap.Error(err))
continue
}
wakuMessages := append(wakuChatMessages, wakuPinMessages...)
topics, err := m.archiveManager.GetCommunityChatsTopics(discordCommunity.ID())
if err != nil {
m.logger.Error("failed to get community chat topics", zap.Error(err))
continue
}
startDate := time.Unix(int64(exportData.OldestMessageTimestamp), 0)
endDate := time.Now()
_, err = m.archiveManager.CreateHistoryArchiveTorrentFromMessages(
discordCommunity.ID(),
wakuMessages,
topics,
startDate,
endDate,
messageArchiveInterval,
discordCommunity.Encrypted(),
)
if err != nil {
m.logger.Error("failed to create history archive torrent", zap.Error(err))
continue
}
if m.archiveManager.IsReady() && communitySettings.HistoryArchiveSupportEnabled {
err = m.archiveManager.SeedHistoryArchiveTorrent(discordCommunity.ID())
if err != nil {
m.logger.Error("failed to seed history archive", zap.Error(err))
}
go m.archiveManager.StartHistoryArchiveTasksInterval(discordCommunity, messageArchiveInterval)
}
}
err = m.publishOrg(discordCommunity, false)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.Stop()
progressUpdates <- importProgress
return
}
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
// Chats need to be saved after the community has been published,
// hence we make this part of the `InitCommunityTask`
err = m.saveChats(chatsToSave)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.Stop()
progressUpdates <- importProgress
return
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, 0.15)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
// Init the community filter so we can receive messages on the community
_, err = m.InitCommunityFilters([]transport.CommunityFilterToInitialize{{
Shard: discordCommunity.Shard(),
PrivKey: discordCommunity.PrivateKey(),
}})
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
return
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, 0.25)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
_, err = m.transport.InitPublicFilters(m.DefaultFilters(discordCommunity))
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
return
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, 0.5)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
filters := m.transport.Filters()
_, err = m.scheduleSyncFilters(filters)
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
return
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, 0.75)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
err = m.reregisterForPushNotifications()
if err != nil {
m.cleanUpImport(communityID)
importProgress.AddTaskError(discord.InitCommunityTask, discord.Error(err.Error()))
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
return
}
importProgress.UpdateTaskProgress(discord.InitCommunityTask, 1)
progressUpdates <- importProgress
if m.DiscordImportMarkedAsCancelled(communityID) {
importProgress.StopTask(discord.InitCommunityTask)
progressUpdates <- importProgress
cancel <- communityID
return
}
m.config.messengerSignalsHandler.DiscordCommunityImportFinished(communityID)
close(done)
}()
}