Move receipt sending to own JetStream producer

clientapi/clientapi.go

@@ -55,7 +55,8 @@ func AddPublicRoutes(
 
 	syncProducer := &producers.SyncAPIProducer{
 		JetStream:         js,
-		Topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData),
+		TopicClientData:   cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData),
+		TopicReceiptEvent: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
 	}
 
 	routing.Setup(

clientapi/producers/syncapi.go

@@ -15,24 +15,28 @@
 package producers
 
 import (
+	"context"
 	"encoding/json"
+	"strconv"
 
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/nats-io/nats.go"
 	log "github.com/sirupsen/logrus"
 )
 
 // SyncAPIProducer produces events for the sync API server to consume
 type SyncAPIProducer struct {
-	Topic     string
+	TopicClientData   string
+	TopicReceiptEvent string
 	JetStream         nats.JetStreamContext
 }
 
 // SendData sends account data to the sync API server
 func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string, readMarker *eventutil.ReadMarkerJSON) error {
 	m := &nats.Msg{
-		Subject: p.Topic,
+		Subject: p.TopicClientData,
 		Header:  nats.Header{},
 	}
 	m.Header.Set(jetstream.UserID, userID)
@@ -52,8 +56,27 @@ func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string
 		"user_id":   userID,
 		"room_id":   roomID,
 		"data_type": dataType,
-	}).Tracef("Producing to topic '%s'", p.Topic)
+	}).Tracef("Producing to topic '%s'", p.TopicClientData)
 
 	_, err = p.JetStream.PublishMsg(m)
 	return err
 }
+
+func (p *SyncAPIProducer) SendReceipt(
+	ctx context.Context,
+	userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
+) error {
+	m := &nats.Msg{
+		Subject: p.TopicReceiptEvent,
+		Header:  nats.Header{},
+	}
+	m.Header.Set(jetstream.UserID, userID)
+	m.Header.Set(jetstream.RoomID, roomID)
+	m.Header.Set(jetstream.EventID, eventID)
+	m.Header.Set("type", receiptType)
+	m.Header.Set("timestamp", strconv.Itoa(int(timestamp)))
+
+	log.WithFields(log.Fields{}).Tracef("Producing to topic '%s'", p.TopicReceiptEvent)
+	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
+	return err
+}

clientapi/routing/account_data.go

