161f145176
* Add NATS JetStream support Update shopify/sarama * Fix addresses * Don't change Addresses in Defaults * Update saramajetstream * Add missing error check Keep typing events for at least one minute * Use all configured NATS addresses * Update saramajetstream * Try setting up with NATS * Make sure NATS uses own persistent directory (TODO: make this configurable) * Update go.mod/go.sum * Jetstream package * Various other refactoring * Build fixes * Config tweaks, make random jetstream storage path for CI * Disable interest policies * Try to sane default on jetstream base path * Try to use in-memory for CI * Restore storage/retention * Update nats.go dependency * Adapt changes to config * Remove unneeded TopicFor * Dep update * Revert "Remove unneeded TopicFor" This reverts commitf5a4e4a339
. * Revert changes made to streams * Fix build problems * Update nats-server * Update go.mod/go.sum * Roomserver input API queuing using NATS * Fix topic naming * Prometheus metrics * More refactoring to remove saramajetstream * Add missing topic * Don't try to populate map that doesn't exist * Roomserver output topic * Update go.mod/go.sum * Message acknowledgements * Ack tweaks * Try to resume transaction re-sends * Try to resume transaction re-sends * Update to matrix-org/gomatrixserverlib@91dadfb * Remove internal.PartitionStorer from components that don't consume keychanges * Try to reduce re-allocations a bit in resolveConflictsV2 * Tweak delivery options on RS input * Publish send-to-device messages into correct JetStream subject * Async and sync roomserver input * Update dendrite-config.yaml * Remove roomserver tests for now (they need rewriting) * Remove roomserver test again (was merged back in) * Update documentation * Docker updates * More Docker updates * Update Docker readme again * Fix lint issues * Send final event in `processEvent` synchronously (since this might stop Sytest from being so upset) * Don't report event rejection errors via `/send`, since apparently this is upsetting tests that don't expect that * Go 1.16 instead of Go 1.13 for upgrade tests and Complement * Revert "Don't report event rejection errors via `/send`, since apparently this is upsetting tests that don't expect that" This reverts commit368675283f
. * Don't report any errors on `/send` to see what fun that creates * Fix panics on closed channel sends * Enforce state key matches sender * Do the same for leave * Various tweaks to make tests happier Squashed commit of the following: commit13f9028e7a
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 15:47:14 2022 +0000 Do the same for leave commite6be7f05c3
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 15:33:42 2022 +0000 Enforce state key matches sender commit85ede6d64b
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 14:07:04 2022 +0000 Fix panics on closed channel sends commit9755494a98
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 13:38:22 2022 +0000 Don't report any errors on `/send` to see what fun that creates commit3bb4f87b5d
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 13:00:26 2022 +0000 Revert "Don't report event rejection errors via `/send`, since apparently this is upsetting tests that don't expect that" This reverts commit368675283f
. commitfe2673ed7b
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 12:09:34 2022 +0000 Go 1.16 instead of Go 1.13 for upgrade tests and Complement commit368675283f
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 11:51:45 2022 +0000 Don't report event rejection errors via `/send`, since apparently this is upsetting tests that don't expect that commitb028dfc085
Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Tue Jan 4 10:29:08 2022 +0000 Send final event in `processEvent` synchronously (since this might stop Sytest from being so upset) * Merge in NATS Server v2.6.6 and nats.go v1.13 into the in-process connection fork * Add `jetstream.WithJetStreamMessage` to make ack/nak-ing less messy, use process context in consumers * Fix consumer component name in federation API * Add comment explaining where streams are defined * Tweaks to roomserver input with comments * Finish that sentence that I apparently forgot to finish in INSTALL.md * Bump version number of config to 2 * Add comments around asynchronous sends to roomserver in processEventWithMissingState * More useful error message when the config version does not match * Set version in generate-config * Fix version in config.Defaults Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>
385 lines
11 KiB
Go
385 lines
11 KiB
Go
// 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 (
|
|
"context"
|
|
"net/http"
|
|
"time"
|
|
|
|
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
|
|
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
|
|
"github.com/matrix-org/dendrite/clientapi/httputil"
|
|
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
|
"github.com/matrix-org/dendrite/internal/eventutil"
|
|
"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/dendrite/userapi/storage/accounts"
|
|
"github.com/matrix-org/gomatrixserverlib"
|
|
|
|
"github.com/matrix-org/gomatrix"
|
|
"github.com/matrix-org/util"
|
|
)
|
|
|
|
// GetProfile implements GET /profile/{userID}
|
|
func GetProfile(
|
|
req *http.Request, accountDB accounts.Database, cfg *config.ClientAPI,
|
|
userID string,
|
|
asAPI appserviceAPI.AppServiceQueryAPI,
|
|
federation *gomatrixserverlib.FederationClient,
|
|
) util.JSONResponse {
|
|
profile, err := getProfile(req.Context(), accountDB, cfg, userID, asAPI, federation)
|
|
if err != nil {
|
|
if err == eventutil.ErrProfileNoExists {
|
|
return util.JSONResponse{
|
|
Code: http.StatusNotFound,
|
|
JSON: jsonerror.NotFound("The user does not exist or does not have a profile"),
|
|
}
|
|
}
|
|
|
|
util.GetLogger(req.Context()).WithError(err).Error("getProfile failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
return util.JSONResponse{
|
|
Code: http.StatusOK,
|
|
JSON: eventutil.ProfileResponse{
|
|
AvatarURL: profile.AvatarURL,
|
|
DisplayName: profile.DisplayName,
|
|
},
|
|
}
|
|
}
|
|
|
|
// GetAvatarURL implements GET /profile/{userID}/avatar_url
|
|
func GetAvatarURL(
|
|
req *http.Request, accountDB accounts.Database, cfg *config.ClientAPI,
|
|
userID string, asAPI appserviceAPI.AppServiceQueryAPI,
|
|
federation *gomatrixserverlib.FederationClient,
|
|
) util.JSONResponse {
|
|
profile, err := getProfile(req.Context(), accountDB, cfg, userID, asAPI, federation)
|
|
if err != nil {
|
|
if err == eventutil.ErrProfileNoExists {
|
|
return util.JSONResponse{
|
|
Code: http.StatusNotFound,
|
|
JSON: jsonerror.NotFound("The user does not exist or does not have a profile"),
|
|
}
|
|
}
|
|
|
|
util.GetLogger(req.Context()).WithError(err).Error("getProfile failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
return util.JSONResponse{
|
|
Code: http.StatusOK,
|
|
JSON: eventutil.AvatarURL{
|
|
AvatarURL: profile.AvatarURL,
|
|
},
|
|
}
|
|
}
|
|
|
|
// SetAvatarURL implements PUT /profile/{userID}/avatar_url
|
|
func SetAvatarURL(
|
|
req *http.Request, accountDB accounts.Database,
|
|
device *userapi.Device, userID string, cfg *config.ClientAPI, rsAPI api.RoomserverInternalAPI,
|
|
) util.JSONResponse {
|
|
if userID != device.UserID {
|
|
return util.JSONResponse{
|
|
Code: http.StatusForbidden,
|
|
JSON: jsonerror.Forbidden("userID does not match the current user"),
|
|
}
|
|
}
|
|
|
|
var r eventutil.AvatarURL
|
|
if resErr := httputil.UnmarshalJSONRequest(req, &r); resErr != nil {
|
|
return *resErr
|
|
}
|
|
if r.AvatarURL == "" {
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.BadJSON("'avatar_url' must be supplied."),
|
|
}
|
|
}
|
|
|
|
localpart, _, err := gomatrixserverlib.SplitID('@', userID)
|
|
if err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
evTime, err := httputil.ParseTSParam(req)
|
|
if err != nil {
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.InvalidArgumentValue(err.Error()),
|
|
}
|
|
}
|
|
|
|
oldProfile, err := accountDB.GetProfileByLocalpart(req.Context(), localpart)
|
|
if err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("accountDB.GetProfileByLocalpart failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
if err = accountDB.SetAvatarURL(req.Context(), localpart, r.AvatarURL); err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("accountDB.SetAvatarURL failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
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()
|
|
}
|
|
|
|
newProfile := authtypes.Profile{
|
|
Localpart: localpart,
|
|
DisplayName: oldProfile.DisplayName,
|
|
AvatarURL: r.AvatarURL,
|
|
}
|
|
|
|
events, err := buildMembershipEvents(
|
|
req.Context(), res.RoomIDs, newProfile, userID, cfg, evTime, rsAPI,
|
|
)
|
|
switch e := err.(type) {
|
|
case nil:
|
|
case gomatrixserverlib.BadJSONError:
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.BadJSON(e.Error()),
|
|
}
|
|
default:
|
|
util.GetLogger(req.Context()).WithError(err).Error("buildMembershipEvents failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
return util.JSONResponse{
|
|
Code: http.StatusOK,
|
|
JSON: struct{}{},
|
|
}
|
|
}
|
|
|
|
// GetDisplayName implements GET /profile/{userID}/displayname
|
|
func GetDisplayName(
|
|
req *http.Request, accountDB accounts.Database, cfg *config.ClientAPI,
|
|
userID string, asAPI appserviceAPI.AppServiceQueryAPI,
|
|
federation *gomatrixserverlib.FederationClient,
|
|
) util.JSONResponse {
|
|
profile, err := getProfile(req.Context(), accountDB, cfg, userID, asAPI, federation)
|
|
if err != nil {
|
|
if err == eventutil.ErrProfileNoExists {
|
|
return util.JSONResponse{
|
|
Code: http.StatusNotFound,
|
|
JSON: jsonerror.NotFound("The user does not exist or does not have a profile"),
|
|
}
|
|
}
|
|
|
|
util.GetLogger(req.Context()).WithError(err).Error("getProfile failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
return util.JSONResponse{
|
|
Code: http.StatusOK,
|
|
JSON: eventutil.DisplayName{
|
|
DisplayName: profile.DisplayName,
|
|
},
|
|
}
|
|
}
|
|
|
|
// SetDisplayName implements PUT /profile/{userID}/displayname
|
|
func SetDisplayName(
|
|
req *http.Request, accountDB accounts.Database,
|
|
device *userapi.Device, userID string, cfg *config.ClientAPI, rsAPI api.RoomserverInternalAPI,
|
|
) util.JSONResponse {
|
|
if userID != device.UserID {
|
|
return util.JSONResponse{
|
|
Code: http.StatusForbidden,
|
|
JSON: jsonerror.Forbidden("userID does not match the current user"),
|
|
}
|
|
}
|
|
|
|
var r eventutil.DisplayName
|
|
if resErr := httputil.UnmarshalJSONRequest(req, &r); resErr != nil {
|
|
return *resErr
|
|
}
|
|
if r.DisplayName == "" {
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.BadJSON("'displayname' must be supplied."),
|
|
}
|
|
}
|
|
|
|
localpart, _, err := gomatrixserverlib.SplitID('@', userID)
|
|
if err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
evTime, err := httputil.ParseTSParam(req)
|
|
if err != nil {
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.InvalidArgumentValue(err.Error()),
|
|
}
|
|
}
|
|
|
|
oldProfile, err := accountDB.GetProfileByLocalpart(req.Context(), localpart)
|
|
if err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("accountDB.GetProfileByLocalpart failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
if err = accountDB.SetDisplayName(req.Context(), localpart, r.DisplayName); err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("accountDB.SetDisplayName failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
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()
|
|
}
|
|
|
|
newProfile := authtypes.Profile{
|
|
Localpart: localpart,
|
|
DisplayName: r.DisplayName,
|
|
AvatarURL: oldProfile.AvatarURL,
|
|
}
|
|
|
|
events, err := buildMembershipEvents(
|
|
req.Context(), res.RoomIDs, newProfile, userID, cfg, evTime, rsAPI,
|
|
)
|
|
switch e := err.(type) {
|
|
case nil:
|
|
case gomatrixserverlib.BadJSONError:
|
|
return util.JSONResponse{
|
|
Code: http.StatusBadRequest,
|
|
JSON: jsonerror.BadJSON(e.Error()),
|
|
}
|
|
default:
|
|
util.GetLogger(req.Context()).WithError(err).Error("buildMembershipEvents failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil {
|
|
util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed")
|
|
return jsonerror.InternalServerError()
|
|
}
|
|
|
|
return util.JSONResponse{
|
|
Code: http.StatusOK,
|
|
JSON: struct{}{},
|
|
}
|
|
}
|
|
|
|
// getProfile gets the full profile of a user by querying the database or a
|
|
// remote homeserver.
|
|
// Returns an error when something goes wrong or specifically
|
|
// eventutil.ErrProfileNoExists when the profile doesn't exist.
|
|
func getProfile(
|
|
ctx context.Context, accountDB accounts.Database, cfg *config.ClientAPI,
|
|
userID string,
|
|
asAPI appserviceAPI.AppServiceQueryAPI,
|
|
federation *gomatrixserverlib.FederationClient,
|
|
) (*authtypes.Profile, error) {
|
|
localpart, domain, err := gomatrixserverlib.SplitID('@', userID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if domain != cfg.Matrix.ServerName {
|
|
profile, fedErr := federation.LookupProfile(ctx, domain, userID, "")
|
|
if fedErr != nil {
|
|
if x, ok := fedErr.(gomatrix.HTTPError); ok {
|
|
if x.Code == http.StatusNotFound {
|
|
return nil, eventutil.ErrProfileNoExists
|
|
}
|
|
}
|
|
|
|
return nil, fedErr
|
|
}
|
|
|
|
return &authtypes.Profile{
|
|
Localpart: localpart,
|
|
DisplayName: profile.DisplayName,
|
|
AvatarURL: profile.AvatarURL,
|
|
}, nil
|
|
}
|
|
|
|
profile, err := appserviceAPI.RetrieveUserProfile(ctx, userID, asAPI, accountDB)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return profile, nil
|
|
}
|
|
|
|
func buildMembershipEvents(
|
|
ctx context.Context,
|
|
roomIDs []string,
|
|
newProfile authtypes.Profile, userID string, cfg *config.ClientAPI,
|
|
evTime time.Time, rsAPI api.RoomserverInternalAPI,
|
|
) ([]*gomatrixserverlib.HeaderedEvent, error) {
|
|
evs := []*gomatrixserverlib.HeaderedEvent{}
|
|
|
|
for _, roomID := range roomIDs {
|
|
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
|
|
verRes := api.QueryRoomVersionForRoomResponse{}
|
|
if err := rsAPI.QueryRoomVersionForRoom(ctx, &verReq, &verRes); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
builder := gomatrixserverlib.EventBuilder{
|
|
Sender: userID,
|
|
RoomID: roomID,
|
|
Type: "m.room.member",
|
|
StateKey: &userID,
|
|
}
|
|
|
|
content := gomatrixserverlib.MemberContent{
|
|
Membership: gomatrixserverlib.Join,
|
|
}
|
|
|
|
content.DisplayName = newProfile.DisplayName
|
|
content.AvatarURL = newProfile.AvatarURL
|
|
|
|
if err := builder.SetContent(content); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
event, err := eventutil.QueryAndBuildEvent(ctx, &builder, cfg.Matrix, evTime, rsAPI, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
evs = append(evs, event.Headered(verRes.RoomVersion))
|
|
}
|
|
|
|
return evs, nil
|
|
}
|