Send-to-device messages over federation (#1198)

* Initial work to send send-to-device messages over federation

* Wire up send-to-device consumer, message formatting

* Generate random message ID

* Review comments, update sytest whitelist

federationsender/consumers/eduserver.go

@@ -15,6 +15,7 @@ package consumers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	"github.com/Shopify/sarama"
 	"github.com/matrix-org/dendrite/eduserver/api"
@@ -23,53 +24,120 @@ import (
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
 	log "github.com/sirupsen/logrus"
 )
 
-// OutputTypingEventConsumer consumes events that originate in EDU server.
-type OutputTypingEventConsumer struct {
-	consumer   *internal.ContinualConsumer
-	db         storage.Database
-	queues     *queue.OutgoingQueues
-	ServerName gomatrixserverlib.ServerName
+// OutputEDUConsumer consumes events that originate in EDU server.
+type OutputEDUConsumer struct {
+	typingConsumer       *internal.ContinualConsumer
+	sendToDeviceConsumer *internal.ContinualConsumer
+	db                   storage.Database
+	queues               *queue.OutgoingQueues
+	ServerName           gomatrixserverlib.ServerName
+	TypingTopic          string
+	SendToDeviceTopic    string
 }
 
-// NewOutputTypingEventConsumer creates a new OutputTypingEventConsumer. Call Start() to begin consuming from EDU servers.
-func NewOutputTypingEventConsumer(
+// NewOutputEDUConsumer creates a new OutputEDUConsumer. Call Start() to begin consuming from EDU servers.
+func NewOutputEDUConsumer(
 	cfg *config.Dendrite,
 	kafkaConsumer sarama.Consumer,
 	queues *queue.OutgoingQueues,
 	store storage.Database,
-) *OutputTypingEventConsumer {
-	consumer := internal.ContinualConsumer{
-		Topic:          string(cfg.Kafka.Topics.OutputTypingEvent),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
+) *OutputEDUConsumer {
+	c := &OutputEDUConsumer{
+		typingConsumer: &internal.ContinualConsumer{
+			Topic:          string(cfg.Kafka.Topics.OutputTypingEvent),
+			Consumer:       kafkaConsumer,
+			PartitionStore: store,
+		},
+		sendToDeviceConsumer: &internal.ContinualConsumer{
+			Topic:          string(cfg.Kafka.Topics.OutputSendToDeviceEvent),
+			Consumer:       kafkaConsumer,
+			PartitionStore: store,
+		},
+		queues:            queues,
+		db:                store,
+		ServerName:        cfg.Matrix.ServerName,
+		TypingTopic:       string(cfg.Kafka.Topics.OutputTypingEvent),
+		SendToDeviceTopic: string(cfg.Kafka.Topics.OutputSendToDeviceEvent),
 	}
-	c := &OutputTypingEventConsumer{
-		consumer:   &consumer,
-		queues:     queues,
-		db:         store,
-		ServerName: cfg.Matrix.ServerName,
-	}
-	consumer.ProcessMessage = c.onMessage
+	c.typingConsumer.ProcessMessage = c.onTypingEvent
+	c.sendToDeviceConsumer.ProcessMessage = c.onSendToDeviceEvent
 
 	return c
 }
 
 // Start consuming from EDU servers
-func (t *OutputTypingEventConsumer) Start() error {
-	return t.consumer.Start()
+func (t *OutputEDUConsumer) Start() error {
+	if err := t.typingConsumer.Start(); err != nil {
+		return fmt.Errorf("t.typingConsumer.Start: %w", err)
+	}
+	if err := t.sendToDeviceConsumer.Start(); err != nil {
+		return fmt.Errorf("t.sendToDeviceConsumer.Start: %w", err)
+	}
+	return nil
 }
 
-// onMessage is called for OutputTypingEvent received from the EDU servers.
-// Parses the msg, creates a matrix federation EDU and sends it to joined hosts.
-func (t *OutputTypingEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
+// onSendToDeviceEvent is called in response to a message received on the
+// send-to-device events topic from the EDU server.
+func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *sarama.ConsumerMessage) error {
+	// Extract the send-to-device event from msg.
+	var ote api.OutputSendToDeviceEvent
+	if err := json.Unmarshal(msg.Value, &ote); err != nil {
+		log.WithError(err).Errorf("eduserver output log: message parse failed (expected send-to-device)")
+		return nil
+	}
+
+	// only send send-to-device events which originated from us
+	_, originServerName, err := gomatrixserverlib.SplitID('@', ote.Sender)
+	if err != nil {
+		log.WithError(err).WithField("user_id", ote.Sender).Error("Failed to extract domain from send-to-device sender")
+		return nil
+	}
+	if originServerName != t.ServerName {
+		log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
+		return nil
+	}
+
+	_, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID)
+	if err != nil {
+		log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination")
+		return nil
+	}
+
+	// Pack the EDU and marshal it
+	edu := &gomatrixserverlib.EDU{
+		Type:   gomatrixserverlib.MDirectToDevice,
+		Origin: string(t.ServerName),
+	}
+	tdm := gomatrixserverlib.ToDeviceMessage{
+		Sender:    ote.Sender,
+		Type:      ote.Type,
+		MessageID: util.RandomString(32),
+		Messages: map[string]map[string]json.RawMessage{
+			ote.UserID: {
+				ote.DeviceID: ote.Content,
+			},
+		},
+	}
+	if edu.Content, err = json.Marshal(tdm); err != nil {
+		return err
+	}
+
+	log.Infof("Sending send-to-device message into %q destination queue", destServerName)
+	return t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName})
+}
+
+// onTypingEvent is called in response to a message received on the typing
+// events topic from the EDU server.
+func (t *OutputEDUConsumer) onTypingEvent(msg *sarama.ConsumerMessage) error {
 	// Extract the typing event from msg.
 	var ote api.OutputTypingEvent
 	if err := json.Unmarshal(msg.Value, &ote); err != nil {
 		// Skip this msg but continue processing messages.
-		log.WithError(err).Errorf("eduserver output log: message parse failed")
+		log.WithError(err).Errorf("eduserver output log: message parse failed (expected typing)")
 		return nil
 	}
 

federationsender/federationsender.go

@@ -66,7 +66,7 @@ func NewInternalAPI(
 		logrus.WithError(err).Panic("failed to start room server consumer")
 	}
 
-	tsConsumer := consumers.NewOutputTypingEventConsumer(
+	tsConsumer := consumers.NewOutputEDUConsumer(
 		base.Cfg, base.KafkaConsumer, queues, federationSenderDB,
 	)
 	if err := tsConsumer.Start(); err != nil {

federationsender/queue/queue.go

@@ -174,7 +174,7 @@ func (oqs *OutgoingQueues) SendInvite(
 	return nil
 }
 
-// SendEDU sends an EDU event to the destinations
+// SendEDU sends an EDU event to the destinations.
 func (oqs *OutgoingQueues) SendEDU(
 	e *gomatrixserverlib.EDU, origin gomatrixserverlib.ServerName,
 	destinations []gomatrixserverlib.ServerName,

sytest-whitelist

@@ -316,12 +316,12 @@ Can recv a device message using /sync
 Can recv device messages until they are acknowledged 
 Device messages with the same txn_id are deduplicated 
 Device messages wake up /sync 
-# TODO: separate PR for: Can recv device messages over federation 
-# TODO: separate PR for: Device messages over federation wake up /sync 
+Can recv device messages over federation 
+Device messages over federation wake up /sync 
 Can send messages with a wildcard device id 
 Can send messages with a wildcard device id to two devices 
 Wildcard device messages wake up /sync 
-# TODO: separate PR for: Wildcard device messages over federation wake up /sync 
+Wildcard device messages over federation wake up /sync 
 Can send a to-device message to two users which both receive it using /sync
 User can create and send/receive messages in a room with version 6
 local user can join room with version 6