diff --git a/clientapi/clientapi.go b/clientapi/clientapi.go index 75184d3b7..1a2bf3833 100644 --- a/clientapi/clientapi.go +++ b/clientapi/clientapi.go @@ -54,8 +54,9 @@ func AddPublicRoutes( js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream) syncProducer := &producers.SyncAPIProducer{ - JetStream: js, - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + JetStream: js, + TopicClientData: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + TopicReceiptEvent: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), } routing.Setup( diff --git a/clientapi/producers/syncapi.go b/clientapi/producers/syncapi.go index 9ab90391d..034fbb317 100644 --- a/clientapi/producers/syncapi.go +++ b/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 - JetStream nats.JetStreamContext + 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 +} diff --git a/clientapi/routing/account_data.go b/clientapi/routing/account_data.go index d8e982690..873ffaf5d 100644 --- a/clientapi/routing/account_data.go +++ b/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{ diff --git a/clientapi/routing/receipt.go b/clientapi/routing/receipt.go index fe8fe765d..0f9b1b4ff 100644 --- a/clientapi/routing/receipt.go +++ b/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) } diff --git a/clientapi/routing/routing.go b/clientapi/routing/routing.go index d22fbd809..a53328ec6 100644 --- a/clientapi/routing/routing.go +++ b/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) } diff --git a/cmd/dendrite-polylith-multi/personalities/federationapi.go b/cmd/dendrite-polylith-multi/personalities/federationapi.go index 44357d660..17838a434 100644 --- a/cmd/dendrite-polylith-multi/personalities/federationapi.go +++ b/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, diff --git a/eduserver/api/input.go b/eduserver/api/input.go index 2aab107b2..85c367d78 100644 --- a/eduserver/api/input.go +++ b/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 } diff --git a/eduserver/api/wrapper.go b/eduserver/api/wrapper.go index 7907f4d39..c2c4596de 100644 --- a/eduserver/api/wrapper.go +++ b/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) -} diff --git a/eduserver/eduserver.go b/eduserver/eduserver.go index 9fe8704cf..88f6ebef7 100644 --- a/eduserver/eduserver.go +++ b/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, } } diff --git a/eduserver/input/input.go b/eduserver/input/input.go index e58f0dd34..ad8ac320c 100644 --- a/eduserver/input/input.go +++ b/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 -} diff --git a/eduserver/inthttp/client.go b/eduserver/inthttp/client.go index 0690ed827..7d0bc1603 100644 --- a/eduserver/inthttp/client.go +++ b/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) -} diff --git a/eduserver/inthttp/server.go b/eduserver/inthttp/server.go index a34943750..e374513a3 100644 --- a/eduserver/inthttp/server.go +++ b/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} - }), - ) } diff --git a/federationapi/consumers/eduserver.go b/federationapi/consumers/eduserver.go index 1f81fa258..8ea31b68a 100644 --- a/federationapi/consumers/eduserver.go +++ b/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. - var receipt api.OutputReceiptEvent - if err := json.Unmarshal(msg.Data, &receipt); err != nil { - // Skip this msg but continue processing messages. - log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)") + receipt := api.OutputReceiptEvent{ + 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 } + 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") diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go index b7f93ecb9..057d76e8d 100644 --- a/federationapi/federationapi.go +++ b/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, ) } diff --git a/federationapi/federationapi_test.go b/federationapi/federationapi_test.go index c660f12e0..3eef5daf1 100644 --- a/federationapi/federationapi_test.go +++ b/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://")) diff --git a/federationapi/producers/syncapi.go b/federationapi/producers/syncapi.go new file mode 100644 index 000000000..5d05f36fd --- /dev/null +++ b/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 +} diff --git a/federationapi/routing/routing.go b/federationapi/routing/routing.go index 04c88d957..bcb978add 100644 --- a/federationapi/routing/routing.go +++ b/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) diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go index 745e36de9..95f2bc4cc 100644 --- a/federationapi/routing/send.go +++ b/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{ - 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 { + if err := t.producer.SendReceipt(ctx, userID, roomID, eventID, receiptType, timestamp); err != nil { return fmt.Errorf("unable to set receipt event: %w", err) } } diff --git a/federationapi/routing/send_test.go b/federationapi/routing/send_test.go index 4280643e9..cd9f8e454 100644 --- a/federationapi/routing/send_test.go +++ b/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, diff --git a/setup/jetstream/streams.go b/setup/jetstream/streams.go index aa3e95cb8..4d7668955 100644 --- a/setup/jetstream/streams.go +++ b/setup/jetstream/streams.go @@ -7,8 +7,9 @@ import ( ) const ( - UserID = "user_id" - RoomID = "room_id" + UserID = "user_id" + RoomID = "room_id" + EventID = "event_id" ) var ( diff --git a/setup/monolith.go b/setup/monolith.go index fa6d962c4..88bc27166 100644 --- a/setup/monolith.go +++ b/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, ) diff --git a/syncapi/consumers/eduserver_receipts.go b/syncapi/consumers/receipts.go similarity index 92% rename from syncapi/consumers/eduserver_receipts.go rename to syncapi/consumers/receipts.go index 4e4c61c67..71b2f9bb5 100644 --- a/syncapi/consumers/eduserver_receipts.go +++ b/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 - if err := json.Unmarshal(msg.Data, &output); err != nil { + output := api.OutputReceiptEvent{ + 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,