Send bridge state and message checkpoints

config/bridge.go

@@ -31,6 +31,8 @@ type BridgeConfig struct {
 	DisplaynameTemplate string `yaml:"displayname_template"`
 	ChannelnameTemplate string `yaml:"channelname_template"`
 
+	DeliveryReceipts bool `yaml:"delivery_receipts"`
+
 	CommandPrefix string `yaml:"command_prefix"`
 
 	ManagementRoomText bridgeconfig.ManagementRoomTexts `yaml:"management_room_text"`

config/upgrade.go

@@ -29,6 +29,7 @@ func DoUpgrade(helper *up.Helper) {
 	helper.Copy(up.Str, "bridge", "displayname_template")
 	helper.Copy(up.Str, "bridge", "channelname_template")
 	helper.Copy(up.Int, "bridge", "portal_message_buffer")
+	helper.Copy(up.Bool, "bridge", "delivery_receipts")
 	helper.Copy(up.Bool, "bridge", "sync_with_custom_puppets")
 	helper.Copy(up.Bool, "bridge", "sync_direct_chat_list")
 	helper.Copy(up.Bool, "bridge", "federate_rooms")

example-config.yaml

@@ -75,6 +75,8 @@ bridge:
 
     portal_message_buffer: 128
 
+    # Should the bridge send a read receipt from the bridge bot when a message has been sent to Discord?
+    delivery_receipts: false
     # Should the bridge sync with double puppeting to receive EDUs that aren't normally sent to appservices.
     sync_with_custom_puppets: true
     # Should the bridge update the m.direct account data event when double puppeting is enabled.

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
 	github.com/yuin/goldmark v1.4.12
 	maunium.net/go/maulogger/v2 v2.3.2
-	maunium.net/go/mautrix v0.11.1-0.20220529123139-5bc36b2978c1
+	maunium.net/go/mautrix v0.11.1-0.20220530120827-7eec0bd4d3c4
 )
 
 require (

go.sum

@@ -58,5 +58,5 @@ maunium.net/go/mauflag v1.0.0 h1:YiaRc0tEI3toYtJMRIfjP+jklH45uDHtT80nUamyD4M=
 maunium.net/go/mauflag v1.0.0/go.mod h1:nLivPOpTpHnpzEh8jEdSL9UqO9+/KBJFmNRlwKfkPeA=
 maunium.net/go/maulogger/v2 v2.3.2 h1:1XmIYmMd3PoQfp9J+PaHhpt80zpfmMqaShzUTC7FwY0=
 maunium.net/go/maulogger/v2 v2.3.2/go.mod h1:TYWy7wKwz/tIXTpsx8G3mZseIRiC5DoMxSZazOHy68A=
-maunium.net/go/mautrix v0.11.1-0.20220529123139-5bc36b2978c1 h1:HNntVQh0XVyWDAsSQA/0Rk2++1cGOzmm7tH8xILSsak=
-maunium.net/go/mautrix v0.11.1-0.20220529123139-5bc36b2978c1/go.mod h1:CiKpMhAx5QZFHK03jpWb0iKI3sGU8x6+LfsOjDrcO8I=
+maunium.net/go/mautrix v0.11.1-0.20220530120827-7eec0bd4d3c4 h1:/A03e5QOu8nTi6QChiEr0Udg2YE6alB7ZNJwYdHqDQk=
+maunium.net/go/mautrix v0.11.1-0.20220530120827-7eec0bd4d3c4/go.mod h1:CiKpMhAx5QZFHK03jpWb0iKI3sGU8x6+LfsOjDrcO8I=

portal.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -536,6 +537,7 @@ func (portal *Portal) handleDiscordAttachment(intent *appservice.IntentAPI, msgI
 	if threadRelation != nil {
 		threadRelation.InReplyTo.EventID = resp.EventID
 	}
+	go portal.sendDeliveryReceipt(resp.EventID)
 }
 
 func (portal *Portal) handleDiscordMessageCreate(user *User, msg *discordgo.Message, thread *Thread) {
@@ -618,6 +620,7 @@ func (portal *Portal) handleDiscordMessageCreate(user *User, msg *discordgo.Mess
 		if threadRelation != nil {
 			threadRelation.InReplyTo.EventID = resp.EventID
 		}
+		go portal.sendDeliveryReceipt(resp.EventID)
 	}
 
 	for _, attachment := range msg.Attachments {
@@ -714,13 +717,15 @@ func (portal *Portal) handleDiscordMessageUpdate(user *User, msg *discordgo.Mess
 		editTS = msg.EditedTimestamp.UnixMilli()
 	}
 	// TODO figure out some way to deduplicate outgoing edits
-	_, err := portal.sendMatrixMessage(intent, event.EventMessage, &content, nil, editTS)
+	resp, err := portal.sendMatrixMessage(intent, event.EventMessage, &content, nil, editTS)
 	if err != nil {
 		portal.log.Warnfln("failed to send message %q to matrix: %v", msg.ID, err)
 
 		return
 	}
 
+	portal.sendDeliveryReceipt(resp.EventID)
+
 	//ts, _ := msg.Timestamp.Parse()
 	//portal.markMessageHandled(existing, msg.ID, resp.EventID, msg.Author.ID, ts)
 }
@@ -747,11 +752,12 @@ func (portal *Portal) handleDiscordMessageDelete(user *User, msg *discordgo.Mess
 			attachment.Delete()
 		}
 
-		_, err := intent.RedactEvent(portal.MXID, existing.MXID)
+		resp, err := intent.RedactEvent(portal.MXID, existing.MXID)
 		if err != nil {
 			portal.log.Warnfln("Failed to redact message %s: %v", existing.MXID, err)
 		}
 		existing.Delete()
+		portal.sendDeliveryReceipt(resp.EventID)
 	}
 }
 
@@ -914,12 +920,14 @@ func (portal *Portal) startThreadFromMatrix(sender *User, threadRoot id.EventID)
 
 func (portal *Portal) handleMatrixMessage(sender *User, evt *event.Event) {
 	if portal.IsPrivateChat() && sender.DiscordID != portal.Key.Receiver {
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("user is not portal receiver"), true, 0)
 		return
 	}
 
 	content, ok := evt.Content.Parsed.(*event.MessageEventContent)
 	if !ok {
 		portal.log.Debugfln("Failed to handle event %s: unexpected parsed content type %T", evt.ID, evt.Content.Parsed)
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, fmt.Errorf("unexpected parsed content type %T", evt.Content.Parsed), true, 0)
 		return
 	}
 
