Implement federated public rooms

- Make thirdparty endpoint r0 so riot-web loads the public room list

clientapi/routing/routing.go

@@ -390,7 +390,7 @@ func Setup(
 		}),
 	).Methods(http.MethodGet, http.MethodOptions)
 
-	unstableMux.Handle("/thirdparty/protocols",
+	r0mux.Handle("/thirdparty/protocols",
 		common.MakeExternalAPI("thirdparty_protocols", func(req *http.Request) util.JSONResponse {
 			// TODO: Return the third party protcols
 			return util.JSONResponse{

cmd/dendritejs/main.go

@@ -135,7 +135,7 @@ func main() {
 	)
 	federationapi.SetupFederationAPIComponent(base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fedSenderAPI)
 	mediaapi.SetupMediaAPIComponent(base, deviceDB)
-	publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB, query, p2pPublicRoomProvider)
+	publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB, query, federation, p2pPublicRoomProvider)
 	syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query, federation, cfg)
 
 	httpHandler := common.WrapHandlerInCORS(base.APIMux)

cmd/dendritejs/publicrooms.go

@@ -17,8 +17,6 @@
 package main
 
 import (
-	"time"
-
 	"github.com/matrix-org/go-http-js-libp2p/go_http_js_libp2p"
 )
 
@@ -39,10 +37,6 @@ func (p *libp2pPublicRoomsProvider) foundProviders(peerInfos []go_http_js_libp2p
 	p.providers = peerInfos
 }
 
-func (p *libp2pPublicRoomsProvider) PollInterval() time.Duration {
-	return 10 * time.Second
-}
-
 func (p *libp2pPublicRoomsProvider) Homeservers() []string {
 	result := make([]string, len(p.providers))
 	for i := range p.providers {

publicroomsapi/directory/public_rooms.go

@@ -18,10 +18,13 @@ import (
 	"context"
 	"net/http"
 	"strconv"
+	"sync"
+	"time"
 
 	"github.com/matrix-org/dendrite/clientapi/httputil"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/publicroomsapi/storage"
+	"github.com/matrix-org/dendrite/publicroomsapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
 )
@@ -44,7 +47,7 @@ func GetPostPublicRooms(
 	if fillErr := fillPublicRoomsReq(req, &request); fillErr != nil {
 		return *fillErr
 	}
-	response, err := PublicRooms(req.Context(), request, publicRoomDatabase)
+	response, err := publicRooms(req.Context(), request, publicRoomDatabase)
 	if err != nil {
 		return jsonerror.InternalServerError()
 	}
@@ -54,7 +57,123 @@ func GetPostPublicRooms(
 	}
 }
 
-func PublicRooms(ctx context.Context, request PublicRoomReq, publicRoomDatabase storage.Database) (*gomatrixserverlib.RespPublicRooms, error) {
+// GetPostPublicRoomsWithExternal is the same as GetPostPublicRooms but also mixes in public rooms from the provider supplied.
+func GetPostPublicRoomsWithExternal(
+	req *http.Request, publicRoomDatabase storage.Database, fedClient *gomatrixserverlib.FederationClient,
+	extRoomsProvider types.ExternalPublicRoomsProvider,
+) util.JSONResponse {
+	var request PublicRoomReq
+	if fillErr := fillPublicRoomsReq(req, &request); fillErr != nil {
+		return *fillErr
+	}
+	response, err := publicRooms(req.Context(), request, publicRoomDatabase)
+	if err != nil {
+		return jsonerror.InternalServerError()
+	}
+
+	if request.Since != "" {
+		// TODO: handle pagination tokens sensibly rather than ignoring them.
+		// ignore paginated requests since we don't handle them yet over federation.
+		// Only the initial request will contain federated rooms.
+		return util.JSONResponse{
+			Code: http.StatusOK,
+			JSON: response,
+		}
+	}
+
+	// If we have already hit the limit on the number of rooms, bail.
+	var limit int
+	if request.Limit > 0 {
+		limit = int(request.Limit) - len(response.Chunk)
+		if limit <= 0 {
+			return util.JSONResponse{
+				Code: http.StatusOK,
+				JSON: response,
+			}
+		}
+	}
+
+	// downcasting `limit` is safe as we know it isn't bigger than request.Limit which is int16
+	fedRooms := bulkFetchPublicRoomsFromServers(req.Context(), fedClient, extRoomsProvider.Homeservers(), int16(limit))
+	response.Chunk = append(response.Chunk, fedRooms...)
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: response,
+	}
+}
+
+// bulkFetchPublicRoomsFromServers fetches public rooms from the list of homeservers.
+// Returns a list of public rooms up to the limit specified.
+func bulkFetchPublicRoomsFromServers(
+	ctx context.Context, fedClient *gomatrixserverlib.FederationClient, homeservers []string, limit int16,
+) (publicRooms []gomatrixserverlib.PublicRoom) {
+	// follow pipeline semantics, see https://blog.golang.org/pipelines for more info.
+	// goroutines send rooms to this channel
+	roomCh := make(chan gomatrixserverlib.PublicRoom, int(limit))
+	// signalling channel to tell goroutines to stop sending rooms and quit
+	done := make(chan bool)
+	// signalling to say when we can close the room channel
+	var wg sync.WaitGroup
+	wg.Add(len(homeservers))
+	// concurrently query for public rooms
+	for _, hs := range homeservers {
+		go func(homeserverDomain string) {
+			defer wg.Done()
+			util.GetLogger(ctx).WithField("hs", homeserverDomain).Info("Querying HS for public rooms")
+			fres, err := fedClient.GetPublicRooms(ctx, gomatrixserverlib.ServerName(homeserverDomain), int(limit), "", false, "")
+			if err != nil {
+				util.GetLogger(ctx).WithError(err).WithField("hs", homeserverDomain).Warn(
+					"bulkFetchPublicRoomsFromServers: failed to query hs",
+				)
+				return
+			}
+			for _, room := range fres.Chunk {
+				// atomically send a room or stop
+				select {
+				case roomCh <- room:
+				case <-done:
+					util.GetLogger(ctx).WithError(err).WithField("hs", homeserverDomain).Info("Interruped whilst sending rooms")
+					return
+				}
+			}
+		}(hs)
+	}
+
+	// Close the room channel when the goroutines have quit so we don't leak, but don't let it stop the in-flight request.
+	// This also allows the request to fail fast if all HSes experience errors as it will cause the room channel to be
+	// closed.
+	go func() {
+		wg.Wait()
+		util.GetLogger(ctx).Info("Cleaning up resources")
+		close(roomCh)
+	}()
+
+	// fan-in results with timeout. We stop when we reach the limit.
+FanIn:
+	for len(publicRooms) < int(limit) || limit == 0 {
+		// add a room or timeout
+		select {
+		case room, ok := <-roomCh:
+			if !ok {
+				util.GetLogger(ctx).Info("All homeservers have been queried, returning results.")
+				break FanIn
+			}
+			publicRooms = append(publicRooms, room)
+		case <-time.After(15 * time.Second): // we've waited long enough, let's tell the client what we got.
+			util.GetLogger(ctx).Info("Waited 15s for federated public rooms, returning early")
+			break FanIn
+		case <-ctx.Done(): // the client hung up on us, let's stop.
+			util.GetLogger(ctx).Info("Client hung up, returning early")
+			break FanIn
+		}
+	}
+	// tell goroutines to stop
+	close(done)
+
+	return publicRooms
+}
+
+func publicRooms(ctx context.Context, request PublicRoomReq, publicRoomDatabase storage.Database) (*gomatrixserverlib.RespPublicRooms, error) {
 	var response gomatrixserverlib.RespPublicRooms
 	var limit int16
 	var offset int64

publicroomsapi/publicroomsapi.go

@@ -22,6 +22,7 @@ import (
 	"github.com/matrix-org/dendrite/publicroomsapi/storage"
 	"github.com/matrix-org/dendrite/publicroomsapi/types"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
 )
 
@@ -31,6 +32,7 @@ func SetupPublicRoomsAPIComponent(
 	base *basecomponent.BaseDendrite,
 	deviceDB devices.Database,
 	rsQueryAPI roomserverAPI.RoomserverQueryAPI,
+	fedClient *gomatrixserverlib.FederationClient,
 	extRoomsProvider types.ExternalPublicRoomsProvider,
 ) {
 	publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI))
@@ -45,5 +47,5 @@ func SetupPublicRoomsAPIComponent(
 		logrus.WithError(err).Panic("failed to start public rooms server consumer")
 	}
 
-	routing.Setup(base.APIMux, deviceDB, publicRoomsDB, extRoomsProvider)
+	routing.Setup(base.APIMux, deviceDB, publicRoomsDB, fedClient, extRoomsProvider)
 }

publicroomsapi/routing/routing.go

@@ -15,9 +15,7 @@
 package routing
 
 import (
-	"fmt"
 	"net/http"
-	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/clientapi/auth"
@@ -27,6 +25,7 @@ import (
 	"github.com/matrix-org/dendrite/publicroomsapi/directory"
 	"github.com/matrix-org/dendrite/publicroomsapi/storage"
 	"github.com/matrix-org/dendrite/publicroomsapi/types"
+	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
 )
 
@@ -37,7 +36,10 @@ const pathPrefixR0 = "/_matrix/client/r0"
 // Due to Setup being used to call many other functions, a gocyclo nolint is
 // applied:
 // nolint: gocyclo
-func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.Database, extRoomsProvider types.ExternalPublicRoomsProvider) {
+func Setup(
+	apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.Database,
+	fedClient *gomatrixserverlib.FederationClient, extRoomsProvider types.ExternalPublicRoomsProvider,
+) {
 	r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
 
 	authData := auth.Data{
@@ -68,6 +70,7 @@ func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.
 	r0mux.Handle("/publicRooms",
 		common.MakeExternalAPI("public_rooms", func(req *http.Request) util.JSONResponse {
 			if extRoomsProvider != nil {
+				return directory.GetPostPublicRoomsWithExternal(req, publicRoomsDB, fedClient, extRoomsProvider)
 			}
 			return directory.GetPostPublicRooms(req, publicRoomsDB)
 		}),
@@ -79,18 +82,4 @@ func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.
 			return directory.GetPostPublicRooms(req, publicRoomsDB)
 		}),
 	).Methods(http.MethodGet)
-
-	if extRoomsProvider != nil {
-		pollExternalPublicRooms(extRoomsProvider)
-	}
-}
-
-func pollExternalPublicRooms(extRoomsProvider types.ExternalPublicRoomsProvider) {
-	go func() {
-		for {
-			hses := extRoomsProvider.Homeservers()
-			fmt.Println(hses)
-			time.Sleep(extRoomsProvider.PollInterval())
-		}
-	}()
 }

publicroomsapi/storage/sqlite3/public_rooms_table.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/matrix-org/dendrite/common"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
@@ -66,7 +67,7 @@ const selectPublicRoomsWithLimitSQL = "" +
 	"SELECT room_id, joined_members, aliases, canonical_alias, name, topic, world_readable, guest_can_join, avatar_url" +
 	" FROM publicroomsapi_public_rooms WHERE visibility = true" +
 	" ORDER BY joined_members DESC" +
-	" LIMIT $2 OFFSET $1"
+	" LIMIT $1 OFFSET $2"
 
 const selectPublicRoomsWithFilterSQL = "" +
 	"SELECT room_id, joined_members, aliases, canonical_alias, name, topic, world_readable, guest_can_join, avatar_url" +
@@ -192,6 +193,7 @@ func (s *publicRoomsStatements) selectPublicRooms(
 	if err != nil {
 		return []gomatrixserverlib.PublicRoom{}, nil
 	}
+	defer common.CloseAndLogIfError(ctx, rows, "selectPublicRooms failed to close rows")
 
 	rooms := []gomatrixserverlib.PublicRoom{}
 	for rows.Next() {

publicroomsapi/types/types.go

@@ -14,14 +14,11 @@
 
 package types
 
-import "time"
-
 // ExternalPublicRoomsProvider provides a list of homeservers who should be queried
 // periodically for a list of public rooms on their server.
 type ExternalPublicRoomsProvider interface {
-	// The interval at which to check servers
-	PollInterval() time.Duration
 	// The list of homeserver domains to query. These servers will receive a request
 	// via this API: https://matrix.org/docs/spec/server_server/latest#public-room-directory
+	// This will be called -on demand- by clients, so cache appropriately!
 	Homeservers() []string
 }