Avoid unneeded JSON operations (#2698)

We were `json.Unmarshal`ing the EDU and `json.Marshal`ing right before
sending the EDU to the stream. Those are now removed and the consumer
does `json.Unmarshal` once.

2 files changed, 6 insertions, 26 deletions

diff --git a/federationapi/producers/syncapi.go b/federationapi/producers/syncapi.go
index 0825bccb..86c8c10a 100644
--- a/federationapi/producers/syncapi.go
+++ b/federationapi/producers/syncapi.go
@@ -167,15 +167,11 @@ func (p *SyncAPIProducer) SendPresence(
 }
 
 func (p *SyncAPIProducer) SendDeviceListUpdate(
-	ctx context.Context, deviceListUpdate *gomatrixserverlib.DeviceListUpdateEvent,
+	ctx context.Context, deviceListUpdate gomatrixserverlib.RawJSON, origin string,
 ) (err error) {
 	m := nats.NewMsg(p.TopicDeviceListUpdate)
-	m.Header.Set(jetstream.UserID, deviceListUpdate.UserID)
-	m.Data, err = json.Marshal(deviceListUpdate)
-	if err != nil {
-		return fmt.Errorf("json.Marshal: %w", err)
-	}
-
+	m.Header.Set("origin", origin)
+	m.Data = deviceListUpdate
 	log.Debugf("Sending device list update: %+v", m.Header)
 	_, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
 	return err
diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go
index 3d931996..a9714c65 100644
--- a/federationapi/routing/send.go
+++ b/federationapi/routing/send.go
@@ -359,7 +359,9 @@ func (t *txnReq) processEDUs(ctx context.Context) {
 				}
 			}
 		case gomatrixserverlib.MDeviceListUpdate:
-			t.processDeviceListUpdate(ctx, e)
+			if err := t.producer.SendDeviceListUpdate(ctx, e.Content, e.Origin); err != nil {
+				util.GetLogger(ctx).WithError(err).Error("failed to InputDeviceListUpdate")
+			}
 		case gomatrixserverlib.MReceipt:
 			// https://matrix.org/docs/spec/server_server/r0.1.4#receipts
 			payload := map[string]types.FederationReceiptMRead{}
@@ -454,21 +456,3 @@ func (t *txnReq) processReceiptEvent(ctx context.Context,
 
 	return nil
 }
-
-func (t *txnReq) processDeviceListUpdate(ctx context.Context, e gomatrixserverlib.EDU) {
-	var payload gomatrixserverlib.DeviceListUpdateEvent
-	if err := json.Unmarshal(e.Content, &payload); err != nil {
-		util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal device list update event")
-		return
-	}
-	if _, serverName, err := gomatrixserverlib.SplitID('@', payload.UserID); err != nil {
-		return
-	} else if serverName == t.ourServerName {
-		return
-	} else if serverName != t.Origin {
-		return
-	}
-	if err := t.producer.SendDeviceListUpdate(ctx, &payload); err != nil {
-		util.GetLogger(ctx).WithError(err).WithField("user_id", payload.UserID).Error("failed to InputDeviceListUpdate")
-	}
-}