Add support for MSC2409

config/config.go

@@ -61,6 +61,8 @@ type Config struct {
 			Avatar      string `yaml:"avatar"`
 		} `yaml:"bot"`
 
+		EphemeralEvents bool `yaml:"ephemeral_events"`
+
 		ASToken string `yaml:"as_token"`
 		HSToken string `yaml:"hs_token"`
 	} `yaml:"appservice"`

config/registration.go

@@ -61,6 +61,7 @@ func (config *Config) copyToRegistration(registration *appservice.Registration)
 	falseVal := false
 	registration.RateLimited = &falseVal
 	registration.SenderLocalpart = config.AppService.Bot.Username
+	registration.EphemeralEvents = config.AppService.EphemeralEvents
 
 	userIDRegex, err := regexp.Compile(fmt.Sprintf("^@%s:%s$",
 		config.Bridge.FormatUsername("[0-9]+"),

config/upgrade.go

@@ -51,6 +51,7 @@ func (helper *UpgradeHelper) doUpgrade() {
 	helper.Copy(Str, "appservice", "bot", "username")
 	helper.Copy(Str, "appservice", "bot", "displayname")
 	helper.Copy(Str, "appservice", "bot", "avatar")
+	helper.Copy(Bool, "appservice", "ephemeral_events")
 	helper.Copy(Str, "appservice", "as_token")
 	helper.Copy(Str, "appservice", "hs_token")
 

custompuppet.go

@@ -24,8 +24,6 @@ import (
 	"fmt"
 	"time"
 
-	"go.mau.fi/whatsmeow/types"
-
 	"maunium.net/go/mautrix"
 	"maunium.net/go/mautrix/appservice"
 	"maunium.net/go/mautrix/event"
@@ -139,7 +137,6 @@ func (puppet *Puppet) clearCustomMXID() {
 	puppet.CustomMXID = ""
 	puppet.AccessToken = ""
 	puppet.customIntent = nil
-	puppet.customTypingIn = nil
 	puppet.customUser = nil
 }
 
@@ -165,7 +162,6 @@ func (puppet *Puppet) StartCustomMXID(reloginOnFail bool) error {
 		return ErrMismatchingMXID
 	}
 	puppet.customIntent = intent
-	puppet.customTypingIn = make(map[id.RoomID]bool)
 	puppet.customUser = puppet.bridge.GetUserByMXID(puppet.CustomMXID)
 	puppet.startSyncing()
 	return nil
@@ -210,10 +206,10 @@ func (puppet *Puppet) ProcessResponse(resp *mautrix.RespSync, _ string) error {
 			switch evt.Type {
 			case event.EphemeralEventReceipt:
 				if puppet.EnableReceipts {
-					go puppet.handleReceiptEvent(portal, evt)
+					go puppet.bridge.MatrixHandler.HandleReceipt(evt)
 				}
 			case event.EphemeralEventTyping:
-				go puppet.handleTypingEvent(portal, evt)
+				go puppet.bridge.MatrixHandler.HandleTyping(evt)
 			}
 		}
 	}
@@ -226,66 +222,12 @@ func (puppet *Puppet) ProcessResponse(resp *mautrix.RespSync, _ string) error {
 			if err != nil {
 				continue
 			}
-			go puppet.handlePresenceEvent(evt)
+			go puppet.bridge.MatrixHandler.HandlePresence(evt)
 		}
 	}
 	return nil
 }
 
-func (puppet *Puppet) handlePresenceEvent(event *event.Event) {
-	presence := types.PresenceAvailable
-	if event.Content.Raw["presence"].(string) != "online" {
-		presence = types.PresenceUnavailable
-		puppet.customUser.log.Debugln("Marking offline")
-	} else {
-		puppet.customUser.log.Debugln("Marking online")
-	}
-	puppet.customUser.lastPresence = presence
-	if puppet.customUser.Client.Store.PushName != "" {
-		err := puppet.customUser.Client.SendPresence(presence)
-		if err != nil {
-			puppet.customUser.log.Warnln("Failed to set presence:", err)
-		}
-	}
-}
-
-func (puppet *Puppet) handleReceiptEvent(portal *Portal, event *event.Event) {
-	for eventID, receipts := range *event.Content.AsReceipt() {
-		if receipt, ok := receipts.Read[puppet.CustomMXID]; !ok {
-			// Ignore receipt events where this user isn't present.
-		} else if isDoublePuppeted, _ := receipt.Extra[doublePuppetField].(bool); isDoublePuppeted {
-			puppet.customUser.log.Debugfln("Ignoring double puppeted read receipt %+v", event.Content.Raw)
-			// Ignore double puppeted read receipts.
-		} else {
-			portal.HandleMatrixReadReceipt(puppet.customUser, eventID, time.UnixMilli(receipt.Timestamp))
-		}
-	}
-}
-
-func (puppet *Puppet) handleTypingEvent(portal *Portal, evt *event.Event) {
-	isTyping := false
-	for _, userID := range evt.Content.AsTyping().UserIDs {
-		if userID == puppet.CustomMXID {
-			isTyping = true
-			break
-		}
-	}
-	if puppet.customTypingIn[evt.RoomID] != isTyping {
-		puppet.customTypingIn[evt.RoomID] = isTyping
-		presence := types.ChatPresenceComposing
-		if !isTyping {
-			puppet.customUser.log.Debugfln("Marking not typing in %s/%s", portal.Key.JID, portal.MXID)
-			presence = types.ChatPresencePaused
-		} else {
-			puppet.customUser.log.Debugfln("Marking typing in %s/%s", portal.Key.JID, portal.MXID)
-		}
-		err := puppet.customUser.Client.SendChatPresence(presence, portal.Key.JID)
-		if err != nil {
-			puppet.customUser.log.Warnln("Error setting typing:", err)
-		}
-	}
-}
-
 func (puppet *Puppet) tryRelogin(cause error, action string) bool {
 	if !puppet.bridge.Config.CanAutoDoublePuppet(puppet.CustomMXID) {
 		return false

example-config.yaml

@@ -56,6 +56,11 @@ appservice:
         displayname: WhatsApp bridge bot
         avatar: mxc://maunium.net/NeXNQarUbrlYBiPCpprYsRqr
 
+    # Whether or not to receive ephemeral events via appservice transactions.
+    # Requires MSC2409 support (i.e. Synapse 1.22+).
+    # You should disable bridge -> sync_with_custom_puppets when this is enabled.
+    ephemeral_events: false
+
     # Authentication tokens for AS <-> HS communication. Autogenerated; do not modify.
     as_token: "This value is generated when generating the registration"
     hs_token: "This value is generated when generating the registration"

matrix.go

@@ -1,5 +1,5 @@
 // mautrix-whatsapp - A Matrix-WhatsApp puppeting bridge.
-// Copyright (C) 2020 Tulir Asokan
+// Copyright (C) 2021 Tulir Asokan
 //
 // This program is free software: you can redistribute it and/or modify
 // it under the terms of the GNU Affero General Public License as published by
@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	"go.mau.fi/whatsmeow/types"
 	"maunium.net/go/maulogger/v2"
 
 	"maunium.net/go/mautrix"
@@ -57,6 +58,9 @@ func NewMatrixHandler(bridge *Bridge) *MatrixHandler {
 	bridge.EventProcessor.On(event.StateRoomAvatar, handler.HandleRoomMetadata)
 	bridge.EventProcessor.On(event.StateTopic, handler.HandleRoomMetadata)
 	bridge.EventProcessor.On(event.StateEncryption, handler.HandleEncryption)
+	bridge.EventProcessor.On(event.EphemeralEventPresence, handler.HandlePresence)
+	bridge.EventProcessor.On(event.EphemeralEventReceipt, handler.HandleReceipt)
+	bridge.EventProcessor.On(event.EphemeralEventTyping, handler.HandleTyping)
 	return handler
 }
 
@@ -474,3 +478,61 @@ func (mx *MatrixHandler) HandleRedaction(evt *event.Event) {
 		portal.HandleMatrixRedaction(user, evt)
 	}
 }
+
+func (mx *MatrixHandler) HandlePresence(evt *event.Event) {
+	user := mx.bridge.GetUserByMXIDIfExists(evt.Sender)
+	if user == nil || !user.IsLoggedIn() {
+		return
+	}
+	customPuppet := mx.bridge.GetPuppetByCustomMXID(user.MXID)
+	// TODO move this flag to the user and/or portal data
+	if customPuppet != nil && !customPuppet.EnablePresence {
+		return
+	}
+
+	presence := types.PresenceAvailable
+	if evt.Content.AsPresence().Presence != event.PresenceOnline {
+		presence = types.PresenceUnavailable
+		user.log.Debugln("Marking offline")
+	} else {
+		user.log.Debugln("Marking online")
+	}
+	user.lastPresence = presence
+	if user.Client.Store.PushName != "" {
+		err := user.Client.SendPresence(presence)
+		if err != nil {
+			user.log.Warnln("Failed to set presence:", err)
+		}
+	}
+}
+
+func (mx *MatrixHandler) HandleReceipt(evt *event.Event) {
+	portal := mx.bridge.GetPortalByMXID(evt.RoomID)
+	if portal == nil {
+		return
+	}
+
+	for eventID, receipts := range *evt.Content.AsReceipt() {
+		for userID, receipt := range receipts.Read {
+			if user := mx.bridge.GetUserByMXIDIfExists(userID); user == nil {
+				// Not a bridge user
+			} else if customPuppet := mx.bridge.GetPuppetByCustomMXID(user.MXID); customPuppet != nil && !customPuppet.EnableReceipts {
+				// TODO move this flag to the user and/or portal data
+				continue
+			} else if isDoublePuppeted, _ := receipt.Extra[doublePuppetField].(bool); isDoublePuppeted {
+				// Ignore double puppeted read receipts.
+				user.log.Debugfln("Ignoring double puppeted read receipt %+v", evt.Content.Raw)
+			} else {
+				portal.HandleMatrixReadReceipt(user, eventID, time.UnixMilli(receipt.Timestamp))
+			}
+		}
+	}
+}
+
+func (mx *MatrixHandler) HandleTyping(evt *event.Event) {
+	portal := mx.bridge.GetPortalByMXID(evt.RoomID)
+	if portal == nil {
+		return
+	}
+	portal.HandleMatrixTyping(evt.Content.AsTyping().UserIDs)
+}

portal.go

@@ -198,6 +198,9 @@ type Portal struct {
 
 	privateChatBackfillInvitePuppet func()
 
+	currentlyTyping     []id.UserID
+	currentlyTypingLock sync.Mutex
+
 	messages chan PortalMessage
 
 	relayUser *User
@@ -2213,6 +2216,11 @@ func (portal *Portal) HandleMatrixRedaction(sender *User, evt *event.Event) {
 }
 
 func (portal *Portal) HandleMatrixReadReceipt(sender *User, eventID id.EventID, receiptTimestamp time.Time) {
+	if !sender.IsLoggedIn() {
+		portal.log.Debugfln("Ignoring read receipt by %s: user is not connected to WhatsApp", sender.JID)
+		return
+	}
+
 	maxTimestamp := receiptTimestamp
 	if message := portal.bridge.DB.Message.GetByMXID(eventID); message != nil {
 		maxTimestamp = message.Timestamp
@@ -2240,6 +2248,51 @@ func (portal *Portal) HandleMatrixReadReceipt(sender *User, eventID id.EventID,
 	}
 }
 
+func typingDiff(prev, new []id.UserID) (started, stopped []id.UserID) {
+OuterNew:
+	for _, userID := range new {
+		for _, previousUserID := range prev {
+			if userID == previousUserID {
+				continue OuterNew
+			}
+		}
+		started = append(started, userID)
+	}
+OuterPrev:
+	for _, userID := range prev {
+		for _, previousUserID := range new {
+			if userID == previousUserID {
+				continue OuterPrev
+			}
+		}
+		stopped = append(stopped, userID)
+	}
+	return
+}
+
+func (portal *Portal) setTyping(userIDs []id.UserID, state types.ChatPresence) {
+	for _, userID := range userIDs {
+		user := portal.bridge.GetUserByMXIDIfExists(userID)
+		if user == nil || !user.IsLoggedIn() {
+			continue
+		}
+		portal.log.Debugfln("Bridging typing change from %s to chat presence %s", state, user.MXID)
+		err := user.Client.SendChatPresence(state, portal.Key.JID)
+		if err != nil {
+			portal.log.Warnln("Error sending chat presence:", err)
+		}
+	}
+}
+
+func (portal *Portal) HandleMatrixTyping(newTyping []id.UserID) {
+	portal.currentlyTypingLock.Lock()
+	defer portal.currentlyTypingLock.Unlock()
+	startedTyping, stoppedTyping := typingDiff(portal.currentlyTyping, newTyping)
+	portal.currentlyTyping = newTyping
+	portal.setTyping(startedTyping, types.ChatPresenceComposing)
+	portal.setTyping(stoppedTyping, types.ChatPresencePaused)
+}
+
 func (portal *Portal) canBridgeFrom(sender *User, evtType string) bool {
 	if !sender.IsLoggedIn() {
 		if portal.HasRelaybot() {

puppet.go

@@ -160,7 +160,6 @@ type Puppet struct {
 	MXID id.UserID
 
 	customIntent *appservice.IntentAPI
-	customTypingIn map[id.RoomID]bool
 	customUser   *User
 
 	syncLock sync.Mutex

user.go

@@ -66,7 +66,7 @@ type User struct {
 	lastPresence     types.Presence
 }
 
-func (bridge *Bridge) GetUserByMXID(userID id.UserID) *User {
+func (bridge *Bridge) getUserByMXID(userID id.UserID, onlyIfExists bool) *User {
 	_, isPuppet := bridge.ParsePuppetMXID(userID)
 	if isPuppet || userID == bridge.Bot.UserID {
 		return nil
@@ -75,11 +75,23 @@ func (bridge *Bridge) GetUserByMXID(userID id.UserID) *User {
 	defer bridge.usersLock.Unlock()
 	user, ok := bridge.usersByMXID[userID]
 	if !ok {
-		return bridge.loadDBUser(bridge.DB.User.GetByMXID(userID), &userID)
+		userIDPtr := &userID
+		if onlyIfExists {
+			userIDPtr = nil
+		}
+		return bridge.loadDBUser(bridge.DB.User.GetByMXID(userID), userIDPtr)
 	}
 	return user
 }
 
+func (bridge *Bridge) GetUserByMXID(userID id.UserID) *User {
+	return bridge.getUserByMXID(userID, false)
+}
+
+func (bridge *Bridge) GetUserByMXIDIfExists(userID id.UserID) *User {
+	return bridge.getUserByMXID(userID, true)
+}
+
 func (bridge *Bridge) GetUserByJID(jid types.JID) *User {
 	bridge.usersLock.Lock()
 	defer bridge.usersLock.Unlock()