msc2946: handle suggested_only; remove custom database

As the MSC doesn't require reverse lookups, we can just pull
the room state and inspect via the roomserver database. To
handle this, expand QueryCurrentState to support wildcards.

Use all this and handle `?suggested_only`.

roomserver/api/query.go

@@ -313,6 +313,9 @@ type QueryBulkStateContentResponse struct {
 
 type QueryCurrentStateRequest struct {
 	RoomID         string
+	AllowWildcards bool
+	// State key tuples. If a state_key has '*' and AllowWidlcards is true, returns all matching
+	// state events with that event type.
 	StateTuples []gomatrixserverlib.StateKeyTuple
 }
 

roomserver/internal/query/query.go

@@ -621,6 +621,18 @@ func (r *Queryer) QueryPublishedRooms(
 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 {
+		if tuple.StateKey == "*" && req.AllowWildcards {
+			events, err := r.DB.GetStateEventsWithEventType(ctx, req.RoomID, tuple.EventType)
+			if err != nil {
+				return err
+			}
+			for _, e := range events {
+				res.StateEvents[gomatrixserverlib.StateKeyTuple{
+					EventType: e.Type(),
+					StateKey:  *e.StateKey(),
+				}] = e
+			}
+		} else {
 			ev, err := r.DB.GetStateEvent(ctx, req.RoomID, tuple.EventType, tuple.StateKey)
 			if err != nil {
 				return err
@@ -629,6 +641,7 @@ func (r *Queryer) QueryCurrentState(ctx context.Context, req *api.QueryCurrentSt
 				res.StateEvents[tuple] = ev
 			}
 		}
+	}
 	return nil
 }
 

roomserver/storage/interface.go

@@ -146,6 +146,7 @@ type Database interface {
 	// If no event could be found, returns nil
 	// If there was an issue during the retrieval, returns an error
 	GetStateEvent(ctx context.Context, roomID, evType, stateKey string) (*gomatrixserverlib.HeaderedEvent, error)
+	GetStateEventsWithEventType(ctx context.Context, roomID, evType string) ([]*gomatrixserverlib.HeaderedEvent, error)
 	// GetRoomsByMembership returns a list of room IDs matching the provided membership and user ID (as state_key).
 	GetRoomsByMembership(ctx context.Context, userID, membership string) ([]string, error)
 	// GetBulkStateContent returns all state events which match a given room ID and a given state key tuple. Both must be satisfied for a match.

roomserver/storage/shared/storage.go

@@ -979,6 +979,62 @@ func (d *Database) GetStateEvent(ctx context.Context, roomID, evType, stateKey s
 	return nil, nil
 }
 
+// Same as GetStateEvent but returns all matching state events with this event type. Returns no error
+// if there are no events with this event type.
+func (d *Database) GetStateEventsWithEventType(ctx context.Context, roomID, evType string) ([]*gomatrixserverlib.HeaderedEvent, error) {
+	roomInfo, err := d.RoomInfo(ctx, roomID)
+	if err != nil {
+		return nil, err
+	}
+	if roomInfo == nil {
+		return nil, fmt.Errorf("room %s doesn't exist", roomID)
+	}
+	// e.g invited rooms
+	if roomInfo.IsStub {
+		return nil, nil
+	}
+	eventTypeNID, err := d.EventTypesTable.SelectEventTypeNID(ctx, nil, evType)
+	if err == sql.ErrNoRows {
+		// No rooms have an event of this type, otherwise we'd have an event type NID
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	entries, err := d.loadStateAtSnapshot(ctx, roomInfo.StateSnapshotNID)
+	if err != nil {
+		return nil, err
+	}
+	var eventNIDs []types.EventNID
+	for _, e := range entries {
+		if e.EventTypeNID == eventTypeNID {
+			eventNIDs = append(eventNIDs, e.EventNID)
+		}
+	}
+	eventIDs, _ := d.EventsTable.BulkSelectEventID(ctx, nil, eventNIDs)
+	if err != nil {
+		eventIDs = map[types.EventNID]string{}
+	}
+	// return the events requested
+	eventPairs, err := d.EventJSONTable.BulkSelectEventJSON(ctx, nil, eventNIDs)
+	if err != nil {
+		return nil, err
+	}
+	if len(eventPairs) == 0 {
+		return nil, nil
+	}
+	var result []*gomatrixserverlib.HeaderedEvent
+	for _, pair := range eventPairs {
+		ev, err := gomatrixserverlib.NewEventFromTrustedJSONWithEventID(eventIDs[pair.EventNID], pair.EventJSON, false, roomInfo.RoomVersion)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, ev.Headered(roomInfo.RoomVersion))
+	}
+
+	return result, nil
+}
+
 // GetRoomsByMembership returns a list of room IDs matching the provided membership and user ID (as state_key).
 func (d *Database) GetRoomsByMembership(ctx context.Context, userID, membership string) ([]string, error) {
 	var membershipState tables.MembershipState

setup/mscs/msc2946/msc2946.go

@@ -18,7 +18,6 @@ package msc2946
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -29,7 +28,6 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	fs "github.com/matrix-org/dendrite/federationapi/api"
-	"github.com/matrix-org/dendrite/internal/hooks"
 	"github.com/matrix-org/dendrite/internal/httputil"
 	roomserver "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/base"
@@ -56,21 +54,7 @@ func Enable(
 	base *base.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
 	fsAPI fs.FederationInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
 ) error {
-	db, err := NewDatabase(&base.Cfg.MSCs.Database)
-	if err != nil {
-		return fmt.Errorf("cannot enable MSC2946: %w", err)
-	}
-	hooks.Enable()
-	hooks.Attach(hooks.KindNewEventPersisted, func(headeredEvent interface{}) {
-		he := headeredEvent.(*gomatrixserverlib.HeaderedEvent)
-		hookErr := db.StoreReference(context.Background(), he)
-		if hookErr != nil {
-			util.GetLogger(context.Background()).WithError(hookErr).WithField("event_id", he.EventID()).Error(
-				"failed to StoreReference",
-			)
-		}
-	})
-	clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(db, rsAPI, fsAPI, base.Cfg.Global.ServerName))
+	clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, base.Cfg.Global.ServerName))
 	base.PublicClientAPIMux.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
 	base.PublicClientAPIMux.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
 
@@ -88,7 +72,7 @@ func Enable(
 				return util.ErrorResponse(err)
 			}
 			roomID := params["roomID"]
-			return federatedSpacesHandler(req.Context(), fedReq, roomID, db, rsAPI, fsAPI, base.Cfg.Global.ServerName)
+			return federatedSpacesHandler(req.Context(), fedReq, roomID, rsAPI, fsAPI, base.Cfg.Global.ServerName)
 		},
 	)
 	base.PublicFederationAPIMux.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
