historysync.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594
  1. // mautrix-whatsapp - A Matrix-WhatsApp puppeting bridge.
  2. // Copyright (C) 2021 Tulir Asokan
  3. //
  4. // This program is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Affero General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // This program is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Affero General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Affero General Public License
  15. // along with this program. If not, see <https://www.gnu.org/licenses/>.
  16. package main
  17. import (
  18. "sort"
  19. "sync"
  20. "time"
  21. waProto "go.mau.fi/whatsmeow/binary/proto"
  22. "go.mau.fi/whatsmeow/types"
  23. "maunium.net/go/mautrix"
  24. "maunium.net/go/mautrix/appservice"
  25. "maunium.net/go/mautrix/event"
  26. "maunium.net/go/mautrix/id"
  27. "maunium.net/go/mautrix-whatsapp/database"
  28. )
  29. // region User history sync handling
  30. const (
  31. FastBackfillPortalCount = 20
  32. FastBackfillMessageCount = 20
  33. FastBackfillMessageCap = 30
  34. )
  35. type portalToBackfill struct {
  36. portal *Portal
  37. conv *waProto.Conversation
  38. msgs []*waProto.WebMessageInfo
  39. }
  40. type wrappedInfo struct {
  41. *types.MessageInfo
  42. Type database.MessageType
  43. Error database.MessageErrorType
  44. }
  45. type conversationList []*waProto.Conversation
  46. var _ sort.Interface = (conversationList)(nil)
  47. func (c conversationList) Len() int {
  48. return len(c)
  49. }
  50. func (c conversationList) Less(i, j int) bool {
  51. return getConversationTimestamp(c[i]) < getConversationTimestamp(c[j])
  52. }
  53. func (c conversationList) Swap(i, j int) {
  54. c[i], c[j] = c[j], c[i]
  55. }
  56. func (user *User) handleHistorySyncsLoop() {
  57. for evt := range user.historySyncs {
  58. go user.sendBridgeState(BridgeState{StateEvent: StateBackfilling})
  59. user.handleHistorySync(evt.Data)
  60. if len(user.historySyncs) == 0 && user.IsConnected() {
  61. go user.sendBridgeState(BridgeState{StateEvent: StateConnected})
  62. }
  63. }
  64. }
  65. func (user *User) handleHistorySync(evt *waProto.HistorySync) {
  66. if evt == nil || evt.SyncType == nil || evt.GetSyncType() == waProto.HistorySync_INITIAL_STATUS_V3 || evt.GetSyncType() == waProto.HistorySync_PUSH_NAME {
  67. return
  68. }
  69. user.log.Infofln("Handling history sync with type %s, %d conversations, chunk order %d, progress %d%%", evt.GetSyncType(), len(evt.GetConversations()), evt.GetChunkOrder(), evt.GetProgress())
  70. conversations := conversationList(evt.GetConversations())
  71. // We want to handle recent conversations first
  72. sort.Sort(sort.Reverse(conversations))
  73. portalsToBackfill := make(chan portalToBackfill, len(conversations))
  74. var backfillWait, fastBackfillWait sync.WaitGroup
  75. var fastBackfillWaitDoneOnce sync.Once
  76. // We have to add 1 to the waitgroup beforehand to make sure the wait in the goroutine doesn't finish
  77. // before we add the actual numbers.
  78. fastBackfillWait.Add(1)
  79. backfillWait.Add(1)
  80. go func() {
  81. // Wait for the fast parallelized backfill to complete, then start the slow backfill loop (see explanation below)
  82. fastBackfillWait.Wait()
  83. user.slowBackfillLoop(portalsToBackfill, backfillWait.Done)
  84. }()
  85. for i, conv := range conversations {
  86. // This will create portals and start backfilling for them.
  87. //
  88. // The first 20 (FastBackfillPortalCount) portals will be parallelized, where the portal is
  89. // created and recent messages are backfilled in parallel. Other portals will be created
  90. // synchronously (and this will only return when they're created).
  91. //
  92. // For said other portals, and older messages in the parallelized portals, backfilling also
  93. // happens synchronously: the portals and messages to backfill are added to the
  94. // portalsToBackfill channel, which is consumed one-by-one in the slowBackfillLoop method.
  95. // That loop is only started after the fast parallelized backfill is completed.
  96. user.handleHistorySyncConversation(i, conv, &fastBackfillWait, portalsToBackfill)
  97. if i == FastBackfillPortalCount {
  98. // There won't be any more portals going the fast backfill route, so remove the 1 item
  99. // that was added to the wait group at the beginning.
  100. fastBackfillWaitDoneOnce.Do(fastBackfillWait.Done)
  101. }
  102. }
  103. fastBackfillWaitDoneOnce.Do(fastBackfillWait.Done)
  104. // Wait for fast backfill to complete to make sure everything necessary is in the slow backfill queue,
  105. // then close the slow backfill queue and wait for the loop to finish handling the queue.
  106. fastBackfillWait.Wait()
  107. close(portalsToBackfill)
  108. backfillWait.Wait()
  109. user.log.Infofln("Finished handling history sync with type %s, %d conversations, chunk order %d, progress %d%%", evt.GetSyncType(), len(conversations), evt.GetChunkOrder(), evt.GetProgress())
  110. }
  111. func (user *User) slowBackfillLoop(ch chan portalToBackfill, done func()) {
  112. defer done()
  113. for ptb := range ch {
  114. if len(ptb.msgs) > 0 {
  115. user.log.Debugln("Bridging history sync payload for", ptb.portal.Key.JID)
  116. ptb.portal.backfill(user, ptb.msgs)
  117. } else {
  118. user.log.Debugfln("Not backfilling %s: no bridgeable messages found", ptb.portal.Key.JID)
  119. }
  120. if !ptb.conv.GetMarkedAsUnread() && ptb.conv.GetUnreadCount() == 0 {
  121. user.markSelfReadFull(ptb.portal)
  122. }
  123. }
  124. }
  125. func (user *User) handleHistorySyncConversation(index int, conv *waProto.Conversation, fastBackfillWait *sync.WaitGroup, portalsToBackfill chan portalToBackfill) {
  126. jid, err := types.ParseJID(conv.GetId())
  127. if err != nil {
  128. user.log.Warnfln("Failed to parse chat JID '%s' in history sync: %v", conv.GetId(), err)
  129. return
  130. }
  131. // Update the client store with basic chat settings.
  132. muteEnd := time.Unix(int64(conv.GetMuteEndTime()), 0)
  133. if muteEnd.After(time.Now()) {
  134. _ = user.Client.Store.ChatSettings.PutMutedUntil(jid, muteEnd)
  135. }
  136. if conv.GetArchived() {
  137. _ = user.Client.Store.ChatSettings.PutArchived(jid, true)
  138. }
  139. if conv.GetPinned() > 0 {
  140. _ = user.Client.Store.ChatSettings.PutPinned(jid, true)
  141. }
  142. portal := user.GetPortalByJID(jid)
  143. if conv.EphemeralExpiration != nil && portal.ExpirationTime != conv.GetEphemeralExpiration() {
  144. portal.ExpirationTime = conv.GetEphemeralExpiration()
  145. portal.Update()
  146. }
  147. // Check if portal is too old or doesn't contain anything we can bridge.
  148. if !user.shouldCreatePortalForHistorySync(conv, portal) {
  149. return
  150. }
  151. var msgs []*waProto.WebMessageInfo
  152. if user.bridge.Config.Bridge.HistorySync.Backfill {
  153. msgs = filterMessagesToBackfill(conv.GetMessages())
  154. }
  155. ptb := portalToBackfill{portal: portal, conv: conv, msgs: msgs}
  156. if len(portal.MXID) == 0 {
  157. // For the first few chats, do the portal creation and some backfilling in parallel to populate the chat list ASAP
  158. if index < FastBackfillPortalCount {
  159. fastBackfillWait.Add(1)
  160. go user.fastBackfillRoutine(ptb, fastBackfillWait.Done, portalsToBackfill)
  161. return
  162. }
  163. user.log.Debugln("Creating portal for", portal.Key.JID, "as part of history sync handling")
  164. err = portal.CreateMatrixRoom(user, getPartialInfoFromConversation(jid, conv), false)
  165. if err != nil {
  166. user.log.Warnfln("Failed to create room for %s during backfill: %v", portal.Key.JID, err)
  167. return
  168. }
  169. } else {
  170. portal.UpdateMatrixRoom(user, nil)
  171. }
  172. if !user.bridge.Config.Bridge.HistorySync.Backfill {
  173. user.log.Debugln("Backfill is disabled, not bridging history sync payload for", portal.Key.JID)
  174. } else {
  175. portalsToBackfill <- ptb
  176. }
  177. }
  178. func getConversationTimestamp(conv *waProto.Conversation) uint64 {
  179. convTs := conv.GetConversationTimestamp()
  180. if convTs == 0 && len(conv.GetMessages()) > 0 {
  181. convTs = conv.Messages[0].GetMessage().GetMessageTimestamp()
  182. }
  183. return convTs
  184. }
  185. func (user *User) shouldCreatePortalForHistorySync(conv *waProto.Conversation, portal *Portal) bool {
  186. maxAge := user.bridge.Config.Bridge.HistorySync.MaxAge
  187. minLastMsgToCreate := time.Now().Add(-time.Duration(maxAge) * time.Second)
  188. lastMsg := time.Unix(int64(getConversationTimestamp(conv)), 0)
  189. if len(portal.MXID) > 0 {
  190. user.log.Debugfln("Portal for %s already exists, ensuring user is invited", portal.Key.JID)
  191. portal.ensureUserInvited(user)
  192. // Portal exists, let backfill continue
  193. return true
  194. } else if !user.bridge.Config.Bridge.HistorySync.CreatePortals {
  195. user.log.Debugfln("Not creating portal for %s: creating rooms from history sync is disabled", portal.Key.JID)
  196. } else if !containsSupportedMessages(conv) {
  197. user.log.Debugfln("Not creating portal for %s: no interesting messages found", portal.Key.JID)
  198. } else if maxAge > 0 && !lastMsg.After(minLastMsgToCreate) {
  199. user.log.Debugfln("Not creating portal for %s: last message older than limit (%s)", portal.Key.JID, lastMsg)
  200. } else {
  201. // Portal doesn't exist, but should be created
  202. return true
  203. }
  204. // Portal shouldn't be created, reason logged above
  205. return false
  206. }
  207. func (user *User) fastBackfillRoutine(ptb portalToBackfill, done func(), slowBackfillChan chan portalToBackfill) {
  208. defer done()
  209. user.log.Debugln("Asynchronously creating portal for", ptb.portal.Key.JID, "as part of history sync handling")
  210. err := ptb.portal.CreateMatrixRoom(user, getPartialInfoFromConversation(ptb.portal.Key.JID, ptb.conv), false)
  211. if err != nil {
  212. user.log.Warnfln("Failed to create room for %s during backfill: %v", ptb.portal.Key.JID, err)
  213. return
  214. }
  215. if user.bridge.Config.Bridge.HistorySync.Backfill {
  216. if len(ptb.msgs) > FastBackfillMessageCap {
  217. user.log.Debugfln("Bridging first %d messages of history sync payload for %s (async)", FastBackfillMessageCount, ptb.portal.Key.JID)
  218. ptb.portal.backfill(user, ptb.msgs[:FastBackfillMessageCount])
  219. // Send the rest of the messages off to the slow backfill queue
  220. ptb.msgs = ptb.msgs[FastBackfillMessageCount:]
  221. slowBackfillChan <- ptb
  222. } else if len(ptb.msgs) > 0 {
  223. user.log.Debugfln("Bridging all messages (%d) of history sync payload for %s (async)", len(ptb.msgs), ptb.portal.Key.JID)
  224. ptb.portal.backfill(user, ptb.msgs)
  225. } else {
  226. user.log.Debugfln("Not backfilling %s: no bridgeable messages found", ptb.portal.Key.JID)
  227. }
  228. } else {
  229. user.log.Debugln("Backfill is disabled, not bridging history sync payload for", ptb.portal.Key.JID)
  230. }
  231. }
  232. func filterMessagesToBackfill(messages []*waProto.HistorySyncMsg) []*waProto.WebMessageInfo {
  233. filtered := make([]*waProto.WebMessageInfo, 0, len(messages))
  234. for _, msg := range messages {
  235. wmi := msg.GetMessage()
  236. msgType := getMessageType(wmi.GetMessage())
  237. if msgType == "unknown" || msgType == "ignore" || msgType == "unknown_protocol" {
  238. continue
  239. } else {
  240. filtered = append(filtered, wmi)
  241. }
  242. }
  243. return filtered
  244. }
  245. func containsSupportedMessages(conv *waProto.Conversation) bool {
  246. for _, msg := range conv.GetMessages() {
  247. if containsSupportedMessage(msg.GetMessage().GetMessage()) {
  248. return true
  249. }
  250. }
  251. return false
  252. }
  253. func getPartialInfoFromConversation(jid types.JID, conv *waProto.Conversation) *types.GroupInfo {
  254. // TODO broadcast list info?
  255. if jid.Server != types.GroupServer {
  256. return nil
  257. }
  258. participants := make([]types.GroupParticipant, len(conv.GetParticipant()))
  259. for i, pcp := range conv.GetParticipant() {
  260. participantJID, _ := types.ParseJID(pcp.GetUserJid())
  261. participants[i] = types.GroupParticipant{
  262. JID: participantJID,
  263. IsAdmin: pcp.GetRank() == waProto.GroupParticipant_ADMIN,
  264. IsSuperAdmin: pcp.GetRank() == waProto.GroupParticipant_SUPERADMIN,
  265. }
  266. }
  267. return &types.GroupInfo{
  268. JID: jid,
  269. GroupName: types.GroupName{Name: conv.GetName()},
  270. Participants: participants,
  271. }
  272. }
  273. // endregion
  274. // region Portal backfilling
  275. var (
  276. PortalCreationDummyEvent = event.Type{Type: "fi.mau.dummy.portal_created", Class: event.MessageEventType}
  277. BackfillEndDummyEvent = event.Type{Type: "fi.mau.dummy.backfill_end", Class: event.MessageEventType}
  278. PreBackfillDummyEvent = event.Type{Type: "fi.mau.dummy.pre_backfill", Class: event.MessageEventType}
  279. )
  280. func (portal *Portal) backfill(source *User, messages []*waProto.WebMessageInfo) {
  281. portal.backfillLock.Lock()
  282. defer portal.backfillLock.Unlock()
  283. var historyBatch, newBatch mautrix.ReqBatchSend
  284. var historyBatchInfos, newBatchInfos []*wrappedInfo
  285. firstMsgTimestamp := time.Unix(int64(messages[len(messages)-1].GetMessageTimestamp()), 0)
  286. historyBatch.StateEventsAtStart = make([]*event.Event, 0)
  287. newBatch.StateEventsAtStart = make([]*event.Event, 0)
  288. addedMembers := make(map[id.UserID]*event.MemberEventContent)
  289. addMember := func(puppet *Puppet) {
  290. if _, alreadyAdded := addedMembers[puppet.MXID]; alreadyAdded {
  291. return
  292. }
  293. mxid := puppet.MXID.String()
  294. content := event.MemberEventContent{
  295. Membership: event.MembershipJoin,
  296. Displayname: puppet.Displayname,
  297. AvatarURL: puppet.AvatarURL.CUString(),
  298. }
  299. inviteContent := content
  300. inviteContent.Membership = event.MembershipInvite
  301. historyBatch.StateEventsAtStart = append(historyBatch.StateEventsAtStart, &event.Event{
  302. Type: event.StateMember,
  303. Sender: portal.MainIntent().UserID,
  304. StateKey: &mxid,
  305. Timestamp: firstMsgTimestamp.UnixMilli(),
  306. Content: event.Content{Parsed: &inviteContent},
  307. }, &event.Event{
  308. Type: event.StateMember,
  309. Sender: puppet.MXID,
  310. StateKey: &mxid,
  311. Timestamp: firstMsgTimestamp.UnixMilli(),
  312. Content: event.Content{Parsed: &content},
  313. })
  314. addedMembers[puppet.MXID] = &content
  315. }
  316. firstMessage := portal.bridge.DB.Message.GetFirstInChat(portal.Key)
  317. lastMessage := portal.bridge.DB.Message.GetLastInChat(portal.Key)
  318. var historyMaxTs, newMinTs time.Time
  319. if portal.FirstEventID != "" || portal.NextBatchID != "" {
  320. historyBatch.PrevEventID = portal.FirstEventID
  321. historyBatch.BatchID = portal.NextBatchID
  322. if firstMessage == nil && lastMessage == nil {
  323. historyMaxTs = time.Now()
  324. } else {
  325. historyMaxTs = firstMessage.Timestamp
  326. }
  327. }
  328. if lastMessage != nil {
  329. newBatch.PrevEventID = lastMessage.MXID
  330. newMinTs = lastMessage.Timestamp
  331. }
  332. portal.log.Infofln("Processing history sync with %d messages", len(messages))
  333. // The messages are ordered newest to oldest, so iterate them in reverse order.
  334. for i := len(messages) - 1; i >= 0; i-- {
  335. webMsg := messages[i]
  336. msgType := getMessageType(webMsg.GetMessage())
  337. if msgType == "unknown" || msgType == "ignore" || msgType == "unknown_protocol" {
  338. if msgType != "ignore" {
  339. portal.log.Debugfln("Skipping message %s with unknown type in backfill", webMsg.GetKey().GetId())
  340. }
  341. continue
  342. }
  343. info := portal.parseWebMessageInfo(source, webMsg)
  344. if info == nil {
  345. continue
  346. }
  347. var batch *mautrix.ReqBatchSend
  348. var infos *[]*wrappedInfo
  349. if !historyMaxTs.IsZero() && info.Timestamp.Before(historyMaxTs) {
  350. batch, infos = &historyBatch, &historyBatchInfos
  351. } else if !newMinTs.IsZero() && info.Timestamp.After(newMinTs) {
  352. batch, infos = &newBatch, &newBatchInfos
  353. } else {
  354. continue
  355. }
  356. if webMsg.GetPushName() != "" && webMsg.GetPushName() != "-" {
  357. existingContact, _ := source.Client.Store.Contacts.GetContact(info.Sender)
  358. if !existingContact.Found || existingContact.PushName == "" {
  359. changed, _, err := source.Client.Store.Contacts.PutPushName(info.Sender, webMsg.GetPushName())
  360. if err != nil {
  361. source.log.Errorfln("Failed to save push name of %s from historical message in device store: %v", info.Sender, err)
  362. } else if changed {
  363. source.log.Debugfln("Got push name %s for %s from historical message", webMsg.GetPushName(), info.Sender)
  364. }
  365. }
  366. }
  367. puppet := portal.getMessagePuppet(source, info)
  368. intent := puppet.IntentFor(portal)
  369. if intent.IsCustomPuppet && !portal.bridge.Config.CanDoublePuppetBackfill(puppet.CustomMXID) {
  370. intent = puppet.DefaultIntent()
  371. }
  372. converted := portal.convertMessage(intent, source, info, webMsg.GetMessage())
  373. if converted == nil {
  374. portal.log.Debugfln("Skipping unsupported message %s in backfill", info.ID)
  375. continue
  376. }
  377. if !intent.IsCustomPuppet && !portal.bridge.StateStore.IsInRoom(portal.MXID, puppet.MXID) {
  378. addMember(puppet)
  379. }
  380. // TODO this won't work for history
  381. if len(converted.ReplyTo) > 0 {
  382. portal.SetReply(converted.Content, converted.ReplyTo)
  383. }
  384. err := portal.appendBatchEvents(converted, info, &batch.Events, infos)
  385. if err != nil {
  386. portal.log.Errorfln("Error handling message %s during backfill: %v", info.ID, err)
  387. }
  388. }
  389. if (len(historyBatch.Events) > 0 && len(historyBatch.BatchID) == 0) || len(newBatch.Events) > 0 {
  390. portal.log.Debugln("Sending a dummy event to avoid forward extremity errors with backfill")
  391. _, err := portal.MainIntent().SendMessageEvent(portal.MXID, PreBackfillDummyEvent, struct{}{})
  392. if err != nil {
  393. portal.log.Warnln("Error sending pre-backfill dummy event:", err)
  394. }
  395. }
  396. if len(historyBatch.Events) > 0 && len(historyBatch.PrevEventID) > 0 {
  397. portal.log.Infofln("Sending %d historical messages...", len(historyBatch.Events))
  398. historyResp, err := portal.MainIntent().BatchSend(portal.MXID, &historyBatch)
  399. if err != nil {
  400. portal.log.Errorln("Error sending batch of historical messages:", err)
  401. } else {
  402. portal.finishBatch(historyResp.EventIDs, historyBatchInfos)
  403. portal.NextBatchID = historyResp.NextBatchID
  404. portal.Update()
  405. // If batchID is non-empty, it means this is backfilling very old messages, and we don't need a post-backfill dummy.
  406. if historyBatch.BatchID == "" {
  407. portal.sendPostBackfillDummy(time.UnixMilli(historyBatch.Events[len(historyBatch.Events)-1].Timestamp))
  408. }
  409. }
  410. }
  411. if len(newBatch.Events) > 0 && len(newBatch.PrevEventID) > 0 {
  412. portal.log.Infofln("Sending %d new messages...", len(newBatch.Events))
  413. newResp, err := portal.MainIntent().BatchSend(portal.MXID, &newBatch)
  414. if err != nil {
  415. portal.log.Errorln("Error sending batch of new messages:", err)
  416. } else {
  417. portal.finishBatch(newResp.EventIDs, newBatchInfos)
  418. portal.sendPostBackfillDummy(time.UnixMilli(newBatch.Events[len(newBatch.Events)-1].Timestamp))
  419. }
  420. }
  421. }
  422. func (portal *Portal) parseWebMessageInfo(source *User, webMsg *waProto.WebMessageInfo) *types.MessageInfo {
  423. info := types.MessageInfo{
  424. MessageSource: types.MessageSource{
  425. Chat: portal.Key.JID,
  426. IsFromMe: webMsg.GetKey().GetFromMe(),
  427. IsGroup: portal.Key.JID.Server == types.GroupServer,
  428. },
  429. ID: webMsg.GetKey().GetId(),
  430. PushName: webMsg.GetPushName(),
  431. Timestamp: time.Unix(int64(webMsg.GetMessageTimestamp()), 0),
  432. }
  433. var err error
  434. if info.IsFromMe {
  435. info.Sender = source.JID.ToNonAD()
  436. } else if portal.IsPrivateChat() {
  437. info.Sender = portal.Key.JID
  438. } else if webMsg.GetParticipant() != "" {
  439. info.Sender, err = types.ParseJID(webMsg.GetParticipant())
  440. } else if webMsg.GetKey().GetParticipant() != "" {
  441. info.Sender, err = types.ParseJID(webMsg.GetKey().GetParticipant())
  442. }
  443. if info.Sender.IsEmpty() {
  444. portal.log.Warnfln("Failed to get sender of message %s (parse error: %v)", info.ID, err)
  445. return nil
  446. }
  447. return &info
  448. }
  449. func (portal *Portal) appendBatchEvents(converted *ConvertedMessage, info *types.MessageInfo, eventsArray *[]*event.Event, infoArray *[]*wrappedInfo) error {
  450. mainEvt, err := portal.wrapBatchEvent(info, converted.Intent, converted.Type, converted.Content, converted.Extra)
  451. if err != nil {
  452. return err
  453. }
  454. if converted.Caption != nil {
  455. captionEvt, err := portal.wrapBatchEvent(info, converted.Intent, converted.Type, converted.Caption, nil)
  456. if err != nil {
  457. return err
  458. }
  459. *eventsArray = append(*eventsArray, mainEvt, captionEvt)
  460. *infoArray = append(*infoArray, &wrappedInfo{info, database.MsgNormal, converted.Error}, nil)
  461. } else {
  462. *eventsArray = append(*eventsArray, mainEvt)
  463. *infoArray = append(*infoArray, &wrappedInfo{info, database.MsgNormal, converted.Error})
  464. }
  465. if converted.MultiEvent != nil {
  466. for _, subEvtContent := range converted.MultiEvent {
  467. subEvt, err := portal.wrapBatchEvent(info, converted.Intent, converted.Type, subEvtContent, nil)
  468. if err != nil {
  469. return err
  470. }
  471. *eventsArray = append(*eventsArray, subEvt)
  472. *infoArray = append(*infoArray, nil)
  473. }
  474. }
  475. return nil
  476. }
  477. const backfillIDField = "fi.mau.whatsapp.backfill_msg_id"
  478. func (portal *Portal) wrapBatchEvent(info *types.MessageInfo, intent *appservice.IntentAPI, eventType event.Type, content *event.MessageEventContent, extraContent map[string]interface{}) (*event.Event, error) {
  479. if extraContent == nil {
  480. extraContent = map[string]interface{}{}
  481. }
  482. extraContent[backfillIDField] = info.ID
  483. if intent.IsCustomPuppet {
  484. extraContent[doublePuppetKey] = doublePuppetValue
  485. }
  486. wrappedContent := event.Content{
  487. Parsed: content,
  488. Raw: extraContent,
  489. }
  490. newEventType, err := portal.encrypt(&wrappedContent, eventType)
  491. if err != nil {
  492. return nil, err
  493. }
  494. return &event.Event{
  495. Sender: intent.UserID,
  496. Type: newEventType,
  497. Timestamp: info.Timestamp.UnixMilli(),
  498. Content: wrappedContent,
  499. }, nil
  500. }
  501. func (portal *Portal) finishBatch(eventIDs []id.EventID, infos []*wrappedInfo) {
  502. if len(eventIDs) != len(infos) {
  503. portal.log.Errorfln("Length of event IDs (%d) and message infos (%d) doesn't match! Using slow path for mapping event IDs", len(eventIDs), len(infos))
  504. infoMap := make(map[types.MessageID]*wrappedInfo, len(infos))
  505. for _, info := range infos {
  506. infoMap[info.ID] = info
  507. }
  508. for _, eventID := range eventIDs {
  509. if evt, err := portal.MainIntent().GetEvent(portal.MXID, eventID); err != nil {
  510. portal.log.Warnfln("Failed to get event %s to register it in the database: %v", eventID, err)
  511. } else if msgID, ok := evt.Content.Raw[backfillIDField].(string); !ok {
  512. portal.log.Warnfln("Event %s doesn't include the WhatsApp message ID", eventID)
  513. } else if info, ok := infoMap[types.MessageID(msgID)]; !ok {
  514. portal.log.Warnfln("Didn't find info of message %s (event %s) to register it in the database", msgID, eventID)
  515. } else {
  516. portal.markHandled(nil, info.MessageInfo, eventID, true, false, info.Type, info.Error)
  517. }
  518. }
  519. } else {
  520. for i := 0; i < len(infos); i++ {
  521. if infos[i] != nil {
  522. portal.markHandled(nil, infos[i].MessageInfo, eventIDs[i], true, false, infos[i].Type, infos[i].Error)
  523. }
  524. }
  525. portal.log.Infofln("Successfully sent %d events", len(eventIDs))
  526. }
  527. }
  528. func (portal *Portal) sendPostBackfillDummy(lastTimestamp time.Time) {
  529. resp, err := portal.MainIntent().SendMessageEvent(portal.MXID, BackfillEndDummyEvent, struct{}{})
  530. if err != nil {
  531. portal.log.Errorln("Error sending post-backfill dummy event:", err)
  532. return
  533. }
  534. msg := portal.bridge.DB.Message.New()
  535. msg.Chat = portal.Key
  536. msg.MXID = resp.EventID
  537. msg.JID = types.MessageID(resp.EventID)
  538. msg.Timestamp = lastTimestamp.Add(1 * time.Second)
  539. msg.Sent = true
  540. msg.Insert()
  541. }
  542. // endregion