Filter `/members`, return members at given point (#2827)

Makes the tests
```
Can get rooms/{roomId}/members at a given point
Can filter rooms/{roomId}/members
```
pass, by moving `/members` and `/joined_members` to the SyncAPI.

9 files changed, 257 insertions, 3 deletions

diff --git a/syncapi/routing/memberships.go b/syncapi/routing/memberships.go
new file mode 100644
index 00000000..b4e34225
--- /dev/null
+++ b/syncapi/routing/memberships.go
@@ -0,0 +1,131 @@
+// 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 routing
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/storage"
+	"github.com/matrix-org/dendrite/syncapi/types"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
+)
+
+type getMembershipResponse struct {
+	Chunk []gomatrixserverlib.ClientEvent `json:"chunk"`
+}
+
+// https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-rooms-roomid-joined-members
+type getJoinedMembersResponse struct {
+	Joined map[string]joinedMember `json:"joined"`
+}
+
+type joinedMember struct {
+	DisplayName string `json:"display_name"`
+	AvatarURL   string `json:"avatar_url"`
+}
+
+// The database stores 'displayname' without an underscore.
+// Deserialize into this and then change to the actual API response
+type databaseJoinedMember struct {
+	DisplayName string `json:"displayname"`
+	AvatarURL   string `json:"avatar_url"`
+}
+
+// GetMemberships implements
+//
+//	GET /rooms/{roomId}/members
+//	GET /rooms/{roomId}/joined_members
+func GetMemberships(
+	req *http.Request, device *userapi.Device, roomID string,
+	syncDB storage.Database, rsAPI api.SyncRoomserverAPI,
+	joinedOnly bool, membership, notMembership *string, at string,
+) util.JSONResponse {
+	queryReq := api.QueryMembershipForUserRequest{
+		RoomID: roomID,
+		UserID: device.UserID,
+	}
+
+	var queryRes api.QueryMembershipForUserResponse
+	if err := rsAPI.QueryMembershipForUser(req.Context(), &queryReq, &queryRes); err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("rsAPI.QueryMembershipsForRoom failed")
+		return jsonerror.InternalServerError()
+	}
+
+	if !queryRes.HasBeenInRoom {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("You aren't a member of the room and weren't previously a member of the room."),
+		}
+	}
+
+	db, err := syncDB.NewDatabaseSnapshot(req.Context())
+	if err != nil {
+		return jsonerror.InternalServerError()
+	}
+
+	atToken, err := types.NewTopologyTokenFromString(at)
+	if err != nil {
+		if queryRes.HasBeenInRoom && !queryRes.IsInRoom {
+			// If you have left the room then this will be the members of the room when you left.
+			atToken, err = db.EventPositionInTopology(req.Context(), queryRes.EventID)
+		} else {
+			// If you are joined to the room then this will be the current members of the room.
+			atToken, err = db.MaxTopologicalPosition(req.Context(), roomID)
+		}
+		if err != nil {
+			util.GetLogger(req.Context()).WithError(err).Error("unable to get 'atToken'")
+			return jsonerror.InternalServerError()
+		}
+	}
+
+	eventIDs, err := db.SelectMemberships(req.Context(), roomID, atToken, membership, notMembership)
+	if err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("db.SelectMemberships failed")
+		return jsonerror.InternalServerError()
+	}
+
+	result, err := db.Events(req.Context(), eventIDs)
+	if err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("db.Events failed")
+		return jsonerror.InternalServerError()
+	}
+
+	if joinedOnly {
+		var res getJoinedMembersResponse
+		res.Joined = make(map[string]joinedMember)
+		for _, ev := range result {
+			var content databaseJoinedMember
+			if err := json.Unmarshal(ev.Content(), &content); err != nil {
+				util.GetLogger(req.Context()).WithError(err).Error("failed to unmarshal event content")
+				return jsonerror.InternalServerError()
+			}
+			res.Joined[ev.Sender()] = joinedMember(content)
+		}
+		return util.JSONResponse{
+			Code: http.StatusOK,
+			JSON: res,
+		}
+	}
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: getMembershipResponse{gomatrixserverlib.HeaderedToClientEvents(result, gomatrixserverlib.FormatSync)},
+	}
+}
diff --git a/syncapi/routing/routing.go b/syncapi/routing/routing.go
index 71fa93c1..bc3ad238 100644
--- a/syncapi/routing/routing.go
+++ b/syncapi/routing/routing.go
@@ -172,4 +172,37 @@ func Setup(
 			return Search(req, device, syncDB, fts, nextBatch)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
+
+	v3mux.Handle("/rooms/{roomID}/members",
+		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+			var membership, notMembership *string
+			if req.URL.Query().Has("membership") {
+				m := req.URL.Query().Get("membership")
+				membership = &m
+			}
+			if req.URL.Query().Has("not_membership") {
+				m := req.URL.Query().Get("not_membership")
+				notMembership = &m
+			}
+
+			at := req.URL.Query().Get("at")
+			return GetMemberships(req, device, vars["roomID"], syncDB, rsAPI, false, membership, notMembership, at)
+		}),
+	).Methods(http.MethodGet, http.MethodOptions)
+
+	v3mux.Handle("/rooms/{roomID}/joined_members",
+		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+			at := req.URL.Query().Get("at")
+			membership := gomatrixserverlib.Join
+			return GetMemberships(req, device, vars["roomID"], syncDB, rsAPI, true, &membership, nil, at)
+		}),
+	).Methods(http.MethodGet, http.MethodOptions)
 }
diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go
index 02d45f80..af4fce44 100644
--- a/syncapi/storage/interface.go
+++ b/syncapi/storage/interface.go
@@ -178,6 +178,11 @@ type Database interface {
 	ReIndex(ctx context.Context, limit, afterID int64) (map[int64]gomatrixserverlib.HeaderedEvent, error)
 	UpdateRelations(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
 	RedactRelations(ctx context.Context, roomID, redactedEventID string) error
+	SelectMemberships(
+		ctx context.Context,
+		roomID string, pos types.TopologyToken,
+		membership, notMembership *string,
+	) (eventIDs []string, err error)
 }
 
 type Presence interface {
diff --git a/syncapi/storage/postgres/memberships_table.go b/syncapi/storage/postgres/memberships_table.go
index 939d6b3f..b555e845 100644
--- a/syncapi/storage/postgres/memberships_table.go
+++ b/syncapi/storage/postgres/memberships_table.go
@@ -20,11 +20,12 @@ import (
 	"fmt"
 
 	"github.com/lib/pq"
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // The memberships table is designed to track the last time that
@@ -69,11 +70,20 @@ const selectHeroesSQL = "" +
 const selectMembershipBeforeSQL = "" +
 	"SELECT membership, topological_pos FROM syncapi_memberships WHERE room_id = $1 and user_id = $2 AND topological_pos <= $3 ORDER BY topological_pos DESC LIMIT 1"
 
+const selectMembersSQL = `
+SELECT event_id FROM (
+	SELECT DISTINCT ON (room_id, user_id) room_id, user_id, event_id, membership FROM syncapi_memberships WHERE room_id = $1 AND topological_pos <= $2 ORDER BY room_id, user_id, stream_pos DESC  
+) t 
+WHERE ($3::text IS NULL OR t.membership = $3)
+	AND ($4::text IS NULL OR t.membership <> $4)
+`
+
 type membershipsStatements struct {
 	upsertMembershipStmt        *sql.Stmt
 	selectMembershipCountStmt   *sql.Stmt
 	selectHeroesStmt            *sql.Stmt
 	selectMembershipForUserStmt *sql.Stmt
+	selectMembersStmt           *sql.Stmt
 }
 
 func NewPostgresMembershipsTable(db *sql.DB) (tables.Memberships, error) {
@@ -87,6 +97,7 @@ func NewPostgresMembershipsTable(db *sql.DB) (tables.Memberships, error) {
 		{&s.selectMembershipCountStmt, selectMembershipCountSQL},
 		{&s.selectHeroesStmt, selectHeroesSQL},
 		{&s.selectMembershipForUserStmt, selectMembershipBeforeSQL},
+		{&s.selectMembersStmt, selectMembersSQL},
 	}.Prepare(db)
 }
 
@@ -154,3 +165,25 @@ func (s *membershipsStatements) SelectMembershipForUser(
 	}
 	return membership, topologyPos, nil
 }
+
+func (s *membershipsStatements) SelectMemberships(
+	ctx context.Context, txn *sql.Tx,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectMembersStmt)
+	rows, err := stmt.QueryContext(ctx, roomID, pos.Depth, membership, notMembership)
+	if err != nil {
+		return
+	}
+	var (
+		eventID string
+	)
+	for rows.Next() {
+		if err = rows.Scan(&eventID); err != nil {
+			return
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}
diff --git a/syncapi/storage/shared/storage_consumer.go b/syncapi/storage/shared/storage_consumer.go
index bf12203d..23f53d11 100644
--- a/syncapi/storage/shared/storage_consumer.go
+++ b/syncapi/storage/shared/storage_consumer.go
@@ -617,3 +617,11 @@ func (d *Database) RedactRelations(ctx context.Context, roomID, redactedEventID
 		return d.Relations.DeleteRelation(ctx, txn, roomID, redactedEventID)
 	})
 }
+
+func (d *Database) SelectMemberships(
+	ctx context.Context,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	return d.Memberships.SelectMemberships(ctx, nil, roomID, pos, membership, notMembership)
+}
diff --git a/syncapi/storage/sqlite3/memberships_table.go b/syncapi/storage/sqlite3/memberships_table.go
index 0c966fca..7e54fac1 100644
--- a/syncapi/storage/sqlite3/memberships_table.go
+++ b/syncapi/storage/sqlite3/memberships_table.go
@@ -20,11 +20,12 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // The memberships table is designed to track the last time that
@@ -69,12 +70,20 @@ const selectHeroesSQL = "" +
 const selectMembershipBeforeSQL = "" +
 	"SELECT membership, topological_pos FROM syncapi_memberships WHERE room_id = $1 and user_id = $2 AND topological_pos <= $3 ORDER BY topological_pos DESC LIMIT 1"
 
+const selectMembersSQL = `
+SELECT event_id FROM
+ ( SELECT event_id, membership FROM syncapi_memberships WHERE room_id = $1 AND topological_pos <= $2 GROUP BY user_id HAVING(max(stream_pos))) t
+    WHERE ($3 IS NULL OR t.membership = $3)
+		 	AND ($4 IS NULL OR t.membership <> $4)
+`
+
 type membershipsStatements struct {
 	db                        *sql.DB
 	upsertMembershipStmt      *sql.Stmt
 	selectMembershipCountStmt *sql.Stmt
 	//selectHeroesStmt          *sql.Stmt - prepared at runtime due to variadic
 	selectMembershipForUserStmt *sql.Stmt
+	selectMembersStmt           *sql.Stmt
 }
 
 func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
@@ -89,6 +98,7 @@ func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
 		{&s.upsertMembershipStmt, upsertMembershipSQL},
 		{&s.selectMembershipCountStmt, selectMembershipCountSQL},
 		{&s.selectMembershipForUserStmt, selectMembershipBeforeSQL},
+		{&s.selectMembersStmt, selectMembersSQL},
 		// {&s.selectHeroesStmt, selectHeroesSQL}, - prepared at runtime due to variadic
 	}.Prepare(db)
 }
