From 6f2a51410f4e5a3e5944eda316abdce849ab3716 Mon Sep 17 00:00:00 2001 From: Tulir Asokan Date: Tue, 21 May 2019 23:44:14 +0300 Subject: [PATCH] Make message handling more synchronous and fill history on portal create --- go.mod | 4 +- go.sum | 2 + portal.go | 114 +++++++++++++++++++++++++++++------ user.go | 68 +++++++++++---------- whatsapp-ext/chat.go | 7 ++- whatsapp-ext/cmd.go | 7 ++- whatsapp-ext/conn.go | 7 ++- whatsapp-ext/jsonmessage.go | 7 ++- whatsapp-ext/msginfo.go | 7 ++- whatsapp-ext/presence.go | 7 ++- whatsapp-ext/props.go | 7 ++- whatsapp-ext/protomessage.go | 9 ++- whatsapp-ext/stream.go | 7 ++- whatsapp-ext/whatsapp.go | 9 +++ 14 files changed, 198 insertions(+), 64 deletions(-) diff --git a/go.mod b/go.mod index a1c9244..8f14757 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module maunium.net/go/mautrix-whatsapp go 1.12 require ( - github.com/Rhymen/go-whatsapp v0.0.2-0.20190520154255-e9add85056dd + github.com/Rhymen/go-whatsapp v0.0.2-0.20190521203213-17d96f9c79ea github.com/golang/protobuf v1.3.1 // indirect github.com/gorilla/mux v1.7.2 // indirect github.com/lib/pq v1.1.1 @@ -21,4 +21,4 @@ require ( replace gopkg.in/russross/blackfriday.v2 => github.com/russross/blackfriday/v2 v2.0.1 -replace github.com/Rhymen/go-whatsapp => github.com/tulir/go-whatsapp v0.0.2-0.20190520154255-e9add85056dd +replace github.com/Rhymen/go-whatsapp => github.com/tulir/go-whatsapp v0.0.2-0.20190521203213-17d96f9c79ea diff --git a/go.sum b/go.sum index 2357705..a450b63 100644 --- a/go.sum +++ b/go.sum @@ -76,6 +76,8 @@ github.com/skip2/go-qrcode v0.0.0-20190110000554-dc11ecdae0a9 h1:lpEzuenPuO1XNTe github.com/skip2/go-qrcode v0.0.0-20190110000554-dc11ecdae0a9/go.mod h1:PLPIyL7ikehBD1OAjmKKiOEhbvWyHGaNDjquXMcYABo= github.com/tulir/go-whatsapp v0.0.2-0.20190520154255-e9add85056dd h1:NN9d4MwNnEAT6X9WO49W3Q5YmydKx6y5xrPCnsjEuFM= github.com/tulir/go-whatsapp v0.0.2-0.20190520154255-e9add85056dd/go.mod h1:qf/2PQi82Okxw/igghu/oMGzTeUYuKBq1JNo3tdQyNg= +github.com/tulir/go-whatsapp v0.0.2-0.20190521203213-17d96f9c79ea h1:cvrUAxWhx461T2E/vEMGKHy6zuA8/ml7NDYtYH/lx9A= +github.com/tulir/go-whatsapp v0.0.2-0.20190521203213-17d96f9c79ea/go.mod h1:qf/2PQi82Okxw/igghu/oMGzTeUYuKBq1JNo3tdQyNg= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= golang.org/x/crypto v0.0.0-20190131182504-b8fe1690c613/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190222235706-ffb98f73852f h1:qWFY9ZxP3tfI37wYIs/MnIAqK0vlXp1xnYEa5HxFSSY= diff --git a/portal.go b/portal.go index 3d0ea34..c25a3ed 100644 --- a/portal.go +++ b/portal.go @@ -102,18 +102,27 @@ func (bridge *Bridge) GetAllPortals() []*Portal { } func (bridge *Bridge) NewPortal(dbPortal *database.Portal) *Portal { - return &Portal{ + portal := &Portal{ Portal: dbPortal, bridge: bridge, log: bridge.Log.Sub(fmt.Sprintf("Portal/%s", dbPortal.Key)), messageLocks: make(map[types.WhatsAppMessageID]sync.Mutex), recentlyHandled: [recentlyHandledLength]types.WhatsAppMessageID{}, + + messages: make(chan PortalMessage, 128), } + go portal.handleMessageLoop() + return portal } const recentlyHandledLength = 100 +type PortalMessage struct { + source *User + data interface{} +} + type Portal struct { *database.Portal @@ -128,9 +137,47 @@ type Portal struct { recentlyHandledLock sync.Mutex recentlyHandledIndex uint8 + lastMessageTs uint64 + + messages chan PortalMessage + isPrivate *bool } +func (portal *Portal) handleMessageLoop() { + for msg := range portal.messages { + if len(portal.MXID) == 0 { + _, isRevocation := msg.data.(whatsappExt.MessageRevocation) + if isRevocation { + continue + } + err := portal.CreateMatrixRoom(msg.source) + if err != nil { + portal.log.Errorln("Failed to create portal room:", err) + return + } + } + portal.handleMessage(msg) + } +} + +func (portal *Portal) handleMessage(msg PortalMessage) { + switch data := msg.data.(type) { + case whatsapp.TextMessage: + portal.HandleTextMessage(msg.source, data) + case whatsapp.ImageMessage: + portal.HandleMediaMessage(msg.source, data.Download, data.Thumbnail, data.Info, data.Type, data.Caption) + case whatsapp.VideoMessage: + portal.HandleMediaMessage(msg.source, data.Download, data.Thumbnail, data.Info, data.Type, data.Caption) + case whatsapp.AudioMessage: + portal.HandleMediaMessage(msg.source, data.Download, nil, data.Info, data.Type, "") + case whatsapp.DocumentMessage: + portal.HandleMediaMessage(msg.source, data.Download, data.Thumbnail, data.Info, data.Type, data.Title) + case whatsappExt.MessageRevocation: + portal.HandleMessageRevoke(msg.source, data) + } +} + func (portal *Portal) getMessageLock(messageID types.WhatsAppMessageID) sync.Mutex { portal.messageLocksLock.Lock() defer portal.messageLocksLock.Unlock() @@ -194,16 +241,17 @@ func (portal *Portal) markHandled(source *User, message *waProto.WebMessageInfo, portal.recentlyHandled[index] = msg.JID } -func (portal *Portal) startHandling(id types.WhatsAppMessageID) (*sync.Mutex, bool) { - if portal.isRecentlyHandled(id) { +func (portal *Portal) startHandling(info whatsapp.MessageInfo) (*sync.Mutex, bool) { + if portal.lastMessageTs > info.Timestamp+1 || portal.isRecentlyHandled(info.Id) { return nil, false } - lock := portal.getMessageLock(id) + lock := portal.getMessageLock(info.Id) lock.Lock() - if portal.isDuplicate(id) { + if portal.isDuplicate(info.Id) { lock.Unlock() return nil, false } + portal.lastMessageTs = info.Timestamp return &lock, true } @@ -476,6 +524,27 @@ func (portal *Portal) RestrictMetadataChanges(restrict bool) { } } +func (portal *Portal) FillHistory(user *User) error { + resp, err := user.Conn.LoadMessages(portal.Key.JID, "", 50) + if err != nil { + return err + } + messages, ok := resp.Content.([]interface{}) + if !ok { + return fmt.Errorf("history response not list") + } + for _, rawMessage := range messages { + message, ok := rawMessage.(*waProto.WebMessageInfo) + if !ok { + portal.log.Warnln("Unexpected non-WebMessageInfo item in history response:", rawMessage) + continue + } + fmt.Println("Filling history", message.GetKey(), message.GetMessageTimestamp()) + portal.handleMessage(PortalMessage{user, whatsapp.ParseProtoMessage(message)}) + } + return nil +} + func (portal *Portal) CreateMatrixRoom(user *User) error { portal.roomCreateLock.Lock() defer portal.roomCreateLock.Unlock() @@ -523,6 +592,10 @@ func (portal *Portal) CreateMatrixRoom(user *User) error { } portal.MXID = resp.RoomID portal.Update() + err = portal.FillHistory(user) + if err != nil { + portal.log.Errorln("Failed to fill history:", err) + } return nil } @@ -605,18 +678,16 @@ func (portal *Portal) HandleMessageRevoke(user *User, message whatsappExt.Messag } func (portal *Portal) HandleTextMessage(source *User, message whatsapp.TextMessage) { - lock, ok := portal.startHandling(message.Info.Id) + if len(portal.MXID) == 0 { + return + } + + lock, ok := portal.startHandling(message.Info) if !ok { return } defer lock.Unlock() - err := portal.CreateMatrixRoom(source) - if err != nil { - portal.log.Errorln("Failed to create portal room:", err) - return - } - intent := portal.GetMessageIntent(source, message.Info) if intent == nil { return @@ -640,18 +711,16 @@ func (portal *Portal) HandleTextMessage(source *User, message whatsapp.TextMessa } func (portal *Portal) HandleMediaMessage(source *User, download func() ([]byte, error), thumbnail []byte, info whatsapp.MessageInfo, mimeType, caption string) { - lock, ok := portal.startHandling(info.Id) + if len(portal.MXID) == 0 { + return + } + + lock, ok := portal.startHandling(info) if !ok { return } defer lock.Unlock() - err := portal.CreateMatrixRoom(source) - if err != nil { - portal.log.Errorln("Failed to create portal room:", err) - return - } - intent := portal.GetMessageIntent(source, info) if intent == nil { return @@ -1018,6 +1087,9 @@ func (portal *Portal) Cleanup(puppetsOnly bool) { return } for member, _ := range members.Joined { + if member == intent.UserID { + continue + } puppet := portal.bridge.GetPuppetByMXID(member) if puppet != nil { _, err = puppet.Intent().LeaveRoom(portal.MXID) @@ -1031,6 +1103,10 @@ func (portal *Portal) Cleanup(puppetsOnly bool) { } } } + _, err = intent.LeaveRoom(portal.MXID) + if err != nil { + portal.log.Errorln("Error leaving with main intent while cleaning up portal:", err) + } } func (portal *Portal) HandleMatrixLeave(sender *User) { diff --git a/user.go b/user.go index 8b8ed5a..c02b23d 100644 --- a/user.go +++ b/user.go @@ -25,10 +25,11 @@ import ( "github.com/skip2/go-qrcode" log "maunium.net/go/maulogger/v2" - "github.com/Rhymen/go-whatsapp" "maunium.net/go/mautrix" "maunium.net/go/mautrix/format" + "github.com/Rhymen/go-whatsapp" + "maunium.net/go/mautrix-whatsapp/database" "maunium.net/go/mautrix-whatsapp/types" "maunium.net/go/mautrix-whatsapp/whatsapp-ext" @@ -306,6 +307,10 @@ func (user *User) HandleError(err error) { _, _ = user.bridge.Bot.SendMessageEvent(user.ManagementRoom, mautrix.EventMessage, content) } +func (user *User) ShouldCallSynchronously() bool { + return true +} + func (user *User) HandleJSONParseError(err error) { user.log.Errorln("WhatsApp JSON parse error:", err) } @@ -319,33 +324,27 @@ func (user *User) GetPortalByJID(jid types.WhatsAppID) *Portal { } func (user *User) HandleTextMessage(message whatsapp.TextMessage) { - portal := user.GetPortalByJID(message.Info.RemoteJid) - portal.HandleTextMessage(user, message) + user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandleImageMessage(message whatsapp.ImageMessage) { - portal := user.GetPortalByJID(message.Info.RemoteJid) - portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Caption) + user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandleVideoMessage(message whatsapp.VideoMessage) { - portal := user.GetPortalByJID(message.Info.RemoteJid) - portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Caption) + user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandleAudioMessage(message whatsapp.AudioMessage) { - portal := user.GetPortalByJID(message.Info.RemoteJid) - portal.HandleMediaMessage(user, message.Download, nil, message.Info, message.Type, "") + user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandleDocumentMessage(message whatsapp.DocumentMessage) { - portal := user.GetPortalByJID(message.Info.RemoteJid) - portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Title) + user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandleMessageRevoke(message whatsappExt.MessageRevocation) { - portal := user.GetPortalByJID(message.RemoteJid) - portal.HandleMessageRevoke(user, message) + user.GetPortalByJID(message.RemoteJid).messages <- PortalMessage{user, message} } func (user *User) HandlePresence(info whatsappExt.Presence) { @@ -382,17 +381,20 @@ func (user *User) HandleMsgInfo(info whatsappExt.MsgInfo) { return } - intent := user.bridge.GetPuppetByJID(info.SenderJID).Intent() - for _, id := range info.IDs { - msg := user.bridge.DB.Message.GetByJID(portal.Key, id) - if msg == nil { - continue + go func() { + intent := user.bridge.GetPuppetByJID(info.SenderJID).Intent() + for _, id := range info.IDs { + msg := user.bridge.DB.Message.GetByJID(portal.Key, id) + if msg == nil { + continue + } + + err := intent.MarkRead(portal.MXID, msg.MXID) + if err != nil { + user.log.Warnln("Failed to mark message %s as read by %s: %v", msg.MXID, info.SenderJID, err) + } } - err := intent.MarkRead(portal.MXID, msg.MXID) - if err != nil { - user.log.Warnln("Failed to mark message %s as read by %s: %v", msg.MXID, info.SenderJID, err) - } - } + }() } } @@ -401,10 +403,10 @@ func (user *User) HandleCommand(cmd whatsappExt.Command) { case whatsappExt.CommandPicture: if strings.HasSuffix(cmd.JID, whatsappExt.NewUserSuffix) { puppet := user.bridge.GetPuppetByJID(cmd.JID) - puppet.UpdateAvatar(user, cmd.ProfilePicInfo) + go puppet.UpdateAvatar(user, cmd.ProfilePicInfo) } else { portal := user.GetPortalByJID(cmd.JID) - portal.UpdateAvatar(user, cmd.ProfilePicInfo) + go portal.UpdateAvatar(user, cmd.ProfilePicInfo) } case whatsappExt.CommandDisconnect: var msg string @@ -416,7 +418,7 @@ func (user *User) HandleCommand(cmd whatsappExt.Command) { msg = fmt.Sprintf("\u26a0 Your WhatsApp connection was closed by the server (reason code: %s).\n\n"+ "Use the `reconnect` command to reconnect.", cmd.Kind) } - _, _ = user.bridge.Bot.SendMessageEvent(user.ManagementRoom, mautrix.EventMessage, format.RenderMarkdown(msg)) + go user.bridge.Bot.SendMessageEvent(user.ManagementRoom, mautrix.EventMessage, format.RenderMarkdown(msg)) } } @@ -432,19 +434,19 @@ func (user *User) HandleChatUpdate(cmd whatsappExt.ChatUpdate) { switch cmd.Data.Action { case whatsappExt.ChatActionNameChange: - portal.UpdateName(cmd.Data.NameChange.Name, cmd.Data.SenderJID) + go portal.UpdateName(cmd.Data.NameChange.Name, cmd.Data.SenderJID) case whatsappExt.ChatActionAddTopic: - portal.UpdateTopic(cmd.Data.AddTopic.Topic, cmd.Data.SenderJID) + go portal.UpdateTopic(cmd.Data.AddTopic.Topic, cmd.Data.SenderJID) case whatsappExt.ChatActionRemoveTopic: - portal.UpdateTopic("", cmd.Data.SenderJID) + go portal.UpdateTopic("", cmd.Data.SenderJID) case whatsappExt.ChatActionPromote: - portal.ChangeAdminStatus(cmd.Data.PermissionChange.JIDs, true) + go portal.ChangeAdminStatus(cmd.Data.PermissionChange.JIDs, true) case whatsappExt.ChatActionDemote: - portal.ChangeAdminStatus(cmd.Data.PermissionChange.JIDs, false) + go portal.ChangeAdminStatus(cmd.Data.PermissionChange.JIDs, false) case whatsappExt.ChatActionAnnounce: - portal.RestrictMessageSending(cmd.Data.Announce) + go portal.RestrictMessageSending(cmd.Data.Announce) case whatsappExt.ChatActionRestrict: - portal.RestrictMetadataChanges(cmd.Data.Restrict) + go portal.RestrictMetadataChanges(cmd.Data.Restrict) } } diff --git a/whatsapp-ext/chat.go b/whatsapp-ext/chat.go index b601eb4..d8c96f3 100644 --- a/whatsapp-ext/chat.go +++ b/whatsapp-ext/chat.go @@ -142,6 +142,11 @@ func (ext *ExtendedConn) handleMessageChatUpdate(message []byte) { if !ok { continue } - go chatUpdateHandler.HandleChatUpdate(event) + + if ext.shouldCallSynchronously(chatUpdateHandler) { + chatUpdateHandler.HandleChatUpdate(event) + } else { + go chatUpdateHandler.HandleChatUpdate(event) + } } } diff --git a/whatsapp-ext/cmd.go b/whatsapp-ext/cmd.go index debc31b..185a1c5 100644 --- a/whatsapp-ext/cmd.go +++ b/whatsapp-ext/cmd.go @@ -59,6 +59,11 @@ func (ext *ExtendedConn) handleMessageCommand(message []byte) { if !ok { continue } - go commandHandler.HandleCommand(event) + + if ext.shouldCallSynchronously(commandHandler) { + commandHandler.HandleCommand(event) + } else { + go commandHandler.HandleCommand(event) + } } } diff --git a/whatsapp-ext/conn.go b/whatsapp-ext/conn.go index 9f72251..12162b8 100644 --- a/whatsapp-ext/conn.go +++ b/whatsapp-ext/conn.go @@ -55,6 +55,11 @@ func (ext *ExtendedConn) handleMessageConn(message []byte) { if !ok { continue } - connInfoHandler.HandleConnInfo(event) + + if ext.shouldCallSynchronously(connInfoHandler) { + connInfoHandler.HandleConnInfo(event) + } else { + go connInfoHandler.HandleConnInfo(event) + } } } diff --git a/whatsapp-ext/jsonmessage.go b/whatsapp-ext/jsonmessage.go index 1174844..8c95a0f 100644 --- a/whatsapp-ext/jsonmessage.go +++ b/whatsapp-ext/jsonmessage.go @@ -96,7 +96,12 @@ func (ext *ExtendedConn) HandleJsonMessage(message string) { if !ok { continue } - ujmHandler.HandleUnhandledJSONMessage(message) + + if ext.shouldCallSynchronously(ujmHandler) { + ujmHandler.HandleUnhandledJSONMessage(message) + } else { + go ujmHandler.HandleUnhandledJSONMessage(message) + } } } } diff --git a/whatsapp-ext/msginfo.go b/whatsapp-ext/msginfo.go index 2366e3f..d6588ab 100644 --- a/whatsapp-ext/msginfo.go +++ b/whatsapp-ext/msginfo.go @@ -85,6 +85,11 @@ func (ext *ExtendedConn) handleMessageMsgInfo(msgType JSONMessageType, message [ if !ok { continue } - go msgInfoHandler.HandleMsgInfo(event) + + if ext.shouldCallSynchronously(msgInfoHandler) { + msgInfoHandler.HandleMsgInfo(event) + } else { + go msgInfoHandler.HandleMsgInfo(event) + } } } diff --git a/whatsapp-ext/presence.go b/whatsapp-ext/presence.go index c70f4e6..12808e2 100644 --- a/whatsapp-ext/presence.go +++ b/whatsapp-ext/presence.go @@ -62,6 +62,11 @@ func (ext *ExtendedConn) handleMessagePresence(message []byte) { if !ok { continue } - go presenceHandler.HandlePresence(event) + + if ext.shouldCallSynchronously(presenceHandler) { + presenceHandler.HandlePresence(event) + } else { + go presenceHandler.HandlePresence(event) + } } } diff --git a/whatsapp-ext/props.go b/whatsapp-ext/props.go index 8f5dbbd..76d99f5 100644 --- a/whatsapp-ext/props.go +++ b/whatsapp-ext/props.go @@ -63,6 +63,11 @@ func (ext *ExtendedConn) handleMessageProps(message []byte) { if !ok { continue } - go protocolPropsHandler.HandleProtocolProps(event) + + if ext.shouldCallSynchronously(protocolPropsHandler) { + protocolPropsHandler.HandleProtocolProps(event) + } else { + go protocolPropsHandler.HandleProtocolProps(event) + } } } diff --git a/whatsapp-ext/protomessage.go b/whatsapp-ext/protomessage.go index ad7e76e..620ee46 100644 --- a/whatsapp-ext/protomessage.go +++ b/whatsapp-ext/protomessage.go @@ -35,7 +35,7 @@ type MessageRevocation struct { func (ext *ExtendedConn) HandleRawMessage(message *proto.WebMessageInfo) { protoMsg := message.GetMessage().GetProtocolMessage() - if protoMsg.GetType() == proto.ProtocolMessage_REVOKE { + if protoMsg != nil && protoMsg.GetType() == proto.ProtocolMessage_REVOKE { key := protoMsg.GetKey() deletedMessage := MessageRevocation{ Id: key.GetId(), @@ -48,7 +48,12 @@ func (ext *ExtendedConn) HandleRawMessage(message *proto.WebMessageInfo) { if !ok { continue } - mrHandler.HandleMessageRevoke(deletedMessage) + + if ext.shouldCallSynchronously(mrHandler) { + mrHandler.HandleMessageRevoke(deletedMessage) + } else { + go mrHandler.HandleMessageRevoke(deletedMessage) + } } } } diff --git a/whatsapp-ext/stream.go b/whatsapp-ext/stream.go index 85dbbd7..877e7a9 100644 --- a/whatsapp-ext/stream.go +++ b/whatsapp-ext/stream.go @@ -58,6 +58,11 @@ func (ext *ExtendedConn) handleMessageStream(message []json.RawMessage) { if !ok { continue } - go streamHandler.HandleStreamEvent(event) + + if ext.shouldCallSynchronously(streamHandler) { + streamHandler.HandleStreamEvent(event) + } else { + go streamHandler.HandleStreamEvent(event) + } } } diff --git a/whatsapp-ext/whatsapp.go b/whatsapp-ext/whatsapp.go index 5e041ea..f3d0b84 100644 --- a/whatsapp-ext/whatsapp.go +++ b/whatsapp-ext/whatsapp.go @@ -46,6 +46,15 @@ func ExtendConn(conn *whatsapp.Conn) *ExtendedConn { return ext } +func (ext *ExtendedConn) shouldCallSynchronously(handler whatsapp.Handler) bool { + sh, ok := handler.(whatsapp.SyncHandler) + return ok && sh.ShouldCallSynchronously() +} + +func (ext *ExtendedConn) ShouldCallSynchronously() bool { + return true +} + type GroupInfo struct { JID string `json:"jid"` OwnerJID string `json:"owner"`