historysync.go 22 KB

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