Filter `/members`, return members at given point
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.
ADD
clientapi/routing/joined_rooms.go
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// 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 (
+ "net/http"
+
+ "github.com/matrix-org/util"
+
+ "github.com/matrix-org/dendrite/clientapi/jsonerror"
+ "github.com/matrix-org/dendrite/roomserver/api"
+ userapi "github.com/matrix-org/dendrite/userapi/api"
+)
+
+type getJoinedRoomsResponse struct {
+ JoinedRooms []string `json:"joined_rooms"`
+}
+
+func GetJoinedRooms(
+ req *http.Request,
+ device *userapi.Device,
+ rsAPI api.ClientRoomserverAPI,
+) util.JSONResponse {
+ var res api.QueryRoomsForUserResponse
+ err := rsAPI.QueryRoomsForUser(req.Context(), &api.QueryRoomsForUserRequest{
+ UserID: device.UserID,
+ WantMembership: "join",
+ }, &res)
+ if err != nil {
+ util.GetLogger(req.Context()).WithError(err).Error("QueryRoomsForUser failed")
+ return jsonerror.InternalServerError()
+ }
+ if res.RoomIDs == nil {
+ res.RoomIDs = []string{}
+ }
+ return util.JSONResponse{
+ Code: http.StatusOK,
+ JSON: getJoinedRoomsResponse{res.RoomIDs},
+ }
+}

MODIFY
clientapi/routing/routing.go
clientapi/routing/routing.go
@@ -950,26 +950,6 @@ func Setup(
}),
).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)
- }
- return GetMemberships(req, device, vars["roomID"], false, cfg, rsAPI)
- }),
- ).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)
- }
- return GetMemberships(req, device, vars["roomID"], true, cfg, rsAPI)
- }),
- ).Methods(http.MethodGet, http.MethodOptions)
-
v3mux.Handle("/rooms/{roomID}/read_markers",
httputil.MakeAuthAPI("rooms_read_markers", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
if r := rateLimits.Limit(req, device); r != nil {

MODIFY
docs/caddy/polylith/Caddyfile
docs/caddy/polylith/Caddyfile
@@ -74,7 +74,7 @@ matrix.example.com {
# Change the end of each reverse_proxy line to the correct
# address for your various services.
@sync_api {
- path_regexp /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$
+ path_regexp /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$
}
reverse_proxy @sync_api sync_api:8073

MODIFY
docs/hiawatha/polylith-sample.conf
docs/hiawatha/polylith-sample.conf
@@ -23,8 +23,10 @@ VirtualHost {
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}/{eventType}
+ # /_matrix/client/.*/rooms/{roomId}/members
+ # /_matrix/client/.*/rooms/{roomId}/joined_members
# to sync_api
- ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$ http://localhost:8073 600
+ ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$ http://localhost:8073 600
ReverseProxy = /_matrix/client http://localhost:8071 600
ReverseProxy = /_matrix/federation http://localhost:8072 600
ReverseProxy = /_matrix/key http://localhost:8072 600

MODIFY
docs/nginx/polylith-sample.conf
docs/nginx/polylith-sample.conf
@@ -33,8 +33,10 @@ server {
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}
# /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}/{eventType}
+ # /_matrix/client/.*/rooms/{roomId}/members
+ # /_matrix/client/.*/rooms/{roomId}/joined_members
# to sync_api
- location ~ /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$ {
+ location ~ /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$ {
proxy_pass http://sync_api:8073;
}

RENAME
clientapi/routing/memberships.go
syncapi/routing/memberships.go
@@ -18,22 +18,20 @@ 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/setup/config"
+ "github.com/matrix-org/dendrite/syncapi/storage"
+ "github.com/matrix-org/dendrite/syncapi/types"
userapi "github.com/matrix-org/dendrite/userapi/api"
- "github.com/matrix-org/gomatrixserverlib"
- "github.com/matrix-org/util"
)
type getMembershipResponse struct {
Chunk []gomatrixserverlib.ClientEvent `json:"chunk"`
}
-type getJoinedRoomsResponse struct {
- JoinedRooms []string `json:"joined_rooms"`
-}
-
// 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"`
@@ -51,19 +49,22 @@ type databaseJoinedMember struct {
AvatarURL string `json:"avatar_url"`
}
-// GetMemberships implements GET /rooms/{roomId}/members
+// GetMemberships implements
+//
+// GET /rooms/{roomId}/members
+// GET /rooms/{roomId}/joined_members
func GetMemberships(
- req *http.Request, device *userapi.Device, roomID string, joinedOnly bool,
- _ *config.ClientAPI,
- rsAPI api.ClientRoomserverAPI,
+ 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.QueryMembershipsForRoomRequest{
- JoinedOnly: joinedOnly,
+ queryReq := api.QueryMembershipForUserRequest{
RoomID: roomID,
- Sender: device.UserID,
+ UserID: device.UserID,
}
- var queryRes api.QueryMembershipsForRoomResponse
- if err := rsAPI.QueryMembershipsForRoom(req.Context(), &queryReq, &queryRes); err != nil {
+
+ 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()
}
@@ -75,16 +76,48 @@ func GetMemberships(
}
}
+ 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 queryRes.JoinEvents {
+ for _, ev := range result {
var content databaseJoinedMember
- if err := json.Unmarshal(ev.Content, &content); err != nil {
+ 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)
+ res.Joined[ev.Sender()] = joinedMember(content)
}
return util.JSONResponse{
Code: http.StatusOK,
@@ -93,29 +126,6 @@ func GetMemberships(
}
return util.JSONResponse{
Code: http.StatusOK,
- JSON: getMembershipResponse{queryRes.JoinEvents},
- }
-}
-
-func GetJoinedRooms(
- req *http.Request,
- device *userapi.Device,
- rsAPI api.ClientRoomserverAPI,
-) util.JSONResponse {
- var res api.QueryRoomsForUserResponse
- err := rsAPI.QueryRoomsForUser(req.Context(), &api.QueryRoomsForUserRequest{
- UserID: device.UserID,
- WantMembership: "join",
- }, &res)
- if err != nil {
- util.GetLogger(req.Context()).WithError(err).Error("QueryRoomsForUser failed")
- return jsonerror.InternalServerError()
- }
- if res.RoomIDs == nil {
- res.RoomIDs = []string{}
- }
- return util.JSONResponse{
- Code: http.StatusOK,
- JSON: getJoinedRoomsResponse{res.RoomIDs},
+ JSON: getMembershipResponse{gomatrixserverlib.HeaderedToClientEvents(result, gomatrixserverlib.FormatSync)},
}
}

MODIFY
syncapi/routing/routing.go
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)
}

MODIFY
syncapi/storage/interface.go
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 {

MODIFY
syncapi/storage/postgres/memberships_table.go
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()
+}

MODIFY
syncapi/storage/shared/storage_consumer.go
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)
+}

MODIFY
syncapi/storage/sqlite3/memberships_table.go
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()
+}

MODIFY
syncapi/storage/tables/interface.go
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 {

MODIFY
syncapi/streams/stream_pdu.go
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
}

MODIFY
syncapi/types/types.go
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)
}

MODIFY
sytest-whitelist
sytest-whitelist
@@ -755,3 +755,5 @@ Messages that highlight from another user increment unread highlight count
Notifications can be viewed with GET /notifications
Can get rooms/{roomId}/messages for a departed room (SPEC-216)
Local device key changes appear in /keys/changes
+Can get rooms/{roomId}/members at a given point
+Can filter rooms/{roomId}/members
\ No newline at end of file

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.