diff --git a/.github/workflows/dendrite.yml b/.github/workflows/dendrite.yml index f8019b3ea..a8271b675 100644 --- a/.github/workflows/dendrite.yml +++ b/.github/workflows/dendrite.yml @@ -109,6 +109,11 @@ jobs: uses: actions/setup-go@v3 with: go-version: ${{ matrix.go }} + - name: Set up gotestfmt + uses: gotesttools/gotestfmt-action@v2 + with: + # Optional: pass GITHUB_TOKEN to avoid rate limiting. + token: ${{ secrets.GITHUB_TOKEN }} - uses: actions/cache@v3 with: path: | @@ -117,7 +122,7 @@ jobs: key: ${{ runner.os }}-go${{ matrix.go }}-test-${{ hashFiles('**/go.sum') }} restore-keys: | ${{ runner.os }}-go${{ matrix.go }}-test- - - run: go test ./... + - run: go test -json -v ./... 2>&1 | gotestfmt env: POSTGRES_HOST: localhost POSTGRES_USER: postgres diff --git a/CHANGES.md b/CHANGES.md index eea2c3c7c..1ed87824a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,23 @@ # Changelog +## Dendrite 0.10.4 (2022-10-21) + +### Features + +* Various tables belonging to the user API will be renamed so that they are namespaced with the `userapi_` prefix + * Note that, after upgrading to this version, you should not revert to an older version of Dendrite as the database changes **will not** be reverted automatically +* The backoff and retry behaviour in the federation API has been refactored and improved + +### Fixes + +* Private read receipt support is now advertised in the client `/versions` endpoint +* Private read receipts will now clear notification counts properly +* A bug where a false `leave` membership transition was inserted into the timeline after accepting an invite has been fixed +* Some panics caused by concurrent map writes in the key server have been fixed +* The sync API now calculates membership transitions from state deltas more accurately +* Transaction IDs are now scoped to endpoints, which should fix some bugs where transaction ID reuse could cause nonsensical cached responses from some endpoints +* The length of the `type`, `sender`, `state_key` and `room_id` fields in events are now verified by number of bytes rather than codepoints after a spec clarification, reverting a change made in Dendrite 0.9.6 + ## Dendrite 0.10.3 (2022-10-14) ### Features diff --git a/clientapi/routing/joined_rooms.go b/clientapi/routing/joined_rooms.go new file mode 100644 index 000000000..4bb353ea9 --- /dev/null +++ b/clientapi/routing/joined_rooms.go @@ -0,0 +1,52 @@ +// 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}, + } +} diff --git a/clientapi/routing/keys.go b/clientapi/routing/keys.go index 5c3681382..0c12b1117 100644 --- a/clientapi/routing/keys.go +++ b/clientapi/routing/keys.go @@ -99,7 +99,11 @@ func (r *queryKeysRequest) GetTimeout() time.Duration { if r.Timeout == 0 { return 10 * time.Second } - return time.Duration(r.Timeout) * time.Millisecond + timeout := time.Duration(r.Timeout) * time.Millisecond + if timeout > time.Second*20 { + timeout = time.Second * 20 + } + return timeout } func QueryKeys(req *http.Request, keyAPI api.ClientKeyAPI, device *userapi.Device) util.JSONResponse { diff --git a/clientapi/routing/routing.go b/clientapi/routing/routing.go index 4ca8e59c5..e0e3e33d4 100644 --- a/clientapi/routing/routing.go +++ b/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 { diff --git a/docs/caddy/polylith/Caddyfile b/docs/caddy/polylith/Caddyfile index 8aeb9317f..c2d81b49b 100644 --- a/docs/caddy/polylith/Caddyfile +++ b/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 diff --git a/docs/hiawatha/polylith-sample.conf b/docs/hiawatha/polylith-sample.conf index 0093fdcf2..eb1dd4f9a 100644 --- a/docs/hiawatha/polylith-sample.conf +++ b/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 diff --git a/docs/nginx/polylith-sample.conf b/docs/nginx/polylith-sample.conf index 6e81eb5f2..0ad24509a 100644 --- a/docs/nginx/polylith-sample.conf +++ b/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; } diff --git a/internal/version.go b/internal/version.go index c888748a8..5d739a45d 100644 --- a/internal/version.go +++ b/internal/version.go @@ -17,7 +17,7 @@ var build string const ( VersionMajor = 0 VersionMinor = 10 - VersionPatch = 3 + VersionPatch = 4 VersionTag = "" // example: "rc1" ) diff --git a/keyserver/internal/internal.go b/keyserver/internal/internal.go index 49ef03054..ff0968b27 100644 --- a/keyserver/internal/internal.go +++ b/keyserver/internal/internal.go @@ -257,9 +257,6 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques res.UserSigningKeys = make(map[string]gomatrixserverlib.CrossSigningKey) res.Failures = make(map[string]interface{}) - // get cross-signing keys from the database - a.crossSigningKeysFromDatabase(ctx, req, res) - // make a map from domain to device keys domainToDeviceKeys := make(map[string]map[string][]string) domainToCrossSigningKeys := make(map[string]map[string]struct{}) @@ -336,6 +333,10 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques a.queryRemoteKeys(ctx, req.Timeout, res, domainToDeviceKeys, domainToCrossSigningKeys) } + // Now that we've done the potentially expensive work of asking the federation, + // try filling the cross-signing keys from the database that we know about. + a.crossSigningKeysFromDatabase(ctx, req, res) + // Finally, append signatures that we know about // TODO: This is horrible because we need to round-trip the signature from // JSON, add the signatures and marshal it again, for some reason? diff --git a/keyserver/storage/postgres/cross_signing_sigs_table.go b/keyserver/storage/postgres/cross_signing_sigs_table.go index 8b2a865b9..4536b7d80 100644 --- a/keyserver/storage/postgres/cross_signing_sigs_table.go +++ b/keyserver/storage/postgres/cross_signing_sigs_table.go @@ -42,7 +42,7 @@ CREATE INDEX IF NOT EXISTS keyserver_cross_signing_sigs_idx ON keyserver_cross_s const selectCrossSigningSigsForTargetSQL = "" + "SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" + - " WHERE (origin_user_id = $1 OR origin_user_id = target_user_id) AND target_user_id = $2 AND target_key_id = $3" + " WHERE (origin_user_id = $1 OR origin_user_id = $2) AND target_user_id = $2 AND target_key_id = $3" const upsertCrossSigningSigsForTargetSQL = "" + "INSERT INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" + diff --git a/keyserver/storage/sqlite3/cross_signing_sigs_table.go b/keyserver/storage/sqlite3/cross_signing_sigs_table.go index ea431151e..7a153e8fb 100644 --- a/keyserver/storage/sqlite3/cross_signing_sigs_table.go +++ b/keyserver/storage/sqlite3/cross_signing_sigs_table.go @@ -42,7 +42,7 @@ CREATE INDEX IF NOT EXISTS keyserver_cross_signing_sigs_idx ON keyserver_cross_s const selectCrossSigningSigsForTargetSQL = "" + "SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" + - " WHERE (origin_user_id = $1 OR origin_user_id = target_user_id) AND target_user_id = $2 AND target_key_id = $3" + " WHERE (origin_user_id = $1 OR origin_user_id = $2) AND target_user_id = $3 AND target_key_id = $4" const upsertCrossSigningSigsForTargetSQL = "" + "INSERT OR REPLACE INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" + @@ -85,7 +85,7 @@ func NewSqliteCrossSigningSigsTable(db *sql.DB) (tables.CrossSigningSigs, error) func (s *crossSigningSigsStatements) SelectCrossSigningSigsForTarget( ctx context.Context, txn *sql.Tx, originUserID, targetUserID string, targetKeyID gomatrixserverlib.KeyID, ) (r types.CrossSigningSigMap, err error) { - rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, originUserID, targetUserID, targetKeyID) + rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, originUserID, targetUserID, targetUserID, targetKeyID) if err != nil { return nil, err } diff --git a/setup/config/config_global.go b/setup/config/config_global.go index 2efae0d5a..784893d24 100644 --- a/setup/config/config_global.go +++ b/setup/config/config_global.go @@ -170,7 +170,7 @@ type ServerNotices struct { // The displayname to be used when sending notices DisplayName string `yaml:"display_name"` // The avatar of this user - AvatarURL string `yaml:"avatar"` + AvatarURL string `yaml:"avatar_url"` // The roomname to be used when creating messages RoomName string `yaml:"room_name"` } diff --git a/clientapi/routing/memberships.go b/syncapi/routing/memberships.go similarity index 52% rename from clientapi/routing/memberships.go rename to syncapi/routing/memberships.go index 9bdd8a4f4..c9acc5d2b 100644 --- a/clientapi/routing/memberships.go +++ b/syncapi/routing/memberships.go @@ -18,22 +18,20 @@ import ( "encoding/json" "net/http" - "github.com/matrix-org/dendrite/clientapi/jsonerror" - "github.com/matrix-org/dendrite/roomserver/api" - "github.com/matrix-org/dendrite/setup/config" - userapi "github.com/matrix-org/dendrite/userapi/api" "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"` } -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, - RoomID: roomID, - Sender: device.UserID, + queryReq := api.QueryMembershipForUserRequest{ + RoomID: roomID, + 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,54 @@ 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 { + if !queryRes.IsInRoom { + 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."), + } + } 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 +132,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)}, } } diff --git a/syncapi/routing/messages.go b/syncapi/routing/messages.go index 8f3ed3f5b..86cf8e736 100644 --- a/syncapi/routing/messages.go +++ b/syncapi/routing/messages.go @@ -83,18 +83,18 @@ func OnIncomingMessagesRequest( defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err) // check if the user has already forgotten about this room - isForgotten, roomExists, err := checkIsRoomForgotten(req.Context(), roomID, device.UserID, rsAPI) + membershipResp, err := getMembershipForUser(req.Context(), roomID, device.UserID, rsAPI) if err != nil { return jsonerror.InternalServerError() } - if !roomExists { + if !membershipResp.RoomExists { return util.JSONResponse{ Code: http.StatusForbidden, JSON: jsonerror.Forbidden("room does not exist"), } } - if isForgotten { + if membershipResp.IsRoomForgotten { return util.JSONResponse{ Code: http.StatusForbidden, JSON: jsonerror.Forbidden("user already forgot about this room"), @@ -195,6 +195,20 @@ func OnIncomingMessagesRequest( } } + // If the user already left the room, grep events from before that + if membershipResp.Membership == gomatrixserverlib.Leave { + var token types.TopologyToken + token, err = snapshot.EventPositionInTopology(req.Context(), membershipResp.EventID) + if err != nil { + return util.JSONResponse{ + Code: http.StatusInternalServerError, + } + } + if backwardOrdering { + from = token + } + } + mReq := messagesReq{ ctx: req.Context(), db: db, @@ -283,17 +297,16 @@ func (m *messagesResp) applyLazyLoadMembers( } } -func checkIsRoomForgotten(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (forgotten bool, exists bool, err error) { +func getMembershipForUser(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (resp api.QueryMembershipForUserResponse, err error) { req := api.QueryMembershipForUserRequest{ RoomID: roomID, UserID: userID, } - resp := api.QueryMembershipForUserResponse{} if err := rsAPI.QueryMembershipForUser(ctx, &req, &resp); err != nil { - return false, false, err + return api.QueryMembershipForUserResponse{}, err } - return resp.IsRoomForgotten, resp.RoomExists, nil + return resp, nil } // retrieveEvents retrieves events from the local database for a request on @@ -313,7 +326,11 @@ func (r *messagesReq) retrieveEvents() ( } var events []*gomatrixserverlib.HeaderedEvent - util.GetLogger(r.ctx).WithField("start", start).WithField("end", end).Infof("Fetched %d events locally", len(streamEvents)) + util.GetLogger(r.ctx).WithFields(logrus.Fields{ + "start": r.from, + "end": r.to, + "backwards": r.backwardOrdering, + }).Infof("Fetched %d events locally", len(streamEvents)) // There can be two reasons for streamEvents to be empty: either we've // reached the oldest event in the room (or the most recent one, depending diff --git a/syncapi/routing/routing.go b/syncapi/routing/routing.go index 71fa93c1e..bc3ad2384 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 02d45f801..af4fce44e 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 939d6b3f5..b555e8456 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 bf12203db..23f53d11f 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 0c966fca0..7e54fac17 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 e48c050dd..2c4f04ec2 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 9ec2b61cd..90cf8ce53 100644 --- a/syncapi/streams/stream_pdu.go +++ b/syncapi/streams/stream_pdu.go @@ -227,14 +227,10 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse( stateFilter *gomatrixserverlib.StateFilter, req *types.SyncRequest, ) (types.StreamPosition, error) { - if delta.MembershipPos > 0 && delta.Membership == gomatrixserverlib.Leave { - // make sure we don't leak recent events after the leave event. - // TODO: History visibility makes this somewhat complex to handle correctly. For example: - // TODO: This doesn't work for join -> leave in a single /sync request (see events prior to join). - // TODO: This will fail on join -> leave -> sensitive msg -> join -> leave - // in a single /sync request - // This is all "okay" assuming history_visibility == "shared" which it is by default. - r.To = delta.MembershipPos + + originalLimit := eventFilter.Limit + if r.Backwards { + eventFilter.Limit = int(r.From - r.To) } recentStreamEvents, limited, err := snapshot.RecentEvents( ctx, delta.RoomID, r, @@ -303,6 +299,12 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse( logrus.WithError(err).Error("unable to apply history visibility filter") } + if r.Backwards && len(events) > originalLimit { + // We're going backwards and the events are ordered chronologically, so take the last `limit` events + events = events[len(events)-originalLimit:] + limited = true + } + if len(delta.StateEvents) > 0 { updateLatestPosition(delta.StateEvents[len(delta.StateEvents)-1].EventID()) } @@ -473,7 +475,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 57ce7b6ff..295187acc 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) } diff --git a/sytest-whitelist b/sytest-whitelist index 1387838f7..60610929a 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -752,4 +752,9 @@ When user joins a room the state is included in the next sync When user joins a room the state is included in a gapped sync Messages that notify from another user increment notification_count Messages that highlight from another user increment unread highlight count -Notifications can be viewed with GET /notifications \ No newline at end of file +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 +Current state appears in timeline in private history with many messages after \ No newline at end of file