package telegram import ( "fmt" "os" "path/filepath" "strconv" "strings" "sync" "dev.narayana.im/narayana/telegabber/telegram/formatter" "dev.narayana.im/narayana/telegabber/xmpp/gateway" "github.com/google/uuid" 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) 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 cacheUser, ok := c.cache.GetUser(update.User.Id) if ok && (cacheUser.FirstName != update.User.FirstName || cacheUser.LastName != update.User.LastName) { newNickname := c.GetMUCNickname(update.User.Id) c.updateMUCsNickname(update.User.Id, newNickname) } 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, false) } 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, 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) isMUC := c.Session.MUC && c.IsGroup(chat) var jids []string if isMUC { _, jids = c.getMUCJoinedJIDs(update.ChatId, nil, true) } else { c.getCarbonFullJids(true, ignoredResource) } if len(jids) == 0 { log.Info("The only resource is ignored, aborting") return } if update.NewContent.MessageContentType() == client.TypeMessageText { textContent := update.NewContent.(*client.MessageText) log.Debugf("textContent: %#v", textContent.Text) var replaceId string sId := strconv.FormatInt(update.MessageId, 10) var isCarbon bool message, messageErr := c.client.GetMessage(&client.GetMessageRequest{ ChatId: update.ChatId, MessageId: update.MessageId, }) var prefix string if messageErr == nil { if message.EditDate == 0 { return } 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", update.ChatId, update.MessageId) } // 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 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, )) var from string var originalFrom string if isMUC { var nickname string if messageErr == nil { senderId := c.getMessageSenderId(message) nickname = c.GetMUCNickname(senderId) originalFrom = gateway.CHATJID(senderId, true) } else { nickname = "#ERROR#" } from = gateway.MUCJID(update.ChatId) + "/" + nickname } else { from = gateway.CHATNODE(update.ChatId) } id := "e"+sId uuid, err := uuid.NewRandom() if err == nil { id = id+":"+uuid.String() } for _, jid := range jids { gateway.SendMessage(jid, from, text.String(), id, c.xmpp, nil, 0, replaceId, isCarbon, isMUC, false, originalFrom, "") } } } // 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 { if c.Session.IsChatIgnored(update.ChatId) { return } if c.Session.IgnoreGroupDeletions { chatType, _, chatTypeErr := c.GetChatType(update.ChatId) if chatTypeErr == nil && (chatType == ChatTypeBasicGroup || chatType == ChatTypeSupergroup) { return } } var isGroupchat bool chat, _, _ := c.GetContactByID(update.ChatId, nil) 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) } else { fromJid = gateway.CHATNODE(update.ChatId) c.getCarbonFullJids(true, "") } 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.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.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) 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", 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 = make(map[int64]*MUCOccupant) c.updateMUCOccupants(mucState, chatID, update.BasicGroupFullInfo.Members) } c.locks.mucCacheLock.Unlock() } } func (c *Client) updateChatPermissions(update *client.UpdateChatPermissions) { chat, _, _ := c.GetContactByID(update.ChatId, nil) // 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 { for memberID, occupant := range mucState.Occupants { 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(memberID, true)), gateway.SPMUCAffiliation(affiliation), gateway.SPMUCRole(role), ) } } } c.locks.mucCacheLock.Unlock() } }