backfill.go 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284
  1. package main
  2. import (
  3. "crypto/sha256"
  4. "encoding/base64"
  5. "fmt"
  6. "sort"
  7. "github.com/bwmarrin/discordgo"
  8. "github.com/rs/zerolog"
  9. "maunium.net/go/mautrix"
  10. "maunium.net/go/mautrix/bridge/bridgeconfig"
  11. "maunium.net/go/mautrix/event"
  12. "maunium.net/go/mautrix/id"
  13. "go.mau.fi/mautrix-discord/database"
  14. )
  15. func (portal *Portal) forwardBackfillInitial(source *User) {
  16. defer portal.forwardBackfillLock.Unlock()
  17. // This should only be called from CreateMatrixRoom which locks forwardBackfillLock before creating the room.
  18. if portal.forwardBackfillLock.TryLock() {
  19. panic("forwardBackfillInitial() called without locking forwardBackfillLock")
  20. }
  21. limit := portal.bridge.Config.Bridge.Backfill.Limits.Initial.Channel
  22. if portal.GuildID == "" {
  23. limit = portal.bridge.Config.Bridge.Backfill.Limits.Initial.DM
  24. }
  25. if limit == 0 {
  26. return
  27. }
  28. log := portal.zlog.With().
  29. Str("action", "initial backfill").
  30. Int("limit", limit).
  31. Logger()
  32. portal.backfillLimited(log, source, limit, "")
  33. }
  34. func (portal *Portal) ForwardBackfillMissed(source *User, meta *discordgo.Channel) {
  35. limit := portal.bridge.Config.Bridge.Backfill.Limits.Missed.Channel
  36. if portal.GuildID == "" {
  37. limit = portal.bridge.Config.Bridge.Backfill.Limits.Missed.DM
  38. }
  39. if limit == 0 {
  40. return
  41. }
  42. log := portal.zlog.With().
  43. Str("action", "missed event backfill").
  44. Int("limit", limit).
  45. Logger()
  46. portal.forwardBackfillLock.Lock()
  47. defer portal.forwardBackfillLock.Unlock()
  48. lastMessage := portal.bridge.DB.Message.GetLast(portal.Key)
  49. if lastMessage == nil || meta.LastMessageID == "" {
  50. log.Debug().Msg("Not backfilling, no last message in database or no last message in metadata")
  51. return
  52. } else if !shouldBackfill(lastMessage.DiscordID, meta.LastMessageID) {
  53. log.Debug().
  54. Str("last_bridged_message", lastMessage.DiscordID).
  55. Str("last_server_message", meta.LastMessageID).
  56. Msg("Not backfilling, last message in database is newer than last message in metadata")
  57. return
  58. }
  59. log.Debug().
  60. Str("last_bridged_message", lastMessage.DiscordID).
  61. Str("last_server_message", meta.LastMessageID).
  62. Msg("Backfilling missed messages")
  63. if limit < 0 {
  64. portal.backfillUnlimitedMissed(log, source, lastMessage.DiscordID)
  65. } else {
  66. portal.backfillLimited(log, source, limit, lastMessage.DiscordID)
  67. }
  68. }
  69. const messageFetchChunkSize = 50
  70. func (portal *Portal) collectBackfillMessages(log zerolog.Logger, source *User, limit int, until string) ([]*discordgo.Message, bool, error) {
  71. var messages []*discordgo.Message
  72. var before string
  73. var foundAll bool
  74. for {
  75. log.Debug().Str("before_id", before).Msg("Fetching messages for backfill")
  76. newMessages, err := source.Session.ChannelMessages(portal.Key.ChannelID, messageFetchChunkSize, before, "", "")
  77. if err != nil {
  78. return nil, false, err
  79. }
  80. if until != "" {
  81. for i, msg := range newMessages {
  82. if compareMessageIDs(msg.ID, until) <= 0 {
  83. log.Debug().
  84. Str("message_id", msg.ID).
  85. Str("until_id", until).
  86. Msg("Found message that was already bridged")
  87. newMessages = newMessages[:i]
  88. foundAll = true
  89. break
  90. }
  91. }
  92. }
  93. messages = append(messages, newMessages...)
  94. log.Debug().Int("count", len(newMessages)).Msg("Added messages to backfill collection")
  95. if len(newMessages) <= messageFetchChunkSize || len(messages) >= limit {
  96. break
  97. }
  98. before = newMessages[len(newMessages)-1].ID
  99. }
  100. if len(messages) > limit {
  101. messages = messages[:limit]
  102. }
  103. return messages, foundAll, nil
  104. }
  105. func (portal *Portal) backfillLimited(log zerolog.Logger, source *User, limit int, after string) {
  106. messages, foundAll, err := portal.collectBackfillMessages(log, source, limit, after)
  107. if err != nil {
  108. log.Err(err).Msg("Error collecting messages to forward backfill")
  109. return
  110. }
  111. log.Info().
  112. Int("count", len(messages)).
  113. Bool("found_all", foundAll).
  114. Msg("Collected messages to backfill")
  115. sort.Sort(MessageSlice(messages))
  116. if !foundAll {
  117. _, err = portal.sendMatrixMessage(portal.MainIntent(), event.EventMessage, &event.MessageEventContent{
  118. MsgType: event.MsgNotice,
  119. Body: "Some messages may have been missed here while the bridge was offline.",
  120. }, nil, 0)
  121. if err != nil {
  122. log.Warn().Err(err).Msg("Failed to send missed message warning")
  123. } else {
  124. log.Debug().Msg("Sent warning about possibly missed messages")
  125. }
  126. }
  127. portal.sendBackfillBatch(log, source, messages)
  128. }
  129. func (portal *Portal) backfillUnlimitedMissed(log zerolog.Logger, source *User, after string) {
  130. for {
  131. log.Debug().Str("after_id", after).Msg("Fetching chunk of messages to backfill")
  132. messages, err := source.Session.ChannelMessages(portal.Key.ChannelID, messageFetchChunkSize, "", after, "")
  133. if err != nil {
  134. log.Err(err).Msg("Error fetching chunk of messages to forward backfill")
  135. return
  136. }
  137. log.Debug().Int("count", len(messages)).Msg("Fetched chunk of messages to backfill")
  138. sort.Sort(MessageSlice(messages))
  139. portal.sendBackfillBatch(log, source, messages)
  140. if len(messages) < messageFetchChunkSize {
  141. // Assume that was all the missing messages
  142. log.Debug().Msg("Chunk had less than 50 messages, stopping backfill")
  143. return
  144. }
  145. after = messages[len(messages)-1].ID
  146. }
  147. }
  148. func (portal *Portal) sendBackfillBatch(log zerolog.Logger, source *User, messages []*discordgo.Message) {
  149. if portal.bridge.Config.Homeserver.Software == bridgeconfig.SoftwareHungry {
  150. log.Debug().Msg("Using hungryserv, sending messages with batch send endpoint")
  151. portal.forwardBatchSend(log, source, messages)
  152. } else {
  153. log.Debug().Msg("Not using hungryserv, sending messages one by one")
  154. for _, msg := range messages {
  155. portal.handleDiscordMessageCreate(source, msg, nil)
  156. }
  157. }
  158. }
  159. func (portal *Portal) forwardBatchSend(log zerolog.Logger, source *User, messages []*discordgo.Message) {
  160. evts := make([]*event.Event, 0, len(messages))
  161. dbMessages := make([]database.Message, 0, len(messages))
  162. for _, msg := range messages {
  163. for _, mention := range msg.Mentions {
  164. puppet := portal.bridge.GetPuppetByID(mention.ID)
  165. puppet.UpdateInfo(nil, mention)
  166. }
  167. puppet := portal.bridge.GetPuppetByID(msg.Author.ID)
  168. puppet.UpdateInfo(source, msg.Author)
  169. intent := puppet.IntentFor(portal)
  170. replyTo := portal.getReplyTarget(source, msg.MessageReference, true)
  171. ts, _ := discordgo.SnowflakeTimestamp(msg.ID)
  172. parts := portal.convertDiscordMessage(intent, msg)
  173. for i, part := range parts {
  174. if replyTo != nil {
  175. part.Content.RelatesTo = &event.RelatesTo{InReplyTo: replyTo}
  176. // Only set reply for first event
  177. replyTo = nil
  178. }
  179. partName := part.AttachmentID
  180. // Always use blank part name for first part so that replies and other things
  181. // can reference it without knowing about attachments.
  182. if i == 0 {
  183. partName = ""
  184. }
  185. evts = append(evts, &event.Event{
  186. ID: portal.deterministicEventID(msg.ID, partName),
  187. Type: part.Type,
  188. Sender: intent.UserID,
  189. Timestamp: ts.UnixMilli(),
  190. Content: event.Content{
  191. Parsed: part.Content,
  192. Raw: part.Extra,
  193. },
  194. })
  195. dbMessages = append(dbMessages, database.Message{
  196. Channel: portal.Key,
  197. DiscordID: msg.ID,
  198. SenderID: msg.Author.ID,
  199. Timestamp: ts,
  200. AttachmentID: part.AttachmentID,
  201. })
  202. }
  203. }
  204. log.Info().Int("parts", len(evts)).Msg("Converted messages to backfill")
  205. resp, err := portal.MainIntent().BatchSend(portal.MXID, &mautrix.ReqBatchSend{
  206. BeeperNewMessages: true,
  207. Events: evts,
  208. })
  209. if err != nil {
  210. log.Err(err).Msg("Error sending backfill batch")
  211. return
  212. }
  213. for i, evtID := range resp.EventIDs {
  214. dbMessages[i].MXID = evtID
  215. }
  216. portal.bridge.DB.Message.MassInsert(portal.Key, dbMessages)
  217. log.Info().Msg("Inserted backfilled batch to database")
  218. }
  219. func (portal *Portal) deterministicEventID(messageID, partName string) id.EventID {
  220. data := fmt.Sprintf("%s/discord/%s/%s", portal.MXID, messageID, partName)
  221. sum := sha256.Sum256([]byte(data))
  222. return id.EventID(fmt.Sprintf("$%s:discord.com", base64.RawURLEncoding.EncodeToString(sum[:])))
  223. }
  224. // compareMessageIDs compares two Discord message IDs.
  225. //
  226. // If the first ID is lower, -1 is returned.
  227. // If the second ID is lower, 1 is returned.
  228. // If the IDs are equal, 0 is returned.
  229. func compareMessageIDs(id1, id2 string) int {
  230. if id1 == id2 {
  231. return 0
  232. }
  233. if len(id1) < len(id2) {
  234. return -1
  235. } else if len(id2) < len(id1) {
  236. return 1
  237. }
  238. if id1 < id2 {
  239. return -1
  240. }
  241. return 1
  242. }
  243. func shouldBackfill(latestBridgedIDStr, latestIDFromServerStr string) bool {
  244. return compareMessageIDs(latestBridgedIDStr, latestIDFromServerStr) == -1
  245. }
  246. type MessageSlice []*discordgo.Message
  247. var _ sort.Interface = (MessageSlice)(nil)
  248. func (a MessageSlice) Len() int {
  249. return len(a)
  250. }
  251. func (a MessageSlice) Swap(i, j int) {
  252. a[i], a[j] = a[j], a[i]
  253. }
  254. func (a MessageSlice) Less(i, j int) bool {
  255. return compareMessageIDs(a[i].ID, a[j].ID) == -1
  256. }