@@ -170,3 +180,23 @@ func (s *membershipsStatements) SelectMembershipForUser(
 	}
 	return membership, topologyPos, nil
 }
+
+func (s *membershipsStatements) SelectMemberships(
+	ctx context.Context, txn *sql.Tx,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectMembersStmt)
+	rows, err := stmt.QueryContext(ctx, roomID, pos.Depth, membership, notMembership)
+	if err != nil {
+		return
+	}
+	var eventID string
+	for rows.Next() {
+		if err = rows.Scan(&eventID); err != nil {
+			return
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index e48c050d..2c4f04ec 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -187,6 +187,11 @@ type Memberships interface {
 	SelectMembershipCount(ctx context.Context, txn *sql.Tx, roomID, membership string, pos types.StreamPosition) (count int, err error)
 	SelectHeroes(ctx context.Context, txn *sql.Tx, roomID, userID string, memberships []string) (heroes []string, err error)
 	SelectMembershipForUser(ctx context.Context, txn *sql.Tx, roomID, userID string, pos int64) (membership string, topologicalPos int, err error)
+	SelectMemberships(
+		ctx context.Context, txn *sql.Tx,
+		roomID string, pos types.TopologyToken,
+		membership, notMembership *string,
+	) (eventIDs []string, err error)
 }
 
 type NotificationData interface {
diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go
index 9ec2b61c..707dbe8d 100644
--- a/syncapi/streams/stream_pdu.go
+++ b/syncapi/streams/stream_pdu.go
@@ -473,7 +473,13 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	var prevBatch *types.TopologyToken
 	if len(recentStreamEvents) > 0 {
 		var backwardTopologyPos, backwardStreamPos types.StreamPosition
-		backwardTopologyPos, backwardStreamPos, err = snapshot.PositionInTopology(ctx, recentStreamEvents[0].EventID())
+		event := recentStreamEvents[0]
+		// If this is the beginning of the room, we can't go back further. We're going to return
+		// the TopologyToken from the last event instead. (Synapse returns the /sync next_Batch)
+		if event.Type() == gomatrixserverlib.MRoomCreate && event.StateKeyEquals("") {
+			event = recentStreamEvents[len(recentStreamEvents)-1]
+		}
+		backwardTopologyPos, backwardStreamPos, err = snapshot.PositionInTopology(ctx, event.EventID())
 		if err != nil {
 			return
 		}
diff --git a/syncapi/types/types.go b/syncapi/types/types.go
index 57ce7b6f..295187ac 100644
--- a/syncapi/types/types.go
+++ b/syncapi/types/types.go
@@ -234,6 +234,9 @@ func (t *TopologyToken) StreamToken() StreamingToken {
 }
 
 func (t TopologyToken) String() string {
+	if t.Depth <= 0 && t.PDUPosition <= 0 {
+		return ""
+	}
 	return fmt.Sprintf("t%d_%d", t.Depth, t.PDUPosition)
 }