telegabber/telegram/handlers.go

package telegram

import (
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"

	"dev.narayana.im/narayana/telegabber/e2ee"
	"dev.narayana.im/narayana/telegabber/telegram/formatter"
	"dev.narayana.im/narayana/telegabber/xmpp/gateway"

	"github.com/google/uuid"
	"github.com/pkg/errors"
	log "github.com/sirupsen/logrus"
	"github.com/zelenin/go-tdlib/client"
)

func int64SliceToStringSlice(ints []int64) []string {
	strings := make([]string, len(ints))
	wg := sync.WaitGroup{}

	for i, xi := range ints {
		wg.Add(1)
		go func(i int, xi int64) {
			strings[i] = strconv.FormatInt(xi, 10)
			wg.Done()
		}(i, xi)
	}
	wg.Wait()

	return strings
}

func (c *Client) getChatMessageLock(chatID int64) *sync.Mutex {
	lock, ok := c.locks.chatMessageLocks[chatID]
	if !ok {
		lock = &sync.Mutex{}
		c.locks.chatMessageLocks[chatID] = lock
	}

	return lock
}

func (c *Client) cleanTempFile(path string) {
	os.Remove(path)

	dir := filepath.Dir(path)
	dirName := filepath.Base(dir)
	if strings.HasPrefix(dirName, "telegabber-") {
		os.Remove(dir)
	}
}

func (c *Client) sendMarker(chatId, messageId int64, typ gateway.MarkerType) {
	xmppId, err := gateway.IdsDB.GetByTgIds(c.Session.Login, c.jid, chatId, messageId)
	if err != nil {
		xmppId = strconv.FormatInt(messageId, 10)
	}

	var stringType string
	if typ == gateway.MarkerTypeReceived {
		stringType = "received"
	} else if typ == gateway.MarkerTypeDisplayed {
		stringType = "displayed"
	}
	log.WithFields(log.Fields{
		"xmppId": xmppId,
	}).Debugf("marker: %s", stringType)

	gateway.SendMessageMarker(
		c.jid,
		gateway.CHATNODE(chatId),
		c.xmpp,
		typ,
		xmppId,
	)
}

func (c *Client) updateHandler() {
	listener := c.client.GetListener()
	defer listener.Close()

	for update := range listener.Updates {
		if update.GetClass() == client.ClassUpdate {
			switch update.GetType() {
			case client.TypeUpdateUser:
				typedUpdate, _ := update.(*client.UpdateUser)
				c.updateUser(typedUpdate)
				log.Debugf("%#v", typedUpdate.User)
			case client.TypeUpdateUserStatus:
				typedUpdate, _ := update.(*client.UpdateUserStatus)
				c.updateUserStatus(typedUpdate)
				log.Debugf("%#v", typedUpdate.Status)
			case client.TypeUpdateNewChat:
				typedUpdate, _ := update.(*client.UpdateNewChat)
				c.updateNewChat(typedUpdate)
				log.Debugf("%#v", typedUpdate.Chat)
			case client.TypeUpdateChatPosition:
				typedUpdate, _ := update.(*client.UpdateChatPosition)
				c.updateChatPosition(typedUpdate)
				log.Debugf("%#v", typedUpdate)
			case client.TypeUpdateChatLastMessage:
				typedUpdate, _ := update.(*client.UpdateChatLastMessage)
				c.updateChatLastMessage(typedUpdate)
				log.Debugf("%#v", typedUpdate)
			case client.TypeUpdateNewMessage:
				typedUpdate, _ := update.(*client.UpdateNewMessage)
				c.updateNewMessage(typedUpdate)
				log.Debugf("%#v", typedUpdate.Message)
			case client.TypeUpdateMessageContent:
				typedUpdate, _ := update.(*client.UpdateMessageContent)
				c.updateMessageContent(typedUpdate)
				log.Debugf("%#v", typedUpdate.NewContent)
			case client.TypeUpdateDeleteMessages:
				typedUpdate, _ := update.(*client.UpdateDeleteMessages)
				c.updateDeleteMessages(typedUpdate)
			case client.TypeUpdateAuthorizationState:
				typedUpdate, _ := update.(*client.UpdateAuthorizationState)
				c.updateAuthorizationState(typedUpdate)
			case client.TypeUpdateMessageSendSucceeded:
				typedUpdate, _ := update.(*client.UpdateMessageSendSucceeded)
				c.updateMessageSendSucceeded(typedUpdate)
			case client.TypeUpdateMessageSendFailed:
				typedUpdate, _ := update.(*client.UpdateMessageSendFailed)
				c.updateMessageSendFailed(typedUpdate)
			case client.TypeUpdateChatTitle:
				typedUpdate, _ := update.(*client.UpdateChatTitle)
				c.updateChatTitle(typedUpdate)
			case client.TypeUpdateChatReadOutbox:
				typedUpdate, _ := update.(*client.UpdateChatReadOutbox)
				c.updateChatReadOutbox(typedUpdate)
			case client.TypeUpdateBasicGroupFullInfo:
				typedUpdate, _ := update.(*client.UpdateBasicGroupFullInfo)
				c.updateBasicGroupFullInfo(typedUpdate)
			case client.TypeUpdateChatPermissions:
				typedUpdate, _ := update.(*client.UpdateChatPermissions)
				c.updateChatPermissions(typedUpdate)
			case client.TypeUpdateFile:
				typedUpdate, _ := update.(*client.UpdateFile)
				c.updateFile(typedUpdate)
			default:
				// log only handled types
				continue
			}

			log.Debugf("%#v", update)
		}
	}
}