@@ -23,7 +23,6 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/httputil"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/clientapi/producers"
-	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/userapi/api"
@@ -146,7 +145,7 @@ type fullyReadEvent struct {
 // SaveReadMarker implements POST /rooms/{roomId}/read_markers
 func SaveReadMarker(
 	req *http.Request,
-	userAPI api.UserInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI, eduAPI eduserverAPI.EDUServerInputAPI,
+	userAPI api.UserInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI,
 	syncProducer *producers.SyncAPIProducer, device *api.Device, roomID string,
 ) util.JSONResponse {
 	// Verify that the user is a member of this room
@@ -192,7 +191,7 @@ func SaveReadMarker(
 
 	// Handle the read receipt that may be included in the read marker
 	if r.Read != "" {
-		return SetReceipt(req, eduAPI, device, roomID, "m.read", r.Read)
+		return SetReceipt(req, syncProducer, device, roomID, "m.read", r.Read)
 	}
 
 	return util.JSONResponse{

clientapi/routing/receipt.go

@@ -19,21 +19,20 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/matrix-org/dendrite/clientapi/producers"
 	"github.com/matrix-org/gomatrixserverlib"
 
-	"github.com/matrix-org/dendrite/eduserver/api"
-
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/util"
 	"github.com/sirupsen/logrus"
 )
 
-func SetReceipt(req *http.Request, eduAPI api.EDUServerInputAPI, device *userapi.Device, roomId, receiptType, eventId string) util.JSONResponse {
+func SetReceipt(req *http.Request, syncProducer *producers.SyncAPIProducer, device *userapi.Device, roomID, receiptType, eventID string) util.JSONResponse {
 	timestamp := gomatrixserverlib.AsTimestamp(time.Now())
 	logrus.WithFields(logrus.Fields{
-		"roomId":      roomId,
+		"roomID":      roomID,
 		"receiptType": receiptType,
-		"eventId":     eventId,
+		"eventID":     eventID,
 		"userId":      device.UserID,
 		"timestamp":   timestamp,
 	}).Debug("Setting receipt")
@@ -43,7 +42,7 @@ func SetReceipt(req *http.Request, eduAPI api.EDUServerInputAPI, device *userapi
 		return util.MessageResponse(400, fmt.Sprintf("receipt type must be m.read not '%s'", receiptType))
 	}
 
-	if err := api.SendReceipt(req.Context(), eduAPI, device.UserID, roomId, eventId, receiptType, timestamp); err != nil {
+	if err := syncProducer.SendReceipt(req.Context(), device.UserID, roomID, eventID, receiptType, timestamp); err != nil {
 		return util.ErrorResponse(err)
 	}
 

clientapi/routing/routing.go

@@ -942,7 +942,7 @@ func Setup(
 			if err != nil {
 				return util.ErrorResponse(err)
 			}
-			return SaveReadMarker(req, userAPI, rsAPI, eduAPI, syncProducer, device, vars["roomID"])
+			return SaveReadMarker(req, userAPI, rsAPI, syncProducer, device, vars["roomID"])
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 
@@ -1297,7 +1297,7 @@ func Setup(
 				return util.ErrorResponse(err)
 			}
 
-			return SetReceipt(req, eduAPI, device, vars["roomId"], vars["receiptType"], vars["eventId"])
+			return SetReceipt(req, syncProducer, device, vars["roomId"], vars["receiptType"], vars["eventId"])
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 }

cmd/dendrite-polylith-multi/personalities/federationapi.go

@@ -29,7 +29,7 @@ func FederationAPI(base *basepkg.BaseDendrite, cfg *config.Dendrite) {
 	keyRing := fsAPI.KeyRing()
 
 	federationapi.AddPublicRoutes(
-		base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux,
+		base.ProcessContext, base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux,
 		&base.Cfg.FederationAPI, userAPI, federation, keyRing,
 		rsAPI, fsAPI, base.EDUServerClient(), keyAPI,
 		&base.Cfg.MSCs, nil,

eduserver/api/input.go

@@ -59,22 +59,6 @@ type InputSendToDeviceEventRequest struct {
 // InputSendToDeviceEventResponse is a response to InputSendToDeviceEventRequest
 type InputSendToDeviceEventResponse struct{}
 
-type InputReceiptEvent struct {
-	UserID    string                      `json:"user_id"`
-	RoomID    string                      `json:"room_id"`
-	EventID   string                      `json:"event_id"`
-	Type      string                      `json:"type"`
-	Timestamp gomatrixserverlib.Timestamp `json:"timestamp"`
-}
-
-// InputReceiptEventRequest is a request to EDUServerInputAPI
-type InputReceiptEventRequest struct {
-	InputReceiptEvent InputReceiptEvent `json:"input_receipt_event"`
-}
-
-// InputReceiptEventResponse is a response to InputReceiptEventRequest
-type InputReceiptEventResponse struct{}
-
 type InputCrossSigningKeyUpdateRequest struct {
 	CrossSigningKeyUpdate `json:"signing_keys"`
 }
@@ -94,10 +78,4 @@ type EDUServerInputAPI interface {
 		request *InputSendToDeviceEventRequest,
 		response *InputSendToDeviceEventResponse,
 	) error
-
-	InputReceiptEvent(
-		ctx context.Context,
-		request *InputReceiptEventRequest,
-		response *InputReceiptEventResponse,
-	) error
 }

eduserver/api/wrapper.go

@@ -67,22 +67,3 @@ func SendToDevice(
 	response := InputSendToDeviceEventResponse{}
 	return eduAPI.InputSendToDeviceEvent(ctx, &request, &response)
 }
-
-// SendReceipt sends a receipt event to EDU Server
-func SendReceipt(
-	ctx context.Context,
-	eduAPI EDUServerInputAPI, userID, roomID, eventID, receiptType string,
-	timestamp gomatrixserverlib.Timestamp,
-) error {
-	request := InputReceiptEventRequest{
-		InputReceiptEvent: InputReceiptEvent{
-			UserID:    userID,
-			RoomID:    roomID,
-			EventID:   eventID,
-			Type:      receiptType,
-			Timestamp: timestamp,
-		},
-	}
-	response := InputReceiptEventResponse{}
-	return eduAPI.InputReceiptEvent(ctx, &request, &response)
-}

eduserver/eduserver.go

@@ -50,7 +50,6 @@ func NewInternalAPI(
 		JetStream:                    js,
 		OutputTypingEventTopic:       cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
 		OutputSendToDeviceEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
-		OutputReceiptEventTopic:      cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
 		ServerName:                   cfg.Matrix.ServerName,
 	}
 }

eduserver/input/input.go

@@ -37,8 +37,6 @@ type EDUServerInputAPI struct {
 	OutputTypingEventTopic string
 	// The kafka topic to output new send to device events to.
 	OutputSendToDeviceEventTopic string
-	// The kafka topic to output new receipt events to
-	OutputReceiptEventTopic string
 	// kafka producer
 	JetStream nats.JetStreamContext
 	// Internal user query API
@@ -169,30 +167,3 @@ func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) e
 
 	return nil
 }
-
-// InputReceiptEvent implements api.EDUServerInputAPI
-// TODO: Intelligently batch requests sent by the same user (e.g wait a few milliseconds before emitting output events)
-func (t *EDUServerInputAPI) InputReceiptEvent(
-	ctx context.Context,
-	request *api.InputReceiptEventRequest,
-	response *api.InputReceiptEventResponse,
-) error {
-	logrus.WithFields(logrus.Fields{}).Tracef("Producing to topic '%s'", t.OutputReceiptEventTopic)
-	output := &api.OutputReceiptEvent{
-		UserID:    request.InputReceiptEvent.UserID,
-		RoomID:    request.InputReceiptEvent.RoomID,
-		EventID:   request.InputReceiptEvent.EventID,
-		Type:      request.InputReceiptEvent.Type,
-		Timestamp: request.InputReceiptEvent.Timestamp,
-	}
-	js, err := json.Marshal(output)
-	if err != nil {
-		return err
-	}
-
-	_, err = t.JetStream.PublishMsg(&nats.Msg{
-		Subject: t.OutputReceiptEventTopic,
-		Data:    js,
-	})
-	return err
-}

eduserver/inthttp/client.go

@@ -14,7 +14,6 @@ import (
 const (
 	EDUServerInputTypingEventPath       = "/eduserver/input"
 	EDUServerInputSendToDeviceEventPath = "/eduserver/sendToDevice"
-	EDUServerInputReceiptEventPath      = "/eduserver/receipt"
 )
 
 // NewEDUServerClient creates a EDUServerInputAPI implemented by talking to a HTTP POST API.
@@ -55,16 +54,3 @@ func (h *httpEDUServerInputAPI) InputSendToDeviceEvent(
 	apiURL := h.eduServerURL + EDUServerInputSendToDeviceEventPath
 	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
 }
-
-// InputSendToDeviceEvent implements EDUServerInputAPI
-func (h *httpEDUServerInputAPI) InputReceiptEvent(
-	ctx context.Context,
-	request *api.InputReceiptEventRequest,
-	response *api.InputReceiptEventResponse,
-) error {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "InputReceiptEventPath")
-	defer span.Finish()
-
-	apiURL := h.eduServerURL + EDUServerInputReceiptEventPath
-	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
-}

eduserver/inthttp/server.go

@@ -38,17 +38,4 @@ func AddRoutes(t api.EDUServerInputAPI, internalAPIMux *mux.Router) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
-	internalAPIMux.Handle(EDUServerInputReceiptEventPath,
-		httputil.MakeInternalAPI("inputReceiptEvent", func(req *http.Request) util.JSONResponse {
-			var request api.InputReceiptEventRequest
-			var response api.InputReceiptEventResponse
-			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
-				return util.MessageResponse(http.StatusBadRequest, err.Error())
-			}
-			if err := t.InputReceiptEvent(req.Context(), &request, &response); err != nil {
-				return util.ErrorResponse(err)
-			}
-			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
-		}),
-	)
 }

federationapi/consumers/eduserver.go

@@ -17,7 +17,9 @@ package consumers
 import (
 	"context"
 	"encoding/json"
+	"strconv"
 
+	"github.com/getsentry/sentry-go"
 	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/federationapi/queue"
 	"github.com/matrix-org/dendrite/federationapi/storage"
@@ -198,15 +200,25 @@ func (t *OutputEDUConsumer) onTypingEvent(ctx context.Context, msg *nats.Msg) bo
 // onReceiptEvent is called in response to a message received on the receipt
 // events topic from the EDU server.
 func (t *OutputEDUConsumer) onReceiptEvent(ctx context.Context, msg *nats.Msg) bool {
-	// Extract the typing event from msg.
+	receipt := api.OutputReceiptEvent{
-	var receipt api.OutputReceiptEvent
+		UserID:  msg.Header.Get(jetstream.UserID),
-	if err := json.Unmarshal(msg.Data, &receipt); err != nil {
+		RoomID:  msg.Header.Get(jetstream.RoomID),
-		// Skip this msg but continue processing messages.
+		EventID: msg.Header.Get(jetstream.EventID),
-		log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)")
+		Type:    msg.Header.Get("type"),
+	}
+
+	timestamp, err := strconv.Atoi(msg.Header.Get("timestamp"))
+	if err != nil {
+		// If the message was invalid, log it and move on to the next message in the stream
+		log.WithError(err).Errorf("EDU server output log: message parse failure")
+		sentry.CaptureException(err)
 		return true
 	}
 
+	receipt.Timestamp = gomatrixserverlib.Timestamp(timestamp)
+
 	// only send receipt events which originated from us
+	// TODO: We're consuming/producing on the same topic from the federation api, maybe add different topics?
 	_, receiptServerName, err := gomatrixserverlib.SplitID('@', receipt.UserID)
 	if err != nil {
 		log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender")

federationapi/federationapi.go

@@ -22,6 +22,7 @@ import (
 	"github.com/matrix-org/dendrite/federationapi/consumers"
 	"github.com/matrix-org/dendrite/federationapi/internal"
 	"github.com/matrix-org/dendrite/federationapi/inthttp"
+	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/federationapi/queue"
 	"github.com/matrix-org/dendrite/federationapi/statistics"
 	"github.com/matrix-org/dendrite/federationapi/storage"
@@ -31,6 +32,7 @@ import (
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/sirupsen/logrus"
 
@@ -46,6 +48,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.FederationInternalAPI) {
 
 // AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.
 func AddPublicRoutes(
+	process *process.ProcessContext,
 	fedRouter, keyRouter, wellKnownRouter *mux.Router,
 	cfg *config.FederationAPI,
 	userAPI userapi.UserInternalAPI,
@@ -58,11 +61,18 @@ func AddPublicRoutes(
 	mscCfg *config.MSCs,
 	servers federationAPI.ServersInRoomProvider,
 ) {
+
+	js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream)
+	producer := &producers.SyncAPIProducer{
+		JetStream:         js,
+		TopicReceiptEvent: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
+	}
+
 	routing.Setup(
 		fedRouter, keyRouter, wellKnownRouter, cfg, rsAPI,
 		eduAPI, federationAPI, keyRing,
 		federation, userAPI, keyAPI, mscCfg,
-		servers,
+		servers, producer,
 	)
 }
 

federationapi/federationapi_test.go

@@ -30,7 +30,7 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
 	fsAPI := base.FederationAPIHTTPClient()
 	// TODO: This is pretty fragile, as if anything calls anything on these nils this test will break.
 	// Unfortunately, it makes little sense to instantiate these dependencies when we just want to test routing.
-	federationapi.AddPublicRoutes(base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux, &cfg.FederationAPI, nil, nil, keyRing, nil, fsAPI, nil, nil, &cfg.MSCs, nil)
+	federationapi.AddPublicRoutes(base.ProcessContext, base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux, &cfg.FederationAPI, nil, nil, keyRing, nil, fsAPI, nil, nil, &cfg.MSCs, nil)
 	baseURL, cancel := test.ListenAndServe(t, base.PublicFederationAPIMux, true)
 	defer cancel()
 	serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))

federationapi/producers/syncapi.go

@@ -0,0 +1,50 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package producers
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	log "github.com/sirupsen/logrus"
+)
+
+// SyncAPIProducer produces events for the sync API server to consume
+type SyncAPIProducer struct {
+	TopicReceiptEvent string
+	JetStream         nats.JetStreamContext
+}
+
+func (p *SyncAPIProducer) SendReceipt(
+	ctx context.Context,
+	userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
+) error {
+	m := &nats.Msg{
+		Subject: p.TopicReceiptEvent,
+		Header:  nats.Header{},
+	}
+	m.Header.Set(jetstream.UserID, userID)
+	m.Header.Set(jetstream.RoomID, roomID)
+	m.Header.Set(jetstream.EventID, eventID)
+	m.Header.Set("type", receiptType)
+	m.Header.Set("timestamp", strconv.Itoa(int(timestamp)))
+
+	log.WithFields(log.Fields{}).Tracef("Producing to topic '%s'", p.TopicReceiptEvent)
+	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
+	return err
+}

federationapi/routing/routing.go

@@ -21,6 +21,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
 	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
+	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/httputil"
 	keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
@@ -52,6 +53,7 @@ func Setup(
 	keyAPI keyserverAPI.KeyInternalAPI,
 	mscCfg *config.MSCs,
 	servers federationAPI.ServersInRoomProvider,
+	producer *producers.SyncAPIProducer,
 ) {
 	v2keysmux := keyMux.PathPrefix("/v2").Subrouter()
 	v1fedmux := fedMux.PathPrefix("/v1").Subrouter()
@@ -116,7 +118,7 @@ func Setup(
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
 			return Send(
 				httpReq, request, gomatrixserverlib.TransactionID(vars["txnID"]),
-				cfg, rsAPI, eduAPI, keyAPI, keys, federation, mu, servers,
+				cfg, rsAPI, eduAPI, keyAPI, keys, federation, mu, servers, producer,
 			)
 		},
 	)).Methods(http.MethodPut, http.MethodOptions)

federationapi/routing/send.go

@@ -25,6 +25,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
 	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
+	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/internal"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -93,6 +94,7 @@ func Send(
 	federation *gomatrixserverlib.FederationClient,
 	mu *internal.MutexByRoom,
 	servers federationAPI.ServersInRoomProvider,
+	producer *producers.SyncAPIProducer,
 ) util.JSONResponse {
 	// First we should check if this origin has already submitted this
 	// txn ID to us. If they have and the txnIDs map contains an entry,
@@ -133,6 +135,7 @@ func Send(
 		servers:    servers,
 		keyAPI:     keyAPI,
 		roomsMu:    mu,
+		producer:   producer,
 	}
 
 	var txnEvents struct {
@@ -191,6 +194,7 @@ type txnReq struct {
 	federation txnFederationClient
 	roomsMu    *internal.MutexByRoom
 	servers    federationAPI.ServersInRoomProvider
+	producer   *producers.SyncAPIProducer
 }
 
 // A subset of FederationClient functionality that txn requires. Useful for testing.
@@ -430,17 +434,7 @@ func (t *txnReq) processReceiptEvent(ctx context.Context,
 ) error {
 	// store every event
 	for _, eventID := range eventIDs {
-		req := eduserverAPI.InputReceiptEventRequest{
+		if err := t.producer.SendReceipt(ctx, userID, roomID, eventID, receiptType, timestamp); err != nil {
-			InputReceiptEvent: eduserverAPI.InputReceiptEvent{
-				UserID:    userID,
-				RoomID:    roomID,
-				EventID:   eventID,
-				Type:      receiptType,
-				Timestamp: timestamp,
-			},
-		}
-		resp := eduserverAPI.InputReceiptEventResponse{}
-		if err := t.eduAPI.InputReceiptEvent(ctx, &req, &resp); err != nil {
 			return fmt.Errorf("unable to set receipt event: %w", err)
 		}
 	}

federationapi/routing/send_test.go

@@ -75,14 +75,6 @@ func (p *testEDUProducer) InputSendToDeviceEvent(
 	return nil
 }
 
-func (o *testEDUProducer) InputReceiptEvent(
-	ctx context.Context,
-	request *eduAPI.InputReceiptEventRequest,
-	response *eduAPI.InputReceiptEventResponse,
-) error {
-	return nil
-}
-
 func (o *testEDUProducer) InputCrossSigningKeyUpdate(
 	ctx context.Context,
 	request *eduAPI.InputCrossSigningKeyUpdateRequest,

setup/jetstream/streams.go

@@ -9,6 +9,7 @@ import (
 const (
 	UserID  = "user_id"
 	RoomID  = "room_id"
+	EventID = "event_id"
 )
 
 var (

setup/monolith.go

@@ -64,7 +64,7 @@ func (m *Monolith) AddAllPublicRoutes(process *process.ProcessContext, csMux, ss
 		m.ExtPublicRoomsProvider, &m.Config.MSCs,
 	)
 	federationapi.AddPublicRoutes(
-		ssMux, keyMux, wkMux, &m.Config.FederationAPI, m.UserAPI, m.FedClient,
+		process, ssMux, keyMux, wkMux, &m.Config.FederationAPI, m.UserAPI, m.FedClient,
 		m.KeyRing, m.RoomserverAPI, m.FederationAPI,
 		m.EDUInternalAPI, m.KeyAPI, &m.Config.MSCs, nil,
 	)

syncapi/consumers/receipts.go

@@ -17,8 +17,8 @@ package consumers
 import (
 	"context"
 	"database/sql"
-	"encoding/json"
 	"fmt"
+	"strconv"
 
 	"github.com/getsentry/sentry-go"
 	"github.com/matrix-org/dendrite/eduserver/api"
@@ -81,14 +81,23 @@ func (s *OutputReceiptEventConsumer) Start() error {
 }
 
 func (s *OutputReceiptEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
-	var output api.OutputReceiptEvent
+	output := api.OutputReceiptEvent{
-	if err := json.Unmarshal(msg.Data, &output); err != nil {
+		UserID:  msg.Header.Get(jetstream.UserID),
+		RoomID:  msg.Header.Get(jetstream.RoomID),
+		EventID: msg.Header.Get(jetstream.EventID),
+		Type:    msg.Header.Get("type"),
+	}
+
+	timestamp, err := strconv.Atoi(msg.Header.Get("timestamp"))
+	if err != nil {
 		// If the message was invalid, log it and move on to the next message in the stream
 		log.WithError(err).Errorf("EDU server output log: message parse failure")
 		sentry.CaptureException(err)
 		return true
 	}
 
+	output.Timestamp = gomatrixserverlib.Timestamp(timestamp)
+
 	streamPos, err := s.db.StoreReceipt(
 		s.ctx,
 		output.RoomID,