historysync.go 22 KB

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