hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-08-02 06:12:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into matthew/peeking

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-03 21:01:30 +01:00
commit 8712ea337a
79 changed files with 4073 additions and 1928 deletions
Download patch file Download diff file Expand all files Collapse all files

10
roomserver/internal/alias.go
View file

@ -18,6 +18,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/matrix-org/dendrite/roomserver/api"
@ -239,16 +240,19 @@ func (r *RoomserverInternalAPI) sendUpdatedAliasesEvent(
}
builder.AuthEvents = refs
roomVersion, err := r.DB.GetRoomVersionForRoom(ctx, roomID)
roomInfo, err := r.DB.RoomInfo(ctx, roomID)
if err != nil {
return err
}
if roomInfo == nil {
return fmt.Errorf("room %s does not exist", roomID)
}
// Build the event
now := time.Now()
event, err := builder.Build(
now, r.Cfg.Matrix.ServerName, r.Cfg.Matrix.KeyID,
r.Cfg.Matrix.PrivateKey, roomVersion,
r.Cfg.Matrix.PrivateKey, roomInfo.RoomVersion,
)
if err != nil {
return err
@ -257,7 +261,7 @@ func (r *RoomserverInternalAPI) sendUpdatedAliasesEvent(
// Create the request
ire := api.InputRoomEvent{
Kind: api.KindNew,
Event: event.Headered(roomVersion),
Event: event.Headered(roomInfo.RoomVersion),
AuthEventIDs: event.AuthEventIDs(),
SendAsServer: serverName,
}

111
roomserver/internal/api.go
View file

@ -1,26 +1,129 @@
package internal
import (
"sync"
"context"
"github.com/Shopify/sarama"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/roomserver/acls"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/internal/perform"
"github.com/matrix-org/dendrite/roomserver/internal/query"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/gomatrixserverlib"
)
// RoomserverInternalAPI is an implementation of api.RoomserverInternalAPI
type RoomserverInternalAPI struct {
*input.Inputer
*query.Queryer
*perform.Inviter
*perform.Joiner
*perform.Leaver
*perform.Publisher
*perform.Backfiller
DB storage.Database
Cfg *config.RoomServer
Producer sarama.SyncProducer
Cache caching.RoomServerCaches
ServerName gomatrixserverlib.ServerName
KeyRing gomatrixserverlib.JSONVerifier
FedClient *gomatrixserverlib.FederationClient
OutputRoomEventTopic string // Kafka topic for new output room events
mutexes sync.Map // room ID -> *sync.Mutex, protects calls to processRoomEvent
fsAPI fsAPI.FederationSenderInternalAPI
OutputRoomEventTopic string // Kafka topic for new output room events
}
func NewRoomserverAPI(
cfg *config.RoomServer, roomserverDB storage.Database, producer sarama.SyncProducer,
outputRoomEventTopic string, caches caching.RoomServerCaches,
keyRing gomatrixserverlib.JSONVerifier,
) *RoomserverInternalAPI {
a := &RoomserverInternalAPI{
DB: roomserverDB,
Cfg: cfg,
Cache: caches,
ServerName: cfg.Matrix.ServerName,
KeyRing: keyRing,
Queryer: &query.Queryer{
DB: roomserverDB,
Cache: caches,
ServerACLs: acls.NewServerACLs(roomserverDB),
},
Inputer: &input.Inputer{
DB: roomserverDB,
OutputRoomEventTopic: outputRoomEventTopic,
Producer: producer,
ServerName: cfg.Matrix.ServerName,
},
// perform-er structs get initialised when we have a federation sender to use
}
return a
}
// SetFederationSenderInputAPI passes in a federation sender input API reference
// so that we can avoid the chicken-and-egg problem of both the roomserver input API
// and the federation sender input API being interdependent.
func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSenderInternalAPI) {
r.fsAPI = fsAPI
r.Inviter = &perform.Inviter{
DB: r.DB,
Cfg: r.Cfg,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
}
r.Joiner = &perform.Joiner{
ServerName: r.Cfg.Matrix.ServerName,
Cfg: r.Cfg,
DB: r.DB,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
}
r.Leaver = &perform.Leaver{
Cfg: r.Cfg,
DB: r.DB,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
}
r.Publisher = &perform.Publisher{
DB: r.DB,
}
r.Backfiller = &perform.Backfiller{
ServerName: r.ServerName,
DB: r.DB,
FSAPI: r.fsAPI,
KeyRing: r.KeyRing,
}
}
func (r *RoomserverInternalAPI) PerformInvite(
ctx context.Context,
req *api.PerformInviteRequest,
res *api.PerformInviteResponse,
) error {
outputEvents, err := r.Inviter.PerformInvite(ctx, req, res)
if err != nil {
return err
}
if len(outputEvents) == 0 {
return nil
}
return r.WriteOutputEvents(req.Event.RoomID(), outputEvents)
}
func (r *RoomserverInternalAPI) PerformLeave(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse,
) error {
outputEvents, err := r.Leaver.PerformLeave(ctx, req, res)
if err != nil {
return err
}
if len(outputEvents) == 0 {
return nil
}
return r.WriteOutputEvents(req.RoomID, outputEvents)
}

16
roomserver/internal/input_authevents.go → roomserver/internal/helpers/auth.go
View file

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package helpers
import (
"context"
@ -23,9 +23,9 @@ import (
"github.com/matrix-org/gomatrixserverlib"
)
// checkAuthEvents checks that the event passes authentication checks
// CheckAuthEvents checks that the event passes authentication checks
// Returns the numeric IDs for the auth events.
func checkAuthEvents(
func CheckAuthEvents(
ctx context.Context,
db storage.Database,
event gomatrixserverlib.HeaderedEvent,
@ -63,7 +63,7 @@ func checkAuthEvents(
type authEvents struct {
stateKeyNIDMap map[string]types.EventStateKeyNID
state stateEntryMap
events eventMap
events EventMap
}
// Create implements gomatrixserverlib.AuthEventProvider
@ -99,7 +99,7 @@ func (ae *authEvents) lookupEventWithEmptyStateKey(typeNID types.EventTypeNID) *
if !ok {
return nil
}
event, ok := ae.events.lookup(eventNID)
event, ok := ae.events.Lookup(eventNID)
if !ok {
return nil
}
@ -118,7 +118,7 @@ func (ae *authEvents) lookupEvent(typeNID types.EventTypeNID, stateKey string) *
if !ok {
return nil
}
event, ok := ae.events.lookup(eventNID)
event, ok := ae.events.Lookup(eventNID)
if !ok {
return nil
}
@ -224,10 +224,10 @@ func (m stateEntryMap) lookup(stateKey types.StateKeyTuple) (eventNID types.Even
// Map from numeric event ID to event.
// Implemented using binary search on a sorted array.
type eventMap []types.Event
type EventMap []types.Event
// lookup an entry in the event map.
func (m eventMap) lookup(eventNID types.EventNID) (event *types.Event, ok bool) {
func (m EventMap) Lookup(eventNID types.EventNID) (event *types.Event, ok bool) {
// Since the list is sorted we can implement this using binary search.
// This is faster than using a hash map.
// We don't have to worry about pathological cases because the keys are fixed

6
roomserver/internal/input_authevents_test.go → roomserver/internal/helpers/auth_test.go
View file

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package helpers
import (
"testing"
@ -95,7 +95,7 @@ func TestStateEntryMap(t *testing.T) {
}
func TestEventMap(t *testing.T) {
events := eventMap([]types.Event{
events := EventMap([]types.Event{
{EventNID: 1},
{EventNID: 2},
{EventNID: 3},
@ -123,7 +123,7 @@ func TestEventMap(t *testing.T) {
}
for _, testCase := range testCases {
gotEvent, gotOK := events.lookup(testCase.inputEventNID)
gotEvent, gotOK := events.Lookup(testCase.inputEventNID)
if testCase.wantOK != gotOK {
t.Fatalf("eventMap lookup(%v): want ok to be %v, got %v", testCase.inputEventNID, testCase.wantOK, gotOK)
}

379
roomserver/internal/helpers/helpers.go Normal file
View file

@ -0,0 +1,379 @@
package helpers
import (
"context"
"fmt"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/auth"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
// TODO: temporary package which has helper functions used by both internal/perform packages.
// Move these to a more sensible place.
func UpdateToInviteMembership(
mu *shared.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
roomVersion gomatrixserverlib.RoomVersion,
) ([]api.OutputEvent, error) {
// We may have already sent the invite to the user, either because we are
// reprocessing this event, or because the we received this invite from a
// remote server via the federation invite API. In those cases we don't need
// to send the event.
needsSending, err := mu.SetToInvite(*add)
if err != nil {
return nil, err
}
if needsSending {
// We notify the consumers using a special event even though we will
// notify them about the change in current state as part of the normal
// room event stream. This ensures that the consumers only have to
// consider a single stream of events when determining whether a user
// is invited, rather than having to combine multiple streams themselves.
onie := api.OutputNewInviteEvent{
Event: add.Headered(roomVersion),
RoomVersion: roomVersion,
}
updates = append(updates, api.OutputEvent{
Type: api.OutputTypeNewInviteEvent,
NewInviteEvent: &onie,
})
}
return updates, nil
}
func IsServerCurrentlyInRoom(ctx context.Context, db storage.Database, serverName gomatrixserverlib.ServerName, roomID string) (bool, error) {
info, err := db.RoomInfo(ctx, roomID)
if err != nil {
return false, err
}
if info == nil {
return false, fmt.Errorf("unknown room %s", roomID)
}
eventNIDs, err := db.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, true, false)
if err != nil {
return false, err
}
events, err := db.Events(ctx, eventNIDs)
if err != nil {
return false, err
}
gmslEvents := make([]gomatrixserverlib.Event, len(events))
for i := range events {
gmslEvents[i] = events[i].Event
}
return auth.IsAnyUserOnServerWithMembership(serverName, gmslEvents, gomatrixserverlib.Join), nil
}
func IsInvitePending(
ctx context.Context, db storage.Database,
roomID, userID string,
) (bool, string, string, error) {
// Look up the room NID for the supplied room ID.
info, err := db.RoomInfo(ctx, roomID)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.RoomInfo: %w", err)
}
if info == nil {
return false, "", "", fmt.Errorf("cannot get RoomInfo: unknown room ID %s", roomID)
}
// Look up the state key NID for the supplied user ID.
targetUserNIDs, err := db.EventStateKeyNIDs(ctx, []string{userID})
if err != nil {
return false, "", "", fmt.Errorf("r.DB.EventStateKeyNIDs: %w", err)
}
targetUserNID, targetUserFound := targetUserNIDs[userID]
if !targetUserFound {
return false, "", "", fmt.Errorf("missing NID for user %q (%+v)", userID, targetUserNIDs)
}
// Let's see if we have an event active for the user in the room. If
// we do then it will contain a server name that we can direct the
// send_leave to.
senderUserNIDs, eventIDs, err := db.GetInvitesForUser(ctx, info.RoomNID, targetUserNID)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.GetInvitesForUser: %w", err)
}
if len(senderUserNIDs) == 0 {
return false, "", "", nil
}
userNIDToEventID := make(map[types.EventStateKeyNID]string)
for i, nid := range senderUserNIDs {
userNIDToEventID[nid] = eventIDs[i]
}
// Look up the user ID from the NID.
senderUsers, err := db.EventStateKeys(ctx, senderUserNIDs)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.EventStateKeys: %w", err)
}
if len(senderUsers) == 0 {
return false, "", "", fmt.Errorf("no senderUsers")
}
senderUser, senderUserFound := senderUsers[senderUserNIDs[0]]
if !senderUserFound {
return false, "", "", fmt.Errorf("missing user for NID %d (%+v)", senderUserNIDs[0], senderUsers)
}
return true, senderUser, userNIDToEventID[senderUserNIDs[0]], nil
}
// GetMembershipsAtState filters the state events to
// only keep the "m.room.member" events with a "join" membership. These events are returned.
// Returns an error if there was an issue fetching the events.
func GetMembershipsAtState(
ctx context.Context, db storage.Database, stateEntries []types.StateEntry, joinedOnly bool,
) ([]types.Event, error) {
var eventNIDs []types.EventNID
for _, entry := range stateEntries {
// Filter the events to retrieve to only keep the membership events
if entry.EventTypeNID == types.MRoomMemberNID {
eventNIDs = append(eventNIDs, entry.EventNID)
}
}
// Get all of the events in this state
stateEvents, err := db.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
if !joinedOnly {
return stateEvents, nil
}
// Filter the events to only keep the "join" membership events
var events []types.Event
for _, event := range stateEvents {
membership, err := event.Membership()
if err != nil {
return nil, err
}
if membership == gomatrixserverlib.Join {
events = append(events, event)
}
}
return events, nil
}
func StateBeforeEvent(ctx context.Context, db storage.Database, info types.RoomInfo, eventNID types.EventNID) ([]types.StateEntry, error) {
roomState := state.NewStateResolution(db, info)
// Lookup the event NID
eIDs, err := db.EventIDs(ctx, []types.EventNID{eventNID})
if err != nil {
return nil, err
}
eventIDs := []string{eIDs[eventNID]}
prevState, err := db.StateAtEventIDs(ctx, eventIDs)
if err != nil {
return nil, err
}
// Fetch the state as it was when this event was fired
return roomState.LoadCombinedStateAfterEvents(ctx, prevState)
}
func LoadEvents(
ctx context.Context, db storage.Database, eventNIDs []types.EventNID,
) ([]gomatrixserverlib.Event, error) {
stateEvents, err := db.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
result := make([]gomatrixserverlib.Event, len(stateEvents))
for i := range stateEvents {
result[i] = stateEvents[i].Event
}
return result, nil
}
func LoadStateEvents(
ctx context.Context, db storage.Database, stateEntries []types.StateEntry,
) ([]gomatrixserverlib.Event, error) {
eventNIDs := make([]types.EventNID, len(stateEntries))
for i := range stateEntries {
eventNIDs[i] = stateEntries[i].EventNID
}
return LoadEvents(ctx, db, eventNIDs)
}
func CheckServerAllowedToSeeEvent(
ctx context.Context, db storage.Database, info types.RoomInfo, eventID string, serverName gomatrixserverlib.ServerName, isServerInRoom bool,
) (bool, error) {
roomState := state.NewStateResolution(db, info)
stateEntries, err := roomState.LoadStateAtEvent(ctx, eventID)
if err != nil {
return false, err
}
// TODO: We probably want to make it so that we don't have to pull
// out all the state if possible.
stateAtEvent, err := LoadStateEvents(ctx, db, stateEntries)
if err != nil {
return false, err
}
return auth.IsServerAllowed(serverName, isServerInRoom, stateAtEvent), nil
}
// TODO: Remove this when we have tests to assert correctness of this function
// nolint:gocyclo
func ScanEventTree(
ctx context.Context, db storage.Database, info types.RoomInfo, front []string, visited map[string]bool, limit int,
serverName gomatrixserverlib.ServerName,
) ([]types.EventNID, error) {
var resultNIDs []types.EventNID
var err error
var allowed bool
var events []types.Event
var next []string
var pre string
// TODO: add tests for this function to ensure it meets the contract that callers expect (and doc what that is supposed to be)
// Currently, callers like PerformBackfill will call scanEventTree with a pre-populated `visited` map, assuming that by doing
// so means that the events in that map will NOT be returned from this function. That is not currently true, resulting in
// duplicate events being sent in response to /backfill requests.
initialIgnoreList := make(map[string]bool, len(visited))
for k, v := range visited {
initialIgnoreList[k] = v
}
resultNIDs = make([]types.EventNID, 0, limit)
var checkedServerInRoom bool
var isServerInRoom bool
// Loop through the event IDs to retrieve the requested events and go
// through the whole tree (up to the provided limit) using the events'
// "prev_event" key.
BFSLoop:
for len(front) > 0 {
// Prevent unnecessary allocations: reset the slice only when not empty.
if len(next) > 0 {
next = make([]string, 0)
}
// Retrieve the events to process from the database.
events, err = db.EventsFromIDs(ctx, front)
if err != nil {
return resultNIDs, err
}
if !checkedServerInRoom && len(events) > 0 {
// It's nasty that we have to extract the room ID from an event, but many federation requests
// only talk in event IDs, no room IDs at all (!!!)
ev := events[0]
isServerInRoom, err = IsServerCurrentlyInRoom(ctx, db, serverName, ev.RoomID())
if err != nil {
util.GetLogger(ctx).WithError(err).Error("Failed to check if server is currently in room, assuming not.")
}
checkedServerInRoom = true
}
for _, ev := range events {
// Break out of the loop if the provided limit is reached.
if len(resultNIDs) == limit {
break BFSLoop
}
if !initialIgnoreList[ev.EventID()] {
// Update the list of events to retrieve.
resultNIDs = append(resultNIDs, ev.EventNID)
}
// Loop through the event's parents.
for _, pre = range ev.PrevEventIDs() {
// Only add an event to the list of next events to process if it
// hasn't been seen before.
if !visited[pre] {
visited[pre] = true
allowed, err = CheckServerAllowedToSeeEvent(ctx, db, info, pre, serverName, isServerInRoom)
if err != nil {
util.GetLogger(ctx).WithField("server", serverName).WithField("event_id", pre).WithError(err).Error(
"Error checking if allowed to see event",
)
return resultNIDs, err
}
// If the event hasn't been seen before and the HS
// requesting to retrieve it is allowed to do so, add it to
// the list of events to retrieve.
if allowed {
next = append(next, pre)
} else {
util.GetLogger(ctx).WithField("server", serverName).WithField("event_id", pre).Info("Not allowed to see event")
}
}
}
}
// Repeat the same process with the parent events we just processed.
front = next
}
return resultNIDs, err
}
func QueryLatestEventsAndState(
ctx context.Context, db storage.Database,
request *api.QueryLatestEventsAndStateRequest,
response *api.QueryLatestEventsAndStateResponse,
) error {
roomInfo, err := db.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if roomInfo == nil || roomInfo.IsStub {
response.RoomExists = false
return nil
}
roomState := state.NewStateResolution(db, *roomInfo)
response.RoomExists = true
response.RoomVersion = roomInfo.RoomVersion
var currentStateSnapshotNID types.StateSnapshotNID
response.LatestEvents, currentStateSnapshotNID, response.Depth, err =
db.LatestEventIDs(ctx, roomInfo.RoomNID)
if err != nil {
return err
}
var stateEntries []types.StateEntry
if len(request.StateToFetch) == 0 {
// Look up all room state.
stateEntries, err = roomState.LoadStateAtSnapshot(
ctx, currentStateSnapshotNID,
)
} else {
// Look up the current state for the requested tuples.
stateEntries, err = roomState.LoadStateAtSnapshotForStringTuples(
ctx, currentStateSnapshotNID, request.StateToFetch,
)
}
if err != nil {
return err
}
stateEvents, err := LoadStateEvents(ctx, db, stateEntries)
if err != nil {
return err
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(roomInfo.RoomVersion))
}
return nil
}

89
roomserver/internal/input.go
View file

@ -1,89 +0,0 @@
// 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 input contains the code processes new room events
package internal
import (
"context"
"encoding/json"
"sync"
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/roomserver/api"
log "github.com/sirupsen/logrus"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
)
// SetFederationSenderInputAPI passes in a federation sender input API reference
// so that we can avoid the chicken-and-egg problem of both the roomserver input API
// and the federation sender input API being interdependent.
func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSenderInternalAPI) {
r.fsAPI = fsAPI
}
// WriteOutputEvents implements OutputRoomEventWriter
func (r *RoomserverInternalAPI) WriteOutputEvents(roomID string, updates []api.OutputEvent) error {
messages := make([]*sarama.ProducerMessage, len(updates))
for i := range updates {
value, err := json.Marshal(updates[i])
if err != nil {
return err
}
logger := log.WithFields(log.Fields{
"room_id": roomID,
"type": updates[i].Type,
})
if updates[i].NewRoomEvent != nil {
logger = logger.WithFields(log.Fields{
"event_type": updates[i].NewRoomEvent.Event.Type(),
"event_id": updates[i].NewRoomEvent.Event.EventID(),
"adds_state": len(updates[i].NewRoomEvent.AddsStateEventIDs),
"removes_state": len(updates[i].NewRoomEvent.RemovesStateEventIDs),
"send_as_server": updates[i].NewRoomEvent.SendAsServer,
"sender": updates[i].NewRoomEvent.Event.Sender(),
})
}
logger.Infof("Producing to topic '%s'", r.OutputRoomEventTopic)
messages[i] = &sarama.ProducerMessage{
Topic: r.OutputRoomEventTopic,
Key: sarama.StringEncoder(roomID),
Value: sarama.ByteEncoder(value),
}
}
return r.Producer.SendMessages(messages)
}
// InputRoomEvents implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) InputRoomEvents(
ctx context.Context,
request *api.InputRoomEventsRequest,
response *api.InputRoomEventsResponse,
) (err error) {
for i, e := range request.InputRoomEvents {
roomID := "global"
if r.DB.SupportsConcurrentRoomInputs() {
roomID = e.Event.RoomID()
}
mutex, _ := r.mutexes.LoadOrStore(roomID, &sync.Mutex{})
mutex.(*sync.Mutex).Lock()
if response.EventID, err = r.processRoomEvent(ctx, request.InputRoomEvents[i]); err != nil {
mutex.(*sync.Mutex).Unlock()
return err
}
mutex.(*sync.Mutex).Unlock()
}
return nil
}

157
roomserver/internal/input/input.go Normal file
View file

@ -0,0 +1,157 @@
// 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 input contains the code processes new room events
package input
import (
"context"
"encoding/json"
"sync"
"time"
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/gomatrixserverlib"
log "github.com/sirupsen/logrus"
"go.uber.org/atomic"
)
type Inputer struct {
DB storage.Database
Producer sarama.SyncProducer
ServerName gomatrixserverlib.ServerName
OutputRoomEventTopic string
workers sync.Map // room ID -> *inputWorker
}
type inputTask struct {
ctx context.Context
event *api.InputRoomEvent
wg *sync.WaitGroup
err error // written back by worker, only safe to read when all tasks are done
}
type inputWorker struct {
r *Inputer
running atomic.Bool
input chan *inputTask
}
func (w *inputWorker) start() {
if !w.running.CAS(false, true) {
return
}
defer w.running.Store(false)
for {
select {
case task := <-w.input:
_, task.err = w.r.processRoomEvent(task.ctx, task.event)
task.wg.Done()
case <-time.After(time.Second * 5):
return
}
}
}
// WriteOutputEvents implements OutputRoomEventWriter
func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) error {
messages := make([]*sarama.ProducerMessage, len(updates))
for i := range updates {
value, err := json.Marshal(updates[i])
if err != nil {
return err
}
logger := log.WithFields(log.Fields{
"room_id": roomID,
"type": updates[i].Type,
})
if updates[i].NewRoomEvent != nil {
logger = logger.WithFields(log.Fields{
"event_type": updates[i].NewRoomEvent.Event.Type(),
"event_id": updates[i].NewRoomEvent.Event.EventID(),
"adds_state": len(updates[i].NewRoomEvent.AddsStateEventIDs),
"removes_state": len(updates[i].NewRoomEvent.RemovesStateEventIDs),
"send_as_server": updates[i].NewRoomEvent.SendAsServer,
"sender": updates[i].NewRoomEvent.Event.Sender(),
})
}
logger.Infof("Producing to topic '%s'", r.OutputRoomEventTopic)
messages[i] = &sarama.ProducerMessage{
Topic: r.OutputRoomEventTopic,
Key: sarama.StringEncoder(roomID),
Value: sarama.ByteEncoder(value),
}
}
return r.Producer.SendMessages(messages)
}
// InputRoomEvents implements api.RoomserverInternalAPI
func (r *Inputer) InputRoomEvents(
ctx context.Context,
request *api.InputRoomEventsRequest,
response *api.InputRoomEventsResponse,
) error {
// Create a wait group. Each task that we dispatch will call Done on
// this wait group so that we know when all of our events have been
// processed.
wg := &sync.WaitGroup{}
wg.Add(len(request.InputRoomEvents))
tasks := make([]*inputTask, len(request.InputRoomEvents))
for i, e := range request.InputRoomEvents {
// Work out if we are running per-room workers or if we're just doing
// it on a global basis (e.g. SQLite).
roomID := "global"
if r.DB.SupportsConcurrentRoomInputs() {
roomID = e.Event.RoomID()
}
// Look up the worker, or create it if it doesn't exist. This channel
// is buffered to reduce the chance that we'll be blocked by another
// room - the channel will be quite small as it's just pointer types.
w, _ := r.workers.LoadOrStore(roomID, &inputWorker{
r: r,
input: make(chan *inputTask, 10),
})
worker := w.(*inputWorker)
// Create a task. This contains the input event and a reference to
// the wait group, so that the worker can notify us when this specific
// task has been finished.
tasks[i] = &inputTask{
ctx: ctx,
event: &request.InputRoomEvents[i],
wg: wg,
}
// Send the task to the worker.
go worker.start()
worker.input <- tasks[i]
}
// Wait for all of the workers to return results about our tasks.
wg.Wait()
// If any of the tasks returned an error, we should probably report
// that back to the caller.
for _, task := range tasks {
if task.err != nil {
return task.err
}
}
return nil
}

37
roomserver/internal/input_events.go → roomserver/internal/input/input_events.go
View file

@ -14,7 +14,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package input
import (
"context"
@ -22,6 +22,7 @@ import (
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
@ -35,9 +36,9 @@ import (
// state deltas when sending to kafka streams
// TODO: Break up function - we should probably do transaction ID checks before calling this.
// nolint:gocyclo
func (r *RoomserverInternalAPI) processRoomEvent(
func (r *Inputer) processRoomEvent(
ctx context.Context,
input api.InputRoomEvent,
input *api.InputRoomEvent,
) (eventID string, err error) {
// Parse and validate the event JSON
headered := input.Event
@ -45,7 +46,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
// Check that the event passes authentication checks and work out
// the numeric IDs for the auth events.
authEventNIDs, err := checkAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
authEventNIDs, err := helpers.CheckAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
if err != nil {
logrus.WithError(err).WithField("event_id", event.EventID()).WithField("auth_event_ids", input.AuthEventIDs).Error("processRoomEvent.checkAuthEvents failed for event")
return
@ -64,7 +65,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
}
// Store the event.
roomNID, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs)
_, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs)
if err != nil {
return "", fmt.Errorf("r.DB.StoreEvent: %w", err)
}
@ -89,10 +90,18 @@ func (r *RoomserverInternalAPI) processRoomEvent(
return event.EventID(), nil
}
roomInfo, err := r.DB.RoomInfo(ctx, event.RoomID())
if err != nil {
return "", fmt.Errorf("r.DB.RoomInfo: %w", err)
}
if roomInfo == nil {
return "", fmt.Errorf("r.DB.RoomInfo missing for room %s", event.RoomID())
}
if stateAtEvent.BeforeStateSnapshotNID == 0 {
// We haven't calculated a state for this event yet.
// Lets calculate one.
err = r.calculateAndSetState(ctx, input, roomNID, &stateAtEvent, event)
err = r.calculateAndSetState(ctx, input, *roomInfo, &stateAtEvent, event)
if err != nil {
return "", fmt.Errorf("r.calculateAndSetState: %w", err)
}
@ -100,7 +109,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
if err = r.updateLatestEvents(
ctx, // context
roomNID, // room NID to update
roomInfo, // room info for the room being updated
stateAtEvent, // state at event (below)
event, // event
input.SendAsServer, // send as server
@ -132,22 +141,22 @@ func (r *RoomserverInternalAPI) processRoomEvent(
return event.EventID(), nil
}
func (r *RoomserverInternalAPI) calculateAndSetState(
func (r *Inputer) calculateAndSetState(
ctx context.Context,
input api.InputRoomEvent,
roomNID types.RoomNID,
input *api.InputRoomEvent,
roomInfo types.RoomInfo,
stateAtEvent *types.StateAtEvent,
event gomatrixserverlib.Event,
) error {
var err error
roomState := state.NewStateResolution(r.DB)
roomState := state.NewStateResolution(r.DB, roomInfo)
if input.HasState {
// Check here if we think we're in the room already.
stateAtEvent.Overwrite = true
var joinEventNIDs []types.EventNID
// Request join memberships only for local users only.
if joinEventNIDs, err = r.DB.GetMembershipEventNIDsForRoom(ctx, roomNID, true, true); err == nil {
if joinEventNIDs, err = r.DB.GetMembershipEventNIDsForRoom(ctx, roomInfo.RoomNID, true, true); err == nil {
// If we have no local users that are joined to the room then any state about
// the room that we have is quite possibly out of date. Therefore in that case
// we should overwrite it rather than merge it.
@ -161,14 +170,14 @@ func (r *RoomserverInternalAPI) calculateAndSetState(
return err
}
if stateAtEvent.BeforeStateSnapshotNID, err = r.DB.AddState(ctx, roomNID, nil, entries); err != nil {
if stateAtEvent.BeforeStateSnapshotNID, err = r.DB.AddState(ctx, roomInfo.RoomNID, nil, entries); err != nil {
return err
}
} else {
stateAtEvent.Overwrite = false
// We haven't been told what the state at the event is so we need to calculate it from the prev_events
if stateAtEvent.BeforeStateSnapshotNID, err = roomState.CalculateAndStoreStateBeforeEvent(ctx, event, roomNID); err != nil {
if stateAtEvent.BeforeStateSnapshotNID, err = roomState.CalculateAndStoreStateBeforeEvent(ctx, event); err != nil {
return err
}
}

29
roomserver/internal/input_latest_events.go → roomserver/internal/input/input_latest_events.go
View file

@ -14,7 +14,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package input
import (
"bytes"
@ -47,15 +47,15 @@ import (
// 7 <----- latest
//
// Can only be called once at a time
func (r *RoomserverInternalAPI) updateLatestEvents(
func (r *Inputer) updateLatestEvents(
ctx context.Context,
roomNID types.RoomNID,
roomInfo *types.RoomInfo,
stateAtEvent types.StateAtEvent,
event gomatrixserverlib.Event,
sendAsServer string,
transactionID *api.TransactionID,
) (err error) {
updater, err := r.DB.GetLatestEventsForUpdate(ctx, roomNID)
updater, err := r.DB.GetLatestEventsForUpdate(ctx, *roomInfo)
if err != nil {
return fmt.Errorf("r.DB.GetLatestEventsForUpdate: %w", err)
}
@ -66,7 +66,7 @@ func (r *RoomserverInternalAPI) updateLatestEvents(
ctx: ctx,
api: r,
updater: updater,
roomNID: roomNID,
roomInfo: roomInfo,
stateAtEvent: stateAtEvent,
event: event,
sendAsServer: sendAsServer,
@ -87,9 +87,9 @@ func (r *RoomserverInternalAPI) updateLatestEvents(
// when there are so many variables to pass around.
type latestEventsUpdater struct {
ctx context.Context
api *RoomserverInternalAPI
api *Inputer
updater *shared.LatestEventsUpdater
roomNID types.RoomNID
roomInfo *types.RoomInfo
stateAtEvent types.StateAtEvent
event gomatrixserverlib.Event
transactionID *api.TransactionID
@ -196,7 +196,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
return fmt.Errorf("u.api.WriteOutputEvents: %w", err)
}
if err = u.updater.SetLatestEvents(u.roomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil {
if err = u.updater.SetLatestEvents(u.roomInfo.RoomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil {
return fmt.Errorf("u.updater.SetLatestEvents: %w", err)
}
@ -209,7 +209,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
func (u *latestEventsUpdater) latestState() error {
var err error
roomState := state.NewStateResolution(u.api.DB)
roomState := state.NewStateResolution(u.api.DB, *u.roomInfo)
// Get a list of the current latest events.
latestStateAtEvents := make([]types.StateAtEvent, len(u.latest))
@ -221,7 +221,7 @@ func (u *latestEventsUpdater) latestState() error {
// of the state after the events. The snapshot state will be resolved
// using the correct state resolution algorithm for the room.
u.newStateNID, err = roomState.CalculateAndStoreStateAfterEvents(
u.ctx, u.roomNID, latestStateAtEvents,
u.ctx, latestStateAtEvents,
)
if err != nil {
return fmt.Errorf("roomState.CalculateAndStoreStateAfterEvents: %w", err)
@ -303,13 +303,8 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
latestEventIDs[i] = u.latest[i].EventID
}
roomVersion, err := u.api.DB.GetRoomVersionForRoom(u.ctx, u.event.RoomID())
if err != nil {
return nil, err
}
ore := api.OutputNewRoomEvent{
Event: u.event.Headered(roomVersion),
Event: u.event.Headered(u.roomInfo.RoomVersion),
LastSentEventID: u.lastEventIDSent,
LatestEventIDs: latestEventIDs,
TransactionID: u.transactionID,
@ -337,7 +332,7 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
// include extra state events if they were added as nearly every downstream component will care about it
// and we'd rather not have them all hit QueryEventsByID at the same time!
if len(ore.AddsStateEventIDs) > 0 {
ore.AddStateEvents, err = u.extraEventsForIDs(roomVersion, ore.AddsStateEventIDs)
ore.AddStateEvents, err = u.extraEventsForIDs(u.roomInfo.RoomVersion, ore.AddsStateEventIDs)
if err != nil {
return nil, fmt.Errorf("failed to load add_state_events from db: %w", err)
}

47
roomserver/internal/input_membership.go → roomserver/internal/input/input_membership.go
View file

@ -12,13 +12,14 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package input
import (
"context"
"fmt"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
@ -28,7 +29,7 @@ import (
// user affected by a change in the current state of the room.
// Returns a list of output events to write to the kafka log to inform the
// consumers about the invites added or retired by the change in current state.
func (r *RoomserverInternalAPI) updateMemberships(
func (r *Inputer) updateMemberships(
ctx context.Context,
updater *shared.LatestEventsUpdater,
removed, added []types.StateEntry,
@ -59,13 +60,13 @@ func (r *RoomserverInternalAPI) updateMemberships(
var re *gomatrixserverlib.Event
targetUserNID := change.EventStateKeyNID
if change.removedEventNID != 0 {
ev, _ := eventMap(events).lookup(change.removedEventNID)
ev, _ := helpers.EventMap(events).Lookup(change.removedEventNID)
if ev != nil {
re = &ev.Event
}
}
if change.addedEventNID != 0 {
ev, _ := eventMap(events).lookup(change.addedEventNID)
ev, _ := helpers.EventMap(events).Lookup(change.addedEventNID)
if ev != nil {
ae = &ev.Event
}
@ -77,7 +78,7 @@ func (r *RoomserverInternalAPI) updateMemberships(
return updates, nil
}
func (r *RoomserverInternalAPI) updateMembership(
func (r *Inputer) updateMembership(
updater *shared.LatestEventsUpdater,
targetUserNID types.EventStateKeyNID,
remove, add *gomatrixserverlib.Event,
@ -120,7 +121,7 @@ func (r *RoomserverInternalAPI) updateMembership(
switch newMembership {
case gomatrixserverlib.Invite:
return updateToInviteMembership(mu, add, updates, updater.RoomVersion())
return helpers.UpdateToInviteMembership(mu, add, updates, updater.RoomVersion())
case gomatrixserverlib.Join:
return updateToJoinMembership(mu, add, updates)
case gomatrixserverlib.Leave, gomatrixserverlib.Ban:
@ -132,45 +133,15 @@ func (r *RoomserverInternalAPI) updateMembership(
}
}
func (r *RoomserverInternalAPI) isLocalTarget(event *gomatrixserverlib.Event) bool {
func (r *Inputer) isLocalTarget(event *gomatrixserverlib.Event) bool {
isTargetLocalUser := false
if statekey := event.StateKey(); statekey != nil {
_, domain, _ := gomatrixserverlib.SplitID('@', *statekey)
isTargetLocalUser = domain == r.Cfg.Matrix.ServerName
isTargetLocalUser = domain == r.ServerName
}
return isTargetLocalUser
}
func updateToInviteMembership(
mu *shared.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
roomVersion gomatrixserverlib.RoomVersion,
) ([]api.OutputEvent, error) {
// We may have already sent the invite to the user, either because we are
// reprocessing this event, or because the we received this invite from a
// remote server via the federation invite API. In those cases we don't need
// to send the event.
needsSending, err := mu.SetToInvite(*add)
if err != nil {
return nil, err
}
if needsSending {
// We notify the consumers using a special event even though we will
// notify them about the change in current state as part of the normal
// room event stream. This ensures that the consumers only have to
// consider a single stream of events when determining whether a user
// is invited, rather than having to combine multiple streams themselves.
onie := api.OutputNewInviteEvent{
Event: add.Headered(roomVersion),
RoomVersion: roomVersion,
}
updates = append(updates, api.OutputEvent{
Type: api.OutputTypeNewInviteEvent,
NewInviteEvent: &onie,
})
}
return updates, nil
}
func updateToJoinMembership(
mu *shared.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
) ([]api.OutputEvent, error) {

562
roomserver/internal/perform/perform_backfill.go Normal file
View file

@ -0,0 +1,562 @@
// 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 perform
import (
"context"
"fmt"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/auth"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/sirupsen/logrus"
)
type Backfiller struct {
ServerName gomatrixserverlib.ServerName
DB storage.Database
FSAPI federationSenderAPI.FederationSenderInternalAPI
KeyRing gomatrixserverlib.JSONVerifier
}
// PerformBackfill implements api.RoomServerQueryAPI
func (r *Backfiller) PerformBackfill(
ctx context.Context,
request *api.PerformBackfillRequest,
response *api.PerformBackfillResponse,
) error {
// if we are requesting the backfill then we need to do a federation hit
// TODO: we could be more sensible and fetch as many events we already have then request the rest
// which is what the syncapi does already.
if request.ServerName == r.ServerName {
return r.backfillViaFederation(ctx, request, response)
}
// someone else is requesting the backfill, try to service their request.
var err error
var front []string
// The limit defines the maximum number of events to retrieve, so it also
// defines the highest number of elements in the map below.
visited := make(map[string]bool, request.Limit)
// this will include these events which is what we want
front = request.PrevEventIDs()
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info == nil || info.IsStub {
return fmt.Errorf("PerformBackfill: missing room info for room %s", request.RoomID)
}
// Scan the event tree for events to send back.
resultNIDs, err := helpers.ScanEventTree(ctx, r.DB, *info, front, visited, request.Limit, request.ServerName)
if err != nil {
return err
}
// Retrieve events from the list that was filled previously.
var loadedEvents []gomatrixserverlib.Event
loadedEvents, err = helpers.LoadEvents(ctx, r.DB, resultNIDs)
if err != nil {
return err
}
for _, event := range loadedEvents {
response.Events = append(response.Events, event.Headered(info.RoomVersion))
}
return err
}
func (r *Backfiller) backfillViaFederation(ctx context.Context, req *api.PerformBackfillRequest, res *api.PerformBackfillResponse) error {
info, err := r.DB.RoomInfo(ctx, req.RoomID)
if err != nil {
return err
}
if info == nil || info.IsStub {
return fmt.Errorf("backfillViaFederation: missing room info for room %s", req.RoomID)
}
requester := newBackfillRequester(r.DB, r.FSAPI, r.ServerName, req.BackwardsExtremities)
// Request 100 items regardless of what the query asks for.
// We don't want to go much higher than this.
// We can't honour exactly the limit as some sytests rely on requesting more for tests to pass
// (so we don't need to hit /state_ids which the test has no listener for)
// Specifically the test "Outbound federation can backfill events"
events, err := gomatrixserverlib.RequestBackfill(
ctx, requester,
r.KeyRing, req.RoomID, info.RoomVersion, req.PrevEventIDs(), 100)
if err != nil {
return err
}
logrus.WithField("room_id", req.RoomID).Infof("backfilled %d events", len(events))
// persist these new events - auth checks have already been done
roomNID, backfilledEventMap := persistEvents(ctx, r.DB, events)
if err != nil {
return err
}
for _, ev := range backfilledEventMap {
// now add state for these events
stateIDs, ok := requester.eventIDToBeforeStateIDs[ev.EventID()]
if !ok {
// this should be impossible as all events returned must have pass Step 5 of the PDU checks
// which requires a list of state IDs.
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to find state IDs for event which passed auth checks")
continue
}
var entries []types.StateEntry
if entries, err = r.DB.StateEntriesForEventIDs(ctx, stateIDs); err != nil {
// attempt to fetch the missing events
r.fetchAndStoreMissingEvents(ctx, info.RoomVersion, requester, stateIDs)
// try again
entries, err = r.DB.StateEntriesForEventIDs(ctx, stateIDs)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to get state entries for event")
return err
}
}
var beforeStateSnapshotNID types.StateSnapshotNID
if beforeStateSnapshotNID, err = r.DB.AddState(ctx, roomNID, nil, entries); err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to persist state entries to get snapshot nid")
return err
}
if err = r.DB.SetState(ctx, ev.EventNID, beforeStateSnapshotNID); err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to persist snapshot nid")
}
}
// TODO: update backwards extremities, as that should be moved from syncapi to roomserver at some point.
res.Events = events
return nil
}
// fetchAndStoreMissingEvents does a best-effort fetch and store of missing events specified in stateIDs. Returns no error as it is just
// best effort.
func (r *Backfiller) fetchAndStoreMissingEvents(ctx context.Context, roomVer gomatrixserverlib.RoomVersion,
backfillRequester *backfillRequester, stateIDs []string) {
servers := backfillRequester.servers
// work out which are missing
nidMap, err := r.DB.EventNIDs(ctx, stateIDs)
if err != nil {
util.GetLogger(ctx).WithError(err).Warn("cannot query missing events")
return
}
missingMap := make(map[string]*gomatrixserverlib.HeaderedEvent) // id -> event
for _, id := range stateIDs {
if _, ok := nidMap[id]; !ok {
missingMap[id] = nil
}
}
util.GetLogger(ctx).Infof("Fetching %d missing state events (from %d possible servers)", len(missingMap), len(servers))
// fetch the events from federation. Loop the servers first so if we find one that works we stick with them
for _, srv := range servers {
for id, ev := range missingMap {
if ev != nil {
continue // already found
}
logger := util.GetLogger(ctx).WithField("server", srv).WithField("event_id", id)
res, err := r.FSAPI.GetEvent(ctx, srv, id)
if err != nil {
logger.WithError(err).Warn("failed to get event from server")
continue
}
loader := gomatrixserverlib.NewEventsLoader(roomVer, r.KeyRing, backfillRequester, backfillRequester.ProvideEvents, false)
result, err := loader.LoadAndVerify(ctx, res.PDUs, gomatrixserverlib.TopologicalOrderByPrevEvents)
if err != nil {
logger.WithError(err).Warn("failed to load and verify event")
continue
}
logger.Infof("returned %d PDUs which made events %+v", len(res.PDUs), result)
for _, res := range result {
if res.Error != nil {
logger.WithError(err).Warn("event failed PDU checks")
continue
}
missingMap[id] = res.Event
}
}
}
var newEvents []gomatrixserverlib.HeaderedEvent
for _, ev := range missingMap {
if ev != nil {
newEvents = append(newEvents, *ev)
}
}
util.GetLogger(ctx).Infof("Persisting %d new events", len(newEvents))
persistEvents(ctx, r.DB, newEvents)
}
// backfillRequester implements gomatrixserverlib.BackfillRequester
type backfillRequester struct {
db storage.Database
fsAPI federationSenderAPI.FederationSenderInternalAPI
thisServer gomatrixserverlib.ServerName
bwExtrems map[string][]string
// per-request state
servers []gomatrixserverlib.ServerName
eventIDToBeforeStateIDs map[string][]string
eventIDMap map[string]gomatrixserverlib.Event
}
func newBackfillRequester(db storage.Database, fsAPI federationSenderAPI.FederationSenderInternalAPI, thisServer gomatrixserverlib.ServerName, bwExtrems map[string][]string) *backfillRequester {
return &backfillRequester{
db: db,
fsAPI: fsAPI,
thisServer: thisServer,
eventIDToBeforeStateIDs: make(map[string][]string),
eventIDMap: make(map[string]gomatrixserverlib.Event),
bwExtrems: bwExtrems,
}
}
func (b *backfillRequester) StateIDsBeforeEvent(ctx context.Context, targetEvent gomatrixserverlib.HeaderedEvent) ([]string, error) {
b.eventIDMap[targetEvent.EventID()] = targetEvent.Unwrap()
if ids, ok := b.eventIDToBeforeStateIDs[targetEvent.EventID()]; ok {
return ids, nil
}
if len(targetEvent.PrevEventIDs()) == 0 && targetEvent.Type() == "m.room.create" && targetEvent.StateKeyEquals("") {
util.GetLogger(ctx).WithField("room_id", targetEvent.RoomID()).Info("Backfilled to the beginning of the room")
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = []string{}
return nil, nil
}
// if we have exactly 1 prev event and we know the state of the room at that prev event, then just roll forward the prev event.
// Else, we have to hit /state_ids because either we don't know the state at all at this event (new backwards extremity) or
// we don't know the result of state res to merge forks (2 or more prev_events)
if len(targetEvent.PrevEventIDs()) == 1 {
prevEventID := targetEvent.PrevEventIDs()[0]
prevEvent, ok := b.eventIDMap[prevEventID]
if !ok {
goto FederationHit
}
prevEventStateIDs, ok := b.eventIDToBeforeStateIDs[prevEventID]
if !ok {
goto FederationHit
}
newStateIDs := b.calculateNewStateIDs(targetEvent.Unwrap(), prevEvent, prevEventStateIDs)
if newStateIDs != nil {
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs, nil
}
// else we failed to calculate the new state, so fallthrough
}
FederationHit:
var lastErr error
logrus.WithField("event_id", targetEvent.EventID()).Info("Requesting /state_ids at event")
for _, srv := range b.servers { // hit any valid server
c := gomatrixserverlib.FederatedStateProvider{
FedClient: b.fsAPI,
RememberAuthEvents: false,
Server: srv,
}
res, err := c.StateIDsBeforeEvent(ctx, targetEvent)
if err != nil {
lastErr = err
continue
}
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = res
return res, nil
}
return nil, lastErr
}
func (b *backfillRequester) calculateNewStateIDs(targetEvent, prevEvent gomatrixserverlib.Event, prevEventStateIDs []string) []string {
newStateIDs := prevEventStateIDs[:]
if prevEvent.StateKey() == nil {
// state is the same as the previous event
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs
}
missingState := false // true if we are missing the info for a state event ID
foundEvent := false // true if we found a (type, state_key) match
// find which state ID to replace, if any
for i, id := range newStateIDs {
ev, ok := b.eventIDMap[id]
if !ok {
missingState = true
continue
}
// The state IDs BEFORE the target event are the state IDs BEFORE the prev_event PLUS the prev_event itself
if ev.Type() == prevEvent.Type() && ev.StateKeyEquals(*prevEvent.StateKey()) {
newStateIDs[i] = prevEvent.EventID()
foundEvent = true
break
}
}
if !foundEvent && !missingState {
// we can be certain that this is new state
newStateIDs = append(newStateIDs, prevEvent.EventID())
foundEvent = true
}
if foundEvent {
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs
}
return nil
}
func (b *backfillRequester) StateBeforeEvent(ctx context.Context, roomVer gomatrixserverlib.RoomVersion,
event gomatrixserverlib.HeaderedEvent, eventIDs []string) (map[string]*gomatrixserverlib.Event, error) {
// try to fetch the events from the database first
events, err := b.ProvideEvents(roomVer, eventIDs)
if err != nil {
// non-fatal, fallthrough
logrus.WithError(err).Info("Failed to fetch events")
} else {
logrus.Infof("Fetched %d/%d events from the database", len(events), len(eventIDs))
if len(events) == len(eventIDs) {
result := make(map[string]*gomatrixserverlib.Event)
for i := range events {
result[events[i].EventID()] = &events[i]
b.eventIDMap[events[i].EventID()] = events[i]
}
return result, nil
}
}
c := gomatrixserverlib.FederatedStateProvider{
FedClient: b.fsAPI,
RememberAuthEvents: false,
Server: b.servers[0],
}
result, err := c.StateBeforeEvent(ctx, roomVer, event, eventIDs)
if err != nil {
return nil, err
}
for eventID, ev := range result {
b.eventIDMap[eventID] = *ev
}
return result, nil
}
// ServersAtEvent is called when trying to determine which server to request from.
// It returns a list of servers which can be queried for backfill requests. These servers
// will be servers that are in the room already. The entries at the beginning are preferred servers
// and will be tried first. An empty list will fail the request.
// nolint:gocyclo
func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID string) []gomatrixserverlib.ServerName {
// eventID will be a prev_event ID of a backwards extremity, meaning we will not have a database entry for it. Instead, use
// its successor, so look it up.
successor := ""
FindSuccessor:
for sucID, prevEventIDs := range b.bwExtrems {
for _, pe := range prevEventIDs {
if pe == eventID {
successor = sucID
break FindSuccessor
}
}
}
if successor == "" {
logrus.WithField("event_id", eventID).Error("ServersAtEvent: failed to find successor of this event to determine room state")
return nil
}
eventID = successor
// getMembershipsBeforeEventNID requires a NID, so retrieving the NID for
// the event is necessary.
NIDs, err := b.db.EventNIDs(ctx, []string{eventID})
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get event NID for event")
return nil
}
info, err := b.db.RoomInfo(ctx, roomID)
if err != nil {
logrus.WithError(err).WithField("room_id", roomID).Error("ServersAtEvent: failed to get RoomInfo for room")
return nil
}
if info == nil || info.IsStub {
logrus.WithField("room_id", roomID).Error("ServersAtEvent: failed to get RoomInfo for room, room is missing")
return nil
}
stateEntries, err := helpers.StateBeforeEvent(ctx, b.db, *info, NIDs[eventID])
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to load state before event")
return nil
}
// possibly return all joined servers depending on history visiblity
memberEventsFromVis, err := joinEventsFromHistoryVisibility(ctx, b.db, roomID, stateEntries)
if err != nil {
logrus.WithError(err).Error("ServersAtEvent: failed calculate servers from history visibility rules")
return nil
}
logrus.Infof("ServersAtEvent including %d current events from history visibility", len(memberEventsFromVis))
// Retrieve all "m.room.member" state events of "join" membership, which
// contains the list of users in the room before the event, therefore all
// the servers in it at that moment.
memberEvents, err := helpers.GetMembershipsAtState(ctx, b.db, stateEntries, true)
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get memberships before event")
return nil
}
memberEvents = append(memberEvents, memberEventsFromVis...)
// Store the server names in a temporary map to avoid duplicates.
serverSet := make(map[gomatrixserverlib.ServerName]bool)
for _, event := range memberEvents {
serverSet[event.Origin()] = true
}
var servers []gomatrixserverlib.ServerName
for server := range serverSet {
if server == b.thisServer {
continue
}
servers = append(servers, server)
}
b.servers = servers
return servers
}
// Backfill performs a backfill request to the given server.
// https://matrix.org/docs/spec/server_server/latest#get-matrix-federation-v1-backfill-roomid
func (b *backfillRequester) Backfill(ctx context.Context, server gomatrixserverlib.ServerName, roomID string,
limit int, fromEventIDs []string) (gomatrixserverlib.Transaction, error) {
tx, err := b.fsAPI.Backfill(ctx, server, roomID, limit, fromEventIDs)
return tx, err
}
func (b *backfillRequester) ProvideEvents(roomVer gomatrixserverlib.RoomVersion, eventIDs []string) ([]gomatrixserverlib.Event, error) {
ctx := context.Background()
nidMap, err := b.db.EventNIDs(ctx, eventIDs)
if err != nil {
logrus.WithError(err).WithField("event_ids", eventIDs).Error("Failed to find events")
return nil, err
}
eventNIDs := make([]types.EventNID, len(nidMap))
i := 0
for _, nid := range nidMap {
eventNIDs[i] = nid
i++
}
eventsWithNids, err := b.db.Events(ctx, eventNIDs)
if err != nil {
logrus.WithError(err).WithField("event_nids", eventNIDs).Error("Failed to load events")
return nil, err
}
events := make([]gomatrixserverlib.Event, len(eventsWithNids))
for i := range eventsWithNids {
events[i] = eventsWithNids[i].Event
}
return events, nil
}
// joinEventsFromHistoryVisibility returns all CURRENTLY joined members if the provided state indicated a 'shared' history visibility.
// TODO: Long term we probably want a history_visibility table which stores eventNID | visibility_enum so we can just
// pull all events and then filter by that table.
func joinEventsFromHistoryVisibility(
ctx context.Context, db storage.Database, roomID string, stateEntries []types.StateEntry) ([]types.Event, error) {
var eventNIDs []types.EventNID
for _, entry := range stateEntries {
// Filter the events to retrieve to only keep the membership events
if entry.EventTypeNID == types.MRoomHistoryVisibilityNID && entry.EventStateKeyNID == types.EmptyStateKeyNID {
eventNIDs = append(eventNIDs, entry.EventNID)
break
}
}
// Get all of the events in this state
stateEvents, err := db.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
events := make([]gomatrixserverlib.Event, len(stateEvents))
for i := range stateEvents {
events[i] = stateEvents[i].Event
}
visibility := auth.HistoryVisibilityForRoom(events)
if visibility != "shared" {
logrus.Infof("ServersAtEvent history visibility not shared: %s", visibility)
return nil, nil
}
// get joined members
info, err := db.RoomInfo(ctx, roomID)
if err != nil {
return nil, err
}
joinEventNIDs, err := db.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, true, false)
if err != nil {
return nil, err
}
return db.Events(ctx, joinEventNIDs)
}
func persistEvents(ctx context.Context, db storage.Database, events []gomatrixserverlib.HeaderedEvent) (types.RoomNID, map[string]types.Event) {
var roomNID types.RoomNID
backfilledEventMap := make(map[string]types.Event)
for j, ev := range events {
nidMap, err := db.EventNIDs(ctx, ev.AuthEventIDs())
if err != nil { // this shouldn't happen as RequestBackfill already found them
logrus.WithError(err).WithField("auth_events", ev.AuthEventIDs()).Error("Failed to find one or more auth events")
continue
}
authNids := make([]types.EventNID, len(nidMap))
i := 0
for _, nid := range nidMap {
authNids[i] = nid
i++
}
var stateAtEvent types.StateAtEvent
var redactedEventID string
var redactionEvent *gomatrixserverlib.Event
roomNID, stateAtEvent, redactionEvent, redactedEventID, err = db.StoreEvent(ctx, ev.Unwrap(), nil, authNids)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to persist event")
continue
}
// If storing this event results in it being redacted, then do so.
// It's also possible for this event to be a redaction which results in another event being
// redacted, which we don't care about since we aren't returning it in this backfill.
if redactedEventID == ev.EventID() {
eventToRedact := ev.Unwrap()
redactedEvent, err := eventutil.RedactEvent(redactionEvent, &eventToRedact)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to redact event")
continue
}
ev = redactedEvent.Headered(ev.RoomVersion)
events[j] = ev
}
backfilledEventMap[ev.EventID()] = types.Event{
EventNID: stateAtEvent.StateEntry.EventNID,
Event: ev.Unwrap(),
}
}
return roomNID, backfilledEventMap
}

74
roomserver/internal/perform_invite.go → roomserver/internal/perform/perform_invite.go
View file

@ -1,11 +1,28 @@
package internal
// 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 perform
import (
"context"
"fmt"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
@ -13,22 +30,29 @@ import (
log "github.com/sirupsen/logrus"
)
type Inviter struct {
DB storage.Database
Cfg *config.RoomServer
FSAPI federationSenderAPI.FederationSenderInternalAPI
Inputer *input.Inputer
}
// nolint:gocyclo
func (r *RoomserverInternalAPI) PerformInvite(
func (r *Inviter) PerformInvite(
ctx context.Context,
req *api.PerformInviteRequest,
res *api.PerformInviteResponse,
) error {
) ([]api.OutputEvent, error) {
event := req.Event
if event.StateKey() == nil {
return fmt.Errorf("invite must be a state event")
return nil, fmt.Errorf("invite must be a state event")
}
roomID := event.RoomID()
targetUserID := *event.StateKey()
info, err := r.DB.RoomInfo(ctx, roomID)
if err != nil {
return fmt.Errorf("Failed to load RoomInfo: %w", err)
return nil, fmt.Errorf("Failed to load RoomInfo: %w", err)
}
log.WithFields(log.Fields{
@ -52,11 +76,11 @@ func (r *RoomserverInternalAPI) PerformInvite(
}
if len(inviteState) == 0 {
if err = event.SetUnsignedField("invite_room_state", struct{}{}); err != nil {
return fmt.Errorf("event.SetUnsignedField: %w", err)
return nil, fmt.Errorf("event.SetUnsignedField: %w", err)
}
} else {
if err = event.SetUnsignedField("invite_room_state", inviteState); err != nil {
return fmt.Errorf("event.SetUnsignedField: %w", err)
return nil, fmt.Errorf("event.SetUnsignedField: %w", err)
}
}
@ -64,7 +88,7 @@ func (r *RoomserverInternalAPI) PerformInvite(
if info != nil {
_, isAlreadyJoined, err = r.DB.GetMembership(ctx, info.RoomNID, *event.StateKey())
if err != nil {
return fmt.Errorf("r.DB.GetMembership: %w", err)
return nil, fmt.Errorf("r.DB.GetMembership: %w", err)
}
}
if isAlreadyJoined {
@ -99,7 +123,7 @@ func (r *RoomserverInternalAPI) PerformInvite(
Code: api.PerformErrorNotAllowed,
Msg: "User is already joined to room",
}
return nil
return nil, nil
}
if isOriginLocal {
@ -107,7 +131,7 @@ func (r *RoomserverInternalAPI) PerformInvite(
// try and see if the user is allowed to make this invite. We can't do
// this for invites coming in over federation - we have to take those on
// trust.
_, err = checkAuthEvents(ctx, r.DB, event, event.AuthEventIDs())
_, err = helpers.CheckAuthEvents(ctx, r.DB, event, event.AuthEventIDs())
if err != nil {
log.WithError(err).WithField("event_id", event.EventID()).WithField("auth_event_ids", event.AuthEventIDs()).Error(
"processInviteEvent.checkAuthEvents failed for event",
@ -117,9 +141,9 @@ func (r *RoomserverInternalAPI) PerformInvite(
Msg: err.Error(),
Code: api.PerformErrorNotAllowed,
}
return nil
return nil, nil
}
return fmt.Errorf("checkAuthEvents: %w", err)
return nil, fmt.Errorf("checkAuthEvents: %w", err)
}
// If the invite originated from us and the target isn't local then we
@ -133,13 +157,13 @@ func (r *RoomserverInternalAPI) PerformInvite(
InviteRoomState: inviteState,
}
fsRes := &federationSenderAPI.PerformInviteResponse{}
if err = r.fsAPI.PerformInvite(ctx, fsReq, fsRes); err != nil {
if err = r.FSAPI.PerformInvite(ctx, fsReq, fsRes); err != nil {
res.Error = &api.PerformError{
Msg: err.Error(),
Code: api.PerformErrorNoOperation,
}
log.WithError(err).WithField("event_id", event.EventID()).Error("r.fsAPI.PerformInvite failed")
return nil
log.WithError(err).WithField("event_id", event.EventID()).Error("r.FSAPI.PerformInvite failed")
return nil, nil
}
event = fsRes.Event
}
@ -159,8 +183,8 @@ func (r *RoomserverInternalAPI) PerformInvite(
},
}
inputRes := &api.InputRoomEventsResponse{}
if err = r.InputRoomEvents(context.Background(), inputReq, inputRes); err != nil {
return fmt.Errorf("r.InputRoomEvents: %w", err)
if err = r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes); err != nil {
return nil, fmt.Errorf("r.InputRoomEvents: %w", err)
}
} else {
// The invite originated over federation. Process the membership
@ -168,25 +192,23 @@ func (r *RoomserverInternalAPI) PerformInvite(
// invite.
updater, err := r.DB.MembershipUpdater(ctx, roomID, targetUserID, isTargetLocal, req.RoomVersion)
if err != nil {
return fmt.Errorf("r.DB.MembershipUpdater: %w", err)
return nil, fmt.Errorf("r.DB.MembershipUpdater: %w", err)
}
unwrapped := event.Unwrap()
outputUpdates, err := updateToInviteMembership(updater, &unwrapped, nil, req.Event.RoomVersion)
outputUpdates, err := helpers.UpdateToInviteMembership(updater, &unwrapped, nil, req.Event.RoomVersion)
if err != nil {
return fmt.Errorf("updateToInviteMembership: %w", err)
return nil, fmt.Errorf("updateToInviteMembership: %w", err)
}
if err = updater.Commit(); err != nil {
return fmt.Errorf("updater.Commit: %w", err)
return nil, fmt.Errorf("updater.Commit: %w", err)
}
if err = r.WriteOutputEvents(roomID, outputUpdates); err != nil {
return fmt.Errorf("r.WriteOutputEvents: %w", err)
}
return outputUpdates, nil
}
return nil
return nil, nil
}
func buildInviteStrippedState(
@ -208,7 +230,7 @@ func buildInviteStrippedState(
StateKey: "",
})
}
roomState := state.NewStateResolution(db)
roomState := state.NewStateResolution(db, *info)
stateEntries, err := roomState.LoadStateAtSnapshotForStringTuples(
ctx, info.StateSnapshotNID, stateWanted,
)

87
roomserver/internal/perform_join.go → roomserver/internal/perform/perform_join.go
View file

@ -1,4 +1,18 @@
package internal
// 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 perform
import (
"context"
@ -8,14 +22,27 @@ import (
"time"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
)
type Joiner struct {
ServerName gomatrixserverlib.ServerName
Cfg *config.RoomServer
FSAPI fsAPI.FederationSenderInternalAPI
DB storage.Database
Inputer *input.Inputer
}
// PerformJoin handles joining matrix rooms, including over federation by talking to the federationsender.
func (r *RoomserverInternalAPI) PerformJoin(
func (r *Joiner) PerformJoin(
ctx context.Context,
req *api.PerformJoinRequest,
res *api.PerformJoinResponse,
@ -34,7 +61,7 @@ func (r *RoomserverInternalAPI) PerformJoin(
res.RoomID = roomID
}
func (r *RoomserverInternalAPI) performJoin(
func (r *Joiner) performJoin(
ctx context.Context,
req *api.PerformJoinRequest,
) (string, error) {
@ -63,7 +90,7 @@ func (r *RoomserverInternalAPI) performJoin(
}
}
func (r *RoomserverInternalAPI) performJoinRoomByAlias(
func (r *Joiner) performJoinRoomByAlias(
ctx context.Context,
req *api.PerformJoinRequest,
) (string, error) {
@ -85,7 +112,7 @@ func (r *RoomserverInternalAPI) performJoinRoomByAlias(
ServerName: domain, // the server to ask
}
dirRes := fsAPI.PerformDirectoryLookupResponse{}
err = r.fsAPI.PerformDirectoryLookup(ctx, &dirReq, &dirRes)
err = r.FSAPI.PerformDirectoryLookup(ctx, &dirReq, &dirRes)
if err != nil {
logrus.WithError(err).Errorf("error looking up alias %q", req.RoomIDOrAlias)
return "", fmt.Errorf("Looking up alias %q over federation failed: %w", req.RoomIDOrAlias, err)
@ -112,7 +139,7 @@ func (r *RoomserverInternalAPI) performJoinRoomByAlias(
// TODO: Break this function up a bit
// nolint:gocyclo
func (r *RoomserverInternalAPI) performJoinRoomByID(
func (r *Joiner) performJoinRoomByID(
ctx context.Context,
req *api.PerformJoinRequest,
) (string, error) {
@ -161,8 +188,8 @@ func (r *RoomserverInternalAPI) performJoinRoomByID(
// where we might think we know about a room in the following
// section but don't know the latest state as all of our users
// have left.
serverInRoom, _ := r.isServerCurrentlyInRoom(ctx, r.ServerName, req.RoomIDOrAlias)
isInvitePending, inviteSender, _, err := r.isInvitePending(ctx, req.RoomIDOrAlias, req.UserID)
serverInRoom, _ := helpers.IsServerCurrentlyInRoom(ctx, r.DB, r.ServerName, req.RoomIDOrAlias)
isInvitePending, inviteSender, _, err := helpers.IsInvitePending(ctx, r.DB, req.RoomIDOrAlias, req.UserID)
if err == nil && isInvitePending && !serverInRoom {
// Check if there's an invite pending.
_, inviterDomain, ierr := gomatrixserverlib.SplitID('@', inviteSender)
@ -188,15 +215,7 @@ func (r *RoomserverInternalAPI) performJoinRoomByID(
// locally on the homeserver.
// TODO: Check what happens if the room exists on the server
// but everyone has since left. I suspect it does the wrong thing.
buildRes := api.QueryLatestEventsAndStateResponse{}
event, err := eventutil.BuildEvent(
ctx, // the request context
&eb, // the template join event
r.Cfg.Matrix, // the server configuration
time.Now(), // the event timestamp to use
r, // the roomserver API to use
&buildRes, // the query response
)
event, buildRes, err := buildEvent(ctx, r.DB, r.Cfg.Matrix, &eb)
switch err {
case nil:
@ -228,7 +247,7 @@ func (r *RoomserverInternalAPI) performJoinRoomByID(
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
var notAllowed *gomatrixserverlib.NotAllowed
if errors.As(err, &notAllowed) {
return "", &api.PerformError{
@ -271,7 +290,7 @@ func (r *RoomserverInternalAPI) performJoinRoomByID(
return req.RoomIDOrAlias, nil
}
func (r *RoomserverInternalAPI) performFederatedJoinRoomByID(
func (r *Joiner) performFederatedJoinRoomByID(
ctx context.Context,
req *api.PerformJoinRequest,
) error {
@ -283,7 +302,7 @@ func (r *RoomserverInternalAPI) performFederatedJoinRoomByID(
Content: req.Content, // the membership event content
}
fedRes := fsAPI.PerformJoinResponse{}
r.fsAPI.PerformJoin(ctx, &fedReq, &fedRes)
r.FSAPI.PerformJoin(ctx, &fedReq, &fedRes)
if fedRes.LastError != nil {
return &api.PerformError{
Code: api.PerformErrRemote,
@ -293,3 +312,31 @@ func (r *RoomserverInternalAPI) performFederatedJoinRoomByID(
}
return nil
}
func buildEvent(
ctx context.Context, db storage.Database, cfg *config.Global, builder *gomatrixserverlib.EventBuilder,
) (*gomatrixserverlib.HeaderedEvent, *api.QueryLatestEventsAndStateResponse, error) {
eventsNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(builder)
if err != nil {
return nil, nil, fmt.Errorf("gomatrixserverlib.StateNeededForEventBuilder: %w", err)
}
if len(eventsNeeded.Tuples()) == 0 {
return nil, nil, errors.New("expecting state tuples for event builder, got none")
}
var queryRes api.QueryLatestEventsAndStateResponse
err = helpers.QueryLatestEventsAndState(ctx, db, &api.QueryLatestEventsAndStateRequest{
RoomID: builder.RoomID,
StateToFetch: eventsNeeded.Tuples(),
}, &queryRes)
if err != nil {
return nil, nil, fmt.Errorf("QueryLatestEventsAndState: %w", err)
}
ev, err := eventutil.BuildEvent(ctx, builder, cfg, time.Now(), &eventsNeeded, &queryRes)
if err != nil {
return nil, nil, err
}
return ev, &queryRes, nil
}

183
roomserver/internal/perform/perform_leave.go Normal file
View file

@ -0,0 +1,183 @@
// 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 perform
import (
"context"
"fmt"
"strings"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/gomatrixserverlib"
)
type Leaver struct {
Cfg *config.RoomServer
DB storage.Database
FSAPI fsAPI.FederationSenderInternalAPI
Inputer *input.Inputer
}
// WriteOutputEvents implements OutputRoomEventWriter
func (r *Leaver) PerformLeave(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse,
) ([]api.OutputEvent, error) {
_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil {
return nil, fmt.Errorf("Supplied user ID %q in incorrect format", req.UserID)
}
if domain != r.Cfg.Matrix.ServerName {
return nil, fmt.Errorf("User %q does not belong to this homeserver", req.UserID)
}
if strings.HasPrefix(req.RoomID, "!") {
return r.performLeaveRoomByID(ctx, req, res)
}
return nil, fmt.Errorf("Room ID %q is invalid", req.RoomID)
}
func (r *Leaver) performLeaveRoomByID(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse, // nolint:unparam
) ([]api.OutputEvent, error) {
// If there's an invite outstanding for the room then respond to
// that.
isInvitePending, senderUser, eventID, err := helpers.IsInvitePending(ctx, r.DB, req.RoomID, req.UserID)
if err == nil && isInvitePending {
return r.performRejectInvite(ctx, req, res, senderUser, eventID)
}
// There's no invite pending, so first of all we want to find out
// if the room exists and if the user is actually in it.
latestReq := api.QueryLatestEventsAndStateRequest{
RoomID: req.RoomID,
StateToFetch: []gomatrixserverlib.StateKeyTuple{
{
EventType: gomatrixserverlib.MRoomMember,
StateKey: req.UserID,
},
},
}
latestRes := api.QueryLatestEventsAndStateResponse{}
if err = helpers.QueryLatestEventsAndState(ctx, r.DB, &latestReq, &latestRes); err != nil {
return nil, err
}
if !latestRes.RoomExists {
return nil, fmt.Errorf("Room %q does not exist", req.RoomID)
}
// Now let's see if the user is in the room.
if len(latestRes.StateEvents) == 0 {
return nil, fmt.Errorf("User %q is not a member of room %q", req.UserID, req.RoomID)
}
membership, err := latestRes.StateEvents[0].Membership()
if err != nil {
return nil, fmt.Errorf("Error getting membership: %w", err)
}
if membership != gomatrixserverlib.Join {
// TODO: should be able to handle "invite" in this case too, if
// it's a case of kicking or banning or such
return nil, fmt.Errorf("User %q is not joined to the room (membership is %q)", req.UserID, membership)
}
// Prepare the template for the leave event.
userID := req.UserID
eb := gomatrixserverlib.EventBuilder{
Type: gomatrixserverlib.MRoomMember,
Sender: userID,
StateKey: &userID,
RoomID: req.RoomID,
Redacts: "",
}
if err = eb.SetContent(map[string]interface{}{"membership": "leave"}); err != nil {
return nil, fmt.Errorf("eb.SetContent: %w", err)
}
if err = eb.SetUnsigned(struct{}{}); err != nil {
return nil, fmt.Errorf("eb.SetUnsigned: %w", err)
}
// We know that the user is in the room at this point so let's build
// a leave event.
// TODO: Check what happens if the room exists on the server
// but everyone has since left. I suspect it does the wrong thing.
event, buildRes, err := buildEvent(ctx, r.DB, r.Cfg.Matrix, &eb)
if err != nil {
return nil, fmt.Errorf("eventutil.BuildEvent: %w", err)
}
// Give our leave event to the roomserver input stream. The
// roomserver will process the membership change and notify
// downstream automatically.
inputReq := api.InputRoomEventsRequest{
InputRoomEvents: []api.InputRoomEvent{
{
Kind: api.KindNew,
Event: event.Headered(buildRes.RoomVersion),
AuthEventIDs: event.AuthEventIDs(),
SendAsServer: string(r.Cfg.Matrix.ServerName),
},
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
return nil, fmt.Errorf("r.InputRoomEvents: %w", err)
}
return nil, nil
}
func (r *Leaver) performRejectInvite(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse, // nolint:unparam
senderUser, eventID string,
) ([]api.OutputEvent, error) {
_, domain, err := gomatrixserverlib.SplitID('@', senderUser)
if err != nil {
return nil, fmt.Errorf("User ID %q invalid: %w", senderUser, err)
}
// Ask the federation sender to perform a federated leave for us.
leaveReq := fsAPI.PerformLeaveRequest{
RoomID: req.RoomID,
UserID: req.UserID,
ServerNames: []gomatrixserverlib.ServerName{domain},
}
leaveRes := fsAPI.PerformLeaveResponse{}
if err := r.FSAPI.PerformLeave(ctx, &leaveReq, &leaveRes); err != nil {
return nil, err
}
// Withdraw the invite, so that the sync API etc are
// notified that we rejected it.
return []api.OutputEvent{
{
Type: api.OutputTypeRetireInviteEvent,
RetireInviteEvent: &api.OutputRetireInviteEvent{
EventID: eventID,
Membership: "leave",
TargetUserID: req.UserID,
},
},
}, nil
}

39
roomserver/internal/perform/perform_publish.go Normal file
View file

@ -0,0 +1,39 @@
// 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 perform
import (
"context"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage"
)
type Publisher struct {
DB storage.Database
}
func (r *Publisher) PerformPublish(
ctx context.Context,
req *api.PerformPublishRequest,
res *api.PerformPublishResponse,
) {
err := r.DB.PublishRoom(ctx, req.RoomID, req.Visibility == "public")
if err != nil {
res.Error = &api.PerformError{
Msg: err.Error(),
}
}
}

305
roomserver/internal/perform_backfill.go
View file

@ -1,305 +0,0 @@
package internal
import (
"context"
"github.com/matrix-org/dendrite/roomserver/auth"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/sirupsen/logrus"
)
// backfillRequester implements gomatrixserverlib.BackfillRequester
type backfillRequester struct {
db storage.Database
fedClient *gomatrixserverlib.FederationClient
thisServer gomatrixserverlib.ServerName
bwExtrems map[string][]string
// per-request state
servers []gomatrixserverlib.ServerName
eventIDToBeforeStateIDs map[string][]string
eventIDMap map[string]gomatrixserverlib.Event
}
func newBackfillRequester(db storage.Database, fedClient *gomatrixserverlib.FederationClient, thisServer gomatrixserverlib.ServerName, bwExtrems map[string][]string) *backfillRequester {
return &backfillRequester{
db: db,
fedClient: fedClient,
thisServer: thisServer,
eventIDToBeforeStateIDs: make(map[string][]string),
eventIDMap: make(map[string]gomatrixserverlib.Event),
bwExtrems: bwExtrems,
}
}
func (b *backfillRequester) StateIDsBeforeEvent(ctx context.Context, targetEvent gomatrixserverlib.HeaderedEvent) ([]string, error) {
b.eventIDMap[targetEvent.EventID()] = targetEvent.Unwrap()
if ids, ok := b.eventIDToBeforeStateIDs[targetEvent.EventID()]; ok {
return ids, nil
}
if len(targetEvent.PrevEventIDs()) == 0 && targetEvent.Type() == "m.room.create" && targetEvent.StateKeyEquals("") {
util.GetLogger(ctx).WithField("room_id", targetEvent.RoomID()).Info("Backfilled to the beginning of the room")
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = []string{}
return nil, nil
}
// if we have exactly 1 prev event and we know the state of the room at that prev event, then just roll forward the prev event.
// Else, we have to hit /state_ids because either we don't know the state at all at this event (new backwards extremity) or
// we don't know the result of state res to merge forks (2 or more prev_events)
if len(targetEvent.PrevEventIDs()) == 1 {
prevEventID := targetEvent.PrevEventIDs()[0]
prevEvent, ok := b.eventIDMap[prevEventID]
if !ok {
goto FederationHit
}
prevEventStateIDs, ok := b.eventIDToBeforeStateIDs[prevEventID]
if !ok {
goto FederationHit
}
newStateIDs := b.calculateNewStateIDs(targetEvent.Unwrap(), prevEvent, prevEventStateIDs)
if newStateIDs != nil {
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs, nil
}
// else we failed to calculate the new state, so fallthrough
}
FederationHit:
var lastErr error
logrus.WithField("event_id", targetEvent.EventID()).Info("Requesting /state_ids at event")
for _, srv := range b.servers { // hit any valid server
c := gomatrixserverlib.FederatedStateProvider{
FedClient: b.fedClient,
RememberAuthEvents: false,
Server: srv,
}
res, err := c.StateIDsBeforeEvent(ctx, targetEvent)
if err != nil {
lastErr = err
continue
}
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = res
return res, nil
}
return nil, lastErr
}
func (b *backfillRequester) calculateNewStateIDs(targetEvent, prevEvent gomatrixserverlib.Event, prevEventStateIDs []string) []string {
newStateIDs := prevEventStateIDs[:]
if prevEvent.StateKey() == nil {
// state is the same as the previous event
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs
}
missingState := false // true if we are missing the info for a state event ID
foundEvent := false // true if we found a (type, state_key) match
// find which state ID to replace, if any
for i, id := range newStateIDs {
ev, ok := b.eventIDMap[id]
if !ok {
missingState = true
continue
}
// The state IDs BEFORE the target event are the state IDs BEFORE the prev_event PLUS the prev_event itself
if ev.Type() == prevEvent.Type() && ev.StateKeyEquals(*prevEvent.StateKey()) {
newStateIDs[i] = prevEvent.EventID()
foundEvent = true
break
}
}
if !foundEvent && !missingState {
// we can be certain that this is new state
newStateIDs = append(newStateIDs, prevEvent.EventID())
foundEvent = true
}
if foundEvent {
b.eventIDToBeforeStateIDs[targetEvent.EventID()] = newStateIDs
return newStateIDs
}
return nil
}
func (b *backfillRequester) StateBeforeEvent(ctx context.Context, roomVer gomatrixserverlib.RoomVersion,
event gomatrixserverlib.HeaderedEvent, eventIDs []string) (map[string]*gomatrixserverlib.Event, error) {
// try to fetch the events from the database first
events, err := b.ProvideEvents(roomVer, eventIDs)
if err != nil {
// non-fatal, fallthrough
logrus.WithError(err).Info("Failed to fetch events")
} else {
logrus.Infof("Fetched %d/%d events from the database", len(events), len(eventIDs))
if len(events) == len(eventIDs) {
result := make(map[string]*gomatrixserverlib.Event)
for i := range events {
result[events[i].EventID()] = &events[i]
b.eventIDMap[events[i].EventID()] = events[i]
}
return result, nil
}
}
c := gomatrixserverlib.FederatedStateProvider{
FedClient: b.fedClient,
RememberAuthEvents: false,
Server: b.servers[0],
}
result, err := c.StateBeforeEvent(ctx, roomVer, event, eventIDs)
if err != nil {
return nil, err
}
for eventID, ev := range result {
b.eventIDMap[eventID] = *ev
}
return result, nil
}
// ServersAtEvent is called when trying to determine which server to request from.
// It returns a list of servers which can be queried for backfill requests. These servers
// will be servers that are in the room already. The entries at the beginning are preferred servers
// and will be tried first. An empty list will fail the request.
func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID string) []gomatrixserverlib.ServerName {
// eventID will be a prev_event ID of a backwards extremity, meaning we will not have a database entry for it. Instead, use
// its successor, so look it up.
successor := ""
FindSuccessor:
for sucID, prevEventIDs := range b.bwExtrems {
for _, pe := range prevEventIDs {
if pe == eventID {
successor = sucID
break FindSuccessor
}
}
}
if successor == "" {
logrus.WithField("event_id", eventID).Error("ServersAtEvent: failed to find successor of this event to determine room state")
return nil
}
eventID = successor
// getMembershipsBeforeEventNID requires a NID, so retrieving the NID for
// the event is necessary.
NIDs, err := b.db.EventNIDs(ctx, []string{eventID})
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get event NID for event")
return nil
}
stateEntries, err := stateBeforeEvent(ctx, b.db, NIDs[eventID])
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to load state before event")
return nil
}
// possibly return all joined servers depending on history visiblity
memberEventsFromVis, err := joinEventsFromHistoryVisibility(ctx, b.db, roomID, stateEntries)
if err != nil {
logrus.WithError(err).Error("ServersAtEvent: failed calculate servers from history visibility rules")
return nil
}
logrus.Infof("ServersAtEvent including %d current events from history visibility", len(memberEventsFromVis))
// Retrieve all "m.room.member" state events of "join" membership, which
// contains the list of users in the room before the event, therefore all
// the servers in it at that moment.
memberEvents, err := getMembershipsAtState(ctx, b.db, stateEntries, true)
if err != nil {
logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get memberships before event")
return nil
}
memberEvents = append(memberEvents, memberEventsFromVis...)
// Store the server names in a temporary map to avoid duplicates.
serverSet := make(map[gomatrixserverlib.ServerName]bool)
for _, event := range memberEvents {
serverSet[event.Origin()] = true
}
var servers []gomatrixserverlib.ServerName
for server := range serverSet {
if server == b.thisServer {
continue
}
servers = append(servers, server)
}
b.servers = servers
return servers
}
// Backfill performs a backfill request to the given server.
// https://matrix.org/docs/spec/server_server/latest#get-matrix-federation-v1-backfill-roomid
func (b *backfillRequester) Backfill(ctx context.Context, server gomatrixserverlib.ServerName, roomID string,
fromEventIDs []string, limit int) (*gomatrixserverlib.Transaction, error) {
tx, err := b.fedClient.Backfill(ctx, server, roomID, limit, fromEventIDs)
return &tx, err
}
func (b *backfillRequester) ProvideEvents(roomVer gomatrixserverlib.RoomVersion, eventIDs []string) ([]gomatrixserverlib.Event, error) {
ctx := context.Background()
nidMap, err := b.db.EventNIDs(ctx, eventIDs)
if err != nil {
logrus.WithError(err).WithField("event_ids", eventIDs).Error("Failed to find events")
return nil, err
}
eventNIDs := make([]types.EventNID, len(nidMap))
i := 0
for _, nid := range nidMap {
eventNIDs[i] = nid
i++
}
eventsWithNids, err := b.db.Events(ctx, eventNIDs)
if err != nil {
logrus.WithError(err).WithField("event_nids", eventNIDs).Error("Failed to load events")
return nil, err
}
events := make([]gomatrixserverlib.Event, len(eventsWithNids))
for i := range eventsWithNids {
events[i] = eventsWithNids[i].Event
}
return events, nil
}
// joinEventsFromHistoryVisibility returns all CURRENTLY joined members if the provided state indicated a 'shared' history visibility.
// TODO: Long term we probably want a history_visibility table which stores eventNID | visibility_enum so we can just
// pull all events and then filter by that table.
func joinEventsFromHistoryVisibility(
ctx context.Context, db storage.Database, roomID string, stateEntries []types.StateEntry) ([]types.Event, error) {
var eventNIDs []types.EventNID
for _, entry := range stateEntries {
// Filter the events to retrieve to only keep the membership events
if entry.EventTypeNID == types.MRoomHistoryVisibilityNID && entry.EventStateKeyNID == types.EmptyStateKeyNID {
eventNIDs = append(eventNIDs, entry.EventNID)
break
}
}
// Get all of the events in this state
stateEvents, err := db.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
events := make([]gomatrixserverlib.Event, len(stateEvents))
for i := range stateEvents {
events[i] = stateEvents[i].Event
}
visibility := auth.HistoryVisibilityForRoom(events)
if visibility != "shared" {
logrus.Infof("ServersAtEvent history visibility not shared: %s", visibility)
return nil, nil
}
// get joined members
info, err := db.RoomInfo(ctx, roomID)
if err != nil {
return nil, err
}
joinEventNIDs, err := db.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, true, false)
if err != nil {
return nil, err
}
return db.Events(ctx, joinEventNIDs)
}

223
roomserver/internal/perform_leave.go
View file

@ -1,223 +0,0 @@
package internal
import (
"context"
"fmt"
"strings"
"time"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
// WriteOutputEvents implements OutputRoomEventWriter
func (r *RoomserverInternalAPI) PerformLeave(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse,
) error {
_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil {
return fmt.Errorf("Supplied user ID %q in incorrect format", req.UserID)
}
if domain != r.Cfg.Matrix.ServerName {
return fmt.Errorf("User %q does not belong to this homeserver", req.UserID)
}
if strings.HasPrefix(req.RoomID, "!") {
return r.performLeaveRoomByID(ctx, req, res)
}
return fmt.Errorf("Room ID %q is invalid", req.RoomID)
}
func (r *RoomserverInternalAPI) performLeaveRoomByID(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse, // nolint:unparam
) error {
// If there's an invite outstanding for the room then respond to
// that.
isInvitePending, senderUser, eventID, err := r.isInvitePending(ctx, req.RoomID, req.UserID)
if err == nil && isInvitePending {
return r.performRejectInvite(ctx, req, res, senderUser, eventID)
}
// There's no invite pending, so first of all we want to find out
// if the room exists and if the user is actually in it.
latestReq := api.QueryLatestEventsAndStateRequest{
RoomID: req.RoomID,
StateToFetch: []gomatrixserverlib.StateKeyTuple{
{
EventType: gomatrixserverlib.MRoomMember,
StateKey: req.UserID,
},
},
}
latestRes := api.QueryLatestEventsAndStateResponse{}
if err = r.QueryLatestEventsAndState(ctx, &latestReq, &latestRes); err != nil {
return err
}
if !latestRes.RoomExists {
return fmt.Errorf("Room %q does not exist", req.RoomID)
}
// Now let's see if the user is in the room.
if len(latestRes.StateEvents) == 0 {
return fmt.Errorf("User %q is not a member of room %q", req.UserID, req.RoomID)
}
membership, err := latestRes.StateEvents[0].Membership()
if err != nil {
return fmt.Errorf("Error getting membership: %w", err)
}
if membership != gomatrixserverlib.Join {
// TODO: should be able to handle "invite" in this case too, if
// it's a case of kicking or banning or such
return fmt.Errorf("User %q is not joined to the room (membership is %q)", req.UserID, membership)
}
// Prepare the template for the leave event.
userID := req.UserID
eb := gomatrixserverlib.EventBuilder{
Type: gomatrixserverlib.MRoomMember,
Sender: userID,
StateKey: &userID,
RoomID: req.RoomID,
Redacts: "",
}
if err = eb.SetContent(map[string]interface{}{"membership": "leave"}); err != nil {
return fmt.Errorf("eb.SetContent: %w", err)
}
if err = eb.SetUnsigned(struct{}{}); err != nil {
return fmt.Errorf("eb.SetUnsigned: %w", err)
}
// We know that the user is in the room at this point so let's build
// a leave event.
// TODO: Check what happens if the room exists on the server
// but everyone has since left. I suspect it does the wrong thing.
buildRes := api.QueryLatestEventsAndStateResponse{}
event, err := eventutil.BuildEvent(
ctx, // the request context
&eb, // the template leave event
r.Cfg.Matrix, // the server configuration
time.Now(), // the event timestamp to use
r, // the roomserver API to use
&buildRes, // the query response
)
if err != nil {
return fmt.Errorf("eventutil.BuildEvent: %w", err)
}
// Give our leave event to the roomserver input stream. The
// roomserver will process the membership change and notify
// downstream automatically.
inputReq := api.InputRoomEventsRequest{
InputRoomEvents: []api.InputRoomEvent{
{
Kind: api.KindNew,
Event: event.Headered(buildRes.RoomVersion),
AuthEventIDs: event.AuthEventIDs(),
SendAsServer: string(r.Cfg.Matrix.ServerName),
},
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
return fmt.Errorf("r.InputRoomEvents: %w", err)
}
return nil
}
func (r *RoomserverInternalAPI) performRejectInvite(
ctx context.Context,
req *api.PerformLeaveRequest,
res *api.PerformLeaveResponse, // nolint:unparam
senderUser, eventID string,
) error {
_, domain, err := gomatrixserverlib.SplitID('@', senderUser)
if err != nil {
return fmt.Errorf("User ID %q invalid: %w", senderUser, err)
}
// Ask the federation sender to perform a federated leave for us.
leaveReq := fsAPI.PerformLeaveRequest{
RoomID: req.RoomID,
UserID: req.UserID,
ServerNames: []gomatrixserverlib.ServerName{domain},
}
leaveRes := fsAPI.PerformLeaveResponse{}
if err := r.fsAPI.PerformLeave(ctx, &leaveReq, &leaveRes); err != nil {
return err
}
// Withdraw the invite, so that the sync API etc are
// notified that we rejected it.
return r.WriteOutputEvents(req.RoomID, []api.OutputEvent{
{
Type: api.OutputTypeRetireInviteEvent,
RetireInviteEvent: &api.OutputRetireInviteEvent{
EventID: eventID,
Membership: "leave",
TargetUserID: req.UserID,
},
},
})
}
func (r *RoomserverInternalAPI) isInvitePending(
ctx context.Context,
roomID, userID string,
) (bool, string, string, error) {
// Look up the room NID for the supplied room ID.
info, err := r.DB.RoomInfo(ctx, roomID)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.RoomInfo: %w", err)
}
if info == nil {
return false, "", "", fmt.Errorf("cannot get RoomInfo: unknown room ID %s", roomID)
}
// Look up the state key NID for the supplied user ID.
targetUserNIDs, err := r.DB.EventStateKeyNIDs(ctx, []string{userID})
if err != nil {
return false, "", "", fmt.Errorf("r.DB.EventStateKeyNIDs: %w", err)
}
targetUserNID, targetUserFound := targetUserNIDs[userID]
if !targetUserFound {
return false, "", "", fmt.Errorf("missing NID for user %q (%+v)", userID, targetUserNIDs)
}
// Let's see if we have an event active for the user in the room. If
// we do then it will contain a server name that we can direct the
// send_leave to.
senderUserNIDs, eventIDs, err := r.DB.GetInvitesForUser(ctx, info.RoomNID, targetUserNID)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.GetInvitesForUser: %w", err)
}
if len(senderUserNIDs) == 0 {
return false, "", "", nil
}
userNIDToEventID := make(map[types.EventStateKeyNID]string)
for i, nid := range senderUserNIDs {
userNIDToEventID[nid] = eventIDs[i]
}
// Look up the user ID from the NID.
senderUsers, err := r.DB.EventStateKeys(ctx, senderUserNIDs)
if err != nil {
return false, "", "", fmt.Errorf("r.DB.EventStateKeys: %w", err)
}
if len(senderUsers) == 0 {
return false, "", "", fmt.Errorf("no senderUsers")
}
senderUser, senderUserFound := senderUsers[senderUserNIDs[0]]
if !senderUserFound {
return false, "", "", fmt.Errorf("missing user for NID %d (%+v)", senderUserNIDs[0], senderUsers)
}
return true, senderUser, userNIDToEventID[senderUserNIDs[0]], nil
}

20
roomserver/internal/perform_publish.go
View file

@ -1,20 +0,0 @@
package internal
import (
"context"
"github.com/matrix-org/dendrite/roomserver/api"
)
func (r *RoomserverInternalAPI) PerformPublish(
ctx context.Context,
req *api.PerformPublishRequest,
res *api.PerformPublishResponse,
) {
err := r.DB.PublishRoom(ctx, req.RoomID, req.Visibility == "public")
if err != nil {
res.Error = &api.PerformError{
Msg: err.Error(),
}
}
}

960
roomserver/internal/query.go
View file

@ -1,960 +0,0 @@
// Copyright 2017 Vector Creations Ltd
// Copyright 2018 New Vector Ltd
// Copyright 2019-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 internal
import (
"context"
"fmt"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/auth"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/roomserver/version"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/sirupsen/logrus"
)
// QueryLatestEventsAndState implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryLatestEventsAndState(
ctx context.Context,
request *api.QueryLatestEventsAndStateRequest,
response *api.QueryLatestEventsAndStateResponse,
) error {
roomVersion, err := r.DB.GetRoomVersionForRoom(ctx, request.RoomID)
if err != nil {
response.RoomExists = false
return nil
}
roomState := state.NewStateResolution(r.DB)
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info.IsStub {
return nil
}
response.RoomExists = true
response.RoomVersion = roomVersion
var currentStateSnapshotNID types.StateSnapshotNID
response.LatestEvents, currentStateSnapshotNID, response.Depth, err =
r.DB.LatestEventIDs(ctx, info.RoomNID)
if err != nil {
return err
}
var stateEntries []types.StateEntry
if len(request.StateToFetch) == 0 {
// Look up all room state.
stateEntries, err = roomState.LoadStateAtSnapshot(
ctx, currentStateSnapshotNID,
)
} else {
// Look up the current state for the requested tuples.
stateEntries, err = roomState.LoadStateAtSnapshotForStringTuples(
ctx, currentStateSnapshotNID, request.StateToFetch,
)
}
if err != nil {
return err
}
stateEvents, err := r.loadStateEvents(ctx, stateEntries)
if err != nil {
return err
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(roomVersion))
}
return nil
}
// QueryStateAfterEvents implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryStateAfterEvents(
ctx context.Context,
request *api.QueryStateAfterEventsRequest,
response *api.QueryStateAfterEventsResponse,
) error {
roomVersion, err := r.DB.GetRoomVersionForRoom(ctx, request.RoomID)
if err != nil {
response.RoomExists = false
return nil
}
roomState := state.NewStateResolution(r.DB)
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info.IsStub {
return nil
}
response.RoomExists = true
response.RoomVersion = roomVersion
prevStates, err := r.DB.StateAtEventIDs(ctx, request.PrevEventIDs)
if err != nil {
switch err.(type) {
case types.MissingEventError:
return nil
default:
return err
}
}
response.PrevEventsExist = true
// Look up the currrent state for the requested tuples.
stateEntries, err := roomState.LoadStateAfterEventsForStringTuples(
ctx, info.RoomNID, prevStates, request.StateToFetch,
)
if err != nil {
return err
}
stateEvents, err := r.loadStateEvents(ctx, stateEntries)
if err != nil {
return err
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(roomVersion))
}
return nil
}
// QueryEventsByID implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryEventsByID(
ctx context.Context,
request *api.QueryEventsByIDRequest,
response *api.QueryEventsByIDResponse,
) error {
eventNIDMap, err := r.DB.EventNIDs(ctx, request.EventIDs)
if err != nil {
return err
}
var eventNIDs []types.EventNID
for _, nid := range eventNIDMap {
eventNIDs = append(eventNIDs, nid)
}
events, err := r.loadEvents(ctx, eventNIDs)
if err != nil {
return err
}
for _, event := range events {
roomVersion, verr := r.DB.GetRoomVersionForRoom(ctx, event.RoomID())
if verr != nil {
return verr
}
response.Events = append(response.Events, event.Headered(roomVersion))
}
return nil
}
func (r *RoomserverInternalAPI) loadStateEvents(
ctx context.Context, stateEntries []types.StateEntry,
) ([]gomatrixserverlib.Event, error) {
eventNIDs := make([]types.EventNID, len(stateEntries))
for i := range stateEntries {
eventNIDs[i] = stateEntries[i].EventNID
}
return r.loadEvents(ctx, eventNIDs)
}
func (r *RoomserverInternalAPI) loadEvents(
ctx context.Context, eventNIDs []types.EventNID,
) ([]gomatrixserverlib.Event, error) {
stateEvents, err := r.DB.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
result := make([]gomatrixserverlib.Event, len(stateEvents))
for i := range stateEvents {
result[i] = stateEvents[i].Event
}
return result, nil
}
// QueryMembershipForUser implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryMembershipForUser(
ctx context.Context,
request *api.QueryMembershipForUserRequest,
response *api.QueryMembershipForUserResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
membershipEventNID, stillInRoom, err := r.DB.GetMembership(ctx, info.RoomNID, request.UserID)
if err != nil {
return err
}
if membershipEventNID == 0 {
response.HasBeenInRoom = false
return nil
}
response.IsInRoom = stillInRoom
response.HasBeenInRoom = true
evs, err := r.DB.Events(ctx, []types.EventNID{membershipEventNID})
if err != nil {
return err
}
if len(evs) != 1 {
return fmt.Errorf("failed to load membership event for event NID %d", membershipEventNID)
}
response.EventID = evs[0].EventID()
response.Membership, err = evs[0].Membership()
return err
}
// QueryMembershipsForRoom implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryMembershipsForRoom(
ctx context.Context,
request *api.QueryMembershipsForRoomRequest,
response *api.QueryMembershipsForRoomResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
membershipEventNID, stillInRoom, err := r.DB.GetMembership(ctx, info.RoomNID, request.Sender)
if err != nil {
return err
}
if membershipEventNID == 0 {
response.HasBeenInRoom = false
response.JoinEvents = nil
return nil
}
response.HasBeenInRoom = true
response.JoinEvents = []gomatrixserverlib.ClientEvent{}
var events []types.Event
var stateEntries []types.StateEntry
if stillInRoom {
var eventNIDs []types.EventNID
eventNIDs, err = r.DB.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, request.JoinedOnly, false)
if err != nil {
return err
}
events, err = r.DB.Events(ctx, eventNIDs)
} else {
stateEntries, err = stateBeforeEvent(ctx, r.DB, membershipEventNID)
if err != nil {
logrus.WithField("membership_event_nid", membershipEventNID).WithError(err).Error("failed to load state before event")
return err
}
events, err = getMembershipsAtState(ctx, r.DB, stateEntries, request.JoinedOnly)
}
if err != nil {
return err
}
for _, event := range events {
clientEvent := gomatrixserverlib.ToClientEvent(event.Event, gomatrixserverlib.FormatAll)
response.JoinEvents = append(response.JoinEvents, clientEvent)
}
return nil
}
func stateBeforeEvent(ctx context.Context, db storage.Database, eventNID types.EventNID) ([]types.StateEntry, error) {
roomState := state.NewStateResolution(db)
// Lookup the event NID
eIDs, err := db.EventIDs(ctx, []types.EventNID{eventNID})
if err != nil {
return nil, err
}
eventIDs := []string{eIDs[eventNID]}
prevState, err := db.StateAtEventIDs(ctx, eventIDs)
if err != nil {
return nil, err
}
// Fetch the state as it was when this event was fired
return roomState.LoadCombinedStateAfterEvents(ctx, prevState)
}
// getMembershipsAtState filters the state events to
// only keep the "m.room.member" events with a "join" membership. These events are returned.
// Returns an error if there was an issue fetching the events.
func getMembershipsAtState(
ctx context.Context, db storage.Database, stateEntries []types.StateEntry, joinedOnly bool,
) ([]types.Event, error) {
var eventNIDs []types.EventNID
for _, entry := range stateEntries {
// Filter the events to retrieve to only keep the membership events
if entry.EventTypeNID == types.MRoomMemberNID {
eventNIDs = append(eventNIDs, entry.EventNID)
}
}
// Get all of the events in this state
stateEvents, err := db.Events(ctx, eventNIDs)
if err != nil {
return nil, err
}
if !joinedOnly {
return stateEvents, nil
}
// Filter the events to only keep the "join" membership events
var events []types.Event
for _, event := range stateEvents {
membership, err := event.Membership()
if err != nil {
return nil, err
}
if membership == gomatrixserverlib.Join {
events = append(events, event)
}
}
return events, nil
}
// QueryServerAllowedToSeeEvent implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryServerAllowedToSeeEvent(
ctx context.Context,
request *api.QueryServerAllowedToSeeEventRequest,
response *api.QueryServerAllowedToSeeEventResponse,
) (err error) {
events, err := r.DB.EventsFromIDs(ctx, []string{request.EventID})
if err != nil {
return
}
if len(events) == 0 {
response.AllowedToSeeEvent = false // event doesn't exist so not allowed to see
return
}
isServerInRoom, err := r.isServerCurrentlyInRoom(ctx, request.ServerName, events[0].RoomID())
if err != nil {
return
}
response.AllowedToSeeEvent, err = r.checkServerAllowedToSeeEvent(
ctx, request.EventID, request.ServerName, isServerInRoom,
)
return
}
func (r *RoomserverInternalAPI) checkServerAllowedToSeeEvent(
ctx context.Context, eventID string, serverName gomatrixserverlib.ServerName, isServerInRoom bool,
) (bool, error) {
roomState := state.NewStateResolution(r.DB)
stateEntries, err := roomState.LoadStateAtEvent(ctx, eventID)
if err != nil {
return false, err
}
// TODO: We probably want to make it so that we don't have to pull
// out all the state if possible.
stateAtEvent, err := r.loadStateEvents(ctx, stateEntries)
if err != nil {
return false, err
}
return auth.IsServerAllowed(serverName, isServerInRoom, stateAtEvent), nil
}
// QueryMissingEvents implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryMissingEvents(
ctx context.Context,
request *api.QueryMissingEventsRequest,
response *api.QueryMissingEventsResponse,
) error {
var front []string
eventsToFilter := make(map[string]bool, len(request.LatestEvents))
visited := make(map[string]bool, request.Limit) // request.Limit acts as a hint to size.
for _, id := range request.EarliestEvents {
visited[id] = true
}
for _, id := range request.LatestEvents {
if !visited[id] {
front = append(front, id)
eventsToFilter[id] = true
}
}
resultNIDs, err := r.scanEventTree(ctx, front, visited, request.Limit, request.ServerName)
if err != nil {
return err
}
loadedEvents, err := r.loadEvents(ctx, resultNIDs)
if err != nil {
return err
}
response.Events = make([]gomatrixserverlib.HeaderedEvent, 0, len(loadedEvents)-len(eventsToFilter))
for _, event := range loadedEvents {
if !eventsToFilter[event.EventID()] {
roomVersion, verr := r.DB.GetRoomVersionForRoom(ctx, event.RoomID())
if verr != nil {
return verr
}
response.Events = append(response.Events, event.Headered(roomVersion))
}
}
return err
}
// PerformBackfill implements api.RoomServerQueryAPI
func (r *RoomserverInternalAPI) PerformBackfill(
ctx context.Context,
request *api.PerformBackfillRequest,
response *api.PerformBackfillResponse,
) error {
// if we are requesting the backfill then we need to do a federation hit
// TODO: we could be more sensible and fetch as many events we already have then request the rest
// which is what the syncapi does already.
if request.ServerName == r.ServerName {
return r.backfillViaFederation(ctx, request, response)
}
// someone else is requesting the backfill, try to service their request.
var err error
var front []string
// The limit defines the maximum number of events to retrieve, so it also
// defines the highest number of elements in the map below.
visited := make(map[string]bool, request.Limit)
// this will include these events which is what we want
front = request.PrevEventIDs()
// Scan the event tree for events to send back.
resultNIDs, err := r.scanEventTree(ctx, front, visited, request.Limit, request.ServerName)
if err != nil {
return err
}
// Retrieve events from the list that was filled previously.
var loadedEvents []gomatrixserverlib.Event
loadedEvents, err = r.loadEvents(ctx, resultNIDs)
if err != nil {
return err
}
for _, event := range loadedEvents {
roomVersion, verr := r.DB.GetRoomVersionForRoom(ctx, event.RoomID())
if verr != nil {
return verr
}
response.Events = append(response.Events, event.Headered(roomVersion))
}
return err
}
func (r *RoomserverInternalAPI) backfillViaFederation(ctx context.Context, req *api.PerformBackfillRequest, res *api.PerformBackfillResponse) error {
roomVer, err := r.DB.GetRoomVersionForRoom(ctx, req.RoomID)
if err != nil {
return fmt.Errorf("backfillViaFederation: unknown room version for room %s : %w", req.RoomID, err)
}
requester := newBackfillRequester(r.DB, r.FedClient, r.ServerName, req.BackwardsExtremities)
// Request 100 items regardless of what the query asks for.
// We don't want to go much higher than this.
// We can't honour exactly the limit as some sytests rely on requesting more for tests to pass
// (so we don't need to hit /state_ids which the test has no listener for)
// Specifically the test "Outbound federation can backfill events"
events, err := gomatrixserverlib.RequestBackfill(
ctx, requester,
r.KeyRing, req.RoomID, roomVer, req.PrevEventIDs(), 100)
if err != nil {
return err
}
logrus.WithField("room_id", req.RoomID).Infof("backfilled %d events", len(events))
// persist these new events - auth checks have already been done
roomNID, backfilledEventMap := persistEvents(ctx, r.DB, events)
if err != nil {
return err
}
for _, ev := range backfilledEventMap {
// now add state for these events
stateIDs, ok := requester.eventIDToBeforeStateIDs[ev.EventID()]
if !ok {
// this should be impossible as all events returned must have pass Step 5 of the PDU checks
// which requires a list of state IDs.
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to find state IDs for event which passed auth checks")
continue
}
var entries []types.StateEntry
if entries, err = r.DB.StateEntriesForEventIDs(ctx, stateIDs); err != nil {
// attempt to fetch the missing events
r.fetchAndStoreMissingEvents(ctx, roomVer, requester, stateIDs)
// try again
entries, err = r.DB.StateEntriesForEventIDs(ctx, stateIDs)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to get state entries for event")
return err
}
}
var beforeStateSnapshotNID types.StateSnapshotNID
if beforeStateSnapshotNID, err = r.DB.AddState(ctx, roomNID, nil, entries); err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to persist state entries to get snapshot nid")
return err
}
if err = r.DB.SetState(ctx, ev.EventNID, beforeStateSnapshotNID); err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("backfillViaFederation: failed to persist snapshot nid")
}
}
// TODO: update backwards extremities, as that should be moved from syncapi to roomserver at some point.
res.Events = events
return nil
}
func (r *RoomserverInternalAPI) isServerCurrentlyInRoom(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID string) (bool, error) {
info, err := r.DB.RoomInfo(ctx, roomID)
if err != nil {
return false, err
}
if info == nil {
return false, fmt.Errorf("unknown room %s", roomID)
}
eventNIDs, err := r.DB.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, true, false)
if err != nil {
return false, err
}
events, err := r.DB.Events(ctx, eventNIDs)
if err != nil {
return false, err
}
gmslEvents := make([]gomatrixserverlib.Event, len(events))
for i := range events {
gmslEvents[i] = events[i].Event
}
return auth.IsAnyUserOnServerWithMembership(serverName, gmslEvents, gomatrixserverlib.Join), nil
}
// fetchAndStoreMissingEvents does a best-effort fetch and store of missing events specified in stateIDs. Returns no error as it is just
// best effort.
func (r *RoomserverInternalAPI) fetchAndStoreMissingEvents(ctx context.Context, roomVer gomatrixserverlib.RoomVersion,
backfillRequester *backfillRequester, stateIDs []string) {
servers := backfillRequester.servers
// work out which are missing
nidMap, err := r.DB.EventNIDs(ctx, stateIDs)
if err != nil {
util.GetLogger(ctx).WithError(err).Warn("cannot query missing events")
return
}
missingMap := make(map[string]*gomatrixserverlib.HeaderedEvent) // id -> event
for _, id := range stateIDs {
if _, ok := nidMap[id]; !ok {
missingMap[id] = nil
}
}
util.GetLogger(ctx).Infof("Fetching %d missing state events (from %d possible servers)", len(missingMap), len(servers))
// fetch the events from federation. Loop the servers first so if we find one that works we stick with them
for _, srv := range servers {
for id, ev := range missingMap {
if ev != nil {
continue // already found
}
logger := util.GetLogger(ctx).WithField("server", srv).WithField("event_id", id)
res, err := r.FedClient.GetEvent(ctx, srv, id)
if err != nil {
logger.WithError(err).Warn("failed to get event from server")
continue
}
loader := gomatrixserverlib.NewEventsLoader(roomVer, r.KeyRing, backfillRequester, backfillRequester.ProvideEvents, false)
result, err := loader.LoadAndVerify(ctx, res.PDUs, gomatrixserverlib.TopologicalOrderByPrevEvents)
if err != nil {
logger.WithError(err).Warn("failed to load and verify event")
continue
}
logger.Infof("returned %d PDUs which made events %+v", len(res.PDUs), result)
for _, res := range result {
if res.Error != nil {
logger.WithError(err).Warn("event failed PDU checks")
continue
}
missingMap[id] = res.Event
}
}
}
var newEvents []gomatrixserverlib.HeaderedEvent
for _, ev := range missingMap {
if ev != nil {
newEvents = append(newEvents, *ev)
}
}
util.GetLogger(ctx).Infof("Persisting %d new events", len(newEvents))
persistEvents(ctx, r.DB, newEvents)
}
// TODO: Remove this when we have tests to assert correctness of this function
// nolint:gocyclo
func (r *RoomserverInternalAPI) scanEventTree(
ctx context.Context, front []string, visited map[string]bool, limit int,
serverName gomatrixserverlib.ServerName,
) ([]types.EventNID, error) {
var resultNIDs []types.EventNID
var err error
var allowed bool
var events []types.Event
var next []string
var pre string
// TODO: add tests for this function to ensure it meets the contract that callers expect (and doc what that is supposed to be)
// Currently, callers like PerformBackfill will call scanEventTree with a pre-populated `visited` map, assuming that by doing
// so means that the events in that map will NOT be returned from this function. That is not currently true, resulting in
// duplicate events being sent in response to /backfill requests.
initialIgnoreList := make(map[string]bool, len(visited))
for k, v := range visited {
initialIgnoreList[k] = v
}
resultNIDs = make([]types.EventNID, 0, limit)
var checkedServerInRoom bool
var isServerInRoom bool
// Loop through the event IDs to retrieve the requested events and go
// through the whole tree (up to the provided limit) using the events'
// "prev_event" key.
BFSLoop:
for len(front) > 0 {
// Prevent unnecessary allocations: reset the slice only when not empty.
if len(next) > 0 {
next = make([]string, 0)
}
// Retrieve the events to process from the database.
events, err = r.DB.EventsFromIDs(ctx, front)
if err != nil {
return resultNIDs, err
}
if !checkedServerInRoom && len(events) > 0 {
// It's nasty that we have to extract the room ID from an event, but many federation requests
// only talk in event IDs, no room IDs at all (!!!)
ev := events[0]
isServerInRoom, err = r.isServerCurrentlyInRoom(ctx, serverName, ev.RoomID())
if err != nil {
util.GetLogger(ctx).WithError(err).Error("Failed to check if server is currently in room, assuming not.")
}
checkedServerInRoom = true
}
for _, ev := range events {
// Break out of the loop if the provided limit is reached.
if len(resultNIDs) == limit {
break BFSLoop
}
if !initialIgnoreList[ev.EventID()] {
// Update the list of events to retrieve.
resultNIDs = append(resultNIDs, ev.EventNID)
}
// Loop through the event's parents.
for _, pre = range ev.PrevEventIDs() {
// Only add an event to the list of next events to process if it
// hasn't been seen before.
if !visited[pre] {
visited[pre] = true
allowed, err = r.checkServerAllowedToSeeEvent(ctx, pre, serverName, isServerInRoom)
if err != nil {
util.GetLogger(ctx).WithField("server", serverName).WithField("event_id", pre).WithError(err).Error(
"Error checking if allowed to see event",
)
return resultNIDs, err
}
// If the event hasn't been seen before and the HS
// requesting to retrieve it is allowed to do so, add it to
// the list of events to retrieve.
if allowed {
next = append(next, pre)
} else {
util.GetLogger(ctx).WithField("server", serverName).WithField("event_id", pre).Info("Not allowed to see event")
}
}
}
}
// Repeat the same process with the parent events we just processed.
front = next
}
return resultNIDs, err
}
// QueryStateAndAuthChain implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryStateAndAuthChain(
ctx context.Context,
request *api.QueryStateAndAuthChainRequest,
response *api.QueryStateAndAuthChainResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info.IsStub {
return nil
}
response.RoomExists = true
response.RoomVersion = info.RoomVersion
stateEvents, err := r.loadStateAtEventIDs(ctx, request.PrevEventIDs)
if err != nil {
return err
}
response.PrevEventsExist = true
// add the auth event IDs for the current state events too
var authEventIDs []string
authEventIDs = append(authEventIDs, request.AuthEventIDs...)
for _, se := range stateEvents {
authEventIDs = append(authEventIDs, se.AuthEventIDs()...)
}
authEventIDs = util.UniqueStrings(authEventIDs) // de-dupe
authEvents, err := getAuthChain(ctx, r.DB.EventsFromIDs, authEventIDs)
if err != nil {
return err
}
if request.ResolveState {
if stateEvents, err = state.ResolveConflictsAdhoc(
info.RoomVersion, stateEvents, authEvents,
); err != nil {
return err
}
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(info.RoomVersion))
}
for _, event := range authEvents {
response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
}
return err
}
func (r *RoomserverInternalAPI) loadStateAtEventIDs(ctx context.Context, eventIDs []string) ([]gomatrixserverlib.Event, error) {
roomState := state.NewStateResolution(r.DB)
prevStates, err := r.DB.StateAtEventIDs(ctx, eventIDs)
if err != nil {
switch err.(type) {
case types.MissingEventError:
return nil, nil
default:
return nil, err
}
}
// Look up the currrent state for the requested tuples.
stateEntries, err := roomState.LoadCombinedStateAfterEvents(
ctx, prevStates,
)
if err != nil {
return nil, err
}
return r.loadStateEvents(ctx, stateEntries)
}
type eventsFromIDs func(context.Context, []string) ([]types.Event, error)
// getAuthChain fetches the auth chain for the given auth events. An auth chain
// is the list of all events that are referenced in the auth_events section, and
// all their auth_events, recursively. The returned set of events contain the
// given events. Will *not* error if we don't have all auth events.
func getAuthChain(
ctx context.Context, fn eventsFromIDs, authEventIDs []string,
) ([]gomatrixserverlib.Event, error) {
// List of event IDs to fetch. On each pass, these events will be requested
// from the database and the `eventsToFetch` will be updated with any new
// events that we have learned about and need to find. When `eventsToFetch`
// is eventually empty, we should have reached the end of the chain.
eventsToFetch := authEventIDs
authEventsMap := make(map[string]gomatrixserverlib.Event)
for len(eventsToFetch) > 0 {
// Try to retrieve the events from the database.
events, err := fn(ctx, eventsToFetch)
if err != nil {
return nil, err
}
// We've now fetched these events so clear out `eventsToFetch`. Soon we may
// add newly discovered events to this for the next pass.
eventsToFetch = eventsToFetch[:0]
for _, event := range events {
// Store the event in the event map - this prevents us from requesting it
// from the database again.
authEventsMap[event.EventID()] = event.Event
// Extract all of the auth events from the newly obtained event. If we
// don't already have a record of the event, record it in the list of
// events we want to request for the next pass.
for _, authEvent := range event.AuthEvents() {
if _, ok := authEventsMap[authEvent.EventID]; !ok {
eventsToFetch = append(eventsToFetch, authEvent.EventID)
}
}
}
}
// We've now retrieved all of the events we can. Flatten them down into an
// array and return them.
var authEvents []gomatrixserverlib.Event
for _, event := range authEventsMap {
authEvents = append(authEvents, event)
}
return authEvents, nil
}
func persistEvents(ctx context.Context, db storage.Database, events []gomatrixserverlib.HeaderedEvent) (types.RoomNID, map[string]types.Event) {
var roomNID types.RoomNID
backfilledEventMap := make(map[string]types.Event)
for j, ev := range events {
nidMap, err := db.EventNIDs(ctx, ev.AuthEventIDs())
if err != nil { // this shouldn't happen as RequestBackfill already found them
logrus.WithError(err).WithField("auth_events", ev.AuthEventIDs()).Error("Failed to find one or more auth events")
continue
}
authNids := make([]types.EventNID, len(nidMap))
i := 0
for _, nid := range nidMap {
authNids[i] = nid
i++
}
var stateAtEvent types.StateAtEvent
var redactedEventID string
var redactionEvent *gomatrixserverlib.Event
roomNID, stateAtEvent, redactionEvent, redactedEventID, err = db.StoreEvent(ctx, ev.Unwrap(), nil, authNids)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to persist event")
continue
}
// If storing this event results in it being redacted, then do so.
// It's also possible for this event to be a redaction which results in another event being
// redacted, which we don't care about since we aren't returning it in this backfill.
if redactedEventID == ev.EventID() {
eventToRedact := ev.Unwrap()
redactedEvent, err := eventutil.RedactEvent(redactionEvent, &eventToRedact)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to redact event")
continue
}
ev = redactedEvent.Headered(ev.RoomVersion)
events[j] = ev
}
backfilledEventMap[ev.EventID()] = types.Event{
EventNID: stateAtEvent.StateEntry.EventNID,
Event: ev.Unwrap(),
}
}
return roomNID, backfilledEventMap
}
// QueryRoomVersionCapabilities implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryRoomVersionCapabilities(
ctx context.Context,
request *api.QueryRoomVersionCapabilitiesRequest,
response *api.QueryRoomVersionCapabilitiesResponse,
) error {
response.DefaultRoomVersion = version.DefaultRoomVersion()
response.AvailableRoomVersions = make(map[gomatrixserverlib.RoomVersion]string)
for v, desc := range version.SupportedRoomVersions() {
if desc.Stable {
response.AvailableRoomVersions[v] = "stable"
} else {
response.AvailableRoomVersions[v] = "unstable"
}
}
return nil
}
// QueryRoomVersionCapabilities implements api.RoomserverInternalAPI
func (r *RoomserverInternalAPI) QueryRoomVersionForRoom(
ctx context.Context,
request *api.QueryRoomVersionForRoomRequest,
response *api.QueryRoomVersionForRoomResponse,
) error {
if roomVersion, ok := r.Cache.GetRoomVersion(request.RoomID); ok {
response.RoomVersion = roomVersion
return nil
}
roomVersion, err := r.DB.GetRoomVersionForRoom(ctx, request.RoomID)
if err != nil {
return err
}
response.RoomVersion = roomVersion
r.Cache.StoreRoomVersion(request.RoomID, response.RoomVersion)
return nil
}
func (r *RoomserverInternalAPI) QueryPublishedRooms(
ctx context.Context,
req *api.QueryPublishedRoomsRequest,
res *api.QueryPublishedRoomsResponse,
) error {
rooms, err := r.DB.GetPublishedRooms(ctx)
if err != nil {
return err
}
res.RoomIDs = rooms
return nil
}

602
roomserver/internal/query/query.go Normal file
View file

@ -0,0 +1,602 @@
// 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 query
import (
"context"
"errors"
"fmt"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/roomserver/acls"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/roomserver/version"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/sirupsen/logrus"
)
type Queryer struct {
DB storage.Database
Cache caching.RoomServerCaches
ServerACLs *acls.ServerACLs
}
// QueryLatestEventsAndState implements api.RoomserverInternalAPI
func (r *Queryer) QueryLatestEventsAndState(
ctx context.Context,
request *api.QueryLatestEventsAndStateRequest,
response *api.QueryLatestEventsAndStateResponse,
) error {
return helpers.QueryLatestEventsAndState(ctx, r.DB, request, response)
}
// QueryStateAfterEvents implements api.RoomserverInternalAPI
func (r *Queryer) QueryStateAfterEvents(
ctx context.Context,
request *api.QueryStateAfterEventsRequest,
response *api.QueryStateAfterEventsResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info == nil || info.IsStub {
return nil
}
roomState := state.NewStateResolution(r.DB, *info)
response.RoomExists = true
response.RoomVersion = info.RoomVersion
prevStates, err := r.DB.StateAtEventIDs(ctx, request.PrevEventIDs)
if err != nil {
switch err.(type) {
case types.MissingEventError:
return nil
default:
return err
}
}
response.PrevEventsExist = true
// Look up the currrent state for the requested tuples.
stateEntries, err := roomState.LoadStateAfterEventsForStringTuples(
ctx, prevStates, request.StateToFetch,
)
if err != nil {
return err
}
stateEvents, err := helpers.LoadStateEvents(ctx, r.DB, stateEntries)
if err != nil {
return err
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(info.RoomVersion))
}
return nil
}
// QueryEventsByID implements api.RoomserverInternalAPI
func (r *Queryer) QueryEventsByID(
ctx context.Context,
request *api.QueryEventsByIDRequest,
response *api.QueryEventsByIDResponse,
) error {
eventNIDMap, err := r.DB.EventNIDs(ctx, request.EventIDs)
if err != nil {
return err
}
var eventNIDs []types.EventNID
for _, nid := range eventNIDMap {
eventNIDs = append(eventNIDs, nid)
}
events, err := helpers.LoadEvents(ctx, r.DB, eventNIDs)
if err != nil {
return err
}
for _, event := range events {
roomVersion, verr := r.roomVersion(event.RoomID())
if verr != nil {
return verr
}
response.Events = append(response.Events, event.Headered(roomVersion))
}
return nil
}
// QueryMembershipForUser implements api.RoomserverInternalAPI
func (r *Queryer) QueryMembershipForUser(
ctx context.Context,
request *api.QueryMembershipForUserRequest,
response *api.QueryMembershipForUserResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
membershipEventNID, stillInRoom, err := r.DB.GetMembership(ctx, info.RoomNID, request.UserID)
if err != nil {
return err
}
if membershipEventNID == 0 {
response.HasBeenInRoom = false
return nil
}
response.IsInRoom = stillInRoom
response.HasBeenInRoom = true
evs, err := r.DB.Events(ctx, []types.EventNID{membershipEventNID})
if err != nil {
return err
}
if len(evs) != 1 {
return fmt.Errorf("failed to load membership event for event NID %d", membershipEventNID)
}
response.EventID = evs[0].EventID()
response.Membership, err = evs[0].Membership()
return err
}
// QueryMembershipsForRoom implements api.RoomserverInternalAPI
func (r *Queryer) QueryMembershipsForRoom(
ctx context.Context,
request *api.QueryMembershipsForRoomRequest,
response *api.QueryMembershipsForRoomResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
membershipEventNID, stillInRoom, err := r.DB.GetMembership(ctx, info.RoomNID, request.Sender)
if err != nil {
return err
}
if membershipEventNID == 0 {
response.HasBeenInRoom = false
response.JoinEvents = nil
return nil
}
response.HasBeenInRoom = true
response.JoinEvents = []gomatrixserverlib.ClientEvent{}
var events []types.Event
var stateEntries []types.StateEntry
if stillInRoom {
var eventNIDs []types.EventNID
eventNIDs, err = r.DB.GetMembershipEventNIDsForRoom(ctx, info.RoomNID, request.JoinedOnly, false)
if err != nil {
return err
}
events, err = r.DB.Events(ctx, eventNIDs)
} else {
stateEntries, err = helpers.StateBeforeEvent(ctx, r.DB, *info, membershipEventNID)
if err != nil {
logrus.WithField("membership_event_nid", membershipEventNID).WithError(err).Error("failed to load state before event")
return err
}
events, err = helpers.GetMembershipsAtState(ctx, r.DB, stateEntries, request.JoinedOnly)
}
if err != nil {
return err
}
for _, event := range events {
clientEvent := gomatrixserverlib.ToClientEvent(event.Event, gomatrixserverlib.FormatAll)
response.JoinEvents = append(response.JoinEvents, clientEvent)
}
return nil
}
// QueryServerAllowedToSeeEvent implements api.RoomserverInternalAPI
func (r *Queryer) QueryServerAllowedToSeeEvent(
ctx context.Context,
request *api.QueryServerAllowedToSeeEventRequest,
response *api.QueryServerAllowedToSeeEventResponse,
) (err error) {
events, err := r.DB.EventsFromIDs(ctx, []string{request.EventID})
if err != nil {
return
}
if len(events) == 0 {
response.AllowedToSeeEvent = false // event doesn't exist so not allowed to see
return
}
roomID := events[0].RoomID()
isServerInRoom, err := helpers.IsServerCurrentlyInRoom(ctx, r.DB, request.ServerName, roomID)
if err != nil {
return
}
info, err := r.DB.RoomInfo(ctx, roomID)
if err != nil {
return err
}
if info == nil {
return fmt.Errorf("QueryServerAllowedToSeeEvent: no room info for room %s", roomID)
}
response.AllowedToSeeEvent, err = helpers.CheckServerAllowedToSeeEvent(
ctx, r.DB, *info, request.EventID, request.ServerName, isServerInRoom,
)
return
}
// QueryMissingEvents implements api.RoomserverInternalAPI
// nolint:gocyclo
func (r *Queryer) QueryMissingEvents(
ctx context.Context,
request *api.QueryMissingEventsRequest,
response *api.QueryMissingEventsResponse,
) error {
var front []string
eventsToFilter := make(map[string]bool, len(request.LatestEvents))
visited := make(map[string]bool, request.Limit) // request.Limit acts as a hint to size.
for _, id := range request.EarliestEvents {
visited[id] = true
}
for _, id := range request.LatestEvents {
if !visited[id] {
front = append(front, id)
eventsToFilter[id] = true
}
}
events, err := r.DB.EventsFromIDs(ctx, front)
if err != nil {
return err
}
if len(events) == 0 {
return nil // we are missing the events being asked to search from, give up.
}
info, err := r.DB.RoomInfo(ctx, events[0].RoomID())
if err != nil {
return err
}
if info == nil || info.IsStub {
return fmt.Errorf("missing RoomInfo for room %s", events[0].RoomID())
}
resultNIDs, err := helpers.ScanEventTree(ctx, r.DB, *info, front, visited, request.Limit, request.ServerName)
if err != nil {
return err
}
loadedEvents, err := helpers.LoadEvents(ctx, r.DB, resultNIDs)
if err != nil {
return err
}
response.Events = make([]gomatrixserverlib.HeaderedEvent, 0, len(loadedEvents)-len(eventsToFilter))
for _, event := range loadedEvents {
if !eventsToFilter[event.EventID()] {
roomVersion, verr := r.roomVersion(event.RoomID())
if verr != nil {
return verr
}
response.Events = append(response.Events, event.Headered(roomVersion))
}
}
return err
}
// QueryStateAndAuthChain implements api.RoomserverInternalAPI
func (r *Queryer) QueryStateAndAuthChain(
ctx context.Context,
request *api.QueryStateAndAuthChainRequest,
response *api.QueryStateAndAuthChainResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info == nil || info.IsStub {
return nil
}
response.RoomExists = true
response.RoomVersion = info.RoomVersion
stateEvents, err := r.loadStateAtEventIDs(ctx, *info, request.PrevEventIDs)
if err != nil {
return err
}
response.PrevEventsExist = true
// add the auth event IDs for the current state events too
var authEventIDs []string
authEventIDs = append(authEventIDs, request.AuthEventIDs...)
for _, se := range stateEvents {
authEventIDs = append(authEventIDs, se.AuthEventIDs()...)
}
authEventIDs = util.UniqueStrings(authEventIDs) // de-dupe
authEvents, err := getAuthChain(ctx, r.DB.EventsFromIDs, authEventIDs)
if err != nil {
return err
}
if request.ResolveState {
if stateEvents, err = state.ResolveConflictsAdhoc(
info.RoomVersion, stateEvents, authEvents,
); err != nil {
return err
}
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(info.RoomVersion))
}
for _, event := range authEvents {
response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
}
return err
}
func (r *Queryer) loadStateAtEventIDs(ctx context.Context, roomInfo types.RoomInfo, eventIDs []string) ([]gomatrixserverlib.Event, error) {
roomState := state.NewStateResolution(r.DB, roomInfo)
prevStates, err := r.DB.StateAtEventIDs(ctx, eventIDs)
if err != nil {
switch err.(type) {
case types.MissingEventError:
return nil, nil
default:
return nil, err
}
}
// Look up the currrent state for the requested tuples.
stateEntries, err := roomState.LoadCombinedStateAfterEvents(
ctx, prevStates,
)
if err != nil {
return nil, err
}
return helpers.LoadStateEvents(ctx, r.DB, stateEntries)
}
type eventsFromIDs func(context.Context, []string) ([]types.Event, error)
// getAuthChain fetches the auth chain for the given auth events. An auth chain
// is the list of all events that are referenced in the auth_events section, and
// all their auth_events, recursively. The returned set of events contain the
// given events. Will *not* error if we don't have all auth events.
func getAuthChain(
ctx context.Context, fn eventsFromIDs, authEventIDs []string,
) ([]gomatrixserverlib.Event, error) {
// List of event IDs to fetch. On each pass, these events will be requested
// from the database and the `eventsToFetch` will be updated with any new
// events that we have learned about and need to find. When `eventsToFetch`
// is eventually empty, we should have reached the end of the chain.
eventsToFetch := authEventIDs
authEventsMap := make(map[string]gomatrixserverlib.Event)
for len(eventsToFetch) > 0 {
// Try to retrieve the events from the database.
events, err := fn(ctx, eventsToFetch)
if err != nil {
return nil, err
}
// We've now fetched these events so clear out `eventsToFetch`. Soon we may
// add newly discovered events to this for the next pass.
eventsToFetch = eventsToFetch[:0]
for _, event := range events {
// Store the event in the event map - this prevents us from requesting it
// from the database again.
authEventsMap[event.EventID()] = event.Event
// Extract all of the auth events from the newly obtained event. If we
// don't already have a record of the event, record it in the list of
// events we want to request for the next pass.
for _, authEvent := range event.AuthEvents() {
if _, ok := authEventsMap[authEvent.EventID]; !ok {
eventsToFetch = append(eventsToFetch, authEvent.EventID)
}
}
}
}
// We've now retrieved all of the events we can. Flatten them down into an
// array and return them.
var authEvents []gomatrixserverlib.Event
for _, event := range authEventsMap {
authEvents = append(authEvents, event)
}
return authEvents, nil
}
// QueryRoomVersionCapabilities implements api.RoomserverInternalAPI
func (r *Queryer) QueryRoomVersionCapabilities(
ctx context.Context,
request *api.QueryRoomVersionCapabilitiesRequest,
response *api.QueryRoomVersionCapabilitiesResponse,
) error {
response.DefaultRoomVersion = version.DefaultRoomVersion()
response.AvailableRoomVersions = make(map[gomatrixserverlib.RoomVersion]string)
for v, desc := range version.SupportedRoomVersions() {
if desc.Stable {
response.AvailableRoomVersions[v] = "stable"
} else {
response.AvailableRoomVersions[v] = "unstable"
}
}
return nil
}
// QueryRoomVersionCapabilities implements api.RoomserverInternalAPI
func (r *Queryer) QueryRoomVersionForRoom(
ctx context.Context,
request *api.QueryRoomVersionForRoomRequest,
response *api.QueryRoomVersionForRoomResponse,
) error {
if roomVersion, ok := r.Cache.GetRoomVersion(request.RoomID); ok {
response.RoomVersion = roomVersion
return nil
}
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info == nil {
return fmt.Errorf("QueryRoomVersionForRoom: missing room info for room %s", request.RoomID)
}
response.RoomVersion = info.RoomVersion
r.Cache.StoreRoomVersion(request.RoomID, response.RoomVersion)
return nil
}
func (r *Queryer) roomVersion(roomID string) (gomatrixserverlib.RoomVersion, error) {
var res api.QueryRoomVersionForRoomResponse
err := r.QueryRoomVersionForRoom(context.Background(), &api.QueryRoomVersionForRoomRequest{
RoomID: roomID,
}, &res)
return res.RoomVersion, err
}
func (r *Queryer) QueryPublishedRooms(
ctx context.Context,
req *api.QueryPublishedRoomsRequest,
res *api.QueryPublishedRoomsResponse,
) error {
rooms, err := r.DB.GetPublishedRooms(ctx)
if err != nil {
return err
}
res.RoomIDs = rooms
return nil
}
func (r *Queryer) QueryCurrentState(ctx context.Context, req *api.QueryCurrentStateRequest, res *api.QueryCurrentStateResponse) error {
res.StateEvents = make(map[gomatrixserverlib.StateKeyTuple]*gomatrixserverlib.HeaderedEvent)
for _, tuple := range req.StateTuples {
ev, err := r.DB.GetStateEvent(ctx, req.RoomID, tuple.EventType, tuple.StateKey)
if err != nil {
return err
}
if ev != nil {
res.StateEvents[tuple] = ev
}
}
return nil
}
func (r *Queryer) QueryRoomsForUser(ctx context.Context, req *api.QueryRoomsForUserRequest, res *api.QueryRoomsForUserResponse) error {
roomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, req.WantMembership)
if err != nil {
return err
}
res.RoomIDs = roomIDs
return nil
}
func (r *Queryer) QueryKnownUsers(ctx context.Context, req *api.QueryKnownUsersRequest, res *api.QueryKnownUsersResponse) error {
users, err := r.DB.GetKnownUsers(ctx, req.UserID, req.SearchString, req.Limit)
if err != nil {
return err
}
for _, user := range users {
res.Users = append(res.Users, authtypes.FullyQualifiedProfile{
UserID: user,
})
}
return nil
}
func (r *Queryer) QueryBulkStateContent(ctx context.Context, req *api.QueryBulkStateContentRequest, res *api.QueryBulkStateContentResponse) error {
events, err := r.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
}
func (r *Queryer) QuerySharedUsers(ctx context.Context, req *api.QuerySharedUsersRequest, res *api.QuerySharedUsersResponse) error {
roomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, "join")
if err != nil {
return err
}
roomIDs = append(roomIDs, req.IncludeRoomIDs...)
excludeMap := make(map[string]bool)
for _, roomID := range req.ExcludeRoomIDs {
excludeMap[roomID] = true
}
// filter out excluded rooms
j := 0
for i := range roomIDs {
// move elements to include to the beginning of the slice
// then trim elements on the right
if !excludeMap[roomIDs[i]] {
roomIDs[j] = roomIDs[i]
j++
}
}
roomIDs = roomIDs[:j]
users, err := r.DB.JoinedUsersSetInRooms(ctx, roomIDs)
if err != nil {
return err
}
res.UserIDsToCount = users
return nil
}
func (r *Queryer) QueryServerBannedFromRoom(ctx context.Context, req *api.QueryServerBannedFromRoomRequest, res *api.QueryServerBannedFromRoomResponse) error {
if r.ServerACLs == nil {
return errors.New("no server ACL tracking")
}
res.Banned = r.ServerACLs.IsServerBannedFromRoom(req.ServerName, req.RoomID)
return nil
}

4
roomserver/internal/query_test.go → roomserver/internal/query/query_test.go
View file

@ -1,4 +1,4 @@
// Copyright 2017 Vector Creations Ltd
// 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.
@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package internal
package query
import (
"context"