// new user discovered
func (c *Client) updateUser(update *client.UpdateUser) {
	// check if MUC nicknames should be updated
	oldCacheUser, ok := c.cache.GetUser(update.User.Id)
	c.cache.SetUser(update.User.Id, update.User)
	if ok && (oldCacheUser.FirstName != update.User.FirstName || oldCacheUser.LastName != update.User.LastName) {
		newNickname := c.GetMUCNickname(update.User.Id)
		c.updateMUCsNickname(update.User.Id, newNickname)
	}

	show, status, presenceType := c.userStatusToText(update.User.Status, update.User.Id)
	go c.ProcessStatusUpdate(update.User.Id, status, show, false, gateway.SPType(presenceType))
}

// user status changed
func (c *Client) updateUserStatus(update *client.UpdateUserStatus) {
	show, status, presenceType := c.userStatusToText(update.Status, update.UserId)
	go c.ProcessStatusUpdate(update.UserId, status, show, false, gateway.SPImmed(false), gateway.SPType(presenceType))
}

// new chat discovered
func (c *Client) updateNewChat(update *client.UpdateNewChat) {
	go func() {
		if update.Chat != nil && update.Chat.Photo != nil && update.Chat.Photo.Small != nil {
			_, err := c.DownloadFile(update.Chat.Photo.Small.Id, 10, true)

			if err != nil {
				log.Error("Failed to download the chat photo")
			}
		}

		c.cache.SetChat(update.Chat.Id, update.Chat, true)

		if update.Chat.Positions != nil && len(update.Chat.Positions) > 0 {
			c.subscribeToID(update.Chat.Id, update.Chat, false)
		}

		if update.Chat.Id < 0 {
			c.ProcessStatusUpdate(update.Chat.Id, update.Chat.Title, "chat", true)
		}
	}()
}

// chat position is updated
func (c *Client) updateChatPosition(update *client.UpdateChatPosition) {
	if update.Position != nil && update.Position.Order != 0 {
		go c.subscribeToID(update.ChatId, nil, false)
	}
}

// chat last message is updated
func (c *Client) updateChatLastMessage(update *client.UpdateChatLastMessage) {
	if update.Positions != nil && len(update.Positions) > 0 {
		go c.subscribeToID(update.ChatId, nil, false)
	}
}