@@ -97,7 +81,7 @@ func Enable(
 }
 
 func federatedSpacesHandler(
-	ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, roomID string, db Database,
+	ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, roomID string,
 	rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
 	thisServer gomatrixserverlib.ServerName,
 ) util.JSONResponse {
@@ -121,7 +105,6 @@ func federatedSpacesHandler(
 		// This is somewhat equivalent to a Client-Server request with a max_depth=1.
 		maxDepth: 1,
 
-		db:                 db,
 		rsAPI:              rsAPI,
 		fsAPI:              fsAPI,
 		inMemoryBatchCache: inMemoryBatchCache,
@@ -130,7 +113,7 @@ func federatedSpacesHandler(
 }
 
 func spacesHandler(
-	db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
+	rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
 	thisServer gomatrixserverlib.ServerName,
 ) func(*http.Request, *userapi.Device) util.JSONResponse {
 	return func(req *http.Request, device *userapi.Device) util.JSONResponse {
@@ -150,7 +133,6 @@ func spacesHandler(
 			thisServer:    thisServer,
 			ctx:           req.Context(),
 
-			db:                 db,
 			rsAPI:              rsAPI,
 			fsAPI:              fsAPI,
 			inMemoryBatchCache: inMemoryBatchCache,
@@ -164,7 +146,6 @@ type walker struct {
 	caller        *userapi.Device
 	serverName    gomatrixserverlib.ServerName
 	thisServer    gomatrixserverlib.ServerName
-	db            Database
 	rsAPI         roomserver.RoomserverInternalAPI
 	fsAPI         fs.FederationInternalAPI
 	ctx           context.Context
@@ -208,6 +189,7 @@ func (w *walker) markSent(id string) {
 type roomVisit struct {
 	roomID string
 	depth  int
+	vias   []string // vias to query this room by
 }
 
 func (w *walker) walk() util.JSONResponse {
@@ -284,7 +266,7 @@ func (w *walker) walk() util.JSONResponse {
 		} else {
 			// attempt to query this room over federation, as either we've never heard of it before
 			// or we've left it and hence are not authorised (but info may be exposed regardless)
-			fedRes, err := w.federatedRoomInfo(rv.roomID)
+			fedRes, err := w.federatedRoomInfo(rv.roomID, rv.vias)
 			if err != nil {
 				util.GetLogger(w.ctx).WithError(err).WithField("room_id", rv.roomID).Errorf("failed to query federated spaces")
 				continue
@@ -308,17 +290,22 @@ func (w *walker) walk() util.JSONResponse {
 			continue
 		}
 
-		uniqueRooms := make(set)
+		uniqueRooms := make(map[string][]string)
 		for _, ev := range discoveredChildEvents {
-			uniqueRooms[ev.StateKey] = true
+			spaceContent := struct {
+				Via []string `json:"via"`
+			}{}
+			_ = json.Unmarshal(ev.Content, &spaceContent)
+			uniqueRooms[ev.StateKey] = spaceContent.Via
 		}
 
 		// For each referenced room ID in the child events being returned to the caller
 		// add the room ID to the queue of unvisited rooms. Loop from the beginning.
-		for roomID := range uniqueRooms {
+		for roomID, vias := range uniqueRooms {
 			unvisited = append(unvisited, roomVisit{
 				roomID: roomID,
 				depth:  rv.depth + 1,
+				vias:   vias,
 			})
 		}
 	}
@@ -378,35 +365,15 @@ func (w *walker) publicRoomsChunk(roomID string) *gomatrixserverlib.PublicRoom {
 
 // federatedRoomInfo returns more of the spaces graph from another server. Returns nil if this was
 // unsuccessful.
-func (w *walker) federatedRoomInfo(roomID string) (*gomatrixserverlib.MSC2946SpacesResponse, error) {
+func (w *walker) federatedRoomInfo(roomID string, vias []string) (*gomatrixserverlib.MSC2946SpacesResponse, error) {
 	// only do federated requests for client requests
 	if w.caller == nil {
 		return nil, nil
 	}
-	// extract events which point to this room ID and extract their vias
-	events, err := w.db.ChildReferences(w.ctx, roomID)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get ChildReferences events: %w", err)
-	}
-	vias := make(set)
-	for _, ev := range events {
-		if ev.StateKeyEquals(roomID) {
-			// event points at this room, extract vias
-			content := struct {
-				Vias []string `json:"via"`
-			}{}
-			if err = json.Unmarshal(ev.Content(), &content); err != nil {
-				continue // silently ignore corrupted state events
-			}
-			for _, v := range content.Vias {
-				vias[v] = true
-			}
-		}
-	}
 	util.GetLogger(w.ctx).Infof("Querying federatedRoomInfo via %+v", vias)
 	ctx := context.Background()
 	// query more of the spaces graph using these servers
-	for serverName := range vias {
+	for _, serverName := range vias {
 		if serverName == string(w.thisServer) {
 			continue
 		}
@@ -525,30 +492,51 @@ func (w *walker) authorisedUser(roomID string) bool {
 
 // references returns all child references pointing to or from this room.
 func (w *walker) childReferences(roomID string) ([]gomatrixserverlib.MSC2946StrippedEvent, error) {
+	createTuple := gomatrixserverlib.StateKeyTuple{
+		EventType: gomatrixserverlib.MRoomCreate,
+		StateKey:  "",
+	}
+	var res roomserver.QueryCurrentStateResponse
+	err := w.rsAPI.QueryCurrentState(context.Background(), &roomserver.QueryCurrentStateRequest{
+		RoomID:         roomID,
+		AllowWildcards: true,
+		StateTuples: []gomatrixserverlib.StateKeyTuple{
+			createTuple, {
+				EventType: ConstSpaceChildEventType,
+				StateKey:  "*",
+			},
+		},
+	}, &res)
+	if err != nil {
+		return nil, err
+	}
+
 	// don't return any child refs if the room is not a space room
-	create := w.stateEvent(roomID, gomatrixserverlib.MRoomCreate, "")
-	if create != nil {
+	if res.StateEvents[createTuple] != nil {
 		// escape the `.`s so gjson doesn't think it's nested
-		roomType := gjson.GetBytes(create.Content(), strings.ReplaceAll(ConstCreateEventContentKey, ".", `\.`)).Str
+		roomType := gjson.GetBytes(res.StateEvents[createTuple].Content(), strings.ReplaceAll(ConstCreateEventContentKey, ".", `\.`)).Str
 		if roomType != ConstCreateEventContentValueSpace {
 			return nil, nil
 		}
 	}
+	delete(res.StateEvents, createTuple)
 
-	events, err := w.db.ChildReferences(w.ctx, roomID)
-	if err != nil {
-		return nil, err
-	}
-	el := make([]gomatrixserverlib.MSC2946StrippedEvent, 0, len(events))
-	for _, ev := range events {
+	el := make([]gomatrixserverlib.MSC2946StrippedEvent, 0, len(res.StateEvents))
+	for _, ev := range res.StateEvents {
+		content := gjson.ParseBytes(ev.Content())
 		// only return events that have a `via` key as per MSC1772
 		// else we'll incorrectly walk redacted events (as the link
 		// is in the state_key)
-		if gjson.GetBytes(ev.Content(), "via").Exists() {
+		if content.Get("via").Exists() {
 			strip := stripped(ev.Event)
 			if strip == nil {
 				continue
 			}
+			// if suggested only and this child isn't suggested, skip it.
+			// if suggested only = false we include everything so don't need to check the content.
+			if w.suggestedOnly && !content.Get("suggested").Bool() {
+				continue
+			}
 			el = append(el, *strip)
 		}
 	}

setup/mscs/msc2946/storage.go

@@ -1,182 +0,0 @@
-// Copyright 2021 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 msc2946
-
-import (
-	"context"
-	"database/sql"
-
-	"github.com/matrix-org/dendrite/internal"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
-	"github.com/matrix-org/dendrite/setup/config"
-	"github.com/matrix-org/gomatrixserverlib"
-)
-
-var (
-	relTypes = map[string]int{
-		ConstSpaceChildEventType:  1,
-		ConstSpaceParentEventType: 2,
-	}
-)
-
-type Database interface {
-	// StoreReference persists a child or parent space mapping.
-	StoreReference(ctx context.Context, he *gomatrixserverlib.HeaderedEvent) error
-	// ChildReferences returns all space child events in the given room.
-	ChildReferences(ctx context.Context, roomID string) ([]*gomatrixserverlib.HeaderedEvent, error)
-}
-
-type DB struct {
-	db                    *sql.DB
-	writer                sqlutil.Writer
-	insertEdgeStmt        *sql.Stmt
-	selectEdgesOfTypeStmt *sql.Stmt
-}
-
-// NewDatabase loads the database for msc2836
-func NewDatabase(dbOpts *config.DatabaseOptions) (Database, error) {
-	if dbOpts.ConnectionString.IsPostgres() {
-		return newPostgresDatabase(dbOpts)
-	}
-	return newSQLiteDatabase(dbOpts)
-}
-
-func newPostgresDatabase(dbOpts *config.DatabaseOptions) (Database, error) {
-	d := DB{
-		writer: sqlutil.NewDummyWriter(),
-	}
-	var err error
-	if d.db, err = sqlutil.Open(dbOpts); err != nil {
-		return nil, err
-	}
-	_, err = d.db.Exec(`
-	CREATE TABLE IF NOT EXISTS msc2946_edges (
-		room_version TEXT NOT NULL,
-		-- the room ID of the event, the source of the arrow
-		source_room_id TEXT NOT NULL,
-		-- the target room ID, the arrow destination
-		dest_room_id TEXT NOT NULL,
-		-- the kind of relation, either child or parent (1,2)
-		rel_type SMALLINT NOT NULL,
-		event_json TEXT NOT NULL,
-		CONSTRAINT msc2946_edges_uniq UNIQUE (source_room_id, dest_room_id, rel_type)
-	);
-	`)
-	if err != nil {
-		return nil, err
-	}
-	if d.insertEdgeStmt, err = d.db.Prepare(`
-		INSERT INTO msc2946_edges(room_version, source_room_id, dest_room_id, rel_type, event_json)
-		VALUES($1, $2, $3, $4, $5)
-		ON CONFLICT ON CONSTRAINT msc2946_edges_uniq DO UPDATE SET event_json = $5
-	`); err != nil {
-		return nil, err
-	}
-	if d.selectEdgesOfTypeStmt, err = d.db.Prepare(`
-		SELECT room_version, event_json FROM msc2946_edges
-		WHERE source_room_id = $1 AND rel_type = $2
-	`); err != nil {
-		return nil, err
-	}
-	return &d, err
-}
-
-func newSQLiteDatabase(dbOpts *config.DatabaseOptions) (Database, error) {
-	d := DB{
-		writer: sqlutil.NewExclusiveWriter(),
-	}
-	var err error
-	if d.db, err = sqlutil.Open(dbOpts); err != nil {
-		return nil, err
-	}
-	_, err = d.db.Exec(`
-	CREATE TABLE IF NOT EXISTS msc2946_edges (
-		room_version TEXT NOT NULL,
-		-- the room ID of the event, the source of the arrow
-		source_room_id TEXT NOT NULL,
-		-- the target room ID, the arrow destination
-		dest_room_id TEXT NOT NULL,
-		-- the kind of relation, either child or parent (1,2)
-		rel_type SMALLINT NOT NULL,
-		event_json TEXT NOT NULL,
-		UNIQUE (source_room_id, dest_room_id, rel_type)
-	);
-	`)
-	if err != nil {
-		return nil, err
-	}
-	if d.insertEdgeStmt, err = d.db.Prepare(`
-		INSERT INTO msc2946_edges(room_version, source_room_id, dest_room_id, rel_type, event_json)
-		VALUES($1, $2, $3, $4, $5)
-		ON CONFLICT (source_room_id, dest_room_id, rel_type) DO UPDATE SET event_json = $5
-	`); err != nil {
-		return nil, err
-	}
-	if d.selectEdgesOfTypeStmt, err = d.db.Prepare(`
-		SELECT room_version, event_json FROM msc2946_edges
-		WHERE source_room_id = $1 AND rel_type = $2
-	`); err != nil {
-		return nil, err
-	}
-	return &d, err
-}
-
-func (d *DB) StoreReference(ctx context.Context, he *gomatrixserverlib.HeaderedEvent) error {
-	target := SpaceTarget(he)
-	if target == "" {
-		return nil // malformed event
-	}
-	relType := relTypes[he.Type()]
-	_, err := d.insertEdgeStmt.ExecContext(ctx, he.RoomVersion, he.RoomID(), target, relType, he.JSON())
-	return err
-}
-
-func (d *DB) ChildReferences(ctx context.Context, roomID string) ([]*gomatrixserverlib.HeaderedEvent, error) {
-	rows, err := d.selectEdgesOfTypeStmt.QueryContext(ctx, roomID, relTypes[ConstSpaceChildEventType])
-	if err != nil {
-		return nil, err
-	}
-	defer internal.CloseAndLogIfError(ctx, rows, "failed to close References")
-	refs := make([]*gomatrixserverlib.HeaderedEvent, 0)
-	for rows.Next() {
-		var roomVer string
-		var jsonBytes []byte
-		if err := rows.Scan(&roomVer, &jsonBytes); err != nil {
-			return nil, err
-		}
-		ev, err := gomatrixserverlib.NewEventFromTrustedJSON(jsonBytes, false, gomatrixserverlib.RoomVersion(roomVer))
-		if err != nil {
-			return nil, err
-		}
-		he := ev.Headered(gomatrixserverlib.RoomVersion(roomVer))
-		refs = append(refs, he)
-	}
-	return refs, nil
-}
-
-// SpaceTarget returns the destination room ID for the space event. This is either a child or a parent
-// depending on the event type.
-func SpaceTarget(he *gomatrixserverlib.HeaderedEvent) string {
-	if he.StateKey() == nil {
-		return "" // no-op
-	}
-	switch he.Type() {
-	case ConstSpaceParentEventType:
-		return *he.StateKey()
-	case ConstSpaceChildEventType:
-		return *he.StateKey()
-	}
-	return ""
-}