@@ -972,7 +980,7 @@ func (portal *Portal) handleMatrixMessage(sender *User, evt *event.Event) {
 		data, err := portal.downloadMatrixAttachment(evt.ID, content)
 		if err != nil {
 			portal.log.Errorfln("Failed to download matrix attachment: %v", err)
-
+			portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, err, true, 0)
 			return
 		}
 
@@ -983,12 +991,14 @@ func (portal *Portal) handleMatrixMessage(sender *User, evt *event.Event) {
 		}}
 	default:
 		portal.log.Warnln("Unknown message type", content.MsgType)
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, fmt.Errorf("unsupported msgtype %s", content.MsgType), true, 0)
 		return
 	}
 	sendReq.Nonce = generateNonce()
 	msg, err := sender.Session.ChannelMessageSendComplex(channelID, &sendReq)
 	if err != nil {
 		portal.log.Errorfln("Failed to send message: %v", err)
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, err, true, 0)
 		return
 	}
 
@@ -1001,6 +1011,18 @@ func (portal *Portal) handleMatrixMessage(sender *User, evt *event.Event) {
 		dbMsg.Timestamp, _ = discordgo.SnowflakeTimestamp(msg.ID)
 		dbMsg.ThreadID = threadID
 		dbMsg.Insert()
+		portal.log.Debugfln("Handled Matrix event %s", evt.ID)
+		portal.bridge.SendMessageSuccessCheckpoint(evt, bridge.MsgStepRemote, 0)
+		portal.sendDeliveryReceipt(evt.ID)
+	}
+}
+
+func (portal *Portal) sendDeliveryReceipt(eventID id.EventID) {
+	if portal.bridge.Config.Bridge.DeliveryReceipts {
+		err := portal.bridge.Bot.MarkRead(portal.MXID, eventID)
+		if err != nil {
+			portal.log.Debugfln("Failed to send delivery receipt for %s: %v", eventID, err)
+		}
 	}
 }
 
@@ -1117,12 +1139,14 @@ func (portal *Portal) getMatrixUsers() ([]id.UserID, error) {
 
 func (portal *Portal) handleMatrixReaction(sender *User, evt *event.Event) {
 	if portal.IsPrivateChat() && sender.DiscordID != portal.Key.Receiver {
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("user is not portal receiver"), true, 0)
 		return
 	}
 
 	reaction := evt.Content.AsReaction()
 	if reaction.RelatesTo.Type != event.RelAnnotation {
 		portal.log.Errorfln("Ignoring reaction %s due to unknown m.relates_to data", evt.ID)
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("unknown m.relates_to data"), true, 0)
 		return
 	}
 
@@ -1146,6 +1170,7 @@ func (portal *Portal) handleMatrixReaction(sender *User, evt *event.Event) {
 	if msg == nil {
 		attachment := portal.bridge.DB.Attachment.GetByMatrixID(portal.Key, reaction.RelatesTo.EventID)
 		if attachment == nil {
+			portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("unknown reaction target"), true, 0)
 			return
 		}
 		discordID = attachment.MessageID
@@ -1162,6 +1187,7 @@ func (portal *Portal) handleMatrixReaction(sender *User, evt *event.Event) {
 		emoji := portal.bridge.DB.Emoji.GetByMatrixURL(uri)
 		if emoji == nil {
 			portal.log.Errorfln("Couldn't find emoji corresponding to %s", emojiID)
+			portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("unknown emoji"), true, 0)
 			return
 		}
 
@@ -1176,6 +1202,7 @@ func (portal *Portal) handleMatrixReaction(sender *User, evt *event.Event) {
 	err := sender.Session.MessageReactionAdd(channelID, discordID, emojiID)
 	if err != nil {
 		portal.log.Debugf("Failed to send reaction to %s: %v", discordID, err)
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, err, true, 0)
 		return
 	}
 
@@ -1187,6 +1214,9 @@ func (portal *Portal) handleMatrixReaction(sender *User, evt *event.Event) {
 	dbReaction.ThreadID = threadID
 	dbReaction.MXID = evt.ID
 	dbReaction.Insert()
+	portal.log.Debugfln("Handled Matrix reaction %s", evt.ID)
+	portal.bridge.SendMessageSuccessCheckpoint(evt, bridge.MsgStepRemote, 0)
+	portal.sendDeliveryReceipt(evt.ID)
 }
 
 func (portal *Portal) handleDiscordReaction(user *User, reaction *discordgo.MessageReaction, add bool, thread *Thread) {
@@ -1243,13 +1273,13 @@ func (portal *Portal) handleDiscordReaction(user *User, reaction *discordgo.Mess
 			return
 		}
 
-		_, err := intent.RedactEvent(portal.MXID, existing.MXID)
+		resp, err := intent.RedactEvent(portal.MXID, existing.MXID)
 		if err != nil {
 			portal.log.Warnfln("Failed to remove reaction from %s: %v", portal.MXID, err)
 		}
 
 		existing.Delete()
-
+		go portal.sendDeliveryReceipt(resp.EventID)
 		return
 	} else if existing != nil {
 		portal.log.Debugfln("Ignoring duplicate reaction %s from %s to %s", discordID, reaction.UserID, message.DiscordID)
@@ -1287,11 +1317,13 @@ func (portal *Portal) handleDiscordReaction(user *User, reaction *discordgo.Mess
 			dbReaction.ThreadID = thread.ID
 		}
 		dbReaction.Insert()
+		portal.sendDeliveryReceipt(dbReaction.MXID)
 	}
 }
 
 func (portal *Portal) handleMatrixRedaction(sender *User, evt *event.Event) {
 	if portal.IsPrivateChat() && sender.DiscordID != portal.Key.Receiver {
+		portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("user is not portal receiver"), true, 0)
 		return
 	}
 
@@ -1301,8 +1333,11 @@ func (portal *Portal) handleMatrixRedaction(sender *User, evt *event.Event) {
 		err := sender.Session.ChannelMessageDelete(message.DiscordProtoChannelID(), message.DiscordID)
 		if err != nil {
 			portal.log.Debugfln("Failed to delete discord message %s: %v", message.DiscordID, err)
+			portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, err, true, 0)
 		} else {
 			message.Delete()
+			portal.bridge.SendMessageSuccessCheckpoint(evt, bridge.MsgStepRemote, 0)
+			portal.sendDeliveryReceipt(evt.ID)
 		}
 		return
 	}
@@ -1313,14 +1348,18 @@ func (portal *Portal) handleMatrixRedaction(sender *User, evt *event.Event) {
 		err := sender.Session.MessageReactionRemove(reaction.DiscordProtoChannelID(), reaction.MessageID, reaction.EmojiName, reaction.Sender)
 		if err != nil {
 			portal.log.Debugfln("Failed to delete reaction %s from %s: %v", reaction.EmojiName, reaction.MessageID, err)
+			portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, err, true, 0)
 		} else {
 			reaction.Delete()
+			portal.bridge.SendMessageSuccessCheckpoint(evt, bridge.MsgStepRemote, 0)
+			portal.sendDeliveryReceipt(evt.ID)
 		}
 
 		return
 	}
 
 	portal.log.Warnfln("Failed to redact %s: no event found", evt.Redacts)
+	portal.bridge.SendMessageErrorCheckpoint(evt, bridge.MsgStepRemote, errors.New("redaction target not found"), true, 0)
 }
 
 func (portal *Portal) UpdateName(name string) bool {

user.go

@@ -44,6 +44,19 @@ type User struct {
 	dmSpaceMembershipChecked bool
 
 	Session *discordgo.Session
+
+	BridgeState *bridge.BridgeStateQueue
+}
+
+func (user *User) GetRemoteID() string {
+	return user.DiscordID
+}
+
+func (user *User) GetRemoteName() string {
+	if user.Session != nil && user.Session.State != nil && user.Session.State.User != nil {
+		return fmt.Sprintf("%s#%s", user.Session.State.User.Username, user.Session.State.User.Discriminator)
+	}
+	return user.DiscordID
 }
 
 var discordLog log.Logger
@@ -177,6 +190,7 @@ func (br *DiscordBridge) NewUser(dbUser *database.User) *User {
 	}
 
 	user.PermissionLevel = br.Config.Bridge.Permissions.Get(user.MXID)
+	user.BridgeState = br.NewBridgeStateQueue(user, user.log)
 
 	return user
 }
@@ -202,14 +216,20 @@ func (br *DiscordBridge) getAllUsersWithToken() []*User {
 func (br *DiscordBridge) startUsers() {
 	br.Log.Debugln("Starting users")
 
-	for _, u := range br.getAllUsersWithToken() {
+	usersWithToken := br.getAllUsersWithToken()
+	for _, u := range usersWithToken {
 		go func(user *User) {
+			user.BridgeState.Send(bridge.State{StateEvent: bridge.StateConnecting})
 			err := user.Connect()
 			if err != nil {
 				user.log.Errorfln("Error connecting: %v", err)
+				user.BridgeState.Send(bridge.State{StateEvent: bridge.StateUnknownError, Message: err.Error()})
 			}
 		}(u)
 	}
+	if len(usersWithToken) == 0 {
+		br.SendGlobalBridgeState(bridge.State{StateEvent: bridge.StateUnconfigured}.Fill(nil))
+	}
 
 	br.Log.Debugln("Starting custom puppets")
 	for _, customPuppet := range br.GetAllPuppetsWithCustomMXID() {
@@ -499,6 +519,7 @@ func (user *User) readyHandler(_ *discordgo.Session, r *discordgo.Ready) {
 		user.DiscordID = r.User.ID
 		user.Update()
 	}
+	user.BridgeState.Send(bridge.State{StateEvent: bridge.StateBackfilling})
 
 	updateTS := time.Now()
 	portalsInSpace := make(map[string]bool)
@@ -514,6 +535,7 @@ func (user *User) readyHandler(_ *discordgo.Session, r *discordgo.Ready) {
 		portal := user.GetPortalByMeta(ch)
 		user.handlePrivateChannel(portal, ch, updateTS, i < maxCreate, portalsInSpace[portal.Key.String()])
 	}
+	user.BridgeState.Send(bridge.State{StateEvent: bridge.StateConnected})
 }
 
 func (user *User) handlePrivateChannel(portal *Portal, meta *discordgo.Channel, timestamp time.Time, create, isInSpace bool) {
@@ -592,6 +614,7 @@ func (user *User) connectedHandler(_ *discordgo.Session, c *discordgo.Connect) {
 
 func (user *User) disconnectedHandler(_ *discordgo.Session, d *discordgo.Disconnect) {
 	user.log.Debugln("Disconnected from discord")
+	user.BridgeState.Send(bridge.State{StateEvent: bridge.StateTransientDisconnect})
 }
 
 func (user *User) guildCreateHandler(_ *discordgo.Session, g *discordgo.GuildCreate) {