mirror of
https://dev.narayana.im/narayana/telegabber.git
synced 2026-08-05 12:17:06 +00:00
565 lines
16 KiB
Go
565 lines
16 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"
|
|
|
|
"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 {
|
|
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
|
|
|
|
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 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()
|
|
}
|
|
for _, jid := range jids {
|
|
if safeToSend {
|
|
gateway.SendMessage(jid, from, text.String(), id, c.xmpp, nil, 0, replaceId, isCarbon, isMUC, false, originalFrom, "")
|
|
} 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 {
|
|
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)
|
|
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.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.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)
|
|
|
|
// 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.key, true)),
|
|
gateway.SPMUCAffiliation(affiliation),
|
|
gateway.SPMUCRole(role),
|
|
gateway.SPToJids(toJids),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
c.locks.mucCacheLock.Unlock()
|
|
}
|
|
}
|