// message received
func (c *Client) updateNewMessage(update *client.UpdateNewMessage) {
	chatId := update.Message.ChatId
	if c.Session.IsChatIgnored(chatId) {
		return
	}

	// guarantee sequential message delivering per chat
	lock := c.getChatMessageLock(chatId)
	go func() {
		lock.Lock()
		defer lock.Unlock()

		var forceCmd bool
		if c.LastBotCmdString != "" && update.Message.IsOutgoing {
			if update.Message.Content.MessageContentType() == client.TypeMessageText {
				textMessage, _ := update.Message.Content.(*client.MessageText)

				if textMessage.Text != nil && textMessage.Text.Text == c.LastBotCmdString {
					forceCmd = true
					c.LastBotCmdString = ""
				}
			}
		}

		// ignore self outgoing messages
		if update.Message.IsOutgoing &&
			update.Message.SendingState != nil &&
			update.Message.SendingState.MessageSendingStateType() == client.TypeMessageSendingStatePending &&
			!forceCmd {
			return
		}

		log.WithFields(log.Fields{
			"chat_id": chatId,
		}).Warn("New message from chat")

		c.ProcessIncomingMessage(chatId, update.Message)
	}()
}

// message content updated
func (c *Client) updateMessageContent(update *client.UpdateMessageContent) {
	if c.Session.IsChatIgnored(update.ChatId) {
		return
	}

	markupFunction := c.getFormatter()

	log.Debugf("newContent: %#v", update.NewContent)

	lock := c.getChatMessageLock(update.ChatId)
	lock.Lock()
	lock.Unlock()
	c.SendMessageLock.Lock()
	c.SendMessageLock.Unlock()

	xmppId, xmppIdErr := gateway.IdsDB.GetByTgIds(c.Session.Login, c.jid, update.ChatId, update.MessageId)
	var ignoredResource string
	if xmppIdErr == nil {
		ignoredResource = c.popFromEditOutbox(xmppId)
	} else {
		log.Infof("Couldn't retrieve XMPP message ids for %v, an echo may happen", update.MessageId)
	}
	log.Infof("ignoredResource: %v", ignoredResource)

	chat, _, _ := c.GetContactByID(update.ChatId, nil, true)
	isMUC := c.Session.MUC && c.IsGroup(chat)

	var jids []string
	if isMUC {
		_, jids = c.getMUCJoinedJIDs(update.ChatId, nil, true)
	} else {
		jids = c.GetCarbonFullJids(true, ignoredResource, true)
	}
	if len(jids) == 0 {
		log.Info("The only resource is ignored, aborting")
		return
	}

	if update.NewContent.MessageContentType() == client.TypeMessageText {
		safeToSend := true

		textContent := update.NewContent.(*client.MessageText)
		log.Debugf("textContent: %#v", textContent.Text)

		var replaceId string
		sId := strconv.FormatInt(update.MessageId, 10)
		var isCarbon bool

		go func() {
			message, messageErr := c.client.GetMessage(&client.GetMessageRequest{
				ChatId:    update.ChatId,
				MessageId: update.MessageId,
			})
			if messageErr != nil {
				// odnako za vremya puti
				// sobaka mogla podrasti
				c.MessageIdChangesLock.Lock()
				idsMap, idsMapOk := c.MessageIdChanges[update.ChatId]
				hadNoId := false
				if idsMapOk {
					newId, newIdOk := idsMap[update.MessageId]
					if newIdOk {
						if newId.Id == 0 {
							hadNoId = true
							c.MessageIdChangesLock.Unlock()
							newId.Lock()
						}
						log.Infof("falling back to updated message id: %v/%v->%v", update.ChatId, update.MessageId, newId.Id)
						message, messageErr = c.client.GetMessage(&client.GetMessageRequest{
							ChatId:    update.ChatId,
							MessageId: newId.Id,
						})
					}
				}
				if !hadNoId {
					c.MessageIdChangesLock.Unlock()
				}
			}

			var prefix string
			if messageErr == nil {
				if message.EditDate == 0 {
					return
				}
				log.Debugf("editDate: %v", message.EditDate)

				isCarbon = c.isCarbonsEnabled() && message.IsOutgoing && !isMUC
				// reply correction support in clients is suboptimal yet, so cut them out for now
				prefix, _ = c.messageToPrefix(message, "", "", true)
			} else {
				log.Errorf("No message %v/%v found, cannot reliably determine if it is a carbon and if it is edited: %v", update.ChatId, update.MessageId, messageErr.Error())
			}

			// use XEP-0308 edits only if the last message is edited for sure, fallback otherwise
			if c.Session.NativeEdits {
				lastXmppId, ok := c.getLastChatMessageId(update.ChatId)
				if xmppIdErr != nil {
					xmppId = sId
				}
				if ok && lastXmppId == xmppId {
					replaceId = xmppId
				} else {
					log.Infof("Mismatching message ids: %v %v, falling back to separate edit message", lastXmppId, xmppId)
				}
			}

			var forceFallback bool

			var from string
			var originalFrom string
			var nickname string
			if isMUC {
				if messageErr == nil {
					senderId := c.getMessageSenderId(message)
					nickname = c.GetMUCNickname(senderId)
					originalFrom = gateway.CHATJID(senderId, true)

					safeToSend = c.assureMUCOccupant(update.ChatId, senderId, message.SenderId, chat)
					from = gateway.MUCJID(update.ChatId) + "/" + nickname
				} else {
					nickname = "#ERROR#"
					forceFallback = true
					from = gateway.MUCJID(update.ChatId)
				}
			} else {
				from = gateway.CHATNODE(update.ChatId)
			}

			var text strings.Builder

			if replaceId == "" || forceFallback {
				var editChar string
				if c.Session.AsciiArrows {
					editChar = "e"
				} else {
					editChar = "✎"
				}
				text.WriteString(fmt.Sprintf("%s %v | ", editChar, update.MessageId))
			} else if prefix != "" {
				text.WriteString(prefix)
				text.WriteString(c.getPrefixSeparator(update.ChatId))
			}

			text.WriteString(formatter.Format(
				textContent.Text.Text,
				textContent.Text.Entities,
				markupFunction,
				c,
			))

			id := "e"+sId
			uuid, err := uuid.NewRandom()
			if err == nil {
				id = id+":"+uuid.String()
			}

			// OMEMO encrypt hook - same rules as SendMessageToGateway's (see
			// its doc comment): personal-chat pseudo-JIDs only, gated on the
			// per-chat active flag, once per logical edit rather than per
			// jids resource copy.
			var envelope *e2ee.Envelope
			if !isMUC {
				if backend, ok := gateway.E2EE.Backend(); ok {
					owner := e2ee.OwnedPeer(c.Session.Login, gateway.CHATJID(update.ChatId, false))
					if active, _ := backend.Enabled(owner); active {
						peer := e2ee.PeerID(c.jid)
						if err := c.ensureOMEMOSession(backend, owner, peer); err != nil {
							log.Error(errors.Wrap(err, "Failed to establish OMEMO session"))
						} else if env, err := backend.Encrypt(owner, []e2ee.PeerID{peer}, []byte(text.String())); err != nil {
							log.Error(errors.Wrap(err, "Failed to encrypt OMEMO message"))
						} else {
							envelope = &env
						}
					}
				}
			}

			for _, jid := range jids {
				if safeToSend {
					gateway.SendMessage(jid, from, c.xmpp,
						gateway.SMBody(text.String()), gateway.SMId(id), gateway.SMReplaceId(replaceId),
						gateway.SMIsCarbon(isCarbon), gateway.SMIsGroupchat(isMUC), gateway.SMOriginalFrom(originalFrom),
						gateway.SMOMEMOEnvelope(envelope),
					)
				} else {
					gateway.SendMUCAnnouncement(jid, from, text.String(), nickname, id, c.xmpp)
				}
			}
		}()
	}
}

