Make message handling more synchronous and fill history on portal create

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

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=

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) {
+func (portal *Portal) startHandling(info whatsapp.MessageInfo) (*sync.Mutex, bool) {
-	if portal.isRecentlyHandled(id) {
+	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) {

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)
+	user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleTextMessage(user, message)
 }
 
 func (user *User) HandleImageMessage(message whatsapp.ImageMessage) {
-	portal := user.GetPortalByJID(message.Info.RemoteJid)
+	user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Caption)
 }
 
 func (user *User) HandleVideoMessage(message whatsapp.VideoMessage) {
-	portal := user.GetPortalByJID(message.Info.RemoteJid)
+	user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Caption)
 }
 
 func (user *User) HandleAudioMessage(message whatsapp.AudioMessage) {
-	portal := user.GetPortalByJID(message.Info.RemoteJid)
+	user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleMediaMessage(user, message.Download, nil, message.Info, message.Type, "")
 }
 
 func (user *User) HandleDocumentMessage(message whatsapp.DocumentMessage) {
-	portal := user.GetPortalByJID(message.Info.RemoteJid)
+	user.GetPortalByJID(message.Info.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleMediaMessage(user, message.Download, message.Thumbnail, message.Info, message.Type, message.Title)
 }
 
 func (user *User) HandleMessageRevoke(message whatsappExt.MessageRevocation) {
-	portal := user.GetPortalByJID(message.RemoteJid)
+	user.GetPortalByJID(message.RemoteJid).messages <- PortalMessage{user, message}
-	portal.HandleMessageRevoke(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()
+		go func() {
-		for _, id := range info.IDs {
+			intent := user.bridge.GetPuppetByJID(info.SenderJID).Intent()
-			msg := user.bridge.DB.Message.GetByJID(portal.Key, id)
+			for _, id := range info.IDs {
-			if msg == nil {
+				msg := user.bridge.DB.Message.GetByJID(portal.Key, id)
-				continue
+				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)
 	}
 }
 

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)
+		}
 	}
 }

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)
+		}
 	}
 }

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)
+		}
 	}
 }

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)
+			}
 		}
 	}
 }

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)
+		}
 	}
 }

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)
+		}
 	}
 }

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)
+		}
 	}
 }

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)
+			}
 		}
 	}
 }

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)
+		}
 	}
 }

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"`