messagetracking.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327
  1. // mautrix-whatsapp - A Matrix-WhatsApp puppeting bridge.
  2. // Copyright (C) 2022 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. "context"
  19. "errors"
  20. "fmt"
  21. "sync"
  22. "time"
  23. log "maunium.net/go/maulogger/v2"
  24. "go.mau.fi/whatsmeow"
  25. "maunium.net/go/mautrix"
  26. "maunium.net/go/mautrix/bridge"
  27. "maunium.net/go/mautrix/event"
  28. "maunium.net/go/mautrix/id"
  29. )
  30. var (
  31. errUserNotConnected = errors.New("you are not connected to WhatsApp")
  32. errDifferentUser = errors.New("user is not the recipient of this private chat portal")
  33. errUserNotLoggedIn = errors.New("user is not logged in and chat has no relay bot")
  34. errMNoticeDisabled = errors.New("bridging m.notice messages is disabled")
  35. errUnexpectedParsedContentType = errors.New("unexpected parsed content type")
  36. errInvalidGeoURI = errors.New("invalid `geo:` URI in message")
  37. errUnknownMsgType = errors.New("unknown msgtype")
  38. errMediaDownloadFailed = errors.New("failed to download media")
  39. errMediaDecryptFailed = errors.New("failed to decrypt media")
  40. errMediaConvertFailed = errors.New("failed to convert media")
  41. errMediaWhatsAppUploadFailed = errors.New("failed to upload media to WhatsApp")
  42. errTargetNotFound = errors.New("target event not found")
  43. errReactionDatabaseNotFound = errors.New("reaction database entry not found")
  44. errReactionTargetNotFound = errors.New("reaction target message not found")
  45. errTargetIsFake = errors.New("target is a fake event")
  46. errTargetSentBySomeoneElse = errors.New("target is a fake event")
  47. errBroadcastReactionNotSupported = errors.New("reacting to status messages is not currently supported")
  48. errBroadcastSendDisabled = errors.New("sending status messages is disabled")
  49. errMessageDisconnected = &whatsmeow.DisconnectedError{Action: "message send"}
  50. errMessageRetryDisconnected = &whatsmeow.DisconnectedError{Action: "message send (retry)"}
  51. errMessageTakingLong = errors.New("bridging the message is taking longer than usual")
  52. errTimeoutBeforeHandling = errors.New("message timed out before handling was started")
  53. )
  54. func errorToStatusReason(err error) (reason event.MessageStatusReason, isCertain, canRetry, sendNotice bool) {
  55. switch {
  56. case errors.Is(err, whatsmeow.ErrBroadcastListUnsupported),
  57. errors.Is(err, errUnexpectedParsedContentType),
  58. errors.Is(err, errUnknownMsgType),
  59. errors.Is(err, errInvalidGeoURI),
  60. errors.Is(err, whatsmeow.ErrUnknownServer),
  61. errors.Is(err, whatsmeow.ErrRecipientADJID),
  62. errors.Is(err, errBroadcastReactionNotSupported),
  63. errors.Is(err, errBroadcastSendDisabled):
  64. return event.MessageStatusUnsupported, true, false, true
  65. case errors.Is(err, errTimeoutBeforeHandling):
  66. return event.MessageStatusTooOld, true, true, true
  67. case errors.Is(err, context.DeadlineExceeded), errors.Is(err, errMessageTakingLong):
  68. return event.MessageStatusTooOld, false, true, true
  69. case errors.Is(err, errTargetNotFound),
  70. errors.Is(err, errTargetIsFake),
  71. errors.Is(err, errReactionDatabaseNotFound),
  72. errors.Is(err, errReactionTargetNotFound),
  73. errors.Is(err, errTargetSentBySomeoneElse):
  74. return event.MessageStatusGenericError, true, false, false
  75. case errors.Is(err, whatsmeow.ErrNotConnected),
  76. errors.Is(err, errUserNotConnected):
  77. return event.MessageStatusGenericError, true, true, true
  78. case errors.Is(err, errUserNotLoggedIn),
  79. errors.Is(err, errDifferentUser):
  80. return event.MessageStatusGenericError, true, true, false
  81. case errors.Is(err, errMessageDisconnected),
  82. errors.Is(err, errMessageRetryDisconnected):
  83. return event.MessageStatusGenericError, false, true, true
  84. default:
  85. return event.MessageStatusGenericError, false, true, true
  86. }
  87. }
  88. func (portal *Portal) sendErrorMessage(evt *event.Event, err error, confirmed bool, editID id.EventID) id.EventID {
  89. if !portal.bridge.Config.Bridge.MessageErrorNotices {
  90. return ""
  91. }
  92. certainty := "may not have been"
  93. if confirmed {
  94. certainty = "was not"
  95. }
  96. msg := fmt.Sprintf("\u26a0 Your message %s bridged: %v", certainty, err)
  97. if errors.Is(err, errMessageTakingLong) {
  98. msg = "\u26a0 Bridging your message is taking longer than usual"
  99. }
  100. content := &event.MessageEventContent{
  101. MsgType: event.MsgNotice,
  102. Body: msg,
  103. }
  104. if editID != "" {
  105. content.SetEdit(editID)
  106. } else {
  107. content.SetReply(evt)
  108. }
  109. resp, err := portal.sendMainIntentMessage(content)
  110. if err != nil {
  111. portal.log.Warnfln("Failed to send bridging error message:", err)
  112. return ""
  113. }
  114. return resp.EventID
  115. }
  116. func (portal *Portal) sendStatusEvent(evtID, lastRetry id.EventID, err error) {
  117. if !portal.bridge.Config.Bridge.MessageStatusEvents {
  118. return
  119. }
  120. if lastRetry == evtID {
  121. lastRetry = ""
  122. }
  123. intent := portal.bridge.Bot
  124. if !portal.Encrypted {
  125. // Bridge bot isn't present in unencrypted DMs
  126. intent = portal.MainIntent()
  127. }
  128. content := event.BeeperMessageStatusEventContent{
  129. Network: portal.getBridgeInfoStateKey(),
  130. RelatesTo: event.RelatesTo{
  131. Type: event.RelReference,
  132. EventID: evtID,
  133. },
  134. Success: err == nil,
  135. LastRetry: lastRetry,
  136. }
  137. if !content.Success {
  138. reason, isCertain, canRetry, _ := errorToStatusReason(err)
  139. content.Reason = reason
  140. content.IsCertain = &isCertain
  141. content.CanRetry = &canRetry
  142. content.StillWorking = errors.Is(err, errMessageTakingLong)
  143. content.Error = err.Error()
  144. }
  145. _, err = intent.SendMessageEvent(portal.MXID, event.BeeperMessageStatus, &content)
  146. if err != nil {
  147. portal.log.Warnln("Failed to send message status event:", err)
  148. }
  149. }
  150. func (portal *Portal) sendDeliveryReceipt(eventID id.EventID) {
  151. if portal.bridge.Config.Bridge.DeliveryReceipts {
  152. err := portal.bridge.Bot.MarkRead(portal.MXID, eventID)
  153. if err != nil {
  154. portal.log.Debugfln("Failed to send delivery receipt for %s: %v", eventID, err)
  155. }
  156. }
  157. }
  158. func (portal *Portal) sendMessageMetrics(evt *event.Event, err error, part string, ms *metricSender) {
  159. var msgType string
  160. switch evt.Type {
  161. case event.EventMessage:
  162. msgType = "message"
  163. case event.EventReaction:
  164. msgType = "reaction"
  165. case event.EventRedaction:
  166. msgType = "redaction"
  167. default:
  168. msgType = "unknown event"
  169. }
  170. evtDescription := evt.ID.String()
  171. if evt.Type == event.EventRedaction {
  172. evtDescription += fmt.Sprintf(" of %s", evt.Redacts)
  173. }
  174. origEvtID := evt.ID
  175. if retryMeta := evt.Content.AsMessage().MessageSendRetry; retryMeta != nil {
  176. origEvtID = retryMeta.OriginalEventID
  177. }
  178. if err != nil {
  179. level := log.LevelError
  180. if part == "Ignoring" {
  181. level = log.LevelDebug
  182. }
  183. portal.log.Logfln(level, "%s %s %s from %s: %v", part, msgType, evtDescription, evt.Sender, err)
  184. reason, isCertain, _, sendNotice := errorToStatusReason(err)
  185. status := bridge.ReasonToCheckpointStatus(reason)
  186. if errors.Is(err, errMessageTakingLong) {
  187. status = bridge.MsgStatusWillRetry
  188. }
  189. portal.bridge.SendMessageCheckpoint(evt, bridge.MsgStepRemote, err, status, ms.getRetryNum())
  190. if sendNotice {
  191. ms.setNoticeID(portal.sendErrorMessage(evt, err, isCertain, ms.getNoticeID()))
  192. }
  193. portal.sendStatusEvent(origEvtID, evt.ID, err)
  194. } else {
  195. portal.log.Debugfln("Handled Matrix %s %s", msgType, evtDescription)
  196. portal.sendDeliveryReceipt(evt.ID)
  197. portal.bridge.SendMessageSuccessCheckpoint(evt, bridge.MsgStepRemote, ms.getRetryNum())
  198. portal.sendStatusEvent(origEvtID, evt.ID, nil)
  199. if prevNotice := ms.popNoticeID(); prevNotice != "" {
  200. _, _ = portal.MainIntent().RedactEvent(portal.MXID, prevNotice, mautrix.ReqRedact{
  201. Reason: "error resolved",
  202. })
  203. }
  204. }
  205. if ms != nil {
  206. portal.log.Debugfln("Timings for %s: %s", evt.ID, ms.timings.String())
  207. }
  208. }
  209. type messageTimings struct {
  210. initReceive time.Duration
  211. decrypt time.Duration
  212. implicitRR time.Duration
  213. portalQueue time.Duration
  214. totalReceive time.Duration
  215. preproc time.Duration
  216. convert time.Duration
  217. whatsmeow whatsmeow.MessageDebugTimings
  218. totalSend time.Duration
  219. }
  220. func niceRound(dur time.Duration) time.Duration {
  221. switch {
  222. case dur < time.Millisecond:
  223. return dur
  224. case dur < time.Second:
  225. return dur.Round(100 * time.Microsecond)
  226. default:
  227. return dur.Round(time.Millisecond)
  228. }
  229. }
  230. func (mt *messageTimings) String() string {
  231. mt.initReceive = niceRound(mt.initReceive)
  232. mt.decrypt = niceRound(mt.decrypt)
  233. mt.portalQueue = niceRound(mt.portalQueue)
  234. mt.totalReceive = niceRound(mt.totalReceive)
  235. mt.implicitRR = niceRound(mt.implicitRR)
  236. mt.preproc = niceRound(mt.preproc)
  237. mt.convert = niceRound(mt.convert)
  238. mt.whatsmeow.Queue = niceRound(mt.whatsmeow.Queue)
  239. mt.whatsmeow.Marshal = niceRound(mt.whatsmeow.Marshal)
  240. mt.whatsmeow.GetParticipants = niceRound(mt.whatsmeow.GetParticipants)
  241. mt.whatsmeow.GetDevices = niceRound(mt.whatsmeow.GetDevices)
  242. mt.whatsmeow.GroupEncrypt = niceRound(mt.whatsmeow.GroupEncrypt)
  243. mt.whatsmeow.PeerEncrypt = niceRound(mt.whatsmeow.PeerEncrypt)
  244. mt.whatsmeow.Send = niceRound(mt.whatsmeow.Send)
  245. mt.whatsmeow.Resp = niceRound(mt.whatsmeow.Resp)
  246. mt.whatsmeow.Retry = niceRound(mt.whatsmeow.Retry)
  247. mt.totalSend = niceRound(mt.totalSend)
  248. whatsmeowTimings := "N/A"
  249. if mt.totalSend > 0 {
  250. format := "queue: %[1]s, marshal: %[2]s, ske: %[3]s, pcp: %[4]s, dev: %[5]s, encrypt: %[6]s, send: %[7]s, resp: %[8]s"
  251. if mt.whatsmeow.GetParticipants == 0 && mt.whatsmeow.GroupEncrypt == 0 {
  252. format = "queue: %[1]s, marshal: %[2]s, dev: %[5]s, encrypt: %[6]s, send: %[7]s, resp: %[8]s"
  253. }
  254. if mt.whatsmeow.Retry > 0 {
  255. format += ", retry: %[9]s"
  256. }
  257. whatsmeowTimings = fmt.Sprintf(format, mt.whatsmeow.Queue, mt.whatsmeow.Marshal, mt.whatsmeow.GroupEncrypt, mt.whatsmeow.GetParticipants, mt.whatsmeow.GetDevices, mt.whatsmeow.PeerEncrypt, mt.whatsmeow.Send, mt.whatsmeow.Resp, mt.whatsmeow.Retry)
  258. }
  259. return fmt.Sprintf("BRIDGE: receive: %s, decrypt: %s, queue: %s, total hs->portal: %s, implicit rr: %s -- PORTAL: preprocess: %s, convert: %s, total send: %s -- WHATSMEOW: %s", mt.initReceive, mt.decrypt, mt.implicitRR, mt.portalQueue, mt.totalReceive, mt.preproc, mt.convert, mt.totalSend, whatsmeowTimings)
  260. }
  261. type metricSender struct {
  262. portal *Portal
  263. previousNotice id.EventID
  264. lock sync.Mutex
  265. completed bool
  266. retryNum int
  267. timings *messageTimings
  268. }
  269. func (ms *metricSender) getRetryNum() int {
  270. if ms != nil {
  271. return ms.retryNum
  272. }
  273. return 0
  274. }
  275. func (ms *metricSender) getNoticeID() id.EventID {
  276. if ms == nil {
  277. return ""
  278. }
  279. return ms.previousNotice
  280. }
  281. func (ms *metricSender) popNoticeID() id.EventID {
  282. if ms == nil {
  283. return ""
  284. }
  285. evtID := ms.previousNotice
  286. ms.previousNotice = ""
  287. return evtID
  288. }
  289. func (ms *metricSender) setNoticeID(evtID id.EventID) {
  290. if ms != nil && ms.previousNotice == "" {
  291. ms.previousNotice = evtID
  292. }
  293. }
  294. func (ms *metricSender) sendMessageMetrics(evt *event.Event, err error, part string, completed bool) {
  295. ms.lock.Lock()
  296. defer ms.lock.Unlock()
  297. if !completed && ms.completed {
  298. return
  299. }
  300. ms.portal.sendMessageMetrics(evt, err, part, ms)
  301. ms.retryNum++
  302. ms.completed = completed
  303. }