// message(s) deleted
func (c *Client) updateDeleteMessages(update *client.UpdateDeleteMessages) {
	c.locks.pinOutboxLock.Lock()
	for _, messageId := range update.MessageIds {
		ch, chOk := c.pinOutbox[IntPair{update.ChatId, messageId}]
		if chOk {
			ch <-0
		}
	}
	c.locks.pinOutboxLock.Unlock()

	if update.IsPermanent {
		for _, deleteId := range update.MessageIds {
			c.tryUnlockMessageId(update.ChatId, deleteId)
		}

		if c.Session.IsChatIgnored(update.ChatId) {
			return
		}
		if c.Session.IgnoreGroupDeletions {
			chatType, _, chatTypeErr := c.GetChatType(update.ChatId, false)
			if chatTypeErr == nil && (chatType == ChatTypeBasicGroup || chatType == ChatTypeSupergroup) {
				return
			}
		}

		var isGroupchat bool
		chat, _, _ := c.GetContactByID(update.ChatId, nil, false)
		if c.Session.MUC && c.IsGroup(chat) {
			isGroupchat = true
		}

		var deleteChar string
		if c.Session.AsciiArrows {
			deleteChar = "X "
		} else {
			deleteChar = "✗ "
		}
		text := deleteChar + strings.Join(int64SliceToStringSlice(update.MessageIds), ",")

		var fromJid string
		var jids []string
		if isGroupchat {
			fromJid = gateway.MUCJID(update.ChatId)
			_, jids = c.getMUCJoinedJIDs(update.ChatId, nil, true)
			var nickname string
			if chat != nil {
				nickname = chat.Title
			}
			for _, jid := range jids {
				gateway.SendMUCAnnouncement(jid, fromJid, text, nickname, "", c.xmpp)
			}
		} else {
			fromJid = gateway.CHATNODE(update.ChatId)
			jids = c.GetCarbonFullJids(true, "", false)
			for _, jid := range jids {
				gateway.SendTextMessage(jid, fromJid, text, c.xmpp, isGroupchat)
			}
		}
	}
}

