Only care about wildcard when targeted locally

cmd/dendrite-client-api-server/main.go

@@ -39,7 +39,7 @@ func main() {
 	rsAPI := base.CreateHTTPRoomserverAPIs()
 	fsAPI := base.CreateHTTPFederationSenderAPIs()
 	rsAPI.SetFederationSenderAPI(fsAPI)
-	eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New())
+	eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New(), deviceDB)
 
 	clientapi.SetupClientAPIComponent(
 		base, deviceDB, accountDB, federation, keyRing,

cmd/dendrite-demo-libp2p/main.go

@@ -148,7 +148,7 @@ func main() {
 		&base.Base, keyRing, federation,
 	)
 	eduInputAPI := eduserver.SetupEDUServerComponent(
-		&base.Base, cache.New(),
+		&base.Base, cache.New(), deviceDB,
 	)
 	asAPI := appservice.SetupAppServiceAPIComponent(
 		&base.Base, accountDB, deviceDB, federation, rsAPI, transactions.New(),

cmd/dendrite-edu-server/main.go

@@ -29,8 +29,9 @@ func main() {
 			logrus.WithError(err).Warn("BaseDendrite close failed")
 		}
 	}()
+	deviceDB := base.CreateDeviceDB()
 
-	eduserver.SetupEDUServerComponent(base, cache.New())
+	eduserver.SetupEDUServerComponent(base, cache.New(), deviceDB)
 
 	base.SetupAndServeHTTP(string(base.Cfg.Bind.EDUServer), string(base.Cfg.Listen.EDUServer))
 

cmd/dendrite-federation-api-server/main.go

@@ -39,7 +39,7 @@ func main() {
 	rsAPI := base.CreateHTTPRoomserverAPIs()
 	asAPI := base.CreateHTTPAppServiceAPIs()
 	rsAPI.SetFederationSenderAPI(fsAPI)
-	eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New())
+	eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New(), deviceDB)
 	eduProducer := producers.NewEDUServerProducer(eduInputAPI)
 
 	federationapi.SetupFederationAPIComponent(

cmd/dendrite-monolith-server/main.go

@@ -87,7 +87,7 @@ func main() {
 	}
 
 	eduInputAPI := eduserver.SetupEDUServerComponent(
-		base, cache.New(),
+		base, cache.New(), deviceDB,
 	)
 	if base.EnableHTTPAPIs {
 		eduInputAPI = base.CreateHTTPEDUServerAPIs()

eduserver/eduserver.go

@@ -13,6 +13,7 @@
 package eduserver
 
 import (
+	"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
 	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/eduserver/cache"
 	"github.com/matrix-org/dendrite/eduserver/input"
@@ -26,12 +27,15 @@ import (
 func SetupEDUServerComponent(
 	base *basecomponent.BaseDendrite,
 	eduCache *cache.EDUCache,
+	deviceDB devices.Database,
 ) api.EDUServerInputAPI {
 	inputAPI := &input.EDUServerInputAPI{
 		Cache:                        eduCache,
+		DeviceDB:                     deviceDB,
 		Producer:                     base.KafkaProducer,
 		OutputTypingEventTopic:       string(base.Cfg.Kafka.Topics.OutputTypingEvent),
 		OutputSendToDeviceEventTopic: string(base.Cfg.Kafka.Topics.OutputSendToDeviceEventTopic),
+		ServerName:                   base.Cfg.Matrix.ServerName,
 	}
 
 	inputAPI.SetupHTTP(base.InternalAPIMux)

eduserver/input/input.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/gorilla/mux"
+	"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
 	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/eduserver/cache"
 	"github.com/matrix-org/dendrite/internal"
@@ -38,6 +39,10 @@ type EDUServerInputAPI struct {
 	OutputSendToDeviceEventTopic string
 	// kafka producer
 	Producer sarama.SyncProducer
+	// device database
+	DeviceDB devices.Database
+	// our server name
+	ServerName gomatrixserverlib.ServerName
 }
 
 // InputTypingEvent implements api.EDUServerInputAPI
@@ -104,35 +109,57 @@ func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error {
 }
 
 func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) error {
-	ote := &api.OutputSendToDeviceEvent{
+	devices := []string{}
-		UserID:            ise.UserID,
+	localpart, domain, err := gomatrixserverlib.SplitID('@', ise.UserID)
-		DeviceID:          ise.DeviceID,
-		SendToDeviceEvent: ise.SendToDeviceEvent,
-	}
-
-	logrus.WithFields(logrus.Fields{
-		"user_id":    ise.UserID,
-		"device_id":  ise.DeviceID,
-		"event_type": ise.Type,
-	}).Info("handling send-to-device message")
-
-	eventJSON, err := json.Marshal(ote)
 	if err != nil {
-		logrus.WithError(err).Error("sendToDevice failed json.Marshal")
 		return err
 	}
 
-	m := &sarama.ProducerMessage{
+	if domain == t.ServerName && ise.DeviceID == "*" {
-		Topic: string(t.OutputSendToDeviceEventTopic),
+		devs, err := t.DeviceDB.GetDevicesByLocalpart(context.TODO(), localpart)
-		Key:   sarama.StringEncoder(ote.UserID),
+		if err != nil {
-		Value: sarama.ByteEncoder(eventJSON),
+			return err
+		}
+		for _, dev := range devs {
+			devices = append(devices, dev.ID)
+		}
+	} else {
+		devices = append(devices, ise.DeviceID)
 	}
 
-	_, _, err = t.Producer.SendMessage(m)
+	for _, device := range devices {
-	if err != nil {
+		ote := &api.OutputSendToDeviceEvent{
-		logrus.WithError(err).Error("sendToDevice failed t.Producer.SendMessage")
+			UserID:            ise.UserID,
+			DeviceID:          device,
+			SendToDeviceEvent: ise.SendToDeviceEvent,
+		}
+
+		logrus.WithFields(logrus.Fields{
+			"user_id":    ise.UserID,
+			"device_id":  ise.DeviceID,
+			"event_type": ise.Type,
+		}).Info("handling send-to-device message")
+
+		eventJSON, err := json.Marshal(ote)
+		if err != nil {
+			logrus.WithError(err).Error("sendToDevice failed json.Marshal")
+			return err
+		}
+
+		m := &sarama.ProducerMessage{
+			Topic: string(t.OutputSendToDeviceEventTopic),
+			Key:   sarama.StringEncoder(ote.UserID),
+			Value: sarama.ByteEncoder(eventJSON),
+		}
+
+		_, _, err = t.Producer.SendMessage(m)
+		if err != nil {
+			logrus.WithError(err).Error("sendToDevice failed t.Producer.SendMessage")
+			return err
+		}
 	}
-	return err
+
+	return nil
 }
 
 // SetupHTTP adds the EDUServerInputAPI handlers to the http.ServeMux.

syncapi/consumers/eduserver_sendtodevice.go

@@ -19,7 +19,6 @@ import (
 	"encoding/json"
 
 	"github.com/Shopify/sarama"
-	"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
 	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/config"
@@ -35,7 +34,6 @@ import (
 type OutputSendToDeviceEventConsumer struct {
 	sendToDeviceConsumer *internal.ContinualConsumer
 	db                   storage.Database
-	deviceDB             devices.Database
 	serverName           gomatrixserverlib.ServerName // our server name
 	notifier             *sync.Notifier
 }
@@ -47,7 +45,6 @@ func NewOutputSendToDeviceEventConsumer(
 	kafkaConsumer sarama.Consumer,
 	n *sync.Notifier,
 	store storage.Database,
-	deviceDB devices.Database,
 ) *OutputSendToDeviceEventConsumer {
 
 	consumer := internal.ContinualConsumer{
@@ -59,7 +56,6 @@ func NewOutputSendToDeviceEventConsumer(
 	s := &OutputSendToDeviceEventConsumer{
 		sendToDeviceConsumer: &consumer,
 		db:                   store,
-		deviceDB:             deviceDB,
 		serverName:           cfg.Matrix.ServerName,
 		notifier:             n,
 	}
@@ -82,7 +78,7 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 		return err
 	}
 
-	localpart, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
+	_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
 	if err != nil {
 		return err
 	}
@@ -107,22 +103,9 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 		return err
 	}
 
-	devices := []string{}
-	if output.DeviceID == "*" {
-		devs, err := s.deviceDB.GetDevicesByLocalpart(context.TODO(), localpart)
-		if err != nil {
-			return err
-		}
-		for _, dev := range devs {
-			devices = append(devices, dev.ID)
-		}
-	} else {
-		devices = append(devices, output.DeviceID)
-	}
-
 	s.notifier.OnNewSendToDevice(
 		output.UserID,
-		devices,
+		[]string{output.DeviceID},
 		types.NewStreamToken(0, streamPos),
 	)
 

syncapi/syncapi.go

@@ -82,7 +82,7 @@ func SetupSyncAPIComponent(
 	}
 
 	sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer(
-		base.Cfg, base.KafkaConsumer, notifier, syncDB, deviceDB,
+		base.Cfg, base.KafkaConsumer, notifier, syncDB,
 	)
 	if err = sendToDeviceConsumer.Start(); err != nil {
 		logrus.WithError(err).Panicf("failed to start send-to-device consumer")

sytest-whitelist

@@ -300,3 +300,4 @@ 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 
 Wildcard device messages over federation wake up /sync 
+Can send a to-device message to two users which both receive it using /sync