telegabber/telegram/handlers.go

421 lines
12 KiB
Go

package telegram
import (
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"dev.narayana.im/narayana/telegabber/telegram/formatter"
"dev.narayana.im/narayana/telegabber/xmpp/gateway"
log "github.com/sirupsen/logrus"
"github.com/zelenin/go-tdlib/client"
)
func uhOh() {
log.Fatal("Update type mismatch")
}
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,
strconv.FormatInt(chatId, 10),
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, ok := update.(*client.UpdateUser)
if !ok {
uhOh()
}
c.updateUser(typedUpdate)
log.Debugf("%#v", typedUpdate.User)
case client.TypeUpdateUserStatus:
typedUpdate, ok := update.(*client.UpdateUserStatus)
if !ok {
uhOh()
}
c.updateUserStatus(typedUpdate)
log.Debugf("%#v", typedUpdate.Status)
case client.TypeUpdateNewChat:
typedUpdate, ok := update.(*client.UpdateNewChat)
if !ok {
uhOh()
}
c.updateNewChat(typedUpdate)
log.Debugf("%#v", typedUpdate.Chat)
case client.TypeUpdateChatPosition:
typedUpdate, ok := update.(*client.UpdateChatPosition)
if !ok {
uhOh()
}
c.updateChatPosition(typedUpdate)
log.Debugf("%#v", typedUpdate)
case client.TypeUpdateChatLastMessage:
typedUpdate, ok := update.(*client.UpdateChatLastMessage)
if !ok {
uhOh()
}
c.updateChatLastMessage(typedUpdate)
log.Debugf("%#v", typedUpdate)
case client.TypeUpdateNewMessage:
typedUpdate, ok := update.(*client.UpdateNewMessage)
if !ok {
uhOh()
}
c.updateNewMessage(typedUpdate)
log.Debugf("%#v", typedUpdate.Message)
case client.TypeUpdateMessageContent:
typedUpdate, ok := update.(*client.UpdateMessageContent)
if !ok {
uhOh()
}
c.updateMessageContent(typedUpdate)
log.Debugf("%#v", typedUpdate.NewContent)
case client.TypeUpdateDeleteMessages:
typedUpdate, ok := update.(*client.UpdateDeleteMessages)
if !ok {
uhOh()
}
c.updateDeleteMessages(typedUpdate)
case client.TypeUpdateAuthorizationState:
typedUpdate, ok := update.(*client.UpdateAuthorizationState)
if !ok {
uhOh()
}
c.updateAuthorizationState(typedUpdate)
case client.TypeUpdateMessageSendSucceeded:
typedUpdate, ok := update.(*client.UpdateMessageSendSucceeded)
if !ok {
uhOh()
}
c.updateMessageSendSucceeded(typedUpdate)
case client.TypeUpdateMessageSendFailed:
typedUpdate, ok := update.(*client.UpdateMessageSendFailed)
if !ok {
uhOh()
}
c.updateMessageSendFailed(typedUpdate)
case client.TypeUpdateChatTitle:
typedUpdate, ok := update.(*client.UpdateChatTitle)
if !ok {
uhOh()
}
c.updateChatTitle(typedUpdate)
case client.TypeUpdateChatReadOutbox:
typedUpdate, ok := update.(*client.UpdateChatReadOutbox)
if !ok {
uhOh()
}
c.updateChatReadOutbox(typedUpdate)
default:
// log only handled types
continue
}
log.Debugf("%#v", update)
}
}
}
// new user discovered
func (c *Client) updateUser(update *client.UpdateUser) {
c.cache.SetUser(update.User.Id, update.User)
show, status, presenceType := c.userStatusToText(update.User.Status, update.User.Id)
go c.ProcessStatusUpdate(update.User.Id, status, show, 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, 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)
if update.Chat.Positions != nil && len(update.Chat.Positions) > 0 {
c.subscribeToID(update.Chat.Id, update.Chat)
}
if update.Chat.Id < 0 {
c.ProcessStatusUpdate(update.Chat.Id, update.Chat.Title, "chat")
}
}()
}
// 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)
}
}
// 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)
}
}
// message received
func (c *Client) updateNewMessage(update *client.UpdateNewMessage) {
chatId := update.Message.ChatId
// guarantee sequential message delivering per chat
lock := c.getChatMessageLock(chatId)
go func() {
lock.Lock()
defer lock.Unlock()
c.updateLastMessageHash(update.Message.ChatId, update.Message.Id, update.Message.Content)
// ignore self outgoing messages
if update.Message.IsOutgoing &&
update.Message.SendingState != nil &&
update.Message.SendingState.MessageSendingStateType() == client.TypeMessageSendingStatePending {
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) {
markupFunction := c.getFormatter()
defer c.updateLastMessageHash(update.ChatId, update.MessageId, update.NewContent)
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)
jids := c.getCarbonFullJids(true, ignoredResource)
if len(jids) == 0 {
log.Info("The only resource is ignored, aborting")
return
}
if update.NewContent.MessageContentType() == client.TypeMessageText && c.hasLastMessageHashChanged(update.ChatId, update.MessageId, update.NewContent) {
textContent := update.NewContent.(*client.MessageText)
log.Debugf("textContent: %#v", textContent.Text)
var replaceId string
sId := strconv.FormatInt(update.MessageId, 10)
var isCarbon bool
// 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)
}
}
message, messageErr := c.client.GetMessage(&client.GetMessageRequest{
ChatId: update.ChatId,
MessageId: update.MessageId,
})
var prefix string
if messageErr == nil {
isCarbon = c.isCarbonsEnabled() && message.IsOutgoing
// 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's a carbon", update.ChatId, update.MessageId)
}
var text strings.Builder
if replaceId == "" {
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,
))
sChatId := strconv.FormatInt(update.ChatId, 10)
for _, jid := range jids {
gateway.SendMessage(jid, sChatId, text.String(), "e"+sId, c.xmpp, nil, replaceId, isCarbon, false)
}
}
}
// message(s) deleted
func (c *Client) updateDeleteMessages(update *client.UpdateDeleteMessages) {
if update.IsPermanent {
var deleteChar string
if c.Session.AsciiArrows {
deleteChar = "X "
} else {
deleteChar = "✗ "
}
text := deleteChar + strings.Join(int64SliceToStringSlice(update.MessageIds), ",")
gateway.SendTextMessage(c.jid, strconv.FormatInt(update.ChatId, 10), text, c.xmpp)
}
}
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) {
// 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.updateLastMessageHash(update.Message.ChatId, update.Message.Id, update.Message.Content)
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) {
// 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) {
gateway.SetNickname(c.jid, strconv.FormatInt(update.ChatId, 10), update.Title, c.xmpp)
// set also the status (for group chats only)
chat, user, _ := c.GetContactByID(update.ChatId, nil)
if user == nil {
c.ProcessStatusUpdate(update.ChatId, update.Title, "chat", 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)
}