func (c *Client) updateAuthorizationState(update *client.UpdateAuthorizationState) {
	switch update.AuthorizationState.AuthorizationStateType() {
	case client.TypeAuthorizationStateClosing:
		log.Warn("Closing the updates listener")
	case client.TypeAuthorizationStateClosed:
		log.Warn("Closed the updates listener")
		c.forceClose()
	}
}

func (c *Client) updateMessageSendSucceeded(update *client.UpdateMessageSendSucceeded) {
	c.locks.pinOutboxLock.Lock()
	ch, chOk := c.pinOutbox[IntPair{update.Message.ChatId, update.OldMessageId}]
	if chOk {
		ch <-update.Message.Id
	}
	c.locks.pinOutboxLock.Unlock()

	// replace message ID in local database
	log.Debugf("replace message %v with %v", update.OldMessageId, update.Message.Id)
	if err := gateway.IdsDB.ReplaceTgId(c.Session.Login, c.jid, update.Message.ChatId, update.OldMessageId, update.Message.Id); err != nil {
		log.Errorf("failed to replace %v with %v: %v", update.OldMessageId, update.Message.Id, err.Error())
	}

	c.MessageIdChangesLock.Lock()
	idsMap, ok := c.MessageIdChanges[update.Message.ChatId]
	if !ok {
		idsMap = make(map[int64]*newId)
		c.MessageIdChanges[update.Message.ChatId] = idsMap
	}
	id, ok := idsMap[update.OldMessageId]
	if !ok {
		id = newNewId()
		idsMap[update.OldMessageId] = id
	}
	id.Id = update.Message.Id
	c.MessageIdChangesLock.Unlock()

	c.sendMarker(update.Message.ChatId, update.Message.Id, gateway.MarkerTypeReceived)

	// clean uploaded files
	file, _ := c.contentToFile(update.Message.Content)
	if file != nil && file.Local != nil {
		c.cleanTempFile(file.Local.Path)
	}
}
func (c *Client) updateMessageSendFailed(update *client.UpdateMessageSendFailed) {
	c.tryUnlockMessageId(update.Message.ChatId, update.OldMessageId)

	c.locks.pinOutboxLock.Lock()
	ch, chOk := c.pinOutbox[IntPair{update.Message.ChatId, update.OldMessageId}]
	if chOk {
		ch <-0
	}
	c.locks.pinOutboxLock.Unlock()

	// clean uploaded files
	file, _ := c.contentToFile(update.Message.Content)
	if file != nil && file.Local != nil {
		c.cleanTempFile(file.Local.Path)
	}
}

