Refactor a bit

clientapi/producers/eduserver.go

@@ -56,7 +56,7 @@ func (p *EDUServerProducer) SendTyping(
 
 // SendToDevice sends a typing event to EDU server
 func (p *EDUServerProducer) SendToDevice(
-	ctx context.Context, userID, deviceID, eventType string,
+	ctx context.Context, sender, userID, deviceID, eventType string,
 	message interface{},
 ) error {
 	js, err := json.Marshal(message)
@@ -64,11 +64,12 @@ func (p *EDUServerProducer) SendToDevice(
 		return err
 	}
 	requestData := api.InputSendToDeviceEvent{
+		UserID:   userID,
+		DeviceID: deviceID,
 		SendToDeviceEvent: gomatrixserverlib.SendToDeviceEvent{
-			UserID:    userID,
+			Sender:  sender,
-			DeviceID:  deviceID,
+			Type:    eventType,
-			EventType: eventType,
+			Content: js,
-			Message:   js,
 		},
 	}
 	request := api.InputSendToDeviceEventRequest{

clientapi/routing/sendtodevice.go

@@ -51,7 +51,7 @@ func SendToDevice(
 	for userID, byUser := range httpReq.Messages {
 		for deviceID, message := range byUser {
 			if err := eduProducer.SendToDevice(
-				req.Context(), userID, deviceID, eventType, message,
+				req.Context(), device.UserID, userID, deviceID, eventType, message,
 			); err != nil {
 				util.GetLogger(req.Context()).WithError(err).Error("eduProducer.SendToDevice failed")
 				return jsonerror.InternalServerError()

eduserver/api/input.go

@@ -38,6 +38,8 @@ type InputTypingEvent struct {
 }
 
 type InputSendToDeviceEvent struct {
+	UserID   string `json:"user_id"`
+	DeviceID string `json:"device_id"`
 	gomatrixserverlib.SendToDeviceEvent
 }
 

eduserver/api/output.go

@@ -41,5 +41,7 @@ type TypingEvent struct {
 // This contains the event with extra fields used to create 'm.typing' event
 // in clientapi & federation.
 type OutputSendToDeviceEvent struct {
+	UserID   string `json:"user_id"`
+	DeviceID string `json:"device_id"`
 	gomatrixserverlib.SendToDeviceEvent
 }

eduserver/input/input.go

@@ -105,18 +105,15 @@ func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error {
 
 func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) error {
 	ote := &api.OutputSendToDeviceEvent{
-		SendToDeviceEvent: gomatrixserverlib.SendToDeviceEvent{
+		UserID:            ise.UserID,
-			UserID:    ise.UserID,
+		DeviceID:          ise.DeviceID,
-			DeviceID:  ise.DeviceID,
+		SendToDeviceEvent: ise.SendToDeviceEvent,
-			EventType: ise.EventType,
-			Message:   ise.Message,
-		},
 	}
 
 	logrus.WithFields(logrus.Fields{
 		"user_id":    ise.UserID,
 		"device_id":  ise.DeviceID,
-		"event_type": ise.EventType,
+		"event_type": ise.Type,
 	}).Info("handling send-to-device message")
 
 	eventJSON, err := json.Marshal(ote)

federationapi/routing/send.go

@@ -267,12 +267,7 @@ func (t *txnReq) processEDUs(edus []gomatrixserverlib.EDU) {
 			}
 		case gomatrixserverlib.MDirectToDevice:
 			// https://matrix.org/docs/spec/server_server/r0.1.3#m-direct-to-device-schema
-			var directPayload struct {
+			var directPayload gomatrixserverlib.ToDeviceMessage
-				Sender    string                                `json:"sender"`
-				EventType string                                `json:"type"`
-				MessageID string                                `json:"message_id"`
-				Messages  map[string]map[string]json.RawMessage `json:"message"`
-			}
 			if err := json.Unmarshal(e.Content, &directPayload); err != nil {
 				util.GetLogger(t.context).WithError(err).Error("Failed to unmarshal send-to-device events")
 				continue
@@ -280,7 +275,7 @@ func (t *txnReq) processEDUs(edus []gomatrixserverlib.EDU) {
 			for userID, byUser := range directPayload.Messages {
 				for deviceID, message := range byUser {
 					// TODO: check that the user and the device actually exist here
-					if err := t.eduProducer.SendToDevice(t.context, userID, deviceID, directPayload.EventType, message); err != nil {
+					if err := t.eduProducer.SendToDevice(t.context, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil {
 						util.GetLogger(t.context).WithError(err).Error("Failed to send send-to-device event to edu server")
 					}
 				}

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3
 	github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20200528131445-0aa540ad74d1
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20200528171922-504ef926060e
 	github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f
 	github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7
 	github.com/mattn/go-sqlite3 v2.0.2+incompatible

go.sum

@@ -360,6 +360,8 @@ github.com/matrix-org/gomatrixserverlib v0.0.0-20200528122156-fbb320a2ee61 h1:3r
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200528122156-fbb320a2ee61/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200528131445-0aa540ad74d1 h1:ueCm+xtIYsPn0MKsGvd61EC/mRnaThNOYEhmPXIPGr4=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200528131445-0aa540ad74d1/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20200528171922-504ef926060e h1:OivrqPGYy5NYqGZqf+DE9dX1xiJtJ0ip8TJgt3jqcrA=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20200528171922-504ef926060e/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
 github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y=
 github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f/go.mod h1:y0oDTjZDv5SM9a2rp3bl+CU+bvTRINQsdb7YlDql5Go=
 github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 h1:ntrLa/8xVzeSs8vHFHK25k0C+NV74sYMJnNSg5NoSRo=

syncapi/consumers/eduserver_sendtodevice.go

@@ -81,10 +81,12 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 	log.WithFields(log.Fields{
 		"user_id":    output.UserID,
 		"device_id":  output.DeviceID,
-		"event_type": output.EventType,
+		"event_type": output.Type,
 	}).Debug("received send-to-device event from EDU server")
 
-	newPos, err := s.db.StoreNewSendForDeviceMessage(context.TODO(), output.SendToDeviceEvent)
+	newPos, err := s.db.StoreNewSendForDeviceMessage(
+		context.TODO(), output.UserID, output.DeviceID, output.SendToDeviceEvent,
+	)
 	if err != nil {
 		log.WithError(err).Errorf("failed to store send-to-device message")
 		return err

syncapi/storage/interface.go

@@ -109,5 +109,5 @@ type Database interface {
 	// that we can clean up old events properly.
 	SendToDeviceUpdatesForSync(ctx context.Context, userID, deviceID string, token types.StreamingToken) ([]types.SendToDeviceEvent, error)
 	// StoreNewSendForDeviceMessage stores a new send-to-device event for a user's device.
-	StoreNewSendForDeviceMessage(ctx context.Context, event gomatrixserverlib.SendToDeviceEvent) (types.StreamPosition, error)
+	StoreNewSendForDeviceMessage(ctx context.Context, userID, deviceID string, event gomatrixserverlib.SendToDeviceEvent) (types.StreamPosition, error)
 }

syncapi/storage/postgres/send_to_device_table.go

@@ -23,7 +23,6 @@ import (
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 const sendToDeviceSchema = `
@@ -33,12 +32,12 @@ CREATE SEQUENCE IF NOT EXISTS syncapi_send_to_device_id;
 CREATE TABLE IF NOT EXISTS syncapi_send_to_device (
 	-- The ID that uniquely identifies this message.
 	id BIGINT PRIMARY KEY DEFAULT nextval('syncapi_send_to_device_id'),
+	-- The sender of the message.
+	sender TEXT NOT NULL,
 	-- The user ID to send the message to.
 	user_id TEXT NOT NULL,
 	-- The device ID to send the message to.
 	device_id TEXT NOT NULL,
-	-- The event type.
-	event_type TEXT NOT NULL,
 	-- The event content JSON.
 	content TEXT NOT NULL,
 	-- The sync token that was supplied when we tried to send the message,
@@ -48,12 +47,12 @@ CREATE TABLE IF NOT EXISTS syncapi_send_to_device (
 `
 
 const insertSendToDeviceMessageSQL = `
-	INSERT INTO syncapi_send_to_device (user_id, device_id, event_type, content)
+	INSERT INTO syncapi_send_to_device (user_id, device_id, content)
-	  VALUES ($1, $2, $3, $4)
+	  VALUES ($1, $2, $3)
 `
 
 const selectSendToDeviceMessagesSQL = `
-	SELECT id, user_id, device_id, event_type, content, sent_by_token
+	SELECT id, sender, user_id, device_id, content, sent_by_token
 	  FROM syncapi_send_to_device
 	  WHERE user_id = $1 AND device_id = $2
 `
@@ -96,9 +95,9 @@ func NewPostgresSendToDeviceTable(db *sql.DB) (tables.SendToDevice, error) {
 }
 
 func (s *sendToDeviceStatements) InsertSendToDeviceMessage(
-	ctx context.Context, txn *sql.Tx, userID, deviceID, eventType, content string,
+	ctx context.Context, txn *sql.Tx, userID, deviceID, content string,
 ) (err error) {
-	_, err = internal.TxStmt(txn, s.insertSendToDeviceMessageStmt).ExecContext(ctx, userID, deviceID, eventType, content)
+	_, err = internal.TxStmt(txn, s.insertSendToDeviceMessageStmt).ExecContext(ctx, userID, deviceID, content)
 	return
 }
 
@@ -113,19 +112,18 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
 
 	for rows.Next() {
 		var id types.SendToDeviceNID
-		var userID, deviceID, eventType, message string
+		var userID, deviceID, content string
 		var sentByToken *string
-		if err = rows.Scan(&id, &userID, &deviceID, &eventType, &message, &sentByToken); err != nil {
+		if err = rows.Scan(&id, &userID, &deviceID, &content, &sentByToken); err != nil {
 			return
 		}
 		event := types.SendToDeviceEvent{
-			ID: id,
+			ID:       id,
-			SendToDeviceEvent: gomatrixserverlib.SendToDeviceEvent{
+			UserID:   userID,
-				UserID:    userID,
+			DeviceID: deviceID,
-				DeviceID:  deviceID,
+		}
-				EventType: eventType,
+		if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil {
-				Message:   json.RawMessage(message),
+			return
-			},
 		}
 		if sentByToken != nil {
 			if token, err := types.NewStreamTokenFromString(*sentByToken); err == nil {

syncapi/storage/shared/syncserver.go

@@ -1031,18 +1031,18 @@ func (d *Database) currentStateStreamEventsForRoom(
 
 func (d *Database) AddSendToDeviceEvent(
 	ctx context.Context, txn *sql.Tx,
-	userID, deviceID, eventType, message string,
+	userID, deviceID, eventType, content string,
 ) error {
 	return d.SendToDevice.InsertSendToDeviceMessage(
-		ctx, txn, userID, deviceID, eventType, message,
+		ctx, txn, userID, deviceID, content,
 	)
 }
 
 func (d *Database) StoreNewSendForDeviceMessage(
-	ctx context.Context, event gomatrixserverlib.SendToDeviceEvent,
+	ctx context.Context, userID, deviceID string, event gomatrixserverlib.SendToDeviceEvent,
 ) (types.StreamPosition, error) {
 	err := d.AddSendToDeviceEvent(
-		ctx, nil, event.UserID, event.DeviceID, event.EventType, string(event.Message),
+		ctx, nil, userID, deviceID, event.Type, string(event.Content),
 	)
 	if err != nil {
 		return 0, err

syncapi/storage/sqlite3/send_to_device_table.go

@@ -23,7 +23,6 @@ import (
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 const sendToDeviceSchema = `
@@ -31,12 +30,12 @@ const sendToDeviceSchema = `
 CREATE TABLE IF NOT EXISTS syncapi_send_to_device (
 	-- The ID that uniquely identifies this message.
 	id INTEGER PRIMARY KEY AUTOINCREMENT,
+	-- The sender of the message.
+	sender TEXT NOT NULL,
 	-- The user ID to send the message to.
 	user_id TEXT NOT NULL,
 	-- The device ID to send the message to.
 	device_id TEXT NOT NULL,
-	-- The event type.
-	event_type TEXT NOT NULL,
 	-- The event content JSON.
 	content TEXT NOT NULL,
 	-- The sync token that was supplied when we tried to send the message,
@@ -46,12 +45,12 @@ CREATE TABLE IF NOT EXISTS syncapi_send_to_device (
 `
 
 const insertSendToDeviceMessageSQL = `
-	INSERT INTO syncapi_send_to_device (user_id, device_id, event_type, content)
+	INSERT INTO syncapi_send_to_device (user_id, device_id, content)
-	  VALUES ($1, $2, $3, $4)
+	  VALUES ($1, $2, $3)
 `
 
 const selectSendToDeviceMessagesSQL = `
-	SELECT id, user_id, device_id, event_type, content, sent_by_token
+	SELECT id, user_id, device_id, content, sent_by_token
 	  FROM syncapi_send_to_device
 	  WHERE user_id = $1 AND device_id = $2
 `
@@ -86,9 +85,9 @@ func NewSqliteSendToDeviceTable(db *sql.DB) (tables.SendToDevice, error) {
 }
 
 func (s *sendToDeviceStatements) InsertSendToDeviceMessage(
-	ctx context.Context, txn *sql.Tx, userID, deviceID, eventType, content string,
+	ctx context.Context, txn *sql.Tx, userID, deviceID, content string,
 ) (err error) {
-	_, err = internal.TxStmt(txn, s.insertSendToDeviceMessageStmt).ExecContext(ctx, userID, deviceID, eventType, content)
+	_, err = internal.TxStmt(txn, s.insertSendToDeviceMessageStmt).ExecContext(ctx, userID, deviceID, content)
 	return
 }
 
@@ -103,19 +102,18 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
 
 	for rows.Next() {
 		var id types.SendToDeviceNID
-		var userID, deviceID, eventType, message string
+		var sender, userID, deviceID, content string
 		var sentByToken *string
-		if err = rows.Scan(&id, &userID, &deviceID, &eventType, &message, &sentByToken); err != nil {
+		if err = rows.Scan(&id, &sender, &userID, &deviceID, &content, &sentByToken); err != nil {
 			return
 		}
 		event := types.SendToDeviceEvent{
-			ID: id,
+			ID:       id,
-			SendToDeviceEvent: gomatrixserverlib.SendToDeviceEvent{
+			UserID:   userID,
-				UserID:    userID,
+			DeviceID: deviceID,
-				DeviceID:  deviceID,
+		}
-				EventType: eventType,
+		if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil {
-				Message:   json.RawMessage(message),
+			return
-			},
 		}
 		if sentByToken != nil {
 			if token, err := types.NewStreamTokenFromString(*sentByToken); err == nil {

syncapi/storage/storage_test.go

@@ -528,11 +528,10 @@ func TestSendToDeviceBehaviour(t *testing.T) {
 	}
 
 	// Try sending a message.
-	streamPos, err := db.StoreNewSendForDeviceMessage(ctx, gomatrixserverlib.SendToDeviceEvent{
+	streamPos, err := db.StoreNewSendForDeviceMessage(ctx, "alice", "one", gomatrixserverlib.SendToDeviceEvent{
-		UserID:    "alice",
+		Sender:  "bob",
-		DeviceID:  "one",
+		Type:    "m.type",
-		EventType: "m.type",
+		Content: json.RawMessage("{}"),
-		Message:   json.RawMessage("{}"),
 	})
 	if err != nil {
 		t.Fatal(err)

syncapi/storage/tables/interface.go

@@ -96,7 +96,7 @@ type BackwardsExtremities interface {
 }
 
 type SendToDevice interface {
-	InsertSendToDeviceMessage(ctx context.Context, txn *sql.Tx, userID, deviceID, eventType, content string) (err error)
+	InsertSendToDeviceMessage(ctx context.Context, txn *sql.Tx, userID, deviceID, content string) (err error)
 	SelectSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string) (events []types.SendToDeviceEvent, err error)
 	UpdateSentSendToDeviceMessages(ctx context.Context, txn *sql.Tx, token string, nids []types.SendToDeviceNID) (err error)
 	DeleteSendToDeviceMessages(ctx context.Context, txn *sql.Tx, nids []types.SendToDeviceNID) (err error)

syncapi/types/types.go

@@ -404,5 +404,7 @@ type SendToDeviceNID int
 type SendToDeviceEvent struct {
 	gomatrixserverlib.SendToDeviceEvent
 	ID          SendToDeviceNID
+	UserID      string
+	DeviceID    string
 	SentByToken *StreamingToken
 }