Fix storing and retrieving of send-to-device messages

clientapi/routing/sendtodevice.go

@@ -22,7 +22,6 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/producers"
 	"github.com/matrix-org/dendrite/internal/transactions"
 	"github.com/matrix-org/util"
-	"github.com/sirupsen/logrus"
 )
 
 // SendToDevice handles PUT /_matrix/client/r0/sendToDevice/{eventType}/{txnId}
@@ -56,14 +55,6 @@ func SendToDevice(
 			); err != nil {
 				util.GetLogger(req.Context()).WithError(err).Error("eduProducer.SendToDevice failed")
 				return jsonerror.InternalServerError()
-			} else {
-				util.GetLogger(req.Context()).WithFields(logrus.Fields{
-					"sender":     device.UserID,
-					"user_id":    userID,
-					"device_id":  deviceID,
-					"event_type": eventType,
-					"message":    string(message),
-				}).Info("client API processed send-to-device message")
 			}
 		}
 	}

eduserver/cache/cache.go

@@ -117,7 +117,7 @@ func (t *EDUCache) AddSendToDeviceMessage() int64 {
 	defer t.Unlock()
 
 	t.latestSyncPosition++
-	return t.latestSyncPosition
+	return t.latestSyncPosition - 1
 }
 
 // addUser with mutex lock & replace the previous timer.

syncapi/consumers/eduserver_sendtodevice.go

@@ -25,7 +25,7 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/sync"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/sirupsen/logrus"
+	"github.com/matrix-org/util"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -64,13 +64,10 @@ func NewOutputSendToDeviceEventConsumer(
 
 // Start consuming from EDU api
 func (s *OutputSendToDeviceEventConsumer) Start() error {
-	logrus.Info("syncapi starting sendToDevice consumer")
 	return s.sendToDeviceConsumer.Start()
 }
 
 func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
-	logrus.Info("syncapi received sendToDevice event")
-
 	var output api.OutputSendToDeviceEvent
 	if err := json.Unmarshal(msg.Value, &output); err != nil {
 		// If the message was invalid, log it and move on to the next message in the stream
@@ -78,13 +75,12 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 		return err
 	}
 
-	log.WithFields(log.Fields{
+	util.GetLogger(context.TODO()).WithFields(log.Fields{
 		"sender":     output.Sender,
 		"user_id":    output.UserID,
 		"device_id":  output.DeviceID,
 		"event_type": output.Type,
-		"content":    string(output.Content),
+	}).Info("sync API received send-to-device event from EDU server")
-	}).Debug("sync API received send-to-device event from EDU server")
 
 	newPos, err := s.db.StoreNewSendForDeviceMessage(
 		context.TODO(), output.UserID, output.DeviceID, output.SendToDeviceEvent,

syncapi/storage/shared/syncserver.go

@@ -1041,8 +1041,12 @@ func (d *Database) AddSendToDeviceEvent(
 func (d *Database) StoreNewSendForDeviceMessage(
 	ctx context.Context, userID, deviceID string, event gomatrixserverlib.SendToDeviceEvent,
 ) (types.StreamPosition, error) {
-	err := d.AddSendToDeviceEvent(
+	j, err := json.Marshal(event)
-		ctx, nil, userID, deviceID, string(event.Content),
+	if err != nil {
+		return 0, err
+	}
+	err = d.AddSendToDeviceEvent(
+		ctx, nil, userID, deviceID, string(j),
 	)
 	if err != nil {
 		return 0, err

syncapi/sync/requestpool.go

@@ -55,12 +55,13 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *authtype
 			JSON: jsonerror.Unknown(err.Error()),
 		}
 	}
+
 	logger := util.GetLogger(req.Context()).WithFields(log.Fields{
-		"userID":   device.UserID,
+		"user_id":   device.UserID,
-		"deviceID": device.ID,
+		"device_id": device.ID,
-		"since":    syncReq.since,
+		"since":     syncReq.since,
-		"timeout":  syncReq.timeout,
+		"timeout":   syncReq.timeout,
-		"limit":    syncReq.limit,
+		"limit":     syncReq.limit,
 	})
 
 	currPos := rp.notifier.CurrentPosition()