// chat title changed
func (c *Client) updateChatTitle(update *client.UpdateChatTitle) {
	chat, user, _ := c.GetContactByID(update.ChatId, nil, false)
	if c.Session.MUC && c.IsGroup(chat) {
		return
	}

	gateway.SetNickname(c.jid, gateway.CHATNODE(update.ChatId), update.Title, c.xmpp)

	// set also the status (for group chats only)
	if user == nil {
		c.ProcessStatusUpdate(update.ChatId, update.Title, "chat", false, gateway.SPImmed(true))
	}

	// update chat title in the cache
	if chat != nil {
		chat.Title = update.Title
	}
}

func (c *Client) updateChatReadOutbox(update *client.UpdateChatReadOutbox) {
	c.sendMarker(update.ChatId, update.LastReadOutboxMessageId, gateway.MarkerTypeDisplayed)
}

func (c *Client) updateBasicGroupFullInfo(update *client.UpdateBasicGroupFullInfo) {
	if c.Session.MUC && update.BasicGroupFullInfo != nil {
		chatID := -update.BasicGroupId

		c.locks.mucCacheLock.Lock()

		mucState, ok := c.mucCache[chatID]
		if ok && mucState != nil {
			mucState.Occupants.Clear()
			c.updateMUCOccupants(mucState, chatID, update.BasicGroupFullInfo.Members)
		}

		c.locks.mucCacheLock.Unlock()
	}
}

func (c *Client) updateChatPermissions(update *client.UpdateChatPermissions) {
	chat, _, _ := c.GetContactByID(update.ChatId, nil, false)

	// update chat permissions in the cache
	if chat != nil {
		chat.Permissions = update.Permissions
	}

	if c.Session.MUC {
		c.locks.mucCacheLock.Lock()

		mucState, ok := c.mucCache[update.ChatId]
		if ok && mucState != nil {
			_, toJids := c.getMUCJoinedJIDs(update.ChatId, mucState, false)
			for occupant := range mucState.Occupants.Range() {
				affiliation, role := c.memberStatusToAffiliationAndRole(occupant.Status, chat)
				if affiliation != occupant.Affiliation || role != occupant.Role {
					occupant.Affiliation = affiliation
					occupant.Role = role

					c.sendPresence(
						gateway.SPFrom(gateway.MUCNODE(update.ChatId)),
						gateway.SPResource(occupant.Nickname),
						gateway.SPImmed(true),
						gateway.SPMUCJid(gateway.CHATJID(occupant.id, true)),
						gateway.SPMUCAffiliation(affiliation),
						gateway.SPMUCRole(role),
						gateway.SPToJids(toJids),
					)
				}
			}
		}

		c.locks.mucCacheLock.Unlock()
	}
}

func (c *Client) updateFile(update *client.UpdateFile) {
	if update.File != nil && update.File.Local != nil {
		// not really needed, why did I even write this then lol (TODO: maybe clean by some heur anyway)
		/* c.locks.uploadingFilesLock.Lock()
		if _, ok := c.uploadingFiles[update.File.Id]; ok && update.File.Local.CanBeDeleted && update.File.Local.Path != "" {
			err := os.Remove(update.File.Local.Path)
			if err != nil {
				log.Warningf("Couldn't delete uploaded file: %v", err.Error())
			}
			delete(c.uploadingFiles, update.File.Id)
		}
		c.locks.uploadingFilesLock.Unlock() */
	}
}

func (c *Client) tryUnlockMessageId(chatId, messageId int64) {
	c.MessageIdChangesLock.Lock()
	idsMap, ok := c.MessageIdChanges[chatId]
	if ok {
		id, ok := idsMap[messageId]
		if ok {
			id.Unlock()
		}
	}
	c.MessageIdChangesLock.Unlock()
}