CS API: Support for /messages, fixes for /sync (#847)

* Merge forward

* Tidy up a bit

* TODO: What to do with NextBatch here?

* Replace SyncPosition with PaginationToken throughout syncapi

* Fix PaginationTokens

* Fix lint errors

* Add a couple of missing functions into the syncapi external storage interface

* Some updates based on review comments from @babolivier

* Some updates based on review comments from @babolivier

* argh whitespacing

* Fix opentracing span

* Remove dead code

* Don't overshadow err (fix lint issue)

* Handle extremities after inserting event into topology

* Try insert event topology as ON CONFLICT DO NOTHING

* Prevent OOB error in addRoomDeltaToResponse

* Thwarted by gocyclo again

* Fix NewPaginationTokenFromString, define unit test for it

* Update pagination token test

* Update sytest-whitelist

* Hopefully fix some of the sync batch tokens

* Remove extraneous sync position func

* Revert to topology tokens in addRoomDeltaToResponse etc

* Fix typo

* Remove prevPDUPos as dead now that backwardTopologyPos is used instead

* Fix selectEventsWithEventIDsSQL

* Update sytest-blacklist

* Update sytest-whitelist

1 files changed, 49 insertions, 0 deletions

diff --git a/roomserver/query/query.go b/roomserver/query/query.go
index caa7a95b..da8fe23e 100644
--- a/roomserver/query/query.go
+++ b/roomserver/query/query.go
@@ -660,6 +660,41 @@ func getAuthChain(
 	return authEvents, nil
 }
 
+// QueryServersInRoomAtEvent implements api.RoomserverQueryAPI
+func (r *RoomserverQueryAPI) QueryServersInRoomAtEvent(
+	ctx context.Context,
+	request *api.QueryServersInRoomAtEventRequest,
+	response *api.QueryServersInRoomAtEventResponse,
+) error {
+	// getMembershipsBeforeEventNID requires a NID, so retrieving the NID for
+	// the event is necessary.
+	NIDs, err := r.DB.EventNIDs(ctx, []string{request.EventID})
+	if err != nil {
+		return err
+	}
+
+	// Retrieve all "m.room.member" state events of "join" membership, which
+	// contains the list of users in the room before the event, therefore all
+	// the servers in it at that moment.
+	events, err := r.getMembershipsBeforeEventNID(ctx, NIDs[request.EventID], true)
+	if err != nil {
+		return err
+	}
+
+	// Store the server names in a temporary map to avoid duplicates.
+	servers := make(map[gomatrixserverlib.ServerName]bool)
+	for _, event := range events {
+		servers[event.Origin()] = true
+	}
+
+	// Populate the response.
+	for server := range servers {
+		response.Servers = append(response.Servers, server)
+	}
+
+	return nil
+}
+
 // SetupHTTP adds the RoomserverQueryAPI handlers to the http.ServeMux.
 // nolint: gocyclo
 func (r *RoomserverQueryAPI) SetupHTTP(servMux *http.ServeMux) {
@@ -803,4 +838,18 @@ func (r *RoomserverQueryAPI) SetupHTTP(servMux *http.ServeMux) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	servMux.Handle(
+		api.RoomserverQueryServersInRoomAtEventPath,
+		common.MakeInternalAPI("QueryServersInRoomAtEvent", func(req *http.Request) util.JSONResponse {
+			var request api.QueryServersInRoomAtEventRequest
+			var response api.QueryServersInRoomAtEventResponse
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.ErrorResponse(err)
+			}
+			if err := r.QueryServersInRoomAtEvent(req.Context(), &request, &response); err != nil {
+				return util.ErrorResponse(err)
+			}
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 }