mirror of
https://github.com/hoernschen/dendrite.git
synced 2024-12-26 15:08:28 +00:00
Remove QueryBulkStateContent from current state server (#1404)
* Remove QueryBulkStateContent from current state server Expected fail due to db impl not existing * Implement query bulk state content * Fix up rejecting invites over federation * Fix bulk content marshalling
This commit is contained in:
Kegsay
GitHub
parent
895ead8048
commit
7913759921
19 changed files with 198 additions and 239 deletions
|
|
@ -26,7 +26,6 @@ import (
|
|||
"github.com/matrix-org/dendrite/clientapi/api"
|
||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||
currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
"github.com/matrix-org/dendrite/internal/config"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
|
@ -51,7 +50,7 @@ type filter struct {
|
|||
|
||||
// GetPostPublicRooms implements GET and POST /publicRooms
|
||||
func GetPostPublicRooms(
|
||||
req *http.Request, rsAPI roomserverAPI.RoomserverInternalAPI, stateAPI currentstateAPI.CurrentStateInternalAPI,
|
||||
req *http.Request, rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||
extRoomsProvider api.ExtraPublicRoomsProvider,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
cfg *config.ClientAPI,
|
||||
|
|
@ -75,7 +74,7 @@ func GetPostPublicRooms(
|
|||
}
|
||||
}
|
||||
|
||||
response, err := publicRooms(req.Context(), request, rsAPI, stateAPI, extRoomsProvider)
|
||||
response, err := publicRooms(req.Context(), request, rsAPI, extRoomsProvider)
|
||||
if err != nil {
|
||||
util.GetLogger(req.Context()).WithError(err).Errorf("failed to work out public rooms")
|
||||
return jsonerror.InternalServerError()
|
||||
|
|
@ -86,8 +85,8 @@ func GetPostPublicRooms(
|
|||
}
|
||||
}
|
||||
|
||||
func publicRooms(ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||
stateAPI currentstateAPI.CurrentStateInternalAPI, extRoomsProvider api.ExtraPublicRoomsProvider,
|
||||
func publicRooms(
|
||||
ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI.RoomserverInternalAPI, extRoomsProvider api.ExtraPublicRoomsProvider,
|
||||
) (*gomatrixserverlib.RespPublicRooms, error) {
|
||||
|
||||
response := gomatrixserverlib.RespPublicRooms{
|
||||
|
|
@ -110,7 +109,7 @@ func publicRooms(ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI
|
|||
|
||||
var rooms []gomatrixserverlib.PublicRoom
|
||||
if request.Since == "" {
|
||||
rooms = refreshPublicRoomCache(ctx, rsAPI, extRoomsProvider, stateAPI)
|
||||
rooms = refreshPublicRoomCache(ctx, rsAPI, extRoomsProvider)
|
||||
} else {
|
||||
rooms = getPublicRoomsFromCache()
|
||||
}
|
||||
|
|
@ -226,7 +225,6 @@ func sliceInto(slice []gomatrixserverlib.PublicRoom, since int64, limit int16) (
|
|||
|
||||
func refreshPublicRoomCache(
|
||||
ctx context.Context, rsAPI roomserverAPI.RoomserverInternalAPI, extRoomsProvider api.ExtraPublicRoomsProvider,
|
||||
stateAPI currentstateAPI.CurrentStateInternalAPI,
|
||||
) []gomatrixserverlib.PublicRoom {
|
||||
cacheMu.Lock()
|
||||
defer cacheMu.Unlock()
|
||||
|
|
@ -241,7 +239,7 @@ func refreshPublicRoomCache(
|
|||
util.GetLogger(ctx).WithError(err).Error("QueryPublishedRooms failed")
|
||||
return publicRoomsCache
|
||||
}
|
||||
pubRooms, err := currentstateAPI.PopulatePublicRooms(ctx, queryRes.RoomIDs, stateAPI)
|
||||
pubRooms, err := roomserverAPI.PopulatePublicRooms(ctx, queryRes.RoomIDs, rsAPI)
|
||||
if err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Error("PopulatePublicRooms failed")
|
||||
return publicRoomsCache
|
||||
|
|
|
|||
|
|
@ -336,7 +336,7 @@ func Setup(
|
|||
).Methods(http.MethodPut, http.MethodOptions)
|
||||
r0mux.Handle("/publicRooms",
|
||||
httputil.MakeExternalAPI("public_rooms", func(req *http.Request) util.JSONResponse {
|
||||
return GetPostPublicRooms(req, rsAPI, stateAPI, extRoomsProvider, federation, cfg)
|
||||
return GetPostPublicRooms(req, rsAPI, extRoomsProvider, federation, cfg)
|
||||
}),
|
||||
).Methods(http.MethodGet, http.MethodPost, http.MethodOptions)
|
||||
|
||||
|
|
|
|||
|
|
@ -106,7 +106,7 @@ func (p *publicRoomsProvider) AdvertiseRooms() error {
|
|||
util.GetLogger(ctx).WithError(err).Error("QueryPublishedRooms failed")
|
||||
return err
|
||||
}
|
||||
ourRooms, err := currentstateAPI.PopulatePublicRooms(ctx, queryRes.RoomIDs, p.stateAPI)
|
||||
ourRooms, err := roomserverAPI.PopulatePublicRooms(ctx, queryRes.RoomIDs, p.rsAPI)
|
||||
if err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Error("PopulatePublicRooms failed")
|
||||
return err
|
||||
|
|
|
|||
|
|
@ -14,37 +14,5 @@
|
|||
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
type CurrentStateInternalAPI interface {
|
||||
// QueryBulkStateContent does a bulk query for state event content in the given rooms.
|
||||
QueryBulkStateContent(ctx context.Context, req *QueryBulkStateContentRequest, res *QueryBulkStateContentResponse) error
|
||||
}
|
||||
|
||||
type QueryBulkStateContentRequest struct {
|
||||
// Returns state events in these rooms
|
||||
RoomIDs []string
|
||||
// If true, treats the '*' StateKey as "all state events of this type" rather than a literal value of '*'
|
||||
AllowWildcards bool
|
||||
// The state events to return. Only a small subset of tuples are allowed in this request as only certain events
|
||||
// have their content fields extracted. Specifically, the tuple Type must be one of:
|
||||
// m.room.avatar
|
||||
// m.room.create
|
||||
// m.room.canonical_alias
|
||||
// m.room.guest_access
|
||||
// m.room.history_visibility
|
||||
// m.room.join_rules
|
||||
// m.room.member
|
||||
// m.room.name
|
||||
// m.room.topic
|
||||
// Any other tuple type will result in the query failing.
|
||||
StateTuples []gomatrixserverlib.StateKeyTuple
|
||||
}
|
||||
type QueryBulkStateContentResponse struct {
|
||||
// map of room ID -> tuple -> content_value
|
||||
Rooms map[string]map[gomatrixserverlib.StateKeyTuple]string
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,89 +0,0 @@
|
|||
// Copyright 2020 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 api
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
// PopulatePublicRooms extracts PublicRoom information for all the provided room IDs. The IDs are not checked to see if they are visible in the
|
||||
// published room directory.
|
||||
// due to lots of switches
|
||||
// nolint:gocyclo
|
||||
func PopulatePublicRooms(ctx context.Context, roomIDs []string, stateAPI CurrentStateInternalAPI) ([]gomatrixserverlib.PublicRoom, error) {
|
||||
avatarTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.avatar", StateKey: ""}
|
||||
nameTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.name", StateKey: ""}
|
||||
canonicalTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomCanonicalAlias, StateKey: ""}
|
||||
topicTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.topic", StateKey: ""}
|
||||
guestTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.guest_access", StateKey: ""}
|
||||
visibilityTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomHistoryVisibility, StateKey: ""}
|
||||
joinRuleTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomJoinRules, StateKey: ""}
|
||||
|
||||
var stateRes QueryBulkStateContentResponse
|
||||
err := stateAPI.QueryBulkStateContent(ctx, &QueryBulkStateContentRequest{
|
||||
RoomIDs: roomIDs,
|
||||
AllowWildcards: true,
|
||||
StateTuples: []gomatrixserverlib.StateKeyTuple{
|
||||
nameTuple, canonicalTuple, topicTuple, guestTuple, visibilityTuple, joinRuleTuple, avatarTuple,
|
||||
{EventType: gomatrixserverlib.MRoomMember, StateKey: "*"},
|
||||
},
|
||||
}, &stateRes)
|
||||
if err != nil {
|
||||
util.GetLogger(ctx).WithError(err).Error("QueryBulkStateContent failed")
|
||||
return nil, err
|
||||
}
|
||||
chunk := make([]gomatrixserverlib.PublicRoom, len(roomIDs))
|
||||
i := 0
|
||||
for roomID, data := range stateRes.Rooms {
|
||||
pub := gomatrixserverlib.PublicRoom{
|
||||
RoomID: roomID,
|
||||
}
|
||||
joinCount := 0
|
||||
var joinRule, guestAccess string
|
||||
for tuple, contentVal := range data {
|
||||
if tuple.EventType == gomatrixserverlib.MRoomMember && contentVal == "join" {
|
||||
joinCount++
|
||||
continue
|
||||
}
|
||||
switch tuple {
|
||||
case avatarTuple:
|
||||
pub.AvatarURL = contentVal
|
||||
case nameTuple:
|
||||
pub.Name = contentVal
|
||||
case topicTuple:
|
||||
pub.Topic = contentVal
|
||||
case canonicalTuple:
|
||||
pub.CanonicalAlias = contentVal
|
||||
case visibilityTuple:
|
||||
pub.WorldReadable = contentVal == "world_readable"
|
||||
// need both of these to determine whether guests can join
|
||||
case joinRuleTuple:
|
||||
joinRule = contentVal
|
||||
case guestTuple:
|
||||
guestAccess = contentVal
|
||||
}
|
||||
}
|
||||
if joinRule == gomatrixserverlib.Public && guestAccess == "can_join" {
|
||||
pub.GuestCanJoin = true
|
||||
}
|
||||
pub.JoinedMembersCount = joinCount
|
||||
chunk[i] = pub
|
||||
i++
|
||||
}
|
||||
return chunk, nil
|
||||
}
|
||||
|
|
@ -15,33 +15,9 @@
|
|||
package internal
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
"github.com/matrix-org/dendrite/currentstateserver/storage"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
type CurrentStateInternalAPI struct {
|
||||
DB storage.Database
|
||||
}
|
||||
|
||||
func (a *CurrentStateInternalAPI) QueryBulkStateContent(ctx context.Context, req *api.QueryBulkStateContentRequest, res *api.QueryBulkStateContentResponse) error {
|
||||
events, err := a.DB.GetBulkStateContent(ctx, req.RoomIDs, req.StateTuples, req.AllowWildcards)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
res.Rooms = make(map[string]map[gomatrixserverlib.StateKeyTuple]string)
|
||||
for _, ev := range events {
|
||||
if res.Rooms[ev.RoomID] == nil {
|
||||
res.Rooms[ev.RoomID] = make(map[gomatrixserverlib.StateKeyTuple]string)
|
||||
}
|
||||
room := res.Rooms[ev.RoomID]
|
||||
room[gomatrixserverlib.StateKeyTuple{
|
||||
EventType: ev.EventType,
|
||||
StateKey: ev.StateKey,
|
||||
}] = ev.ContentValue
|
||||
res.Rooms[ev.RoomID] = room
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,13 +15,10 @@
|
|||
package inthttp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
// HTTP paths for the internal HTTP APIs
|
||||
|
|
@ -49,15 +46,3 @@ type httpCurrentStateInternalAPI struct {
|
|||
apiURL string
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
func (h *httpCurrentStateInternalAPI) QueryBulkStateContent(
|
||||
ctx context.Context,
|
||||
request *api.QueryBulkStateContentRequest,
|
||||
response *api.QueryBulkStateContentResponse,
|
||||
) error {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "QueryBulkStateContent")
|
||||
defer span.Finish()
|
||||
|
||||
apiURL := h.apiURL + QueryBulkStateContentPath
|
||||
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,27 +15,10 @@
|
|||
package inthttp
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
func AddRoutes(internalAPIMux *mux.Router, intAPI api.CurrentStateInternalAPI) {
|
||||
internalAPIMux.Handle(QueryBulkStateContentPath,
|
||||
httputil.MakeInternalAPI("queryBulkStateContent", func(req *http.Request) util.JSONResponse {
|
||||
request := api.QueryBulkStateContentRequest{}
|
||||
response := api.QueryBulkStateContentResponse{}
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.MessageResponse(http.StatusBadRequest, err.Error())
|
||||
}
|
||||
if err := intAPI.QueryBulkStateContent(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ import (
|
|||
|
||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||
currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
|
|
@ -24,7 +23,7 @@ type filter struct {
|
|||
}
|
||||
|
||||
// GetPostPublicRooms implements GET and POST /publicRooms
|
||||
func GetPostPublicRooms(req *http.Request, rsAPI roomserverAPI.RoomserverInternalAPI, stateAPI currentstateAPI.CurrentStateInternalAPI) util.JSONResponse {
|
||||
func GetPostPublicRooms(req *http.Request, rsAPI roomserverAPI.RoomserverInternalAPI) util.JSONResponse {
|
||||
var request PublicRoomReq
|
||||
if fillErr := fillPublicRoomsReq(req, &request); fillErr != nil {
|
||||
return *fillErr
|
||||
|
|
@ -32,7 +31,7 @@ func GetPostPublicRooms(req *http.Request, rsAPI roomserverAPI.RoomserverInterna
|
|||
if request.Limit == 0 {
|
||||
request.Limit = 50
|
||||
}
|
||||
response, err := publicRooms(req.Context(), request, rsAPI, stateAPI)
|
||||
response, err := publicRooms(req.Context(), request, rsAPI)
|
||||
if err != nil {
|
||||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
|
@ -42,8 +41,9 @@ func GetPostPublicRooms(req *http.Request, rsAPI roomserverAPI.RoomserverInterna
|
|||
}
|
||||
}
|
||||
|
||||
func publicRooms(ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||
stateAPI currentstateAPI.CurrentStateInternalAPI) (*gomatrixserverlib.RespPublicRooms, error) {
|
||||
func publicRooms(
|
||||
ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||
) (*gomatrixserverlib.RespPublicRooms, error) {
|
||||
|
||||
var response gomatrixserverlib.RespPublicRooms
|
||||
var limit int16
|
||||
|
|
@ -80,7 +80,7 @@ func publicRooms(ctx context.Context, request PublicRoomReq, rsAPI roomserverAPI
|
|||
nextIndex = len(queryRes.RoomIDs)
|
||||
}
|
||||
roomIDs := queryRes.RoomIDs[offset:nextIndex]
|
||||
response.Chunk, err = fillInRooms(ctx, roomIDs, stateAPI)
|
||||
response.Chunk, err = fillInRooms(ctx, roomIDs, rsAPI)
|
||||
return &response, err
|
||||
}
|
||||
|
||||
|
|
@ -112,7 +112,7 @@ func fillPublicRoomsReq(httpReq *http.Request, request *PublicRoomReq) *util.JSO
|
|||
|
||||
// due to lots of switches
|
||||
// nolint:gocyclo
|
||||
func fillInRooms(ctx context.Context, roomIDs []string, stateAPI currentstateAPI.CurrentStateInternalAPI) ([]gomatrixserverlib.PublicRoom, error) {
|
||||
func fillInRooms(ctx context.Context, roomIDs []string, rsAPI roomserverAPI.RoomserverInternalAPI) ([]gomatrixserverlib.PublicRoom, error) {
|
||||
avatarTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.avatar", StateKey: ""}
|
||||
nameTuple := gomatrixserverlib.StateKeyTuple{EventType: "m.room.name", StateKey: ""}
|
||||
canonicalTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomCanonicalAlias, StateKey: ""}
|
||||
|
|
@ -121,8 +121,8 @@ func fillInRooms(ctx context.Context, roomIDs []string, stateAPI currentstateAPI
|
|||
visibilityTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomHistoryVisibility, StateKey: ""}
|
||||
joinRuleTuple := gomatrixserverlib.StateKeyTuple{EventType: gomatrixserverlib.MRoomJoinRules, StateKey: ""}
|
||||
|
||||
var stateRes currentstateAPI.QueryBulkStateContentResponse
|
||||
err := stateAPI.QueryBulkStateContent(ctx, ¤tstateAPI.QueryBulkStateContentRequest{
|
||||
var stateRes roomserverAPI.QueryBulkStateContentResponse
|
||||
err := rsAPI.QueryBulkStateContent(ctx, &roomserverAPI.QueryBulkStateContentRequest{
|
||||
RoomIDs: roomIDs,
|
||||
AllowWildcards: true,
|
||||
StateTuples: []gomatrixserverlib.StateKeyTuple{
|
||||
|
|
|
|||
|
|
@ -390,7 +390,7 @@ func Setup(
|
|||
|
||||
v1fedmux.Handle("/publicRooms",
|
||||
httputil.MakeExternalAPI("federation_public_rooms", func(req *http.Request) util.JSONResponse {
|
||||
return GetPostPublicRooms(req, rsAPI, stateAPI)
|
||||
return GetPostPublicRooms(req, rsAPI)
|
||||
}),
|
||||
).Methods(http.MethodGet)
|
||||
|
||||
|
|
|
|||
|
|
@ -303,6 +303,39 @@ type QueryServerBannedFromRoomResponse struct {
|
|||
Banned bool `json:"banned"`
|
||||
}
|
||||
|
||||
// MarshalJSON stringifies the room ID and StateKeyTuple keys so they can be sent over the wire in HTTP API mode.
|
||||
func (r *QueryBulkStateContentResponse) MarshalJSON() ([]byte, error) {
|
||||
se := make(map[string]string)
|
||||
for roomID, tupleToEvent := range r.Rooms {
|
||||
for tuple, event := range tupleToEvent {
|
||||
// use 0x1F (unit separator) as the delimiter between room ID/type/state key,
|
||||
se[fmt.Sprintf("%s\x1F%s\x1F%s", roomID, tuple.EventType, tuple.StateKey)] = event
|
||||
}
|
||||
}
|
||||
return json.Marshal(se)
|
||||
}
|
||||
|
||||
func (r *QueryBulkStateContentResponse) UnmarshalJSON(data []byte) error {
|
||||
wireFormat := make(map[string]string)
|
||||
err := json.Unmarshal(data, &wireFormat)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
r.Rooms = make(map[string]map[gomatrixserverlib.StateKeyTuple]string)
|
||||
for roomTuple, value := range wireFormat {
|
||||
fields := strings.Split(roomTuple, "\x1F")
|
||||
roomID := fields[0]
|
||||
if r.Rooms[roomID] == nil {
|
||||
r.Rooms[roomID] = make(map[gomatrixserverlib.StateKeyTuple]string)
|
||||
}
|
||||
r.Rooms[roomID][gomatrixserverlib.StateKeyTuple{
|
||||
EventType: fields[1],
|
||||
StateKey: fields[2],
|
||||
}] = value
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarshalJSON stringifies the StateKeyTuple keys so they can be sent over the wire in HTTP API mode.
|
||||
func (r *QueryCurrentStateResponse) MarshalJSON() ([]byte, error) {
|
||||
se := make(map[string]*gomatrixserverlib.HeaderedEvent, len(r.StateEvents))
|
||||
|
|
|
|||
|
|
@ -140,6 +140,9 @@ func (r *Queryer) QueryMembershipForUser(
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info == nil {
|
||||
return fmt.Errorf("QueryMembershipForUser: unknown room %s", request.RoomID)
|
||||
}
|
||||
|
||||
membershipEventNID, stillInRoom, err := r.DB.GetMembership(ctx, info.RoomNID, request.UserID)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -17,9 +17,9 @@ package storage
|
|||
import (
|
||||
"context"
|
||||
|
||||
"github.com/matrix-org/dendrite/currentstateserver/storage/tables"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage/shared"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ import (
|
|||
"fmt"
|
||||
"sort"
|
||||
|
||||
csstables "github.com/matrix-org/dendrite/currentstateserver/storage/tables"
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
|
@ -799,8 +798,90 @@ func (d *Database) GetRoomsByMembership(ctx context.Context, userID, membership
|
|||
|
||||
// GetBulkStateContent returns all state events which match a given room ID and a given state key tuple. Both must be satisfied for a match.
|
||||
// If a tuple has the StateKey of '*' and allowWildcards=true then all state events with the EventType should be returned.
|
||||
func (d *Database) GetBulkStateContent(ctx context.Context, roomIDs []string, tuples []gomatrixserverlib.StateKeyTuple, allowWildcards bool) ([]csstables.StrippedEvent, error) {
|
||||
return nil, fmt.Errorf("not implemented yet")
|
||||
// nolint:gocyclo
|
||||
func (d *Database) GetBulkStateContent(ctx context.Context, roomIDs []string, tuples []gomatrixserverlib.StateKeyTuple, allowWildcards bool) ([]tables.StrippedEvent, error) {
|
||||
eventTypes := make([]string, 0, len(tuples))
|
||||
for _, tuple := range tuples {
|
||||
eventTypes = append(eventTypes, tuple.EventType)
|
||||
}
|
||||
// we don't bother failing the request if we get asked for event types we don't know about, as all that would result in is no matches which
|
||||
// isn't a failure.
|
||||
eventTypeNIDMap, err := d.EventTypesTable.BulkSelectEventTypeNID(ctx, eventTypes)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to map event type nids: %w", err)
|
||||
}
|
||||
typeNIDSet := make(map[types.EventTypeNID]bool)
|
||||
for _, nid := range eventTypeNIDMap {
|
||||
typeNIDSet[nid] = true
|
||||
}
|
||||
|
||||
allowWildcard := make(map[types.EventTypeNID]bool)
|
||||
eventStateKeys := make([]string, 0, len(tuples))
|
||||
for _, tuple := range tuples {
|
||||
if allowWildcards && tuple.StateKey == "*" {
|
||||
allowWildcard[eventTypeNIDMap[tuple.EventType]] = true
|
||||
continue
|
||||
}
|
||||
eventStateKeys = append(eventStateKeys, tuple.StateKey)
|
||||
|
||||
}
|
||||
|
||||
eventStateKeyNIDMap, err := d.EventStateKeysTable.BulkSelectEventStateKeyNID(ctx, eventStateKeys)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to map state key nids: %w", err)
|
||||
}
|
||||
stateKeyNIDSet := make(map[types.EventStateKeyNID]bool)
|
||||
for _, nid := range eventStateKeyNIDMap {
|
||||
stateKeyNIDSet[nid] = true
|
||||
}
|
||||
|
||||
var eventNIDs []types.EventNID
|
||||
eventNIDToVer := make(map[types.EventNID]gomatrixserverlib.RoomVersion)
|
||||
// TODO: This feels like this is going to be really slow...
|
||||
for _, roomID := range roomIDs {
|
||||
roomInfo, err2 := d.RoomInfo(ctx, roomID)
|
||||
if err2 != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to load room info for room %s : %w", roomID, err2)
|
||||
}
|
||||
// for unknown rooms or rooms which we don't have the current state, skip them.
|
||||
if roomInfo == nil || roomInfo.IsStub {
|
||||
continue
|
||||
}
|
||||
entries, err2 := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
|
||||
if err2 != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to load state for room %s : %w", roomID, err2)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if typeNIDSet[entry.EventTypeNID] {
|
||||
if allowWildcard[entry.EventTypeNID] || stateKeyNIDSet[entry.EventStateKeyNID] {
|
||||
eventNIDs = append(eventNIDs, entry.EventNID)
|
||||
eventNIDToVer[entry.EventNID] = roomInfo.RoomVersion
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
events, err := d.EventJSONTable.BulkSelectEventJSON(ctx, eventNIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to load event JSON for event nids: %w", err)
|
||||
}
|
||||
result := make([]tables.StrippedEvent, len(events))
|
||||
for i := range events {
|
||||
roomVer := eventNIDToVer[events[i].EventNID]
|
||||
ev, err := gomatrixserverlib.NewEventFromTrustedJSON(events[i].EventJSON, false, roomVer)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetBulkStateContent: failed to load event JSON for event NID %v : %w", events[i].EventNID, err)
|
||||
}
|
||||
hev := ev.Headered(roomVer)
|
||||
result[i] = tables.StrippedEvent{
|
||||
EventType: ev.Type(),
|
||||
RoomID: ev.RoomID(),
|
||||
StateKey: *ev.StateKey(),
|
||||
ContentValue: tables.ExtractContentValue(&hev),
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// JoinedUsersSetInRooms returns all joined users in the rooms given, along with the count of how many times they appear.
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
|
||||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
type EventJSONPair struct {
|
||||
|
|
@ -155,3 +156,45 @@ type Redactions interface {
|
|||
// successfully redacted the event JSON.
|
||||
MarkRedactionValidated(ctx context.Context, txn *sql.Tx, redactionEventID string, validated bool) error
|
||||
}
|
||||
|
||||
// StrippedEvent represents a stripped event for returning extracted content values.
|
||||
type StrippedEvent struct {
|
||||
RoomID string
|
||||
EventType string
|
||||
StateKey string
|
||||
ContentValue string
|
||||
}
|
||||
|
||||
// ExtractContentValue from the given state event. For example, given an m.room.name event with:
|
||||
// content: { name: "Foo" }
|
||||
// this returns "Foo".
|
||||
func ExtractContentValue(ev *gomatrixserverlib.HeaderedEvent) string {
|
||||
content := ev.Content()
|
||||
key := ""
|
||||
switch ev.Type() {
|
||||
case gomatrixserverlib.MRoomCreate:
|
||||
key = "creator"
|
||||
case gomatrixserverlib.MRoomCanonicalAlias:
|
||||
key = "alias"
|
||||
case gomatrixserverlib.MRoomHistoryVisibility:
|
||||
key = "history_visibility"
|
||||
case gomatrixserverlib.MRoomJoinRules:
|
||||
key = "join_rule"
|
||||
case gomatrixserverlib.MRoomMember:
|
||||
key = "membership"
|
||||
case gomatrixserverlib.MRoomName:
|
||||
key = "name"
|
||||
case "m.room.avatar":
|
||||
key = "url"
|
||||
case "m.room.topic":
|
||||
key = "topic"
|
||||
case "m.room.guest_access":
|
||||
key = "guest_access"
|
||||
}
|
||||
result := gjson.GetBytes(content, key)
|
||||
if !result.Exists() {
|
||||
return ""
|
||||
}
|
||||
// this returns the empty string if this is not a string type
|
||||
return result.Str
|
||||
}
|
||||
|
|
|
|||
|
|
@ -133,7 +133,7 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er
|
|||
func (s *OutputKeyChangeEventConsumer) OnJoinEvent(ev *gomatrixserverlib.HeaderedEvent) {
|
||||
// work out who we are now sharing rooms with which we previously were not and notify them about the joining
|
||||
// users keys:
|
||||
changed, _, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, s.stateAPI, *ev.StateKey(), []string{ev.RoomID()}, nil)
|
||||
changed, _, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, *ev.StateKey(), []string{ev.RoomID()}, nil)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("OnJoinEvent: failed to work out changed users")
|
||||
return
|
||||
|
|
@ -146,7 +146,7 @@ func (s *OutputKeyChangeEventConsumer) OnJoinEvent(ev *gomatrixserverlib.Headere
|
|||
|
||||
func (s *OutputKeyChangeEventConsumer) OnLeaveEvent(ev *gomatrixserverlib.HeaderedEvent) {
|
||||
// work out who we are no longer sharing any rooms with and notify them about the leaving user
|
||||
_, left, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, s.stateAPI, *ev.StateKey(), nil, []string{ev.RoomID()})
|
||||
_, left, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, *ev.StateKey(), nil, []string{ev.RoomID()})
|
||||
if err != nil {
|
||||
log.WithError(err).Error("OnLeaveEvent: failed to work out left users")
|
||||
return
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ import (
|
|||
"strings"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
"github.com/matrix-org/dendrite/keyserver/api"
|
||||
keyapi "github.com/matrix-org/dendrite/keyserver/api"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
|
@ -50,7 +49,6 @@ func DeviceOTKCounts(ctx context.Context, keyAPI keyapi.KeyInternalAPI, userID,
|
|||
// nolint:gocyclo
|
||||
func DeviceListCatchup(
|
||||
ctx context.Context, keyAPI keyapi.KeyInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||
stateAPI currentstateAPI.CurrentStateInternalAPI,
|
||||
userID string, res *types.Response, from, to types.StreamingToken,
|
||||
) (hasNew bool, err error) {
|
||||
|
||||
|
|
@ -58,7 +56,7 @@ func DeviceListCatchup(
|
|||
newlyJoinedRooms := joinedRooms(res, userID)
|
||||
newlyLeftRooms := leftRooms(res)
|
||||
if len(newlyJoinedRooms) > 0 || len(newlyLeftRooms) > 0 {
|
||||
changed, left, err := TrackChangedUsers(ctx, rsAPI, stateAPI, userID, newlyJoinedRooms, newlyLeftRooms)
|
||||
changed, left, err := TrackChangedUsers(ctx, rsAPI, userID, newlyJoinedRooms, newlyLeftRooms)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
|
@ -144,7 +142,7 @@ func DeviceListCatchup(
|
|||
// TrackChangedUsers calculates the values of device_lists.changed|left in the /sync response.
|
||||
// nolint:gocyclo
|
||||
func TrackChangedUsers(
|
||||
ctx context.Context, rsAPI roomserverAPI.RoomserverInternalAPI, stateAPI currentstateAPI.CurrentStateInternalAPI, userID string, newlyJoinedRooms, newlyLeftRooms []string,
|
||||
ctx context.Context, rsAPI roomserverAPI.RoomserverInternalAPI, userID string, newlyJoinedRooms, newlyLeftRooms []string,
|
||||
) (changed, left []string, err error) {
|
||||
// process leaves first, then joins afterwards so if we join/leave/join/leave we err on the side of including users.
|
||||
|
||||
|
|
@ -161,8 +159,8 @@ func TrackChangedUsers(
|
|||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
var stateRes currentstateAPI.QueryBulkStateContentResponse
|
||||
err = stateAPI.QueryBulkStateContent(ctx, ¤tstateAPI.QueryBulkStateContentRequest{
|
||||
var stateRes roomserverAPI.QueryBulkStateContentResponse
|
||||
err = rsAPI.QueryBulkStateContent(ctx, &roomserverAPI.QueryBulkStateContentRequest{
|
||||
RoomIDs: newlyLeftRooms,
|
||||
StateTuples: []gomatrixserverlib.StateKeyTuple{
|
||||
{
|
||||
|
|
@ -202,7 +200,7 @@ func TrackChangedUsers(
|
|||
if err != nil {
|
||||
return nil, left, err
|
||||
}
|
||||
err = stateAPI.QueryBulkStateContent(ctx, ¤tstateAPI.QueryBulkStateContentRequest{
|
||||
err = rsAPI.QueryBulkStateContent(ctx, &roomserverAPI.QueryBulkStateContentRequest{
|
||||
RoomIDs: newlyJoinedRooms,
|
||||
StateTuples: []gomatrixserverlib.StateKeyTuple{
|
||||
{
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ import (
|
|||
"testing"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
stateapi "github.com/matrix-org/dendrite/currentstateserver/api"
|
||||
keyapi "github.com/matrix-org/dendrite/keyserver/api"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
|
@ -108,25 +107,6 @@ func (s *mockRoomserverAPI) QuerySharedUsers(ctx context.Context, req *api.Query
|
|||
return nil
|
||||
}
|
||||
|
||||
type mockStateAPI struct {
|
||||
rsAPI *mockRoomserverAPI
|
||||
}
|
||||
|
||||
// QueryBulkStateContent does a bulk query for state event content in the given rooms.
|
||||
func (s *mockStateAPI) QueryBulkStateContent(ctx context.Context, req *stateapi.QueryBulkStateContentRequest, res *stateapi.QueryBulkStateContentResponse) error {
|
||||
var res2 api.QueryBulkStateContentResponse
|
||||
err := s.rsAPI.QueryBulkStateContent(ctx, &api.QueryBulkStateContentRequest{
|
||||
RoomIDs: req.RoomIDs,
|
||||
AllowWildcards: req.AllowWildcards,
|
||||
StateTuples: req.StateTuples,
|
||||
}, &res2)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
res.Rooms = res2.Rooms
|
||||
return nil
|
||||
}
|
||||
|
||||
type wantCatchup struct {
|
||||
hasNew bool
|
||||
changed []string
|
||||
|
|
@ -200,7 +180,7 @@ func TestKeyChangeCatchupOnJoinShareNewUser(t *testing.T) {
|
|||
"!another:room": {syncingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("DeviceListCatchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -223,7 +203,7 @@ func TestKeyChangeCatchupOnLeaveShareLeftUser(t *testing.T) {
|
|||
"!another:room": {syncingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("DeviceListCatchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -246,7 +226,7 @@ func TestKeyChangeCatchupOnJoinShareNoNewUsers(t *testing.T) {
|
|||
"!another:room": {syncingUser, existingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("Catchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -268,7 +248,7 @@ func TestKeyChangeCatchupOnLeaveShareNoUsers(t *testing.T) {
|
|||
"!another:room": {syncingUser, existingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("DeviceListCatchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -327,7 +307,7 @@ func TestKeyChangeCatchupNoNewJoinsButMessages(t *testing.T) {
|
|||
roomID: {syncingUser, existingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("DeviceListCatchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -355,7 +335,7 @@ func TestKeyChangeCatchupChangeAndLeft(t *testing.T) {
|
|||
"!another:room": {syncingUser},
|
||||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
hasNew, err := DeviceListCatchup(context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken)
|
||||
if err != nil {
|
||||
t.Fatalf("Catchup returned an error: %s", err)
|
||||
}
|
||||
|
|
@ -441,7 +421,7 @@ func TestKeyChangeCatchupChangeAndLeftSameRoom(t *testing.T) {
|
|||
},
|
||||
}
|
||||
hasNew, err := DeviceListCatchup(
|
||||
context.Background(), &mockKeyAPI{}, rsAPI, &mockStateAPI{rsAPI}, syncingUser, syncResponse, emptyToken, newestToken,
|
||||
context.Background(), &mockKeyAPI{}, rsAPI, syncingUser, syncResponse, emptyToken, newestToken,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("DeviceListCatchup returned an error: %s", err)
|
||||
|
|
|
|||
|
|
@ -267,7 +267,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea
|
|||
func (rp *RequestPool) appendDeviceLists(
|
||||
data *types.Response, userID string, since, to types.StreamingToken,
|
||||
) (*types.Response, error) {
|
||||
_, err := internal.DeviceListCatchup(context.Background(), rp.keyAPI, rp.rsAPI, rp.stateAPI, userID, data, since, to)
|
||||
_, err := internal.DeviceListCatchup(context.Background(), rp.keyAPI, rp.rsAPI, userID, data, since, to)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("internal.DeviceListCatchup: %w", err)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue