iartamonov/mautrix-telegram
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

ids: add channel ID to message ID to ensure uniqueness (#25)

Browse Source
This commit is contained in:
Tulir Asokan
2024-08-26 20:42:06 +03:00
committed by GitHub
parent d7508579e5
commit e3e709eec6
8 changed files with 111 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/connector/backfill.go
+3 -3
View File
@@ -31,7 +31,7 @@ func (t *TelegramClient) FetchMessages(ctx context.Context, fetchParams bridgev2
Limit: fetchParams.Count,
}
if fetchParams.AnchorMessage != nil && !fetchParams.Forward {
req.MaxID, err = ids.ParseMessageID(fetchParams.AnchorMessage.ID)
_, req.MaxID, err = ids.ParseMessageID(fetchParams.AnchorMessage.ID)
if err != nil {
return nil, err
}
@@ -61,7 +61,7 @@ func (t *TelegramClient) FetchMessages(ctx context.Context, fetchParams bridgev2
var stopAt int
if fetchParams.AnchorMessage != nil && fetchParams.Forward {
stopAt, err = ids.ParseMessageID(fetchParams.AnchorMessage.ID)
_, stopAt, err = ids.ParseMessageID(fetchParams.AnchorMessage.ID)
if err != nil {
return nil, err
}
@@ -100,7 +100,7 @@ func (t *TelegramClient) FetchMessages(ctx context.Context, fetchParams bridgev2
backfillMessage := bridgev2.BackfillMessage{
ConvertedMessage: converted,
Sender: sender,
ID: ids.MakeMessageID(message.ID),
ID: ids.GetMessageIDFromMessage(message),
Timestamp: time.Unix(int64(message.Date), 0),
}
pkg/connector/client.go
+4 -4
View File
@@ -127,10 +127,10 @@ func NewTelegramClient(ctx context.Context, tc *TelegramConnector, login *bridge
})
dispatcher.OnUserName(client.onUserName)
dispatcher.OnDeleteMessages(func(ctx context.Context, e tg.Entities, update *tg.UpdateDeleteMessages) error {
return client.onDeleteMessages(ctx, update)
return client.onDeleteMessages(ctx, 0, update)
})
dispatcher.OnDeleteChannelMessages(func(ctx context.Context, e tg.Entities, update *tg.UpdateDeleteChannelMessages) error {
return client.onDeleteMessages(ctx, update)
return client.onDeleteMessages(ctx, update.ChannelID, update)
})
dispatcher.OnEditMessage(func(ctx context.Context, e tg.Entities, update *tg.UpdateEditMessage) error {
return client.onMessageEdit(ctx, update)
@@ -272,13 +272,13 @@ func NewTelegramClient(ctx context.Context, tc *TelegramConnector, login *bridge
return url
}
message, err := tc.Bridge.DB.Message.GetFirstPartByID(ctx, client.loginID, ids.MakeMessageID(msgID))
message, err := tc.Bridge.DB.Message.GetFirstPartByID(ctx, client.loginID, ids.MakeMessageID(portalKey, msgID))
if err != nil {
log.Err(err).Msg("error getting message")
return url
}
return fmt.Sprintf("https://matrix.to/#/%s/%s", portal.MXID, message.MXID)
return portal.MXID.EventURI(message.MXID, tc.Bridge.Matrix.ServerName()).MatrixToURL()
},
}
client.matrixParser = &matrixfmt.HTMLParser{
pkg/connector/ids/ids.go
+65 -4
View File
@@ -25,16 +25,64 @@ func MakeUserLoginID(userID int64) networkid.UserLoginID {
return networkid.UserLoginID(strconv.FormatInt(userID, 10))
}
func MakeMessageID(messageID int) networkid.MessageID {
return networkid.MessageID(strconv.Itoa(messageID))
func GetMessageIDFromMessage(message tg.MessageClass) networkid.MessageID {
var peer tg.PeerClass
switch typedMsg := message.(type) {
case *tg.MessageEmpty:
peer, _ = typedMsg.GetPeerID()
case *tg.Message:
peer = typedMsg.GetPeerID()
case *tg.MessageService:
peer = typedMsg.GetPeerID()
default:
panic(fmt.Sprintf("unexpected message type %T", message))
}
return MakeMessageID(peer, message.GetID())
}
func MakeMessageID(rawChatID any, messageID int) networkid.MessageID {
var channelID int64
switch typedChatID := rawChatID.(type) {
case networkid.PortalKey:
if typedChatID.Receiver == "" {
_, channelID, _ = ParsePortalID(typedChatID.ID)
}
case *tg.PeerChannel:
channelID = typedChatID.ChannelID
case int64:
channelID = typedChatID
case *tg.PeerUser, *tg.PeerChat:
// No channel ID
case nil:
// Also no channel ID
default:
panic(fmt.Sprintf("unexpected chat ID type %T", rawChatID))
}
if channelID != 0 {
return networkid.MessageID(fmt.Sprintf("%d.%d", channelID, messageID))
}
return networkid.MessageID(fmt.Sprintf("%d", messageID))
}
func MakePaginationCursorID(messageID int) networkid.PaginationCursor {
return networkid.PaginationCursor(strconv.Itoa(messageID))
}
func ParseMessageID(messageID networkid.MessageID) (int, error) {
return strconv.Atoi(string(messageID))
func ParseMessageID(networkID networkid.MessageID) (channelID int64, messageID int, err error) {
parts := strings.Split(string(networkID), ".")
if len(parts) == 1 {
messageID, err = strconv.Atoi(parts[0])
} else if len(parts) == 2 {
channelID, err = strconv.ParseInt(parts[0], 10, 64)
if err != nil {
err = fmt.Errorf("failed to parse chat ID: %w", err)
return
}
messageID, err = strconv.Atoi(parts[1])
} else {
err = fmt.Errorf("invalid number of parts in message ID")
}
return
}
type PeerType string
@@ -81,6 +129,19 @@ func (pt PeerType) AsPortalKey(chatID int64, receiver networkid.UserLoginID) net
return portalKey
}
func GetChatID(peer tg.PeerClass) int64 {
switch v := peer.(type) {
case *tg.PeerUser:
return v.UserID
case *tg.PeerChat:
return v.ChatID
case *tg.PeerChannel:
return v.ChannelID
default:
panic(fmt.Errorf("unknown peer class type %T", v))
}
}
func MakePortalKey(peer tg.PeerClass, receiver networkid.UserLoginID) networkid.PortalKey {
switch v := peer.(type) {
case *tg.PeerUser:
pkg/connector/matrix.go
+7 -7
View File
@@ -93,7 +93,7 @@ func (t *TelegramClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.
var replyTo tg.InputReplyToClass
if msg.ReplyTo != nil {
messageID, err := ids.ParseMessageID(msg.ReplyTo.ID)
_, messageID, err := ids.ParseMessageID(msg.ReplyTo.ID)
if err != nil {
return nil, err
}
@@ -182,7 +182,7 @@ func (t *TelegramClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.
resp = &bridgev2.MatrixMessageResponse{
DB: &database.Message{
ID: ids.MakeMessageID(tgMessageID),
ID: ids.MakeMessageID(msg.Portal, tgMessageID),
MXID: msg.Event.ID,
Room: msg.Portal.PortalKey,
SenderID: t.userID,
@@ -207,7 +207,7 @@ func (t *TelegramClient) HandleMatrixEdit(ctx context.Context, msg *bridgev2.Mat
return err
}
targetID, err := ids.ParseMessageID(msg.EditTarget.ID)
_, targetID, err := ids.ParseMessageID(msg.EditTarget.ID)
if err != nil {
return err
}
@@ -271,7 +271,7 @@ func (t *TelegramClient) HandleMatrixEdit(ctx context.Context, msg *bridgev2.Mat
func (t *TelegramClient) HandleMatrixMessageRemove(ctx context.Context, msg *bridgev2.MatrixMessageRemove) error {
if dbMsg, err := t.main.Bridge.DB.Message.GetPartByMXID(ctx, msg.TargetMessage.MXID); err != nil {
return err
} else if messageID, err := ids.ParseMessageID(dbMsg.ID); err != nil {
} else if _, messageID, err := ids.ParseMessageID(dbMsg.ID); err != nil {
return err
} else if peer, err := t.inputPeerForPortalID(ctx, msg.Portal.ID); err != nil {
return err
@@ -333,7 +333,7 @@ func (t *TelegramClient) HandleMatrixReaction(ctx context.Context, msg *bridgev2
if err != nil {
return nil, err
}
targetMessageID, err := ids.ParseMessageID(msg.TargetMessage.ID)
_, targetMessageID, err := ids.ParseMessageID(msg.TargetMessage.ID)
if err != nil {
return nil, err
}
@@ -384,7 +384,7 @@ func (t *TelegramClient) HandleMatrixReactionRemove(ctx context.Context, msg *br
}
}
messageID, err := ids.ParseMessageID(msg.TargetReaction.MessageID)
_, messageID, err := ids.ParseMessageID(msg.TargetReaction.MessageID)
if err != nil {
return err
}
@@ -444,7 +444,7 @@ func (t *TelegramClient) HandleMatrixReadReceipt(ctx context.Context, msg *bridg
}
}
var maxID int
maxID, readMessagesErr = ids.ParseMessageID(message.ID)
_, maxID, readMessagesErr = ids.ParseMessageID(message.ID)
if readMessagesErr != nil {
return
}
pkg/connector/reactions.go
+1 -1
View File
@@ -95,7 +95,7 @@ func (t *TelegramClient) handleTelegramReactions(ctx context.Context, msg *tg.Me
Int("message_id", msg.ID).
Logger()
dbMsg, err := t.main.Bridge.DB.Message.GetFirstPartByID(ctx, t.loginID, ids.MakeMessageID(msg.ID))
dbMsg, err := t.main.Bridge.DB.Message.GetFirstPartByID(ctx, t.loginID, ids.GetMessageIDFromMessage(msg))
if err != nil {
log.Err(err).Msg("failed to get message from database")
return
pkg/connector/sync.go
+3 -3
View File
@@ -46,7 +46,7 @@ func (t *TelegramClient) SyncChats(ctx context.Context) error {
}
messages := map[networkid.MessageID]tg.MessageClass{}
for _, message := range dialogs.GetMessages() {
messages[ids.MakeMessageID(message.GetID())] = message
messages[ids.GetMessageIDFromMessage(message)] = message
}
var created int
@@ -79,7 +79,7 @@ func (t *TelegramClient) SyncChats(ctx context.Context) error {
if portal == nil || portal.MXID == "" {
// Check what the latest message is
topMessage := messages[ids.MakeMessageID(dialog.TopMessage)]
topMessage := messages[ids.MakeMessageID(dialog.Peer, dialog.TopMessage)]
if topMessage.TypeID() == tg.MessageServiceTypeID {
action := topMessage.(*tg.MessageService).Action
if action.TypeID() == tg.MessageActionContactSignUpTypeID || action.TypeID() == tg.MessageActionHistoryClearTypeID {
@@ -104,7 +104,7 @@ func (t *TelegramClient) SyncChats(ctx context.Context) error {
CreatePortal: true,
},
CheckNeedsBackfillFunc: func(ctx context.Context, latestMessage *database.Message) (bool, error) {
latestMessageID, err := ids.ParseMessageID(latestMessage.ID)
_, latestMessageID, err := ids.ParseMessageID(latestMessage.ID)
if err != nil {
return false, err
}
pkg/connector/telegram.go
+25 -15
View File
@@ -59,7 +59,7 @@ func (t *TelegramClient) onUpdateNewMessage(ctx context.Context, update IGetMess
CreatePortal: true,
Timestamp: time.Unix(int64(msg.Date), 0),
},
ID: ids.MakeMessageID(msg.ID),
ID: ids.GetMessageIDFromMessage(msg),
Data: msg,
ConvertMessageFunc: t.convertToMatrix,
})
@@ -207,14 +207,22 @@ func (t *TelegramClient) onUserName(ctx context.Context, e tg.Entities, update *
return nil
}
func (t *TelegramClient) onDeleteMessages(ctx context.Context, update IGetMessages) error {
func (t *TelegramClient) onDeleteMessages(ctx context.Context, channelID int64, update IGetMessages) error {
for _, messageID := range update.GetMessages() {
parts, err := t.main.Bridge.DB.Message.GetAllPartsByID(ctx, t.loginID, ids.MakeMessageID(messageID))
if err != nil {
return err
}
if len(parts) == 0 {
return fmt.Errorf("no parts found for message %d", messageID)
var portalKey networkid.PortalKey
if channelID == 0 {
// TODO have mautrix-go do this part too?
parts, err := t.main.Bridge.DB.Message.GetAllPartsByID(ctx, t.loginID, ids.MakeMessageID(channelID, messageID))
if err != nil {
return err
}
if len(parts) == 0 {
return fmt.Errorf("no parts found for message %d", messageID)
}
// TODO can deletes happen across rooms?
portalKey = parts[0].Room
} else {
portalKey = ids.MakePortalKey(&tg.PeerChannel{ChannelID: channelID}, t.loginID)
}
t.main.Bridge.QueueRemoteEvent(t.userLogin, &simplevent.MessageRemove{
EventMeta: simplevent.EventMeta{
@@ -224,10 +232,10 @@ func (t *TelegramClient) onDeleteMessages(ctx context.Context, update IGetMessag
Str("action", "delete message").
Int("message_id", messageID)
},
PortalKey: parts[0].Room,
PortalKey: portalKey,
CreatePortal: false,
},
TargetMessage: ids.MakeMessageID(messageID),
TargetMessage: ids.MakeMessageID(channelID, messageID),
})
}
return nil
@@ -285,8 +293,8 @@ func (t *TelegramClient) onMessageEdit(ctx context.Context, update IGetMessage)
PortalKey: ids.MakePortalKey(msg.PeerID, t.loginID),
Timestamp: time.Unix(int64(msg.EditDate), 0),
},
ID: ids.MakeMessageID(msg.ID),
TargetMessage: ids.MakeMessageID(msg.ID),
ID: ids.GetMessageIDFromMessage(msg),
TargetMessage: ids.GetMessageIDFromMessage(msg),
Data: msg,
ConvertEditFunc: func(ctx context.Context, portal *bridgev2.Portal, intent bridgev2.MatrixAPI, existing []*database.Message, data *tg.Message) (*bridgev2.ConvertedEdit, error) {
converted, err := t.convertToMatrix(ctx, portal, intent, msg)
@@ -335,7 +343,9 @@ func (t *TelegramClient) handleTyping(portal networkid.PortalKey, userID int64,
func (t *TelegramClient) updateReadReceipt(update *tg.UpdateReadHistoryOutbox) error {
user, ok := update.Peer.(*tg.PeerUser)
if !ok {
return fmt.Errorf("unsupported peer type %T", update.Peer)
// Read receipts from other users are meaningless in chats/channels
// (they only say "someone read the message" and not who)
return nil
}
t.main.Bridge.QueueRemoteEvent(t.userLogin, &simplevent.Receipt{
EventMeta: simplevent.EventMeta{
@@ -346,7 +356,7 @@ func (t *TelegramClient) updateReadReceipt(update *tg.UpdateReadHistoryOutbox) e
Sender: ids.MakeUserID(user.UserID),
},
},
LastTarget: ids.MakeMessageID(update.MaxID),
LastTarget: ids.MakeMessageID(update.Peer, update.MaxID),
})
return nil
}
@@ -358,7 +368,7 @@ func (t *TelegramClient) onOwnReadReceipt(portalKey networkid.PortalKey, maxID i
PortalKey: portalKey,
Sender: t.mySender(),
},
LastTarget: ids.MakeMessageID(maxID),
LastTarget: ids.MakeMessageID(portalKey, maxID),
})
return nil
}
pkg/connector/tomatrix.go
+3 -3
View File
@@ -156,7 +156,7 @@ func (c *TelegramClient) convertToMatrix(ctx context.Context, portal *bridgev2.P
switch replyTo := replyTo.(type) {
case *tg.MessageReplyHeader:
cm.ReplyTo = &networkid.MessageOptionalPartID{
MessageID: ids.MakeMessageID(replyTo.ReplyToMsgID),
MessageID: ids.MakeMessageID(replyTo.ReplyToPeerID, replyTo.ReplyToMsgID),
}
default:
log.Warn().Type("reply_to", replyTo).Msg("unhandled reply to type")
@@ -415,8 +415,8 @@ func (c *TelegramClient) convertContact(media tg.MessageMediaClass) *bridgev2.Co
if contact.UserID > 0 {
content.Format = event.FormatHTML
content.FormattedBody = fmt.Sprintf(
`Shared contact info for <a href="https://matrix.to/#/%s">%s</a>: %s`,
c.main.Bridge.Matrix.GhostIntent(ids.MakeUserID(contact.UserID)).GetMXID(),
`Shared contact info for <a href="%s">%s</a>: %s`,
c.main.Bridge.Matrix.GhostIntent(ids.MakeUserID(contact.UserID)).GetMXID().URI().MatrixToURL(),
html.EscapeString(name),
html.EscapeString(formattedPhone),
)